diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index ff59b6a209..370f6f455b 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -201,12 +201,14 @@ jobs: timeout-minutes: ${{ contains(runner.name, 'nsc') && 2 || 20 }} env: RAYLIB_BACKEND: headless + STARPILOT_TEST_DIAGNOSTICS: ci-results/host-test-diagnostics/unittest run: tools/op.sh test --json-output unit-test-results.json - name: Test custom module-level pytest cases if: ${{ !cancelled() && steps.build.outcome == 'success' }} timeout-minutes: 15 env: RAYLIB_BACKEND: headless + STARPILOT_TEST_DIAGNOSTICS: ci-results/host-test-diagnostics/pytest run: python3 tools/ci/run_host_tests.py --pytest - name: Test GM vehicle configurations and CAN behavior if: ${{ !cancelled() && steps.build.outcome == 'success' }} @@ -232,13 +234,14 @@ jobs: STARPILOT_PYTHON: ${{ github.workspace }}/.venv/bin/python run: go test -mod=readonly -race -run '^Test(MsgqNativeLayout|Shadow(GoPythonIPC|ExecutableHostGpsIPC))$' -count=1 -v - name: Save complete unit test results - if: ${{ always() && (hashFiles('unit-test-results.json') != '' || hashFiles('gm-test-results.json') != '') }} + if: ${{ always() && (hashFiles('unit-test-results.json') != '' || hashFiles('gm-test-results.json') != '' || hashFiles('ci-results/host-test-diagnostics/**') != '') }} uses: actions/upload-artifact@v4 with: name: unit-test-results path: | unit-test-results.json gm-test-results.json + ci-results/host-test-diagnostics/ if-no-files-found: error aol_lifecycles: diff --git a/prebuilt.json b/prebuilt.json index b6fa56e532..36bd411986 100644 --- a/prebuilt.json +++ b/prebuilt.json @@ -1 +1 @@ -{"protected_inventory_sha256": "a8cf3e523e217d01bce30adf58c7478a7ca9a4216a754d6729da51111304172a", "schema_version": 1} +{"protected_inventory_sha256": "1d615b48ef003de1c0a8230556091c2fc264b2815f7eaac9abb8508931116e6f", "schema_version": 1} diff --git a/tools/ci/pytest_progress.py b/tools/ci/pytest_progress.py new file mode 100644 index 0000000000..1cc7a0ceef --- /dev/null +++ b/tools/ci/pytest_progress.py @@ -0,0 +1,35 @@ +"""Explicit pytest plugin retaining identities/outcomes when a host job is interrupted.""" +import json +import os +from pathlib import Path + +from tools.ci import worker_diagnostics + + +def pytest_sessionstart(session): + worker_diagnostics.initialize() + + +def pytest_collection_finish(session): + directory = os.environ.get('STARPILOT_TEST_DIAGNOSTICS') + if directory: + root = Path(directory).resolve() + root.mkdir(parents=True, exist_ok=True) + (root / 'pytest-collection.json').write_text(json.dumps([item.nodeid for item in session.items]) + '\n') + + +def pytest_runtest_logstart(nodeid, location): + worker_diagnostics.node_started(nodeid) + + +def pytest_runtest_logreport(report): + row = {'id': report.nodeid, 'phase': report.when, 'status': report.outcome, 'duration': report.duration} + if report.failed: + row['failure'] = str(report.longrepr)[:4096] + worker_diagnostics.event('pytest_report', **row) + if report.when == 'teardown': + worker_diagnostics.update(active_test=None, phase='between_tests') + + +def pytest_sessionfinish(session, exitstatus): + worker_diagnostics.update(active_test=None, phase='session_complete', exitstatus=int(exitstatus)) diff --git a/tools/ci/run_host_tests.py b/tools/ci/run_host_tests.py index 7c8528849b..8d08a069f8 100644 --- a/tools/ci/run_host_tests.py +++ b/tools/ci/run_host_tests.py @@ -50,7 +50,13 @@ def main(arguments=None): environment.pop('PYTEST_ADDOPTS', None) environment.pop('PYTEST_PLUGINS', None) environment['PYTEST_DISABLE_PLUGIN_AUTOLOAD'] = '1' - command = ([sys.executable, '-m', 'pytest', '-q', '--import-mode=importlib', *arguments[1:], *files] if pytest_mode + diagnostics = environment.get('STARPILOT_TEST_DIAGNOSTICS') + evidence = [] + if pytest_mode and diagnostics: + directory = Path(diagnostics).resolve() + directory.mkdir(parents=True, exist_ok=True) + evidence = ['-p', 'tools.ci.pytest_progress', '--junitxml=' + str(directory / 'pytest.xml')] + command = ([sys.executable, '-m', 'pytest', '-q', '--import-mode=importlib', *evidence, *arguments[1:], *files] if pytest_mode else [sys.executable, str(RUNNER), *arguments]) result = subprocess.run(command, cwd=ROOT, env=environment, check=False) return result.returncode if result.returncode >= 0 else 128 - result.returncode diff --git a/tools/ci/tests/test_host_failure_diagnostics.py b/tools/ci/tests/test_host_failure_diagnostics.py new file mode 100644 index 0000000000..c130417c0a --- /dev/null +++ b/tools/ci/tests/test_host_failure_diagnostics.py @@ -0,0 +1,127 @@ +"""Actual child death must preserve the last test and failed selected outcomes.""" +import json +import os +from pathlib import Path +import signal +import subprocess +import sys +import tempfile +import time +import unittest +from unittest.mock import patch +from types import SimpleNamespace + +from tools.ci import run_host_tests +from xml.etree import ElementTree + +ROOT = Path(__file__).resolve().parents[3] + + +def child_environment(directory, diagnostics): + return dict(os.environ, PYTHONPATH=os.pathsep.join((str(ROOT), str(directory))), + PYTEST_DISABLE_PLUGIN_AUTOLOAD='1', PYTEST_ADDOPTS='', STARPILOT_TEST_DIAGNOSTICS=str(diagnostics)) + + +class TestHostFailureDiagnostics(unittest.TestCase): + @unittest.skipUnless(os.name == 'posix', 'actual POSIX child signal evidence') + def test_aborted_pool_retains_fault_identity_exit_and_errors(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + diagnostics = root / 'diagnostics' + (root / 'signal_fixture.py').write_text( + 'import os, resource, unittest\nresource.setrlimit(resource.RLIMIT_CORE, (0, 0))\n' + + 'class TestSignal(unittest.TestCase):\n' + + ' def test_abort(self): os.abort()\n' + + ' def test_other(self): self.assertTrue(True)\n') + selected = ['signal_fixture.TestSignal.test_abort', 'signal_fixture.TestSignal.test_other'] + result = subprocess.run([sys.executable, str(ROOT / 'tools/test_runner.py'), '-j', '2', + '--json-output', str(root / 'outcomes.json'), *selected], + cwd=root, env=child_environment(root, diagnostics), capture_output=True, text=True, timeout=15) + self.assertEqual(result.returncode, 1, result.stderr) + outcomes = json.loads((root / 'outcomes.json').read_text()) + self.assertEqual(set(outcomes['selected_test_ids']), set(selected)) + records = {row['id']: row for row in outcomes['results']} + self.assertEqual(records[selected[0]]['status'], 'error') + self.assertEqual(len(records), 2) + witness = json.loads((diagnostics / 'pool-failure.json').read_text()) + self.assertTrue(any(row['exitcode'] == -signal.SIGABRT for row in witness['processes'])) + self.assertTrue(any(row['active_test'] == selected[0] and row['batch'] == [selected[0]] for row in witness['workers'])) + self.assertIn('BrokenProcessPool', witness['detail']) + self.assertLess(len(records[selected[0]]['detail']), 1024) + fault = '\n'.join(path.read_text() for path in diagnostics.glob('*.fault.log')) + self.assertIn('Aborted', fault) + self.assertIn('signal_fixture.py', fault) + + @unittest.skipUnless(os.name == 'posix', 'actual POSIX interruption evidence') + def test_pytest_interruption_retains_collection_reports_and_active_node(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + diagnostics = root / 'diagnostics' + fixture = root / 'test_interrupted_fixture.py' + fixture.write_text('import time\ndef test_a_pass(): assert True\ndef test_b_fail(): assert False\n' + + 'def test_c_pending(): time.sleep(60)\n') + process = subprocess.Popen([sys.executable, '-m', 'pytest', '-q', '-p', 'tools.ci.pytest_progress', str(fixture)], + cwd=root, env=child_environment(root, diagnostics), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + try: + deadline = time.monotonic() + 10 + states = [] + while time.monotonic() < deadline: + states = [json.loads(path.read_text()) for path in diagnostics.glob('worker-*.json')] + if any(str(row.get('active_test', '')).endswith('test_c_pending') for row in states): + break + if process.poll() is not None: + self.fail('pytest exited before the interruption fixture') + time.sleep(0.02) + self.assertTrue(any(str(row.get('active_test', '')).endswith('test_c_pending') for row in states)) + process.terminate() + self.assertEqual(process.wait(timeout=5), -signal.SIGTERM) + selected = json.loads((diagnostics / 'pytest-collection.json').read_text()) + self.assertEqual(len(selected), 3) + reports = [json.loads(line) for path in diagnostics.glob('worker-*.jsonl') for line in path.read_text().splitlines()] + call = {row['id']: row for row in reports if row['event'] == 'pytest_report' and row['phase'] == 'call'} + self.assertEqual(call[selected[0]]['status'], 'passed') + self.assertEqual(call[selected[1]]['status'], 'failed') + self.assertIn('assert False', call[selected[1]]['failure']) + self.assertNotIn(selected[2], call) + finally: + if process.poll() is None: + process.kill() + process.wait(timeout=5) + + def test_completed_pytest_keeps_junit_and_failed_exit_status(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + diagnostics = root / 'diagnostics' + diagnostics.mkdir() + fixture = root / 'test_outcome_fixture.py' + fixture.write_text('def test_pass(): assert True\ndef test_fail(): assert False\n') + result = subprocess.run([sys.executable, '-m', 'pytest', '-q', '-p', 'tools.ci.pytest_progress', + '--junitxml=' + str(diagnostics / 'pytest.xml'), str(fixture)], + cwd=root, env=child_environment(root, diagnostics), capture_output=True, text=True, timeout=15) + self.assertEqual(result.returncode, 1, result.stderr) + report = ElementTree.parse(diagnostics / 'pytest.xml') + self.assertEqual(len(report.findall('.//testcase')), 2) + self.assertEqual(len(report.findall('.//failure')), 1) + states = [json.loads(path.read_text()) for path in diagnostics.glob('worker-*.json')] + self.assertEqual(states[0]['phase'], 'session_complete') + self.assertEqual(states[0]['exitstatus'], 1) + + def test_host_wrapper_preserves_catalog_and_failure_with_diagnostics(self): + with tempfile.TemporaryDirectory() as directory: + diagnostics = Path(directory) / 'diagnostics' + selected = ['first/test_owner.py', 'second/test_owner.py'] + seen = [] + def invoke(command, *, cwd, env, check): + seen.append((command, env)) + self.assertTrue(Path(env['PARAMS_ROOT']).is_dir()) + self.assertEqual(env['PYTEST_DISABLE_PLUGIN_AUTOLOAD'], '1') + return SimpleNamespace(returncode=7) + with patch.dict(os.environ, {'STARPILOT_TEST_DIAGNOSTICS': str(diagnostics)}), \ + patch.object(run_host_tests, 'pytest_files', return_value=selected), \ + patch.object(run_host_tests.subprocess, 'run', side_effect=invoke): + self.assertEqual(run_host_tests.main(['--pytest']), 7) + command, env = seen[0] + self.assertEqual(command[-2:], selected) + self.assertIn('tools.ci.pytest_progress', command) + self.assertIn('--junitxml=' + str(diagnostics.resolve() / 'pytest.xml'), command) + self.assertFalse(Path(env['PARAMS_ROOT']).exists()) diff --git a/tools/ci/worker_diagnostics.py b/tools/ci/worker_diagnostics.py new file mode 100644 index 0000000000..5314ed9906 --- /dev/null +++ b/tools/ci/worker_diagnostics.py @@ -0,0 +1,80 @@ +"""Durable test identities and native fault stacks; never changes test outcomes.""" +import faulthandler +import hashlib +import json +import os +from pathlib import Path +import time +import warnings + +_ROOT = None +_PID = None +_FAULT = None +_STATE = {} + + +def initialize(warning_action=None): + global _ROOT, _PID, _FAULT, _STATE + if warning_action is not None: + warnings.simplefilter(warning_action) + directory = os.environ.get('STARPILOT_TEST_DIAGNOSTICS') + if not directory or _PID == os.getpid(): + return + _ROOT, _PID = Path(directory).resolve(), os.getpid() + _ROOT.mkdir(parents=True, exist_ok=True) + _FAULT = (_ROOT / f'worker-{_PID}.fault.log').open('ab', buffering=0) + # faulthandler bounds native dumps to 100 threads and 100 frames per thread. + # Its separate fd remains open when per-test stdout/stderr capture is lost. + faulthandler.enable(file=_FAULT, all_threads=True) + _STATE = {'pid': _PID, 'batch': [], 'active_test': None, 'phase': 'initialized'} + update() + + +def update(**fields): + if _ROOT is None or _PID != os.getpid(): + return + _STATE.update(fields, monotonic_ns=time.monotonic_ns()) + path = _ROOT / f'worker-{_PID}.json' + temporary = path.with_suffix('.tmp') + temporary.write_text(json.dumps(_STATE, sort_keys=True) + '\n') + os.replace(temporary, path) + + +def event(kind, **fields): + if _ROOT is None or _PID != os.getpid(): + return + # Only identities/status/timestamps, never captured output or exception chains. + row = {'event': kind, 'pid': _PID, 'monotonic_ns': time.monotonic_ns(), **fields} + with (_ROOT / f'worker-{_PID}.jsonl').open('a') as stream: + stream.write(json.dumps(row, sort_keys=True) + '\n') + + +def batch_started(identities): + initialize() + update(batch=list(identities), active_test=None, phase='batch') + + +def node_started(identity): + update(active_test=identity, phase='test') + event('start', id=identity) + + +def node_finished(identity, status): + event('finish', id=identity, status=status) + update(active_test=None, phase='between_tests') + + +def pool_failure(processes, detail): + directory = os.environ.get('STARPILOT_TEST_DIAGNOSTICS') + if not directory: + return None + root = Path(directory).resolve() + root.mkdir(parents=True, exist_ok=True) + path = root / 'pool-failure.json' + if not path.exists(): + rows = [{'pid': process.pid, 'exitcode': process.exitcode} for process in processes] + value = {'detail': detail, 'detail_sha256': hashlib.sha256(detail.encode()).hexdigest(), + 'processes': rows, 'workers': [json.loads(item.read_text()) for item in sorted(root.glob('worker-*.json'))], + 'scope': 'Observed worker exits and last durable test identities; no inferred OOM or culprit for alive workers'} + path.write_text(json.dumps(value, sort_keys=True, indent=2) + '\n') + return path diff --git a/tools/test_runner.py b/tools/test_runner.py index 419ad5e537..90fe98bb34 100755 --- a/tools/test_runner.py +++ b/tools/test_runner.py @@ -13,6 +13,9 @@ import unittest import warnings ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) +from tools.ci import worker_diagnostics + IGNORED = ( ROOT / "openpilot/selfdrive/test/process_replay/test_processes.py", ROOT / "openpilot/tools/sim", @@ -79,18 +82,22 @@ class Result(unittest.TestResult): self.capture = Capture(capture_output) def startTest(self, test): + worker_diagnostics.node_started(test.id()) self.current = make_record(test.id()) self.started = time.monotonic() self.capture.start() def stopTest(self, test): - self.current["time"] = time.monotonic() - self.started - keep = self.current["status"] in FAILURES + current = self.current + assert current is not None + current["time"] = time.monotonic() - self.started + keep = current["status"] in FAILURES stdout, stderr = self.capture.stop(keep) if keep: - self.current["stdout"] = stdout - self.current["stderr"] = stderr - self.records.append(self.current) + current["stdout"] = stdout + current["stderr"] = stderr + self.records.append(current) + worker_diagnostics.node_finished(current["id"], current["status"]) self.current = None def mark(self, test, status, detail=""): @@ -245,6 +252,7 @@ def account_for_tests(test_ids, records, tests=()): def run_batch(test_ids, capture_output): + worker_diagnostics.batch_started(test_ids) result = Result(capture_output) outside = Capture(capture_output) tests = [] @@ -267,17 +275,22 @@ def run_batch(test_ids, capture_output): if failures: # attach class/module fixture output to the first related failure failures[0]["stdout"] = stdout + failures[0]["stdout"] failures[0]["stderr"] = stderr + failures[0]["stderr"] + worker_diagnostics.update(active_test=None, phase="batch_complete") return result.records def run_parallel(batches, workers, warning_action, capture_output): - with ProcessPoolExecutor(max_workers=workers, initializer=warnings.simplefilter, initargs=(warning_action,)) as pool: + with ProcessPoolExecutor(max_workers=workers, initializer=worker_diagnostics.initialize, initargs=(warning_action,)) as pool: futures = {pool.submit(run_batch, batch, capture_output): batch for batch in batches} + processes = tuple(pool._processes.values()) for future in as_completed(futures): try: yield future.result() except Exception: - detail = traceback.format_exc() + detail = traceback.format_exc(chain=False) + witness = worker_diagnostics.pool_failure(processes, detail) + if witness is not None: + detail = f"Worker batch failed; original exception and observed exits/current tests: {witness}" yield [make_record(test_id, "error", detail) for test_id in futures[future]]