Mock serial ownership inspection in Ethernet opener fixtures

test/test_ethernet_hil_serial_open.py

@@ -5,6 +5,7 @@ import importlib.util
 import io
 import os
 from pathlib import Path
+import subprocess
 import sys
 from types import ModuleType, SimpleNamespace
 import unittest
@@ -15,6 +16,7 @@ BASE = Path(__file__).resolve().parents[1] / "tools/hil"
 sys.path.insert(0, str(BASE))
 import ethernet_kiss_probe as probe
 import ethernet_tap_relay as tap
+import serial_session as session
 
 
 class FakeSerial:
@@ -25,6 +27,7 @@ class FakeSerial:
         self.driver_rts = True
         self.driver_dtr = True
         self.is_open = False
+        self.opens = 0
         self.resets = 0
         self.writes = []
         self.trace = []
@@ -57,6 +60,7 @@ class FakeSerial:
         self.trace.append(("dtr", value))
 
     def open(self):
+        self.opens += 1
         self.is_open = True
         # POSIX pySerial's real open order: DTR before RTS unless deferred.
         if not self.dsrdtr:
@@ -92,18 +96,47 @@ def serial_modules(port):
 
 @unittest.skipUnless(os.name == "posix", "POSIX CH340 pin-order behavior")
 class EthernetHilSerialTests(unittest.TestCase):
+    def setUp(self):
+        # Exercise the real Linux ownership guard without consulting host
+        # fuser or a nonexistent synthetic port. A quiet rc1 is the exact
+        # successful no-owner contract; diagnostic output must still refuse.
+        for inspector in (patch.object(session.sys, "platform", "linux"),
+                          patch.object(session.shutil, "which", return_value="/mock/fuser")):
+            inspector.start()
+            self.addCleanup(inspector.stop)
+        inspector = patch.object(session.subprocess, "run", return_value=
+                                 subprocess.CompletedProcess([], 1, stdout="", stderr=""))
+        self.inspections = inspector.start()
+        self.addCleanup(inspector.stop)
+
     def assert_safe(self, port):
+        self.assertEqual(port.opens, 1)
         self.assertEqual(port.resets, 0)
         self.assertEqual(port.trace[0], ("rts", False))
         self.assertFalse(port.dsrdtr)
         self.assertFalse(port.driver_rts)
         self.assertFalse(port.driver_dtr)
+        self.inspections.assert_called_once_with(["/mock/fuser", "/dev/ttyUSB-bench"],
+                                                  capture_output=True, text=True, timeout=3)
+
+    def companion_main(self, port):
+        spec = importlib.util.spec_from_file_location("ethernet_bench_serial_under_test", BASE / "ethernet_bench_serial.py")
+        module = importlib.util.module_from_spec(spec)
+        with patch.dict(sys.modules, serial_modules(port)):
+            spec.loader.exec_module(module)
+            with patch.object(sys, "argv", ["ethernet_bench_serial.py", "--port", "/dev/ttyUSB-bench",
+                                             "--command", "get ethernet"]), \
+                    patch.object(module, "read_quiet", return_value="OK"), \
+                    patch.object(module, "read_command_reply", return_value="get ethernet\r\n  -> on\r\n> "), \
+                    contextlib.redirect_stdout(io.StringIO()):
+                module.main()
 
     def test_reset_simulator_detects_original_plain_open_bug(self):
         port = FakeSerial()
         port.dtr = port.rts = False
         port.open()
         self.assertEqual(port.resets, 1)
+        self.inspections.assert_not_called()
 
     def test_raw_kiss_probe_uses_shared_safe_opener(self):
         port = FakeSerial()
@@ -121,20 +154,35 @@ class EthernetHilSerialTests(unittest.TestCase):
 
     def test_companion_cli_uses_shared_safe_opener(self):
         port = FakeSerial()
-        spec = importlib.util.spec_from_file_location("ethernet_bench_serial_under_test", BASE / "ethernet_bench_serial.py")
-        module = importlib.util.module_from_spec(spec)
-        with patch.dict(sys.modules, serial_modules(port)):
-            spec.loader.exec_module(module)
-            with patch.object(sys, "argv", ["ethernet_bench_serial.py", "--port", "/dev/ttyUSB-bench",
-                                             "--command", "get ethernet"]), \
-                    patch.object(module, "read_quiet", return_value="OK"), \
-                    patch.object(module, "read_command_reply", return_value="get ethernet\r\n  -> on\r\n> "), \
-                    contextlib.redirect_stdout(io.StringIO()):
-                module.main()
+        self.companion_main(port)
         self.assert_safe(port)
         self.assertIn(b"get ethernet\n", port.writes)
         self.assertFalse(port.is_open)
 
+    def test_busy_or_ambiguous_ownership_refuses_each_real_constructor_before_open(self):
+        for result in (subprocess.CompletedProcess([], 0, stdout="1234", stderr=""),
+                       subprocess.CompletedProcess([], 1, stdout="", stderr="Usage: fuser\n"),
+                       subprocess.CompletedProcess([], 1, stdout="", stderr="permission denied\n")):
+            for kind in ("raw-kiss", "tap", "companion-cli"):
+                with self.subTest(kind=kind, status=result.returncode, diagnostic=result.stderr):
+                    self.inspections.reset_mock()
+                    self.inspections.return_value = result
+                    port = FakeSerial()
+                    with patch.dict(sys.modules, serial_modules(port)), \
+                            patch.object(probe.threading, "Thread"), \
+                            self.assertRaisesRegex(RuntimeError, "already in use|fuser error"):
+                        if kind == "raw-kiss":
+                            probe.KissEndpoint("serial:///dev/ttyUSB-bench")
+                        elif kind == "tap":
+                            tap.KissStream("serial:///dev/ttyUSB-bench", 115200)
+                        else:
+                            self.companion_main(port)
+                    self.inspections.assert_called_once()
+                    self.assertEqual(port.opens, 0)
+                    self.assertFalse(port.is_open)
+                    self.assertEqual(port.trace, [])
+                    self.assertEqual(port.writes, [])
+
 
 if __name__ == "__main__":
     unittest.main()