feat(vesc): integrate native calibration diagnostics and configuration archives

plugins/vesc/tests/test_receiver.py

@@ -0,0 +1,75 @@
+"""Firmware neutral-band regression and retained native failure evidence."""
+import json
+import struct
+import unittest
+
+import test_motor_test
+from test_motor_test import configuration
+from runtime.configuration import decode
+from runtime.receiver import active, neutral_band
+
+
+class ReceiverTests(unittest.TestCase):
+    def setUp(self):
+        self.case=test_motor_test.PulseTests('test_fixed_wire_commands_no_broadcast_and_no_parameters')
+        self.case.setUp(); self.addCleanup(self.case.tearDown)
+
+    def offset(self, device, level, band=None):
+        original=device.link.query
+        def query(code,timeout=2):
+            if code==31:return bytes([31])+struct.pack('>ii',int(level*1e6),1466000)
+            raw=original(code,timeout)
+            if code==17 and band is not None:
+                return configuration('application',{**decode(raw,'application'),'app_ppm_conf.hyst':band})
+            return raw
+        device.link.query=query
+
+    def test_inside_deadband_allows_test_without_changing_ppm_configuration(self):
+        c=self.case
+        for device in c.devices:self.offset(device,-.066)
+        before=[d.link.query(17) for d in c.devices]
+        result=c.service.execute(c.command_for_pulse())
+        self.assertEqual(result['state'],'complete',result)
+        self.assertEqual(result['result']['outcome'],'duration')
+        self.assertFalse(c.service.motor.latched)
+        self.assertEqual(before,[d.link.query(17) for d in c.devices])
+
+    def test_each_peer_uses_its_own_fresh_band_and_active_input_blocks_motion(self):
+        c=self.case
+        for device in c.devices:self.offset(device,-.066)
+        self.assertEqual(c.service.execute(c.command_for_pulse())['state'],'complete')
+        c.sent.clear()
+        self.offset(c.devices[1],-.066,.05)
+        result=c.service.execute(c.command_for_pulse())
+        self.assertEqual(result['state'],'error')
+        self.assertTrue(c.service.motor.latched)
+        self.assertEqual(c.sent,[])
+
+    def test_unknown_band_and_invalid_input_do_not_assume_neutral(self):
+        with self.assertRaises(ValueError):self.case.service.motor.receiver_active('missing',0)
+        for level,band in [(float('nan'),.15),(0,float('nan')),(0,0),(0,.8)]:
+            with self.assertRaises(ValueError):active(level,band)
+        self.assertFalse(active(.15,.15))
+        self.assertTrue(active(-.150001,.15))
+
+    def test_native_read_and_preflight_failures_keep_transport_diagnostics(self):
+        c=self.case
+        for action,code in [('vesc.input.read',31),('vesc.motor.pulse',17)]:
+            with self.subTest(action=action):
+                device=c.devices[0]; original=device.link.query
+                trace={'command':code,'stage':'native_response','transport':{'request_emitted':True,'packets_received':0}}
+                def query(actual,timeout=2):
+                    if actual==code:
+                        error=OSError('native timeout');error.native_rpc=trace;error.native_history=[trace]
+                        raise error
+                    return original(actual,timeout)
+                device.link.query=query
+                command=c.command_for_pulse() if action=='vesc.motor.pulse' else c.command(action=action)
+                result=c.service.execute(command)
+                self.assertEqual(result['state'],'error')
+                self.assertEqual(result['result']['failure']['native_rpc'],trace)
+                saved=json.loads((c.service.root/(command['operation_id']+'.json')).read_text())
+                self.assertEqual(saved['receipt'],result)
+                self.assertEqual(c.service.execute(command),result)
+                self.assertEqual(c.sent,[])
+                device.link.query=original