76 lines
3.6 KiB
Python
76 lines
3.6 KiB
Python
"""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
|