import json import ast import os from pathlib import Path import tempfile import threading import unittest from types import SimpleNamespace from unittest.mock import patch from contextlib import nullcontext from app.start_diagnostics import StartDiagnostics, setup_failure_summary from app.stream_lifecycle import RoleRunHandle class StartDiagnosticsTest(unittest.TestCase): def test_setup_failure_summary_keeps_location_not_private_payload(self): try: try: raise OSError(12, "rtsp://private:secret@host", "/private/video.mp4") except OSError as cause: raise RuntimeError("private input") from cause except RuntimeError as error: summary = setup_failure_summary(error) self.assertEqual([x['type'] for x in summary], ['RuntimeError', 'OSError']) self.assertEqual(summary[1]['errno'], 12) self.assertEqual(summary[0]['frames'][-1]['file'], 'test_start_diagnostics.py') self.assertGreater(summary[0]['frames'][-1]['line'], 0) text = json.dumps(summary) self.assertNotIn('rtsp://', text) self.assertNotIn('/private/', text) self.assertNotIn('secret', text) self.assertNotIn('private input', text) def test_setup_failure_summary_is_bounded_and_has_no_exception_references(self): def fail(depth): if depth: return fail(depth-1) raise ValueError('not serialized') try: fail(20) except ValueError as error: error.__cause__ = error summary = setup_failure_summary(error) self.assertEqual(len(summary), 1) self.assertEqual(len(summary[0]['frames']), 8) json.dumps(summary) def test_bootstrap_reports_stage_and_still_signals_public_failure(self): from app.stream_lifecycle import make_stream_failure tree = ast.parse((Path(__file__).parents[1] / 'app/stream_manager.py').read_text()) group = next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == 'StreamGroup') method = next(n for n in group.body if isinstance(n, ast.FunctionDef) and n.name == '_begin') logs, failures, closed = [], [], [] def resolve(*args): raise RuntimeError('rtsp://private:secret@host') scope = dict(threading=threading, json=json, setup_failure_summary=setup_failure_summary, StartDiagnostics=lambda role: SimpleNamespace(span=lambda stage: nullcontext(), close=lambda: closed.append(True)), resolve_source=lambda spec: dict(uri='unused', source_type='file'), resolve_from_spec=resolve, SourceError=type('SourceError', (Exception,), {}), ProfileError=type('ProfileError', (Exception,), {}), make_stream_failure=make_stream_failure, _log=logs.append) exec(compile(ast.Module(body=[method], type_ignores=[]), 'stream_manager.py', 'exec'), scope) fake = SimpleNamespace(spec={}, stream_id='test', FIRST_FRAME_TIMEOUT_S=30, _generation_active=lambda *args: True, _fail_run=lambda *args: failures.append(args[-1].as_dict())) scope['_begin'](fake, 1, threading.Event()) self.assertEqual(json.loads(logs[0].split(' ', 1)[1])['setup_stage'], 'profile') self.assertEqual(failures[0]['error_code'], 'CONFIG_INVALID') self.assertEqual(failures[0]['failure_stage'], 'stream_setup') self.assertEqual(closed, [True]) self.assertNotIn('secret', str(logs) + str(failures)) def setUp(self): self.directory = tempfile.TemporaryDirectory() self.root = Path(self.directory.name) self.env = patch.dict(os.environ, {"CONTEXTRA_START_DIAGNOSTICS_DIR": str(self.root)}) self.env.start() def tearDown(self): self.env.stop() self.directory.cleanup() def records(self): return [json.loads(line) for p in self.root.glob("*.jsonl") for line in p.read_text().splitlines()] def test_disabled_and_unknown_role(self): self.assertIsNone(StartDiagnostics("private-camera").fd) with patch.dict(os.environ, {"CONTEXTRA_START_DIAGNOSTICS_DIR": ""}): self.assertIsNone(StartDiagnostics("det").fd) self.assertFalse(list(self.root.iterdir())) def test_allowlist_dedup_and_no_exception_text(self): trace = StartDiagnostics("det") with self.assertRaisesRegex(ValueError, "private"): with trace.span("sdk_start"): raise ValueError("private input") for _ in range(100): trace.mark("sdk_start", "enter") trace.mark("private input", "enter") trace.close() records = self.records() self.assertEqual(["enter", "raised"], [x["phase"] for x in records]) self.assertNotIn("private", str(records)) self.assertTrue(all(p.stat().st_mode & 0o777 == 0o600 for p in self.root.iterdir())) def test_open_failure_does_not_change_call(self): os.chmod(self.root, 0o755) trace = StartDiagnostics("det") self.assertEqual("DIAGNOSTICS_OPEN_FAILED", trace.failure) with trace.span("pipeline"): value = 7 self.assertEqual(7, value) def test_flow_substage_exception_is_bounded(self): trace = StartDiagnostics('det') with trace.span('batch_capture'): pass with self.assertRaises(ValueError): with trace.span('rtp_attach'): raise ValueError('private source configuration') trace.close() records = self.records() self.assertEqual([(r['stage'], r['phase']) for r in records], [('batch_capture', 'enter'), ('batch_capture', 'returned'), ('rtp_attach', 'enter'), ('rtp_attach', 'raised')]) self.assertNotIn('private', str(records)) def test_write_failure_closes_fd(self): trace = StartDiagnostics("det") with patch("app.start_diagnostics.os.write", side_effect=OSError("private")): trace.mark("pipeline", "enter") self.assertIsNone(trace.fd) self.assertEqual("DIAGNOSTICS_WRITE_FAILED", trace.failure) def test_stack_arm_failure_is_recorded(self): with patch("app.start_diagnostics._stack_file", None), patch( "app.start_diagnostics.faulthandler.dump_traceback_later", side_effect=RuntimeError("private") ): trace = StartDiagnostics("bootstrap") trace.mark("source", "enter") trace.close() self.assertEqual("STACK_ARM_FAILED", self.records()[0]["diagnostic_error"]) self.assertNotIn("private", str(self.records())) def test_stack_armed_once_per_process(self): import app.start_diagnostics as module with patch.object(module, "_stack_file", None), patch.object( module.faulthandler, "dump_traceback_later" ) as arm: first, second = StartDiagnostics("bootstrap"), StartDiagnostics("bootstrap") arm.assert_called_once() self.assertEqual((10,), arm.call_args.args) self.assertFalse(arm.call_args.kwargs["repeat"]) first.close() second.close() module._stack_file.close() def test_symlink_directory_rejected(self): link = self.root / "link" link.symlink_to(self.root, target_is_directory=True) with patch.dict(os.environ, {"CONTEXTRA_START_DIAGNOSTICS_DIR": str(link)}): trace = StartDiagnostics("det") self.assertEqual("DIAGNOSTICS_OPEN_FAILED", trace.failure) def test_blocking_start_observation_and_late_stop(self): trace = StartDiagnostics("det") entered, release = threading.Event(), threading.Event() handle = RoleRunHandle("det", 1) calls = [] def start(): with trace.span("sdk_start"): entered.set() release.wait(2) return None handle.bind_pipeline(SimpleNamespace(start=start, stop=lambda: calls.append("stop"))) a = threading.Thread(target=handle.start_pipeline) b = threading.Thread(target=handle.stop) try: a.start() self.assertTrue(entered.wait(1)) self.assertEqual(["enter"], [x["phase"] for x in self.records()]) b.start() # local state reads remain available while the fake SDK start is waiting. self.assertFalse(handle.initialized) finally: release.set() a.join(2) if b.ident is not None: b.join(2) trace.close() self.assertFalse(a.is_alive() or b.is_alive()) self.assertEqual(["stop"], calls) self.assertEqual(["enter", "returned"], [x["phase"] for x in self.records()])