From a0fd5ea0256fddc7fd69a2436b9d05dc9e4c701f Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Thu, 22 Oct 2026 20:47:05 -0500 Subject: [PATCH] Include native pytest cases in safety coverage --- prebuilt.json | 2 +- tools/ci/run_safety_qualification.py | 15 ++- tools/ci/safety_pytest_coverage.py | 100 ++++++++++++++++++++ tools/ci/tests/test_safety_qualification.py | 86 +++++++++++++++++ 4 files changed, 201 insertions(+), 2 deletions(-) create mode 100644 tools/ci/safety_pytest_coverage.py diff --git a/prebuilt.json b/prebuilt.json index 1bc05f92ba..ecee5d084d 100644 --- a/prebuilt.json +++ b/prebuilt.json @@ -1 +1 @@ -{"protected_inventory_sha256": "247b1614504f5dc59522b8de5b95664b43c475cf935146fd8db5b618b8bded67", "schema_version": 1} +{"protected_inventory_sha256": "23934bc7662bfb078308883c1fc94e325eb4230317e13c38306f0f6c00507957", "schema_version": 1} diff --git a/tools/ci/run_safety_qualification.py b/tools/ci/run_safety_qualification.py index 1639b7f01f..1e114e0621 100644 --- a/tools/ci/run_safety_qualification.py +++ b/tools/ci/run_safety_qualification.py @@ -468,6 +468,19 @@ suite.addTests(unittest.TestLoader().loadTestsFromName('opendbc.bluepilot_latera code, text = self.run("unittest", [self.python, "-c", suite_script, str(native)], cwd=SAFETY_TESTS) summary = parse_unittest_summary(text) self.summary = summary + pytest_results = self.output / "pytest-coverage-results.json" + pytest_code, _ = self.run("pytest-coverage", [self.python, str(ROOT / "tools/ci/safety_pytest_coverage.py"), + "--native", str(native), "--results-json", str(pytest_results)], cwd=ROOT) + try: + from tools.ci.safety_pytest_coverage import report_complete, targets_for_coverage + pytest_report = json.loads(pytest_results.read_text()) + pytest_ok = report_complete(pytest_report, targets_for_coverage(), hashlib.sha256(native.read_bytes()).hexdigest()) + except (OSError, ValueError, TypeError, KeyError): + pytest_report, pytest_ok = {}, False + self.summary["pytest_coverage"] = {"complete": pytest_ok, "collected": len(pytest_report.get("nodeids", [])), + "ran": len(pytest_report.get("records", []))} + if pytest_results.is_file(): + self.hash_artifact(pytest_results) if sys.platform == "darwin": found, llvm_cov = self.run("llvm-cov-path", ["xcrun", "--find", "llvm-cov"]) if found: @@ -480,7 +493,7 @@ suite.addTests(unittest.TestLoader().loadTestsFromName('opendbc.bluepilot_latera for source_root in SAFETY_SOURCE_ROOTS: coverage_argv += ["--filter", re.escape(source_root.resolve().as_posix() + "/")] gcovr, _ = self.run("coverage", coverage_argv, cwd=SAFETY_TESTS) - return code == 0 and complete_unittest(summary) and gcovr == 0 + return code == 0 and complete_unittest(summary) and pytest_code == 0 and pytest_ok and gcovr == 0 if self.name in ("opendbc-misra", "panda-misra"): if self.name == "panda-misra": built, _ = self.run("build-h7", ["scons", "-j4", "panda/board/obj/panda_h7/main.elf", diff --git a/tools/ci/safety_pytest_coverage.py b/tools/ci/safety_pytest_coverage.py new file mode 100644 index 0000000000..f50d578852 --- /dev/null +++ b/tools/ci/safety_pytest_coverage.py @@ -0,0 +1,100 @@ +"""Run omitted native pytest cases serially against the gate's instrumented library.""" +import argparse +import ast +import hashlib +import json +from pathlib import Path +import sys +from unittest import mock + +from tools.ci import vehicle_pytest + +ROOT = Path(__file__).resolve().parents[2] +# These denial classes belong to the existing mandatory RELEASE vehicle plan. +RELEASE_ONLY = frozenset(( + 'opendbc_repo/opendbc/safety/tests/test_hyundai_main_rearm.py::TestHkgMainRelease', + 'opendbc_repo/opendbc/safety/tests/test_hyundai_classic_cancel_joined.py::TestClassicCancelReleaseDenied', +)) + + +def targets_for_coverage(root=ROOT): + native_root = root / 'opendbc_repo/opendbc/safety/tests' + selected = set() + # unittest already owns its TestCase subclasses, including decorated classes. + for path in sorted(native_root.glob('test_*.py')): + relative = path.relative_to(root).as_posix() + for node in ast.parse(path.read_text(), filename=str(path)).body: + function = isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name.startswith('test_') + plain_class = isinstance(node, ast.ClassDef) and node.name.startswith('Test') and not node.bases + if function or plain_class: + selected.add(f'{relative}::{node.name}') + # These existing end-to-end callers bind the supplied library. Other vehicle suites + # deliberately compile distinct DEBUG/RELEASE artifacts and retain their own gates. + for relative in ('openpilot/starpilot/car/nissan/tests/test_aol.py', + 'openpilot/starpilot/car/rivian/tests/test_aol_native.py'): + if (root / relative).is_file(): + selected.add(relative) + return sorted(selected - RELEASE_ONLY) + + +def report_complete(report, targets, native_sha256): + if (not isinstance(report, dict) or type(report.get('schema_version')) is not int or + report.get('schema_version') != 1 or report.get('full_catalog') is not True): + return False + ids, records = report.get('nodeids'), report.get('records') + if (report.get('targets') != targets or not targets or report.get('native_sha256') != native_sha256 or + report.get('native_bound_before') is not True or report.get('native_bound_after') is not True or + report.get('collection_errors') != [] or report.get('execution_errors') != [] or + not isinstance(ids, list) or not ids or not all(type(nodeid) is str for nodeid in ids) or len(set(ids)) != len(ids) or not isinstance(records, list)): + return False + if not all(isinstance(row, dict) and row.get('status') in ('passed', 'skipped') for row in records): + return False + actual = [row.get('id') for row in records] + return (all(type(identity) is str for identity in actual) and len(actual) == len(ids) and len(set(actual)) == len(actual) and + set(actual) == {vehicle_pytest.test_id(nodeid) for nodeid in ids} and any(row['status'] == 'passed' for row in records)) + + +def run(native, output, targets=None): + from opendbc.safety.tests.libsafety import libsafety_py + native = Path(native).resolve() + selected = targets_for_coverage() + full_catalog = targets is None + targets = selected if targets is None else list(targets) + if not targets or any(not any(target == item or target.startswith(item + '::') for item in selected) for target in targets): + raise ValueError('Coverage subset must name existing native catalog cases') + digest = hashlib.sha256(native.read_bytes()).hexdigest() + libsafety_py.load(native) + original_load = libsafety_py.load + + def load_pinned(path): + if Path(path).resolve() != native: + raise RuntimeError('Native coverage cases must retain the same instrumented libsafety') + return original_load(native) + + report = {'schema_version': 1, 'full_catalog': full_catalog, 'targets': targets, 'native_sha256': digest, + 'native_path': str(native), 'native_bound_before': Path(libsafety_py.libpath).resolve() == native, + 'nodeids': [], 'records': [], 'collection_errors': [], 'execution_errors': []} + with mock.patch.object(libsafety_py, 'load', side_effect=load_pinned): + nodeids, errors = vehicle_pytest.collect(targets) + report.update(nodeids=nodeids, collection_errors=errors) + if not errors: + records, errors = vehicle_pytest.run(nodeids) + report.update(records=records, execution_errors=errors) + report['native_bound_after'] = Path(libsafety_py.libpath).resolve() == native and hashlib.sha256(native.read_bytes()).hexdigest() == digest + Path(output).write_text(json.dumps(report, indent=2) + '\n') + # The root's bounded native subset is a diagnostic; production always requires the complete catalog. + checked = dict(report, full_catalog=True) + return report_complete(checked, targets, digest) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--native', required=True) + parser.add_argument('--results-json', required=True) + parser.add_argument('--target', action='append') + args = parser.parse_args() + return 0 if run(args.native, args.results_json, args.target) else 1 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/tools/ci/tests/test_safety_qualification.py b/tools/ci/tests/test_safety_qualification.py index 13064f6999..05ad81cba3 100644 --- a/tools/ci/tests/test_safety_qualification.py +++ b/tools/ci/tests/test_safety_qualification.py @@ -92,6 +92,92 @@ class TestSafetyQualificationRunner(TestCase): with self.assertRaises(FileNotFoundError): qualification.imported_module_hashes(log) + def test_coverage_catalog_includes_native_pytest_and_external_callers(self): + from tools.ci import safety_pytest_coverage as coverage + selected = coverage.targets_for_coverage(qualification.ROOT) + for name in ('opendbc_repo/opendbc/safety/tests/test_tesla_preap_pedal.py::test_actual_registered_priming_rate_absolute_bus_and_counter_bounds', + 'openpilot/starpilot/car/nissan/tests/test_aol.py', + 'openpilot/starpilot/car/rivian/tests/test_aol_native.py'): + self.assertIn(name, selected) + import ast + plan = ast.parse((qualification.ROOT / 'tools/ci/run_vehicle_tests.py').read_text()) + release = next(ast.literal_eval(node.value) for node in plan.body if isinstance(node, ast.Assign) and + any(isinstance(target, ast.Name) and target.id == 'RELEASE_TARGETS' for target in node.targets)) + self.assertEqual(len(coverage.RELEASE_ONLY), 2) + for nodeid in coverage.RELEASE_ONLY: + self.assertIn(nodeid, release) + self.assertNotIn(nodeid, selected) + with tempfile.TemporaryDirectory() as td: + root = Path(td) + native = root / 'opendbc_repo/opendbc/safety/tests' + external = root / 'openpilot/starpilot/car/nissan/tests' + native.mkdir(parents=True) + external.mkdir(parents=True) + (native / 'test_function.py').write_text('def test_native(): pass\n') + (native / 'test_ordinary.py').write_text('class TestOrdinary(unittest.TestCase): pass\n') + (external / 'test_aol.py').write_text('from opendbc.safety.tests.libsafety import libsafety_py\nclass TestCaller(unittest.TestCase): pass\n') + (external / 'test_unrelated.py').write_text('def test_settings(): pass\n') + self.assertEqual(coverage.targets_for_coverage(root), + ['opendbc_repo/opendbc/safety/tests/test_function.py::test_native', + 'openpilot/starpilot/car/nissan/tests/test_aol.py']) + + def test_native_pytest_coverage_requires_complete_accounting_and_same_library(self): + from tools.ci import safety_pytest_coverage as coverage + targets = ['native.py'] + report = {'schema_version': 1, 'full_catalog': True, 'targets': targets, 'native_sha256': 'pinned', + 'native_bound_before': True, 'native_bound_after': True, 'collection_errors': [], 'execution_errors': [], + 'nodeids': ['native.py::test_one'], 'records': [{'id': 'native.test_one', 'status': 'passed'}]} + self.assertTrue(coverage.report_complete(report, targets, 'pinned')) + changes = ({'schema_version': True}, {'full_catalog': False}, {'targets': []}, {'native_sha256': 'other'}, + {'native_bound_before': False}, {'native_bound_after': False}, {'collection_errors': ['bad']}, + {'execution_errors': ['bad']}, {'nodeids': []}, {'nodeids': ['native.py::test_one'] * 2}, + {'records': []}, {'records': [{'id': 'other', 'status': 'passed'}]}, + {'records': [{'id': 'native.test_one', 'status': 'skipped'}]}, + {'records': [{'id': 'native.test_one', 'status': 'xfailed'}]}, + {'records': [{'id': 'native.test_one', 'status': 'failed'}]}) + for changed in changes: + with self.subTest(changed=changed): + self.assertFalse(coverage.report_complete(dict(report, **changed), targets, 'pinned')) + + def test_coverage_runs_pytest_before_unchanged_full_coverage_requirement(self): + from tools.ci import safety_pytest_coverage as coverage + for pytest_code, report_present, expected in ((0, True, True), (1, True, False), (0, False, False)): + with self.subTest(code=pytest_code, report=report_present), tempfile.TemporaryDirectory() as td: + directory = Path(td) + original = directory / 'built.so' + original.write_bytes(b'instrumented fixture') + gate = qualification.Gate('coverage', directory, sys.executable) + targets = ['native.py'] + labels = [] + def execute(label, argv, labels=labels, original=original, directory=directory, + report_present=report_present, targets=targets, pytest_code=pytest_code, **kwargs): + labels.append(label) + if label == 'build-libsafety': + return 0, str(original) + '\n' + if label == 'unittest': + return 0, 'Safety qualification collected 1 tests\nRan 1 test in 0.01s\n\nOK\nSafety qualification suppressed 0 methods\n' + if label == 'pytest-coverage': + self.assertEqual(Path(argv[1]), qualification.ROOT / 'tools/ci/safety_pytest_coverage.py') + self.assertIn(str(directory / 'libsafety-debug.so'), argv) + if report_present: + (directory / 'pytest-coverage-results.json').write_text(json.dumps( + {'schema_version': 1, 'full_catalog': True, 'targets': targets, + 'native_sha256': qualification.hashlib.sha256(original.read_bytes()).hexdigest(), + 'native_bound_before': True, 'native_bound_after': True, 'collection_errors': [], 'execution_errors': [], + 'nodeids': ['native.py::test_one'], 'records': [{'id': 'native.test_one', 'status': 'passed'}]})) + return pytest_code, '' + if label == 'coverage': + self.assertIn('--fail-under-line=100', argv) + return 0, '' + with patch.object(gate, 'run', side_effect=execute), patch.object(coverage, 'targets_for_coverage', return_value=targets): + self.assertEqual(gate.execute(), expected) + self.assertLess(labels.index('unittest'), labels.index('pytest-coverage')) + self.assertLess(labels.index('pytest-coverage'), labels.index('coverage')) + self.assertEqual(gate.summary['ran'], 1) + pytest_summary = gate.summary['pytest_coverage'] + assert isinstance(pytest_summary, dict) + self.assertEqual(pytest_summary['complete'], report_present) + def test_source_identity_includes_gate_and_workflow_inputs(self): identity = qualification.source_identity() self.assertFalse(any("/obj/" in path or "/gen/" in path for path in identity))