feat(fleet): preserve operator VESC integration before final driver merge

2026-09-25 16:40:46 +03:00
parent dad11b47d7
commit 71a648fec8
128 changed files with 22390 additions and 59 deletions
@@ -0,0 +1,50 @@
+"""Targeted attachment checks preserve generation and topology rejection."""
+from pathlib import Path
+import sys,tempfile,unittest
+from unittest.mock import patch
+sys.path.insert(0,str(Path(__file__).resolve().parents[1]))
+from runtime.serial import Attachment, attachment_at, check_attachment, discover
+
+class AttachmentTests(unittest.TestCase):
+    def setUp(self):
+        self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup)
+        self.root=Path(self.temp.name);self.device=self.root/'1-2.3';self.device.mkdir()
+        for name,value in {'idVendor':'0483','idProduct':'5740','product':'ChibiOS/RT Virtual COM Port','devnum':'17','speed':'12'}.items():
+            (self.device/name).write_text(value)
+        (self.device/'1-2.3:1.0/tty/ttyACM2').mkdir(parents=True)
+        self.expected=Attachment('1-2.3','17','ttyACM2','12 Мбит/с')
+
+    def test_target_check_never_enumerates_siblings(self):
+        original=Path.iterdir
+        def entries(path):
+            if path==self.root:raise AssertionError('full bus scan in per-request check')
+            return original(path)
+        with patch.object(Path,'iterdir',entries):check_attachment(self.expected,self.root)
+        self.assertEqual(discover(self.root),[self.expected])
+
+    def test_generation_change_or_removed_port_invalidates_old_owner(self):
+        (self.device/'devnum').write_text('18')
+        with self.assertRaises(OSError):check_attachment(self.expected,self.root)
+        (self.device/'devnum').write_text('17')
+        (self.device/'1-2.3:1.0/tty/ttyACM2').rmdir()
+        with self.assertRaises(OSError):check_attachment(self.expected,self.root)
+
+    def test_ambiguous_tty_or_other_vendor_rejected(self):
+        extra=self.device/'1-2.3:1.1/tty/ttyACM3';extra.mkdir(parents=True)
+        self.assertIsNone(attachment_at(self.device))
+        extra.rmdir();(self.device/'idVendor').write_text('1234')
+        self.assertIsNone(attachment_at(self.device))
+
+    def test_driver_sibling_tty_cannot_be_mistaken_for_device_interface(self):
+        (self.device/'driver/tty/ttyACM9').mkdir(parents=True)
+        self.assertEqual(attachment_at(self.device),self.expected)
+        with self.assertRaises(OSError):check_attachment(Attachment('../1-2.3','17','ttyACM2','12 Мбит/с'),self.root)
+
+    def test_generation_change_during_read_is_rejected(self):
+        original=Path.read_text;reads=0
+        def read(path,*args,**kwargs):
+            nonlocal reads
+            if path==self.device/'devnum':
+                reads+=1;return '17' if reads==1 else '18'
+            return original(path,*args,**kwargs)
+        with patch.object(Path,'read_text',read):self.assertIsNone(attachment_at(self.device))