242 lines
14 KiB
Python
Executable File
242 lines
14 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
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") as subprocess_check_call_mock:
|
|
os_stat_mock.return_value = os.stat_result
|
|
stat_mock.stat.S_IFBLK.return_value = True
|
|
subprocess_check_call_mock.side_effect = subprocess.CalledProcessError(0, "")
|
|
self.drive = hddfancontrol.Drive("/dev/sdz", "/dummy", 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_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()
|