Merge branch 'check_os' into 'v3'

Change shutdown command depending on OS

See merge request openflexure/openflexure-microscope-server!574

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"]
+LEGACY_SHUTDOWN_CMD = ["sudo", "shutdown", "-h", "now"]
-REBOOT_CMD = ["sudo", "shutdown", "-r", "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,

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/"

tests/unit_tests/assets/os-release-broken

@@ -0,0 +1 @@
+This doesn't contain the data we are looking for.

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):
+def test_check_shutdown_commands():
-    """A OpenFlexureSystem Thing that always claims to be a Raspberry Pi."""
+    """Check the shutdown and reboot commands are as expected."""
-
+    # Check the shutdown commands are as expected.
-    is_raspberrypi: bool = True
+    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):
+def test_pi_shutdown_and_restart(system_thing, mocker):
-    """A OpenFlexureSystem Thing that never claims to be a Raspberry Pi."""
-
-    is_raspberrypi: bool = False
-
-
-def test_pi_shutdown(mocker):
     """Check that on a Pi will receive the correct shutdown command."""
-    # Check the shutdown command is as expected.
+    # Mock the commands as we don't want to shutdown when running tests.
-    assert system.SHUTDOWN_CMD == ["sudo", "shutdown", "-h", "now"]
-    # Mock the shutdown command 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
+    mocker.patch.object(
-    system_thing = create_thing_without_server(MockPiSystem)
+        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()
-
+    assert result.output.strip() == "ofm shutdown"
-    # Check the result of the mock command was returned
-    assert result.output.strip() == "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
+    # Otherwise the legacy commands are used.
-    assert result.output.strip() == "restart"
+    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