diff --git a/opendbc_repo/opendbc/car/tests/test_car_interfaces.py b/opendbc_repo/opendbc/car/tests/test_car_interfaces.py index a95e8bacd5..5b20214d20 100644 --- a/opendbc_repo/opendbc/car/tests/test_car_interfaces.py +++ b/opendbc_repo/opendbc/car/tests/test_car_interfaces.py @@ -7,6 +7,9 @@ from opendbc.car.fingerprints import FW_VERSIONS from opendbc.car.fw_versions import FW_QUERY_CONFIGS from opendbc.car.interfaces import CarInterfaceBase, get_interface_attr from opendbc.car.values import PLATFORMS +from opendbc.car.volvo.cma_carcontroller import CarControllerCMA +from opendbc.car.volvo.cma_carstate import CarStateCMA +from opendbc.car.volvo.values import CAR as VOLVO from opendbc.testing import Fuzzy, fuzzy_test ALL_ECUS = tuple(sorted({ecu for ecus in FW_VERSIONS.values() for ecu in ecus} | @@ -112,8 +115,12 @@ def _make_optional_hardware_test(car_name): self.assertEqual(params.carFingerprint, car_name) self.assertEqual(fingerprint, gen_empty_fingerprint()) interface = interface_type(params) - self.assertIsInstance(interface.CS, interface_type.CarState) - self.assertIsInstance(interface.CC, interface_type.CarController) + if car_name in (VOLVO.VOLVO_XC40_RECHARGE, VOLVO.VOLVO_S60_RECHARGE, VOLVO.POLESTAR_2): + self.assertIsInstance(interface.CS, CarStateCMA) + self.assertIsInstance(interface.CC, CarControllerCMA) + else: + self.assertIsInstance(interface.CS, interface_type.CarState) + self.assertIsInstance(interface.CC, interface_type.CarController) return test diff --git a/opendbc_repo/opendbc/safety/tests/test_volvo_cma.py b/opendbc_repo/opendbc/safety/tests/test_volvo_cma.py new file mode 100644 index 0000000000..c9212d7dca --- /dev/null +++ b/opendbc_repo/opendbc/safety/tests/test_volvo_cma.py @@ -0,0 +1,373 @@ +"""Native CMA/SPA contracts, also collected by the safety unittest coverage gate. + +The fixtures use the same DBCs and checksum encoders as the existing Volvo car +integration tests. Permissions come from physical RX, never a forced active flag. +""" +import unittest + +from opendbc.can import CANPacker +from opendbc.car.volvo import cma_can +from opendbc.car.volvo.helpers import ( + checksum_lca_2_message, checksum_2_0x69_message, checksum_lca_5_message, + checksum_1_pscm_related_message, checksum_2_pscm_related_message, +) +from opendbc.safety.tests.libsafety import libsafety_py + + +RELAY_BUSES = {0x58: 2, 0x57: 2, 0x69: 2, 0x90: 2, 0x67: 2, 0x97: 2, 0x92: 2, 0x16: 0, 0x17: 0} + + +def checksummed(message): + address, payload, bus = message + data = bytearray(payload) + if address == 0x69: + data[6] = checksum_lca_2_message(data[0], data[5]) + data[2] = checksum_2_0x69_message(data[0], data[1], data[3], data[4]) + elif address == 0x67: + data[2] = checksum_lca_5_message(data[0], data[1], data[3], data[4], data[5]) + elif address == 0x17: + data[0] = checksum_1_pscm_related_message(data[1], data[2]) + data[3] = checksum_2_pscm_related_message(data[2]) + return address, bytes(data), bus + + +class TestVolvoCMASafety(unittest.TestCase): + PARAM = 0 + + def setUp(self): + self.safety = libsafety_py.libsafety + self.mid = CANPacker('volvo_mid_1') + self.pt = CANPacker('volvo_front_1_spa' if self.PARAM else 'volvo_front_1_cma') + self.reset() + + def reset(self): + self.assertEqual(self.safety.set_safety_hooks(36, self.PARAM), 0) + self.safety.init_tests() + self.safety.set_timer(10000) + + def rx(self, message): + address, data, bus = message + return self.safety.safety_rx_hook(libsafety_py.make_CANPacket(address, bus, data)) + + def tx(self, message): + address, data, bus = message + return self.safety.safety_tx_hook(libsafety_py.make_CANPacket(address, bus, data)) + + def sources(self, cruise=True, gear=3, sample=0, parity=0): + definitions: list[tuple[str, int, dict[str, float]]] = [ + ('SPEED', 0, {'SPEED': 3000}), ('SPEED_2', 0, {}), ('GEAR_POSITION', 0, {'GEAR_POSITION': gear}), + ('LCA', 0, {}), ('LCA_2', 0, {'COUNTER_1': (sample * 2 + parity) % 16, 'COUNTER_2': sample * 4 % 16}), + ('LCA_3', 0, {}), ('LCA_4', 0, {}), ('LCA_5', 0, {}), ('LCA_6', 0, {}), ('LCA_7', 0, {}), + ('PSCM', 2, {}), ('DRIVER_INPUT', 2, {}), ('EGSM', 2, {}), + ('PSCM_RELATED', 2, {'NEW_SIGNAL_2': 8}), ('SAS', 2, {}), + ] + messages = [self.mid.make_can_msg(name, bus, values) for name, bus, values in definitions] + messages.append(self.pt.make_can_msg('ECM_1', 1, {'ACCELERATOR_PEDAL_POS': 0 if self.PARAM else 20})) + cruise_signal = 'CRUISE_CONTROL_SPA_ENABLED' if self.PARAM else 'CRUISE_CONTROL_ENABLED' + messages.append(self.pt.make_can_msg('BUS1_CRUISE_CONTROL', 1, {cruise_signal: int(cruise)})) + return [checksummed(message) for message in messages] + + def feed(self, **kwargs): + incoming = self.sources(**kwargs) + for message in incoming: + self.assertTrue(self.rx(message), message) + self.safety.safety_tick() + self.assertTrue(self.safety.safety_config_valid()) + self.assertEqual(bool(self.safety.get_controls_allowed()), kwargs.get('cruise', True)) + return {address: (address, payload, bus) for address, payload, bus in incoming} + + def relay(self, source): + address, data, _ = source + return address, data, RELAY_BUSES[address] + + def lca_command(self, positive=614, negative=-614): + return cma_can.create_lca_message(self.mid, True, 0, { + 'LCA_STEER': 0, 'OPAQUE_BYTE_0': 0, 'OPAQUE_BYTE_2': 0, 'OPAQUE_BYTE_7': 0, + }, authority_pos=positive, authority_neg=negative) + + def angle_command(self, raw): + # LCA_5_STEER is signed 15-bit, 0.05596 degrees/count, in bytes 6-7. + data = bytearray(8) + data[6:8] = (raw & 0x7FFF).to_bytes(2, 'big') + return checksummed((0x67, data, 2)) + + def test_rx_sensor_scales_and_profile_pedals(self): + self.feed() + self.assertAlmostEqual(self.safety.get_vehicle_speed_max(), 3000 * .003977, places=5) + for raw in (-100, 100): + for _ in range(6): + self.assertTrue(self.rx(self.mid.make_can_msg('PSCM', 2, {'PSCM_ANGLE_SENSOR': raw * .05596}))) + self.assertEqual(self.safety.get_angle_meas_min(), raw) + self.assertEqual(self.safety.get_angle_meas_max(), raw) + for torque in (-10, 10): + for _ in range(6): + self.assertTrue(self.rx(self.mid.make_can_msg('DRIVER_INPUT', 2, {'STEERING_DRIVER_INPUT': torque}))) + self.assertEqual(self.safety.get_torque_driver_min(), torque) + self.assertEqual(self.safety.get_torque_driver_max(), torque) + threshold = 1 if self.PARAM else 21 + step = 1 / 256 if self.PARAM else 1 + for value, pressed in ((threshold + step, True), (threshold, False)): + self.assertTrue(self.rx(self.pt.make_can_msg('ECM_1', 1, {'ACCELERATOR_PEDAL_POS': value}))) + self.assertEqual(bool(self.safety.get_gas_pressed_prev()), pressed) + for sample, pressed in enumerate((True, False), 1): + message = self.mid.make_can_msg('LCA_2', 0, {'COUNTER_1': sample * 2, 'COUNTER_2': sample * 4, + 'BRAKE_PEDAL_PRESSED_B': int(pressed)}) + self.assertTrue(self.rx(checksummed(message))) + self.assertEqual(bool(self.safety.get_brake_pressed_prev()), pressed) + self.assertFalse(self.safety.get_controls_allowed()) + + def test_stock_cruise_withdrawal_and_reengagement(self): + for enabled in (False, True, False, True): + signal = 'CRUISE_CONTROL_SPA_ENABLED' if self.PARAM else 'CRUISE_CONTROL_ENABLED' + self.assertTrue(self.rx(self.pt.make_can_msg('BUS1_CRUISE_CONTROL', 1, {signal: int(enabled)}))) + self.assertEqual(bool(self.safety.get_controls_allowed()), enabled) + if not self.PARAM: + self.assertTrue(self.rx(self.pt.make_can_msg('BUS1_CRUISE_CONTROL', 1, {}))) + self.assertTrue(self.rx(self.pt.make_can_msg('BUS1_CRUISE_CONTROL', 1, {'CRUISE_CONTROL_ENABLED_IDLE_TRAFFIC': 1}))) + self.assertTrue(self.safety.get_controls_allowed()) + + def test_stock_only_when_cruise_withdrawn_and_original_age_expires(self): + incoming = self.feed(cruise=False) + for address in RELAY_BUSES: + message = self.relay(incoming[address]) + self.assertTrue(self.tx(message)) + data = bytearray(message[1]) + data[0] ^= 1 + self.assertFalse(self.tx((address, data, message[2]))) + self.safety.set_timer(110000) + for address in RELAY_BUSES: + self.assertTrue(self.tx(self.relay(incoming[address]))) + self.safety.set_timer(110001) + for address in RELAY_BUSES: + self.assertFalse(self.tx(self.relay(incoming[address]))) + + def test_allowlist_pins_bus_and_length(self): + incoming = self.feed(cruise=False) + for address, expected_bus in RELAY_BUSES.items(): + for bus in range(8): + self.assertEqual(bool(self.tx((address, incoming[address][1], bus))), bus == expected_bus) + for length in range(8): + self.assertFalse(self.tx((address, incoming[address][1][:length], bus))) + for address in (0x15, 0x55, 0x60, 0x68, 0x80, 0x45): + for bus in range(3): + self.assertFalse(self.tx((address, bytes(8), bus))) + + def test_active_native_source_bound_commands(self): + incoming = self.feed() + for address in RELAY_BUSES: + self.assertTrue(self.tx(self.relay(incoming[address]))) + self.assertTrue(self.tx(self.lca_command())) + self.assertTrue(self.tx(self.angle_command(1))) + # These two hands-on relay payloads are emitted by create_pscm_message. + for hands_a, hands_b in ((195, 186), (249, 154)): + message = self.mid.make_can_msg('PSCM', 0, {'HANDS_ON_STEERING_WHEEL_A': hands_a, + 'HANDS_ON_STEERING_WHEEL_B': hands_b}) + self.assertTrue(self.tx(message)) + # Counter replication may advance, but its observed echo must be preserved or zeroed. + for counter in range(15): + message = self.mid.make_can_msg('PSCM_RELATED', 0, {'NEW_SIGNAL_2': 8, + 'SIG1_BYTE_1_HI_NIBBLE': counter, 'SIG1_REPLICA_BYTE_2_LO_NIBLE': counter}) + self.assertTrue(self.tx(checksummed(message))) + + def test_opaque_stock_fields_cannot_be_forged_with_recomputed_checksums(self): + incoming = self.feed() + # Start from accepted active templates, so an opaque mutation cannot be + # rejected merely because the stock frame lacks the fixed active fields. + active = { + 0x58: self.lca_command(), + 0x57: cma_can.create_lca_3_message(self.mid, True, 0, {'SPEED_A': 0, 'SPEED_B': 0, 'NEW_SIGNAL_9': 0}, 0), + 0x69: checksummed(self.mid.make_can_msg('LCA_2', 2, {'BYTE_0': 24, 'PILOT_ASSIST_ENGAGED': 1, 'NEW_SIGNAL_3': 3})), + 0x90: self.mid.make_can_msg('LCA_4', 2, {'LCA_ENABLE': 3}), + 0x67: self.angle_command(1), + 0x16: self.mid.make_can_msg('PSCM', 0, {'HANDS_ON_STEERING_WHEEL_A': 195, 'HANDS_ON_STEERING_WHEEL_B': 186}), + 0x17: checksummed(self.mid.make_can_msg('PSCM_RELATED', 0, {'NEW_SIGNAL_2': 8, + 'SIG1_BYTE_1_HI_NIBBLE': 1, 'SIG1_REPLICA_BYTE_2_LO_NIBLE': 1})), + } + for address, byte, mask in ((0x58, 6, 1), (0x57, 2, 1), (0x69, 5, 0x40), (0x90, 4, 1), + (0x67, 0, 1), (0x97, 0, 1), (0x92, 0, 1), (0x16, 7, 1), (0x17, 6, 1)): + with self.subTest(address=address): + baseline = active.get(address, self.relay(incoming[address])) + self.assertTrue(self.tx(baseline)) + data = bytearray(baseline[1]) + data[byte] ^= mask + self.assertFalse(self.tx(checksummed((address, data, RELAY_BUSES[address])))) + + def test_torque_authority_limits(self): + self.feed() + for positive, negative, allowed in ((0, 0, True), (614, -614, True), (615, -614, False), + (614, -615, False), (-1, 0, False), (0, 1, False)): + self.assertEqual(bool(self.tx(self.lca_command(positive, negative))), allowed) + + def test_angle_absolute_and_rate_limits(self): + for raw in (-9651, 9651): + self.reset() + self.feed() + self.safety.set_desired_angle_last(raw) + self.assertFalse(self.tx(self.angle_command(raw))) + for raw in (-1, 1): + self.reset() + self.feed() + self.assertTrue(self.tx(self.angle_command(raw))) + self.reset() + self.feed() + self.assertFalse(self.tx(self.angle_command(1000))) + + def test_pscm_relay_cannot_invent_measured_angle(self): + self.feed() + for raw in (-55, -54, 54, 55): + self.assertTrue(self.rx(self.mid.make_can_msg('PSCM', 2, {'PSCM_ANGLE_SENSOR': raw * .05596}))) + # Keep the source-bound angle intact and isolate the measurement guard. + # The standard measurement fixture does not grant control permission. + self.safety.set_angle_meas(0, 0) + message = self.mid.make_can_msg('PSCM', 0, {'PSCM_ANGLE_SENSOR': raw * .05596, + 'HANDS_ON_STEERING_WHEEL_A': 195, 'HANDS_ON_STEERING_WHEEL_B': 186}) + self.assertEqual(bool(self.tx(message)), abs(raw) <= 54) + for hands_a, hands_b in ((195, 154), (249, 186), (0, 1)): + self.assertFalse(self.tx(self.mid.make_can_msg('PSCM', 0, { + 'HANDS_ON_STEERING_WHEEL_A': hands_a, 'HANDS_ON_STEERING_WHEEL_B': hands_b}))) + + def test_gear_is_physical_current_and_drive_only(self): + for gear in range(8): + with self.subTest(gear=gear): + self.reset() + incoming = self.feed(gear=gear) + self.assertEqual(bool(self.tx(self.lca_command())), gear in (3, 4)) + self.assertTrue(self.tx(self.relay(incoming[0x58]))) + self.reset() + self.feed() + self.safety.set_timer(110001) + for message in self.sources(sample=1): + if message[0] != 0x80: + self.assertTrue(self.rx(message)) + self.assertTrue(self.safety.get_controls_allowed()) + self.assertFalse(self.tx(self.lca_command())) + + def test_each_known_checksum_rejects_corrupt_rx_and_tx(self): + for address, index in ((0x69, 6), (0x69, 2), (0x67, 2), (0x17, 0), (0x17, 3)): + with self.subTest(address=address, index=index): + self.reset() + incoming = self.feed() + data = bytearray(incoming[address][1]) + data[index] ^= 1 + self.assertFalse(self.tx((address, data, RELAY_BUSES[address]))) + self.assertFalse(self.rx((address, data, incoming[address][2]))) + self.assertFalse(self.safety.get_controls_allowed()) + self.assertFalse(self.tx(self.lca_command())) + self.assertTrue(self.tx(self.relay(incoming[0x58]))) + self.safety.safety_tick() + self.assertFalse(self.safety.safety_config_valid()) + + def test_unknown_pscm_counter_domains_rejected_despite_correct_checksum(self): + for first, second in ((0x10, 0x80), (0xF0, 0x8F), (8, 0x80), (0, 0)): + self.reset() + self.feed() + data = bytearray(8) + data[1:3] = bytes((first, second)) + message = checksummed((0x17, data, 0)) + self.assertFalse(self.tx(message)) + self.assertFalse(self.rx((message[0], message[1], 2))) + + def test_documented_lca2_counter_wraps_but_frozen_counter_expires(self): + for parity in (0, 1): + self.reset() + self.feed(parity=parity) + for sample in range(1, 25): + self.safety.set_timer(10000 + sample * 20000) + self.feed(sample=sample, parity=parity) + frozen = next(message for message in self.sources(sample=24, parity=parity) if message[0] == 0x69) + for sample in range(25, 35): + self.safety.set_timer(10000 + sample * 20000) + for message in self.sources(sample=sample, parity=parity): + if message[0] == 0x69: + self.rx(frozen) # The shared wrong-counter tolerance is eventually exhausted. + else: + self.assertTrue(self.rx(message)) + self.safety.safety_tick() + self.assertFalse(self.safety.safety_config_valid()) + self.assertFalse(self.safety.get_controls_allowed()) + + def test_lca2_phase_must_come_from_checksum_valid_physical_rx(self): + for index, mask in ((1, 1), (5, 1), (5, 4)): + self.reset() + incoming = self.feed() + data = bytearray(incoming[0x69][1]) + data[index] ^= mask + self.assertFalse(self.rx(checksummed((0x69, data, 0)))) + self.assertFalse(self.safety.get_controls_allowed()) + self.reset() + message = next(message for message in self.sources() if message[0] == 0x69) + data = bytearray(message[1]) + data[6] ^= 1 + self.assertFalse(self.rx((0x69, data, 0))) + self.assertFalse(self.tx(self.relay(message))) + self.feed(parity=1) + + def test_existing_lca2_special_case_checksum(self): + message = self.mid.make_can_msg('LCA_2', 0, {'BRAKE_PEDAL_PRESSED_A': 0}) + self.assertEqual(message[1][0], 0) + self.assertEqual(message[1][5], 128) + self.assertEqual(checksummed(message)[1][6], 0) + self.assertTrue(self.rx(checksummed(message))) + + def test_every_required_receiver_needed_after_mode_reset(self): + for missing in range(17): + self.reset() + incoming = self.feed() + self.assertTrue(self.tx(self.lca_command())) + self.reset() + self.assertFalse(self.tx(self.relay(incoming[0x58]))) + messages = self.sources() + for index, message in enumerate(messages): + if index != missing: + self.assertTrue(self.rx(message)) + self.assertFalse(self.safety.safety_config_valid()) + self.assertFalse(self.tx(self.lca_command())) + self.assertTrue(self.rx(messages[missing])) + self.assertTrue(self.safety.safety_config_valid()) + self.assertTrue(self.tx(self.lca_command())) + + def test_required_receiver_loss_rejects_modified_commands(self): + for omitted in range(17): + self.reset() + self.feed() + for sample in range(1, 57): + self.safety.set_timer(10000 + sample * 20000) + for index, message in enumerate(self.sources(sample=sample)): + if index != omitted: + self.assertTrue(self.rx(message)) + self.safety.safety_tick() + self.assertFalse(self.safety.safety_config_valid()) + self.assertFalse(self.safety.get_controls_allowed()) + self.assertFalse(self.tx(self.lca_command())) + + def test_stock_relay_history_does_not_cross_mode_reset(self): + incoming = self.feed(cruise=False) + first = incoming[0x97] + for value in range(1, 5): + message = (0x97, bytes([value]) + bytes(7), 0) + self.assertTrue(self.rx(message)) + self.assertTrue(self.tx(self.relay(message))) + self.assertFalse(self.tx(self.relay(first))) + self.reset() + self.assertFalse(self.tx(self.relay(message))) + + def test_relay_malfunction_and_forwarding(self): + self.feed() + for bus in (0, 2): + for address in (*RELAY_BUSES, 0x60): + blocked = RELAY_BUSES.get(address) == 2 - bus + self.assertEqual(self.safety.safety_fwd_hook(bus, address), -1 if blocked else 2 - bus) + for address, bus in RELAY_BUSES.items(): + self.reset() + self.assertTrue(self.rx((address, bytes(8), bus))) + self.assertTrue(self.safety.get_relay_malfunction()) + self.assertFalse(self.tx(self.lca_command())) + self.assertEqual(self.safety.safety_fwd_hook(0, 0x60), -1) + + +class TestVolvoSPASafety(TestVolvoCMASafety): + PARAM = 1 + + +if __name__ == '__main__': + unittest.main() diff --git a/openpilot/starpilot/flm/tests/test_clipping_quality.py b/openpilot/starpilot/flm/tests/test_clipping_quality.py index 6d8befc463..98b89d3cde 100644 --- a/openpilot/starpilot/flm/tests/test_clipping_quality.py +++ b/openpilot/starpilot/flm/tests/test_clipping_quality.py @@ -49,9 +49,9 @@ def test_recorded_contract_is_exact_original_comparison_for_all_signs(): def test_bad_support_versions_and_unversioned_optin_fail_closed(): for position in range(3): for bad in (math.nan, math.inf, True): - values = [-17, -18, -18] + values: list[float] = [-17, -18, -18] values[position] = bad - assert not debt_is_admitted(*values, STATE_ALIGNED_DEBT) + assert not debt_is_admitted(values[0], values[1], values[2], STATE_ALIGNED_DEBT) rejects(lambda: clipping_quality('unknown')) rejects(lambda: event_clipping_quality({'event_version': 2, 'event_clipping_quality': STATE_ALIGNED_DEBT})) rejects(lambda: event_clipping_quality({'event_version': 3, 'event_clipping_quality': 'unknown'})) diff --git a/openpilot/starpilot/flm/tests/test_combined_encoder_policy.py b/openpilot/starpilot/flm/tests/test_combined_encoder_policy.py index 641b384636..b95bb3c2a1 100644 --- a/openpilot/starpilot/flm/tests/test_combined_encoder_policy.py +++ b/openpilot/starpilot/flm/tests/test_combined_encoder_policy.py @@ -50,6 +50,7 @@ def test_real_driver_measured_unwind_relief_is_weaker_and_final_only(platform): EncoderBranch(quality_contract=STATE_ALIGNED_DEBT).step(event, candidate) branch = EncoderBranch(quality_contract=STATE_ALIGNED_RELIEF) assert branch.step(event, candidate) is not None + assert branch.previous is not None assert 0 < branch.previous <= witness.final_raw # Actual recorded no-clipping status cannot be bypassed by relief. with pytest.raises(ValueError, match='limiter/support'): diff --git a/openpilot/starpilot/flm/tests/test_combined_event_policy.py b/openpilot/starpilot/flm/tests/test_combined_event_policy.py index 700391c5bf..a1c7dcc6e9 100644 --- a/openpilot/starpilot/flm/tests/test_combined_event_policy.py +++ b/openpilot/starpilot/flm/tests/test_combined_event_policy.py @@ -20,7 +20,7 @@ from openpilot.starpilot.flm.tests.test_event_fd_boundary import engine def test_exact_unwind_witnesses_and_partial_order_negatives(): assert native_relief_is_admitted(-76, -76, -75, -80, -79, -79, 4, 3, 4) assert native_relief_is_admitted(-19, -19, -18, -28, -28, -28, 9, 9, 10) - valid = [-76, -76, -75, -80, -79, -79, 4, 3, 4] + valid: list[float] = [-76, -76, -75, -80, -79, -79, 4, 3, 4] for index, value in ((2, -77), (2, 0), (2, 1), (5, -81), (5, 79), (6, -4), (7, -3), (8, -4), (8, 0), (1, float('nan'))): changed = valid.copy() changed[index] = value diff --git a/openpilot/starpilot/flm/tests/test_declared_restore.py b/openpilot/starpilot/flm/tests/test_declared_restore.py index 00f5ed09fb..77cfda6447 100644 --- a/openpilot/starpilot/flm/tests/test_declared_restore.py +++ b/openpilot/starpilot/flm/tests/test_declared_restore.py @@ -45,10 +45,13 @@ def payload(): def plan_for(restored): model = fitted_model() base = primitive_plan(model) - values = {**asdict(base), 'version': RESTORE_ACTIVATION_VERSION, - 'baseline_revision': restored.selected_baseline_revision, 'source_epoch': restored.source_epoch, - 'cutoff_ns': 2_005_000_000, 'windows': ((2_000_000_000, 2_020_000_000, 2_010_000_000, 2_010_000_000),)} - return WholeMapRestorePlan(**values, restore_envelope_sha256=restore_envelope_digest(restored)).validate() + return WholeMapRestorePlan( + version=RESTORE_ACTIVATION_VERSION, identity=base.identity, + baseline_revision=restored.selected_baseline_revision, source_epoch=restored.source_epoch, physical_context=base.physical_context, + calibration_train_blocks=base.calibration_train_blocks, calibration_heldout_blocks=base.calibration_heldout_blocks, + cutoff_ns=2_005_000_000, stage_blocks=base.stage_blocks, + windows=((2_000_000_000, 2_020_000_000, 2_010_000_000, 2_010_000_000),), + restore_envelope_sha256=restore_envelope_digest(restored)).validate() def test_restore_only_payload_has_no_shadow_or_operation_fallback(): diff --git a/openpilot/starpilot/flm/tests/test_event_admission_clock.py b/openpilot/starpilot/flm/tests/test_event_admission_clock.py index fbb110237e..0f3dccb470 100644 --- a/openpilot/starpilot/flm/tests/test_event_admission_clock.py +++ b/openpilot/starpilot/flm/tests/test_event_admission_clock.py @@ -20,11 +20,11 @@ def test_response_after_command_is_fresh_only_at_a_trusted_later_decision(): @pytest.mark.parametrize('index', (0, 1, 2)) def test_each_original_stream_must_be_nonfuture_and_within_100ms(index): - valid = [NOW - 2_000_000, NOW - 1_000_000, NOW] + valid: list[int | float] = [NOW - 2_000_000, NOW - 1_000_000, NOW] for stamp in (NOW + 1, NOW - 101_000_000, 0, -1, True, float(NOW)): values = valid.copy() values[index] = stamp - assert not _fresh_current_physical(*values, NOW) + assert not _fresh_current_physical(values[0], values[1], values[2], NOW) assert not _fresh_current_physical(NOW, NOW - 1, NOW, NOW) diff --git a/openpilot/starpilot/flm/tests/test_event_frame_index.py b/openpilot/starpilot/flm/tests/test_event_frame_index.py index 64f9118ab2..c739150d8d 100644 --- a/openpilot/starpilot/flm/tests/test_event_frame_index.py +++ b/openpilot/starpilot/flm/tests/test_event_frame_index.py @@ -1,5 +1,6 @@ """Exact original-frame join; no synthetic replay qualification.""" from types import SimpleNamespace +from typing import TypedDict from unittest import mock import pytest @@ -8,6 +9,20 @@ from openpilot.starpilot.flm.event_raw_fit import _control_frame_index, fit_even from openpilot.starpilot.flm.tests.test_event_activation import fitted_model +class StageSample(TypedDict): + control_ns: int + apply_ns: int + mono_ns: int + control: dict[str, int] + output: dict[str, int] + physical: dict[str, float] + + +class StageEvent(TypedDict): + block_id: str + samples: tuple[StageSample, ...] + + def test_index_preserves_original_frame_identity_and_missing_timestamp(): first = SimpleNamespace(control_ns=10, desired_physical_accel=.02) second = SimpleNamespace(control_ns=20, desired_physical_accel=.03) @@ -29,7 +44,7 @@ def test_actual_fit_missing_stage_frame_is_fail_closed(): # Isolate the stage join after its independently checked upstream operations. # The model comes from the real measured-row fitter, never a qualification flag. model = fitted_model() - events = tuple({'block_id': str(index), 'samples': ({'control_ns': index * 1_000_000_000, + events: tuple[StageEvent, ...] = tuple({'block_id': str(index), 'samples': ({'control_ns': index * 1_000_000_000, 'apply_ns': index * 1_000_000_000 + 1, 'mono_ns': index * 1_000_000_000 + 2, 'control': {'stamp': index * 1_000_000_000}, 'output': {'stamp': index * 1_000_000_000 + 1}, 'physical': {'lateral_accel': .02}},)} for index in range(1, 17)) diff --git a/openpilot/starpilot/flm/tests/test_friction_chatter_contract.py b/openpilot/starpilot/flm/tests/test_friction_chatter_contract.py index 1bdecc1dd4..d84a369d13 100644 --- a/openpilot/starpilot/flm/tests/test_friction_chatter_contract.py +++ b/openpilot/starpilot/flm/tests/test_friction_chatter_contract.py @@ -1,5 +1,5 @@ """Model noise remains ineligible even for explicitly retained controller chatter.""" -from dataclasses import asdict +from dataclasses import replace import pytest @@ -35,4 +35,4 @@ def test_exact_canonical_objective_only_retains_expected_chatter(group): def test_nonfriction_coordinates_cannot_claim_friction_contract(): objective = EffectObjective('friction.left', GROUPS['friction.left'], 'a' * 64, 'b' * 64) with pytest.raises(ValueError, match='canonical event group'): - EffectObjective(**{**asdict(objective), 'names': GROUPS['ff.left']}) + replace(objective, names=GROUPS['ff.left']) diff --git a/openpilot/starpilot/flm/tests/test_native_controls_acceptance.py b/openpilot/starpilot/flm/tests/test_native_controls_acceptance.py index b315cd3bca..bf67a87bff 100644 --- a/openpilot/starpilot/flm/tests/test_native_controls_acceptance.py +++ b/openpilot/starpilot/flm/tests/test_native_controls_acceptance.py @@ -17,7 +17,7 @@ from openpilot.starpilot.flm.learner_factory import TorqueCollectionAdapter, RAW from openpilot.starpilot.flm.runtime_adapter_registry import baseline_values, baseline_revision, selected_descriptor, selected_identity from openpilot.starpilot.flm.tests.native_controls_fixture import native_frame from openpilot.starpilot.flm.tests.test_registered_controls_global import controls_for, published_cp -from openpilot.starpilot.lateral.controller_selection import DOCUMENT_KEY as SELECTION_KEY, replace_mode +from openpilot.starpilot.lateral.controller_selection import DOCUMENT_KEY as SELECTION_KEY, ControllerMode, replace_mode from openpilot.starpilot.lateral.learning_preferences import KEY, replace_preference from openpilot.starpilot.lateral.torque_settings import DOCUMENT_KEY, FieldChoice, PlatformProfile, serialize_document @@ -104,14 +104,15 @@ def test_off_keeps_historical_controller_output_despite_qualified_native_packet( @contextmanager def manual_ioniq(): cp = cp_for(HYUNDAI.HYUNDAI_IONIQ_6) - basis = tuple(float(getattr(cp.lateralTuning.torque, name)) for name in ('latAccelFactor', 'latAccelOffset', 'friction')) + torque = cp.lateralTuning.torque + basis = (float(torque.latAccelFactor), float(torque.latAccelOffset), float(torque.friction)) document = serialize_document({str(cp.carFingerprint): PlatformProfile(basis, FieldChoice('custom', basis[0] * .95), FieldChoice())}) with OpenpilotPrefix(), mock.patch.dict(os.environ, {'REPLAY': '1', 'SIMULATION': '0', 'TORQUE_REPLAY_RUNTIME': '0', 'AOL_REPLAY_RUNTIME': '0', 'LANE_CENTERING_REPLAY_RUNTIME': '0'}), mock.patch('openpilot.selfdrive.controls.controlsd.messaging.PubMaster'): params = Params() params.put('CarParams', published_cp(cp, params).to_bytes(), block=True) params.put(KEY, replace_preference(None, cp, True), block=True) - params.put(SELECTION_KEY, json.loads(replace_mode(None, cp, 'starpilot')), block=True) + params.put(SELECTION_KEY, json.loads(replace_mode(None, cp, ControllerMode.STARPILOT)), block=True) params.put(DOCUMENT_KEY, json.loads(document), block=True) yield Controls(), document, basis diff --git a/openpilot/starpilot/flm/tests/test_state_aligned_encoder.py b/openpilot/starpilot/flm/tests/test_state_aligned_encoder.py index 3359ca97f3..000c1fdca7 100644 --- a/openpilot/starpilot/flm/tests/test_state_aligned_encoder.py +++ b/openpilot/starpilot/flm/tests/test_state_aligned_encoder.py @@ -102,7 +102,10 @@ def test_honda_normalized_pre_lookup_comparison_and_modified_filter_shadow_is_pu assert observed_branch.civic == prior_civic # paired evaluation never advances filter observed_shapes.append(shape) return _shaped_input(row, request, shape) - candidate_shape = None if not modified else before + tuple(witness.shape[4:]) + candidate_shape = None + if modified: + assert before is not None + candidate_shape = before + tuple(witness.shape[4:]) expected_source = _shaped_input(witness, witness.requested_torque, candidate_shape) params = dict(witness.limits) expected = rate_limit(expected_source, branch.previous, diff --git a/openpilot/starpilot/tests/test_volvo_controls_startup.py b/openpilot/starpilot/tests/test_volvo_controls_startup.py index 9f14cea69d..c87833f554 100644 --- a/openpilot/starpilot/tests/test_volvo_controls_startup.py +++ b/openpilot/starpilot/tests/test_volvo_controls_startup.py @@ -23,12 +23,13 @@ def incoming(identity, *, frame=0): if identity != 'VOLVO_V40': return sources(identity, frame=frame) packer = CANPacker('volvo_v40_2017_pt') - return [packer.make_can_msg(name, bus, values) for name, bus, values in [ + messages: list[tuple[str, int, dict[str, float]]] = [ ('VehicleSpeed1', 0, {'VehicleSpeed': 72}), ('CCButtons', 0, {}), ('PSCM1', 0, {'SteeringAngleServo': 12.5, 'LKATorque': 10}), ('PedalandBrake', 0, {}), ('TCM0', 0, {'GearShifter': 3}), ('ACC', 0, {'SpeedTargetACC': 100}), ('MiscCarInfo', 0, {}), ('FSM0', 2, {'ACCStatusOnOff': 1, 'ACCStatusActive': 1}), ('FSM1', 2, {}), - ]] + ] + return [packer.make_can_msg(name, bus, values) for name, bus, values in messages] class TestVolvoActualStartup(unittest.TestCase): diff --git a/openpilot/starpilot/tests/test_volvo_selfdrived_lifecycle.py b/openpilot/starpilot/tests/test_volvo_selfdrived_lifecycle.py index 6b112500c9..5d1245f46a 100644 --- a/openpilot/starpilot/tests/test_volvo_selfdrived_lifecycle.py +++ b/openpilot/starpilot/tests/test_volvo_selfdrived_lifecycle.py @@ -184,7 +184,7 @@ def test_actual_selfdrived_fault_clear_does_not_reengage_without_pcm_edge(identi if lib.safety_config_valid() and fresh_panda: break else: - pytest.fail('Native RX and a fresh healthy Panda receipt did not recover within one source period') + pytest.fail(reason='Native RX and a fresh healthy Panda receipt did not recover within one source period') assert not sd.enabled _, command = cycle(True) assert sd.enabled and sd.active and command.latActive diff --git a/prebuilt.json b/prebuilt.json index 85f77a0210..71639a748d 100644 --- a/prebuilt.json +++ b/prebuilt.json @@ -1 +1 @@ -{"protected_inventory_sha256": "93618a5ca359682ede714b04c6e50d87c0cf63a50e5f5ab05d1d1729cc38a08d", "schema_version": 1} +{"protected_inventory_sha256": "961f82df6e41b06fefdea7ced62e8e635225d545d9a3f3419b6e96f4d410607c", "schema_version": 1}