diff --git a/src/openflexure_microscope_server/things/system.py b/src/openflexure_microscope_server/things/system.py index ff9da423..ff6a2fed 100644 --- a/src/openflexure_microscope_server/things/system.py +++ b/src/openflexure_microscope_server/things/system.py @@ -5,6 +5,7 @@ the microscope, server, and thing states to the web API. """ import os +import re import socket import subprocess import time @@ -19,8 +20,12 @@ import labthings_fastapi as lt from openflexure_microscope_server.utilities import VersionData, robust_version_strings -SHUTDOWN_CMD = ["sudo", "shutdown", "-h", "now"] -REBOOT_CMD = ["sudo", "shutdown", "-r", "now"] +LEGACY_SHUTDOWN_CMD = ["sudo", "shutdown", "-h", "now"] +OFM_SHUTDOWN_CMD = ["ofm", "system-shutdown"] +LEGACY_REBOOT_CMD = ["sudo", "shutdown", "-r", "now"] +OFM_REBOOT_CMD = ["ofm", "system-restart"] + +OS_VERSION_FILE = "/usr/lib/os-release" class CommandOutput(BaseModel): @@ -81,6 +86,24 @@ class OpenFlexureSystem(lt.Thing): """Return True if running on a Raspberry Pi.""" return os.path.exists("/usr/bin/raspi-config") + @lt.property + def os_version(self) -> Optional[str]: + """Return the OS version of the Pi. Returns None if not on a Pi.""" + if not self.is_raspberrypi: + return None + + if not os.path.isfile(OS_VERSION_FILE): + return "unknown" + + with open(OS_VERSION_FILE, "r", encoding="utf-8") as os_file: + os_data = os_file.read() + + version_match = re.search(r"^VERSION_CODENAME=(.+)$", os_data, re.MULTILINE) + if not version_match: + return "unknown" + + return version_match.group(1) + @lt.action def shutdown(self) -> CommandOutput: """Attempt to shutdown the device.""" @@ -93,9 +116,10 @@ class OpenFlexureSystem(lt.Thing): error="Shutdown command sent to server process, but the server has not shutdown.", ) - # On a Raspberry Pi + cmd = OFM_SHUTDOWN_CMD if self.os_version == "trixie" else LEGACY_SHUTDOWN_CMD + p = subprocess.Popen( - SHUTDOWN_CMD, + cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True, @@ -113,8 +137,10 @@ class OpenFlexureSystem(lt.Thing): error="Restart is only available on Raspberry Pi.", ) + cmd = OFM_REBOOT_CMD if self.os_version == "trixie" else LEGACY_REBOOT_CMD + p = subprocess.Popen( - REBOOT_CMD, + cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True, diff --git a/tests/unit_tests/assets/os-release b/tests/unit_tests/assets/os-release new file mode 100644 index 00000000..91ca955a --- /dev/null +++ b/tests/unit_tests/assets/os-release @@ -0,0 +1,10 @@ +PRETTY_NAME="Debian GNU/Linux 13 (trixie)" +NAME="Debian GNU/Linux" +VERSION_ID="13" +VERSION="13 (trixie)" +VERSION_CODENAME=trixie +DEBIAN_VERSION_FULL=13.4 +ID=debian +HOME_URL="https://www.debian.org/" +SUPPORT_URL="https://www.debian.org/support" +BUG_REPORT_URL="https://bugs.debian.org/" diff --git a/tests/unit_tests/assets/os-release-broken b/tests/unit_tests/assets/os-release-broken new file mode 100644 index 00000000..32271058 --- /dev/null +++ b/tests/unit_tests/assets/os-release-broken @@ -0,0 +1 @@ +This doesn't contain the data we are looking for. \ No newline at end of file diff --git a/tests/unit_tests/test_system_thing.py b/tests/unit_tests/test_system_thing.py index fa1729f7..206077d8 100644 --- a/tests/unit_tests/test_system_thing.py +++ b/tests/unit_tests/test_system_thing.py @@ -11,6 +11,11 @@ from labthings_fastapi.testing import create_thing_without_server from openflexure_microscope_server.things import system from openflexure_microscope_server.utilities import VersionData, robust_version_strings +THIS_DIR = os.path.dirname(os.path.abspath(__file__)) +FAKE_OS_FILE = os.path.join(THIS_DIR, "assets", "os-release") +BROKEN_OS_FILE = os.path.join(THIS_DIR, "assets", "os-release-broken") +MISSING_OS_FILE = os.path.join(THIS_DIR, "assets", "this-does-not-exist") + @pytest.fixture def system_thing(): @@ -37,6 +42,33 @@ def test_is_raspberry(system_thing): assert system_thing.is_raspberrypi == _is_raspberrypi() +@pytest.mark.parametrize( + ("version_file", "expected_result"), + [ + (MISSING_OS_FILE, "unknown"), + (FAKE_OS_FILE, "trixie"), + (BROKEN_OS_FILE, "unknown"), + ], +) +def test_os_version_pi(version_file, expected_result, system_thing, mocker): + """Check reading the operating system version from a Pi (mocked!).""" + type(system_thing).is_raspberrypi = mocker.PropertyMock(return_value=True) + + mocker.patch( + "openflexure_microscope_server.things.system.OS_VERSION_FILE", version_file + ) + assert system_thing.os_version == expected_result + + +def test_os_version_not_pi(system_thing, mocker): + """Check reading the operating system when not on a Pi returns None.""" + mocker.patch( + "openflexure_microscope_server.things.system.OS_VERSION_FILE", FAKE_OS_FILE + ) + type(system_thing).is_raspberrypi = mocker.PropertyMock(return_value=False) + assert system_thing.os_version is None + + def test_version_data(system_thing): """Check VersionData is returned. @@ -80,51 +112,55 @@ def test_thing_state(system_thing, mocker): assert state_dict["version_source"] == version_data.version_source -class MockPiSystem(system.OpenFlexureSystem): - """A OpenFlexureSystem Thing that always claims to be a Raspberry Pi.""" - - is_raspberrypi: bool = True +def test_check_shutdown_commands(): + """Check the shutdown and reboot commands are as expected.""" + # Check the shutdown commands are as expected. + assert system.LEGACY_SHUTDOWN_CMD == ["sudo", "shutdown", "-h", "now"] + assert system.OFM_SHUTDOWN_CMD == ["ofm", "system-shutdown"] + # Check the reboot command are as expected. + assert system.LEGACY_REBOOT_CMD == ["sudo", "shutdown", "-r", "now"] + assert system.OFM_REBOOT_CMD == ["ofm", "system-restart"] -class MockNonPiSystem(system.OpenFlexureSystem): - """A OpenFlexureSystem Thing that never claims to be a Raspberry Pi.""" - - is_raspberrypi: bool = False - - -def test_pi_shutdown(mocker): +def test_pi_shutdown_and_restart(system_thing, mocker): """Check that on a Pi will receive the correct shutdown command.""" - # Check the shutdown command is as expected. - assert system.SHUTDOWN_CMD == ["sudo", "shutdown", "-h", "now"] - # Mock the shutdown command as we don't want to shutdown when running tests. + # Mock the commands as we don't want to shutdown when running tests. mocker.patch.object( - system, "SHUTDOWN_CMD", new=["python", "-c", "print('shutdown')"] + system, "LEGACY_SHUTDOWN_CMD", new=["python", "-c", "print('legacy shutdown')"] ) - # Call shutdown on a MockPiSystem - system_thing = create_thing_without_server(MockPiSystem) + mocker.patch.object( + system, "OFM_SHUTDOWN_CMD", new=["python", "-c", "print('ofm shutdown')"] + ) + mocker.patch.object( + system, "LEGACY_REBOOT_CMD", new=["python", "-c", "print('legacy reboot')"] + ) + mocker.patch.object( + system, "OFM_REBOOT_CMD", new=["python", "-c", "print('ofm reboot')"] + ) + + # Pretend to be a pi. + type(system_thing).is_raspberrypi = mocker.PropertyMock(return_value=True) + + # Check on trixie the ofm commands are used. + type(system_thing).os_version = mocker.PropertyMock(return_value="trixie") result = system_thing.shutdown() - - # Check the result of the mock command was returned - assert result.output.strip() == "shutdown" + assert result.output.strip() == "ofm shutdown" assert result.error.strip() == "" - - -def test_pi_reboot(mocker): - """Check that on a Pi will receive the correct reboot command.""" - # Check the reboot command is as expected. - assert system.REBOOT_CMD == ["sudo", "shutdown", "-r", "now"] - # Mock the reboot command as we don't want to reboot when running tests. - mocker.patch.object(system, "REBOOT_CMD", new=["python", "-c", "print('restart')"]) - # Call reboot on a MockPiSystem - system_thing = create_thing_without_server(MockPiSystem) result = system_thing.reboot() + assert result.output.strip() == "ofm reboot" + assert result.error.strip() == "" - # Check the result of the mock command was returned - assert result.output.strip() == "restart" + # Otherwise the legacy commands are used. + type(system_thing).os_version = mocker.PropertyMock(return_value="bookworm") + result = system_thing.shutdown() + assert result.output.strip() == "legacy shutdown" + assert result.error.strip() == "" + result = system_thing.reboot() + assert result.output.strip() == "legacy reboot" assert result.error.strip() == "" -def test_non_pi_shutdown(mocker): +def test_non_pi_shutdown(system_thing, mocker): """Check that a server not on a pi runs os.kill when asked to shutdown. It should do this instead of running the shutdown command in a subprocess. @@ -133,7 +169,9 @@ def test_non_pi_shutdown(mocker): mock_kill = mocker.patch("os.kill") # Get this subprocess pid = os.getpid() - system_thing = create_thing_without_server(MockNonPiSystem) + + type(system_thing).is_raspberrypi = mocker.PropertyMock(return_value=False) + result = system_thing.shutdown() # Check it tried to kill this process with SIGTERM @@ -143,9 +181,9 @@ def test_non_pi_shutdown(mocker): assert "but the server has not shutdown" in result.error -def test_non_pi_reboot(): +def test_non_pi_reboot(system_thing, mocker): """Check that a server not on a pi refuses to restart.""" - system_thing = create_thing_without_server(MockNonPiSystem) + type(system_thing).is_raspberrypi = mocker.PropertyMock(return_value=False) result = system_thing.reboot() # Check output is an appropriate error message