#!/usr/bin/env python3 """Unit tests for `build_contention_manifest.py`. Run as `python3 test_build_manifest.py` — prints `PASS` per test. """ import json import sys import tempfile from pathlib import Path sys.path.insert(0, str(Path(__file__).parent)) from build_contention_manifest import build_manifest, render_summary # noqa: E402 def write_jsonl(lines: list[str]) -> Path: tmp = tempfile.NamedTemporaryFile( mode="w", suffix=".jsonl", delete=False, encoding="utf-8" ) for line in lines: tmp.write(line + "\n") tmp.close() return Path(tmp.name) def mk_event( kind: str, tid: int, idx: int, payload: dict, engine: str = "canary", ) -> str: return json.dumps( { "schema_version": 1, "engine": engine, "kind": kind, "tid": tid, "tid_event_idx": idx, "guest_cycle": 0, "host_ns": 0, "deterministic": True, "payload": payload, } ) def test_basic_extract() -> None: src = write_jsonl([ mk_event("import.call", 6, 0, {"name": "Foo"}), mk_event( "contention.observed", 6, 104664, {"cs_ptr": "0xbc65c890", "site_sid": "c26a128b", "contended": True}, ), mk_event("import.call", 6, 1, {"name": "Bar"}), ]) m = build_manifest(src) assert m["version"] == 1 assert m["summary"]["total_input_events"] == 3 assert m["summary"]["total_contention_events_kept"] == 1 assert m["summary"]["per_tid_counts"] == {"6": 1} e = m["entries"][0] assert e["tid"] == 6 and e["tid_event_idx"] == 104664 assert e["site_sid"] == "c26a128b" and e["cs_ptr"] == "0xbc65c890" assert e["contended"] is True print("PASS test_basic_extract") def test_filters_non_contention_kinds() -> None: src = write_jsonl([ mk_event("handle.create", 6, 0, {"handle_semantic_id": "x"}), mk_event("wait.begin", 6, 1, {"handles_semantic_ids": ["x"]}), mk_event("kernel.call", 6, 2, {"name": "X"}), mk_event( "contention.observed", 7, 42, {"cs_ptr": "0x1000", "site_sid": "deadbeef", "contended": True}, ), ]) m = build_manifest(src) assert m["summary"]["total_contention_events_kept"] == 1 assert m["entries"][0]["tid"] == 7 print("PASS test_filters_non_contention_kinds") def test_filters_contended_false() -> None: # Stage 1's emitter never emits contended=false today, but defensive # filter must skip those if a future variant adds them. src = write_jsonl([ mk_event( "contention.observed", 6, 10, {"cs_ptr": "0xa", "site_sid": "11", "contended": False}, ), mk_event( "contention.observed", 6, 11, {"cs_ptr": "0xa", "site_sid": "11", "contended": True}, ), ]) m = build_manifest(src) assert m["summary"]["total_contention_events_kept"] == 1 assert m["entries"][0]["tid_event_idx"] == 11 print("PASS test_filters_contended_false") def test_sorts_by_tid_then_idx() -> None: src = write_jsonl([ mk_event( "contention.observed", 9, 5, {"cs_ptr": "0x9", "site_sid": "99", "contended": True}, ), mk_event( "contention.observed", 6, 200, {"cs_ptr": "0xb", "site_sid": "bb", "contended": True}, ), mk_event( "contention.observed", 6, 100, {"cs_ptr": "0xa", "site_sid": "aa", "contended": True}, ), ]) m = build_manifest(src) keys = [(e["tid"], e["tid_event_idx"]) for e in m["entries"]] assert keys == [(6, 100), (6, 200), (9, 5)], keys print("PASS test_sorts_by_tid_then_idx") def test_deduplicates_same_tid_idx() -> None: src = write_jsonl([ mk_event( "contention.observed", 6, 42, {"cs_ptr": "0xa", "site_sid": "aa", "contended": True}, ), mk_event( "contention.observed", 6, 42, {"cs_ptr": "0xb", "site_sid": "bb", "contended": True}, ), ]) m = build_manifest(src) assert m["summary"]["total_contention_events_kept"] == 1 assert m["summary"]["skipped_duplicate_keys"] == 1 # Keeps the first occurrence. assert m["entries"][0]["cs_ptr"] == "0xa" print("PASS test_deduplicates_same_tid_idx") def test_skips_missing_fields() -> None: src = write_jsonl([ # Missing site_sid. mk_event( "contention.observed", 6, 1, {"cs_ptr": "0xa", "contended": True}, ), # Missing cs_ptr. mk_event( "contention.observed", 6, 2, {"site_sid": "aa", "contended": True}, ), # Both present — kept. mk_event( "contention.observed", 6, 3, {"cs_ptr": "0xb", "site_sid": "bb", "contended": True}, ), ]) m = build_manifest(src) assert m["summary"]["total_contention_events_kept"] == 1 assert m["summary"]["skipped_bad_lines"] == 2 print("PASS test_skips_missing_fields") def test_handles_bad_json_lines() -> None: src = write_jsonl([ "not-json", mk_event( "contention.observed", 6, 1, {"cs_ptr": "0xa", "site_sid": "aa", "contended": True}, ), "{\"truncated\":", ]) m = build_manifest(src) assert m["summary"]["total_contention_events_kept"] == 1 assert m["summary"]["skipped_bad_lines"] == 2 print("PASS test_handles_bad_json_lines") def test_render_summary_human_readable() -> None: src = write_jsonl([ mk_event( "contention.observed", 6, 1, {"cs_ptr": "0xa", "site_sid": "aa", "contended": True}, ), mk_event( "contention.observed", 14, 100, {"cs_ptr": "0xb", "site_sid": "bb", "contended": True}, ), ]) m = build_manifest(src) out = render_summary(m) assert "contention events kept: 2" in out assert "tid= 6 1" in out assert "tid= 14 1" in out print("PASS test_render_summary_human_readable") def test_empty_input_yields_zero_kept() -> None: src = write_jsonl([mk_event("import.call", 0, 0, {"name": "X"})]) m = build_manifest(src) assert m["summary"]["total_contention_events_kept"] == 0 assert m["entries"] == [] print("PASS test_empty_input_yields_zero_kept") def test_tid_map_translates_canary_to_ours() -> None: src = write_jsonl([ mk_event( "contention.observed", 6, 104664, {"cs_ptr": "0xbc65c890", "site_sid": "c26a128bf45411f7", "contended": True}, ), mk_event( "contention.observed", 7, 10, {"cs_ptr": "0xa", "site_sid": "aa", "contended": True}, ), ]) m = build_manifest(src, tid_map={6: 1, 7: 2}) assert m["entries"][0]["tid"] == 1, m["entries"][0] assert m["entries"][1]["tid"] == 2 print("PASS test_tid_map_translates_canary_to_ours") def test_tid_map_drops_unmapped_canary_tids() -> None: src = write_jsonl([ mk_event( "contention.observed", 6, 100, {"cs_ptr": "0xa", "site_sid": "aa", "contended": True}, ), mk_event( "contention.observed", 99, 200, {"cs_ptr": "0xb", "site_sid": "bb", "contended": True}, ), ]) m = build_manifest(src, tid_map={6: 1}) assert m["summary"]["total_contention_events_kept"] == 1 assert m["summary"]["skipped_unmapped_tids"] == 1 assert m["entries"][0]["tid"] == 1 print("PASS test_tid_map_drops_unmapped_canary_tids") if __name__ == "__main__": tests = [ test_basic_extract, test_filters_non_contention_kinds, test_filters_contended_false, test_sorts_by_tid_then_idx, test_deduplicates_same_tid_idx, test_skips_missing_fields, test_handles_bad_json_lines, test_render_summary_human_readable, test_empty_input_yields_zero_kept, test_tid_map_translates_canary_to_ours, test_tid_map_drops_unmapped_canary_tids, ] for t in tests: t() print(f"\nALL {len(tests)} TESTS PASS")