283 lines
20 KiB
Python
Executable File
283 lines
20 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
import logging
|
|
import os
|
|
import socketserver
|
|
import subprocess
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
import unittest.mock
|
|
|
|
import hddfancontrol
|
|
|
|
|
|
class FakeHddtempDaemon(threading.Thread):
|
|
|
|
outgoing = b""
|
|
|
|
def __init__(self, port):
|
|
socketserver.TCPServer.allow_reuse_address = True
|
|
self.server = socketserver.TCPServer(("127.0.0.1", port), FakeHddtempDaemonHandler)
|
|
super().__init__()
|
|
|
|
def run(self):
|
|
self.server.serve_forever()
|
|
|
|
|
|
class FakeHddtempDaemonHandler(socketserver.StreamRequestHandler):
|
|
|
|
def handle(self):
|
|
self.wfile.write(FakeHddtempDaemon.outgoing)
|
|
|
|
|
|
class TestDrive(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
with unittest.mock.patch("hddfancontrol.os.stat") as os_stat_mock, \
|
|
unittest.mock.patch("hddfancontrol.stat") as stat_mock, \
|
|
unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock, \
|
|
unittest.mock.patch("hddfancontrol.Drive.getPrettyName") as drive_getPrettyName:
|
|
os_stat_mock.return_value = os.stat_result
|
|
stat_mock.stat.S_IFBLK.return_value = True
|
|
subprocess_check_output_mock.side_effect = subprocess.CalledProcessError(0, "")
|
|
drive_getPrettyName.return_value = "drive_name"
|
|
self.drive = hddfancontrol.Drive("/dev/sdz", None)
|
|
self.hddtemp_daemon = None
|
|
|
|
def tearDown(self):
|
|
if self.hddtemp_daemon is not None:
|
|
self.hddtemp_daemon.server.shutdown()
|
|
self.hddtemp_daemon.server.server_close()
|
|
self.hddtemp_daemon.join()
|
|
|
|
def test_getPrettyName(self):
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.return_value = "\n/dev/sdz:\n\nATA device, with non-removable media\n\tModel Number: WDC WD4003FZEX-00Z4SA0 \n\tSerial Number: WD-WMC5D0D4YY1K\n\tFirmware Revision: 01.01A01\n\tTransport: Serial, SATA 1.0a, SATA II Extensions, SATA Rev 2.5, SATA Rev 2.6, SATA Rev 3.0\nStandards:\n\tSupported: 9 8 7 6 5 \n\tLikely used: 9\nConfiguration:\n\tLogical\t\tmax\tcurrent\n\tcylinders\t16383\t16383\n\theads\t\t16\t16\n\tsectors/track\t63\t63\n\t--\n\tCHS current addressable sectors: 16514064\n\tLBA user addressable sectors: 268435455\n\tLBA48 user addressable sectors: 7814037168\n\tLogical Sector size: 512 bytes\n\tPhysical Sector size: 4096 bytes\n\tLogical Sector-0 offset: 0 bytes\n\tdevice size with M = 1024*1024: 3815447 MBytes\n\tdevice size with M = 1000*1000: 4000787 MBytes (4000 GB)\n\tcache/buffer size = unknown\n\tNominal Media Rotation Rate: 7200\nCapabilities:\n\tLBA, IORDY(can be disabled)\n\tQueue depth: 32\n\tStandby timer values: spec'd by Standard, with device specific minimum\n\tR/W multiple sector transfer: Max = 16\tCurrent = 0\n\tDMA: mdma0 mdma1 mdma2 udma0 udma1 udma2 udma3 udma4 udma5 *udma6 \n\t Cycle time: min=120ns recommended=120ns\n\tPIO: pio0 pio1 pio2 pio3 pio4 \n\t Cycle time: no flow control=120ns IORDY flow control=120ns\nCommands/features:\n\tEnabled\tSupported:\n\t *\tSMART feature set\n\t \tSecurity Mode feature set\n\t *\tPower Management feature set\n\t *\tWrite cache\n\t *\tLook-ahead\n\t *\tHost Protected Area feature set\n\t *\tWRITE_BUFFER command\n\t *\tREAD_BUFFER command\n\t *\tNOP cmd\n\t *\tDOWNLOAD_MICROCODE\n\t \tPower-Up In Standby feature set\n\t *\tSET_FEATURES required to spinup after power up\n\t \tSET_MAX security extension\n\t *\t48-bit Address feature set\n\t *\tMandatory FLUSH_CACHE\n\t *\tFLUSH_CACHE_EXT\n\t *\tSMART error logging\n\t *\tSMART self-test\n\t *\tGeneral Purpose Logging feature set\n\t *\t64-bit World wide name\n\t *\t{READ,WRITE}_DMA_EXT_GPL commands\n\t *\tSegmented DOWNLOAD_MICROCODE\n\t *\tGen1 signaling speed (1.5Gb/s)\n\t *\tGen2 signaling speed (3.0Gb/s)\n\t *\tGen3 signaling speed (6.0Gb/s)\n\t *\tNative Command Queueing (NCQ)\n\t *\tHost-initiated interface power management\n\t *\tPhy event counters\n\t *\tNCQ priority information\n\t *\tREAD_LOG_DMA_EXT equivalent to READ_LOG_EXT\n\t *\tDMA Setup Auto-Activate optimization\n\t *\tSoftware settings preservation\n\t *\tSMART Command Transport (SCT) feature set\n\t *\tSCT Write Same (AC2)\n\t *\tSCT Features Control (AC4)\n\t *\tSCT Data Tables (AC5)\n\t \tunknown 206[12] (vendor specific)\n\t \tunknown 206[13] (vendor specific)\n\t \tunknown 206[14] (vendor specific)\nSecurity: \n\tMaster password revision code = 65534\n\t\tsupported\n\tnot\tenabled\n\tnot\tlocked\n\tnot\tfrozen\n\tnot\texpired: security count\n\t\tsupported: enhanced erase\n\t424min for SECURITY ERASE UNIT. 424min for ENHANCED SECURITY ERASE UNIT. \nLogical Unit WWN Device Identifier: 50014ee0593d4632\n\tNAA\t\t: 5\n\tIEEE OUI\t: 0014ee\n\tUnique ID\t: 0593d4632\nChecksum: correct\n"
|
|
self.assertEqual(self.drive.getPrettyName(), "sdz WDC WD4003FZEX-00Z4SA0")
|
|
subprocess_check_output_mock.assert_called_once_with(("hdparm", "-I", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
universal_newlines=True)
|
|
|
|
def test_supportsHitachiTempQuery(self):
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.return_value = "\n/dev/sdz:\n drive temperature (celsius) is: 30\n drive temperature in range: yes"
|
|
self.assertEqual(self.drive.supportsHitachiTempQuery(), True)
|
|
subprocess_check_output_mock.assert_called_once_with(("hdparm", "-H", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.STDOUT,
|
|
universal_newlines=True)
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.return_value = "\n/dev/sdz:\nSG_IO: questionable sense data, results may be incorrect\n drive temperature (celsius) is: -18\n drive temperature in range: yes"
|
|
self.assertEqual(self.drive.supportsHitachiTempQuery(), False)
|
|
subprocess_check_output_mock.assert_called_once_with(("hdparm", "-H", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.STDOUT,
|
|
universal_newlines=True)
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.return_value = "\n/dev/sdz:\nSG_IO: missing sense data, results may be incorrect\n drive temperature (celsius) is: -18\n drive temperature in range: yes"
|
|
self.assertEqual(self.drive.supportsHitachiTempQuery(), False)
|
|
subprocess_check_output_mock.assert_called_once_with(("hdparm", "-H", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.STDOUT,
|
|
universal_newlines=True)
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.return_value = "\n/dev/sdz:\nSG_IO: bad/missing sense data, sb[]: 70 00 05 00 00 00 00 0a 04 51 40 00 21 04 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00\n drive temperature (celsius) is: -18\n drive temperature in range: yes"
|
|
self.assertEqual(self.drive.supportsHitachiTempQuery(), False)
|
|
subprocess_check_output_mock.assert_called_once_with(("hdparm", "-H", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.STDOUT,
|
|
universal_newlines=True)
|
|
|
|
def test_getState(self):
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.return_value = "\n/dev/sdz:\n drive state is: active/idle\n"
|
|
self.assertEqual(self.drive.getState(), hddfancontrol.Drive.DriveState.ACTIVE_IDLE)
|
|
subprocess_check_output_mock.assert_called_once_with(("hdparm", "-C", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
universal_newlines=True)
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.return_value = "\n/dev/sdz:\n drive state is: standby\n"
|
|
self.assertEqual(self.drive.getState(), hddfancontrol.Drive.DriveState.STANDBY)
|
|
subprocess_check_output_mock.assert_called_once_with(("hdparm", "-C", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
universal_newlines=True)
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.return_value = "\n/dev/sdz:\n drive state is: sleeping\n"
|
|
self.assertEqual(self.drive.getState(), hddfancontrol.Drive.DriveState.SLEEPING)
|
|
subprocess_check_output_mock.assert_called_once_with(("hdparm", "-C", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
universal_newlines=True)
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.side_effect = subprocess.CalledProcessError(0, "")
|
|
with self.assertRaises(Exception):
|
|
self.drive.getState()
|
|
subprocess_check_output_mock.assert_called_once_with(("hdparm", "-C", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
universal_newlines=True)
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.return_value = "/dev/sdz: No such file or directory\n"
|
|
with self.assertRaises(Exception):
|
|
self.drive.getState()
|
|
subprocess_check_output_mock.assert_called_once_with(("hdparm", "-C", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
universal_newlines=True)
|
|
|
|
def test_isSleeping(self):
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.return_value = "\n/dev/sdz:\n drive state is: active/idle\n"
|
|
self.assertFalse(self.drive.isSleeping())
|
|
subprocess_check_output_mock.assert_called_once_with(("hdparm", "-C", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
universal_newlines=True)
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.return_value = "\n/dev/sdz:\n drive state is: standby\n"
|
|
self.assertTrue(self.drive.isSleeping())
|
|
subprocess_check_output_mock.assert_called_once_with(("hdparm", "-C", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
universal_newlines=True)
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.return_value = "\n/dev/sdz:\n drive state is: sleeping\n"
|
|
self.assertTrue(self.drive.isSleeping())
|
|
subprocess_check_output_mock.assert_called_once_with(("hdparm", "-C", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
universal_newlines=True)
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.side_effect = subprocess.CalledProcessError(0, "")
|
|
with self.assertRaises(Exception):
|
|
self.drive.isSleeping()
|
|
subprocess_check_output_mock.assert_called_once_with(("hdparm", "-C", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
universal_newlines=True)
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.return_value = "/dev/sdz: No such file or directory\n"
|
|
with self.assertRaises(Exception):
|
|
self.drive.isSleeping()
|
|
subprocess_check_output_mock.assert_called_once_with(("hdparm", "-C", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
universal_newlines=True)
|
|
|
|
def test_getTemperature(self):
|
|
#
|
|
# Temperature querying can be done in 3 different way:
|
|
# * if drive supports Hitachi-style sensor => use hdparm call
|
|
# * if hddtemp daemon is available => use hddtemp daemon
|
|
# * otherwise use a hddtemp call
|
|
#
|
|
|
|
# hddtemp call
|
|
self.drive.supports_hitachi_temp_query = False
|
|
self.drive.hddtemp_daemon_port = None
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.return_value = "30\n"
|
|
self.assertEqual(self.drive.getTemperature(), 30)
|
|
subprocess_check_output_mock.assert_called_once_with(("hddtemp", "-u", "C", "-n", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
universal_newlines=True)
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.side_effect = subprocess.CalledProcessError(0, "")
|
|
with self.assertRaises(Exception):
|
|
self.drive.getTemperature()
|
|
subprocess_check_output_mock.assert_called_once_with(("hddtemp", "-u", "C", "-n", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
universal_newlines=True)
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.return_value = "/dev/sdz: open: No such file or directory\n\n"
|
|
with self.assertRaises(Exception):
|
|
self.drive.getTemperature()
|
|
subprocess_check_output_mock.assert_called_once_with(("hddtemp", "-u", "C", "-n", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
universal_newlines=True)
|
|
|
|
# hdparm call
|
|
self.drive.supports_hitachi_temp_query = True
|
|
for self.drive.hddtemp_daemon_port in (None, 12345):
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.return_value = "/dev/sdz:\n drive temperature (celsius) is: 30\n drive temperature in range: yes\n"
|
|
self.assertEqual(self.drive.getTemperature(), 30)
|
|
subprocess_check_output_mock.assert_called_once_with(("hdparm", "-H", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
universal_newlines=True)
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.side_effect = subprocess.CalledProcessError(0, "")
|
|
with self.assertRaises(Exception):
|
|
self.drive.getTemperature()
|
|
subprocess_check_output_mock.assert_called_once_with(("hdparm", "-H", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
universal_newlines=True)
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_output") as subprocess_check_output_mock:
|
|
subprocess_check_output_mock.return_value = "/dev/sdz: No such file or directory\n"
|
|
with self.assertRaises(Exception):
|
|
self.drive.getTemperature()
|
|
subprocess_check_output_mock.assert_called_once_with(("hdparm", "-H", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
universal_newlines=True)
|
|
|
|
# hddtemp daemon
|
|
self.drive.supports_hitachi_temp_query = False
|
|
self.drive.hddtemp_daemon_port = 12345
|
|
with self.assertRaises(Exception):
|
|
self.drive.getTemperature()
|
|
self.hddtemp_daemon = FakeHddtempDaemon(12345)
|
|
self.hddtemp_daemon.start()
|
|
FakeHddtempDaemon.outgoing = b"|/dev/sdz|DriveSDZ|30|C|"
|
|
self.assertEqual(self.drive.getTemperature(), 30)
|
|
FakeHddtempDaemon.outgoing = b"|/dev/sdy|DriveSDY|31|C||/dev/sdz|DriveSDZ|30|C|"
|
|
self.assertEqual(self.drive.getTemperature(), 30)
|
|
FakeHddtempDaemon.outgoing = b"|/dev/sdy|DriveSDY|31|C||/dev/sdz|DriveSDZ|30|C||/dev/sdx|DriveSDX|32|C|"
|
|
self.assertEqual(self.drive.getTemperature(), 30)
|
|
FakeHddtempDaemon.outgoing = b"|/dev/sdx|DriveSDX|31|C||/dev/sdy|DriveSDY|32|C|"
|
|
with self.assertRaises(RuntimeError):
|
|
self.drive.getTemperature()
|
|
FakeHddtempDaemon.outgoing = b"|/dev/sdz|DriveSDZ|30|F|"
|
|
with self.assertRaises(RuntimeError):
|
|
self.drive.getTemperature()
|
|
FakeHddtempDaemon.outgoing = b""
|
|
with self.assertRaises(Exception):
|
|
self.drive.getTemperature()
|
|
|
|
def test_spinDown(self):
|
|
with unittest.mock.patch("hddfancontrol.subprocess.check_call") as subprocess_check_call_mock:
|
|
self.drive.spinDown()
|
|
subprocess_check_call_mock.assert_called_once_with(("hdparm", "-y", "/dev/sdz"),
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL)
|
|
|
|
def test_getActivityStats(self):
|
|
with self.assertRaises(Exception):
|
|
self.drive.getActivityStats()
|
|
with tempfile.NamedTemporaryFile("wt") as stat_file:
|
|
self.drive.stat_filepath = stat_file.name
|
|
with self.assertRaises(Exception):
|
|
self.drive.getActivityStats()
|
|
stat_file.write(" 21695 7718 2913268 95136 13986 754 932032 55820 0 19032 150940\n")
|
|
stat_file.flush()
|
|
self.assertEqual(self.drive.getActivityStats(),
|
|
(21695, 7718, 2913268, 95136, 13986, 754, 932032, 55820, 0, 19032, 150940))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# disable logging
|
|
logging.basicConfig(level=logging.CRITICAL + 1)
|
|
|
|
# run tests
|
|
unittest.main()
|