Include native pytest cases in safety coverage

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "247b1614504f5dc59522b8de5b95664b43c475cf935146fd8db5b618b8bded67", "schema_version": 1}
+{"protected_inventory_sha256": "23934bc7662bfb078308883c1fc94e325eb4230317e13c38306f0f6c00507957", "schema_version": 1}

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",

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())

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))