Change shutdown command depending on OS versions

src/openflexure_microscope_server/things/system.py

@@ -20,8 +20,10 @@ 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"
 
@@ -114,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,
@@ -134,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/test_system_thing.py

@@ -112,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.
@@ -165,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
@@ -175,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