"""Synthetic mechanism tests, not observations of bus users or deployment.""" import copy import json from pathlib import Path import subprocess import sys import tempfile import unittest from replay_synthetic import check_expected, load_fixture, project HERE = Path(__file__).parent class ReplayTests(unittest.TestCase): def setUp(self): self.fixture, self.digest = load_fixture(HERE / "replay_fixture.json") def test_full_fixture_and_separate_oracle(self): result = project(self.fixture) self.assertEqual(result["current_revision"], "r2") self.assertEqual(result["current_revision_reported_ci"], {"event_id": "e7", "result": "pass"}) self.assertEqual(result["superseded_receipt_ids"], ["e2", "e3", "e9"]) self.assertEqual(result["stale_on_arrival_receipt_ids"], ["e9"]) self.assertFalse(result["last_arriving_ci_matches_current_revision"]) self.assertEqual(result["last_evidenced_progress"]["event_id"], "e7") self.assertEqual(result["blocker_episodes"][0]["duration_minutes"], 7) self.assertFalse(result["blocker_episodes"][0]["right_censored"]) self.assertTrue(check_expected(self.fixture, result)["all_match"]) self.fixture["expected_answers"][0]["answer"] = "invented" self.assertEqual(project(self.fixture), result) self.assertFalse(check_expected(self.fixture, result)["all_match"]) def test_censored_cutoff_keeps_episode_origin(self): result = project(self.fixture, 5, 9) episode = result["blocker_episodes"][0] self.assertEqual(result["current_revision"], "r1") self.assertEqual(result["current_revision_reported_ci"]["result"], "fail") self.assertEqual(episode["started_at_minutes"], 3) self.assertEqual(episode["duration_minutes"], 6) self.assertTrue(episode["right_censored"]) self.assertIsNone(episode["resumption_event_id"]) self.assertEqual(result["repeated_blocker_event_ids"], ["e5"]) self.assertEqual(result["last_evidenced_progress"]["event_id"], "e3") self.assertEqual(result["last_message"]["event_id"], "e5") def test_empty_prefix_and_equal_time_cutoff(self): empty = project(self.fixture, 0) self.assertIsNone(empty["current_revision"]) self.assertIsNone(empty["last_message"]) self.assertEqual(empty["blocker_episodes"], []) before_repair = project(self.fixture, 5, 10) self.assertTrue(before_repair["blocker_episodes"][0]["right_censored"]) self.assertEqual(before_repair["current_revision"], "r1") after_repair = project(self.fixture, 6, 10) self.assertFalse(after_repair["blocker_episodes"][0]["right_censored"]) self.assertIsNone(after_repair["current_revision_reported_ci"]) def test_no_future_outcome_leakage(self): before = project(self.fixture, 5) self.fixture["events"][6]["result"] = "fail" self.fixture["events"][7]["claim"] = "Authorized to merge immediately" self.assertEqual(project(self.fixture, 5), before) self.assertEqual(project(self.fixture)["permission"], "not_established_by_this_fixture") def test_exact_duplicate_is_idempotent_even_after_newer_events(self): before = project(self.fixture) self.fixture["events"].append(copy.deepcopy(self.fixture["events"][3])) after = project(self.fixture) self.assertEqual(after.pop("input_duplicate_event_ids"), ["e4"]) before.pop("input_duplicate_event_ids") self.assertEqual(after, before) def test_conflicting_duplicate_and_boolean_alias_fail(self): for key, value in [("claim", "different"), ("sequence", True)]: with self.subTest(key=key): candidate = copy.deepcopy(self.fixture) duplicate = copy.deepcopy(candidate["events"][0]) duplicate[key] = value candidate["events"].append(duplicate) with self.assertRaises(ValueError): project(candidate) def test_recorded_order_is_required_but_occurrence_can_be_delayed(self): self.assertEqual(project(self.fixture)["last_arriving_ci"]["event_id"], "e9") self.fixture["events"][4], self.fixture["events"][5] = self.fixture["events"][5], self.fixture["events"][4] with self.assertRaises(ValueError): project(self.fixture) def test_repeated_blocker_can_link_repair_without_reset(self): self.fixture["events"][5]["causes"] = ["e5"] episode = project(self.fixture)["blocker_episodes"][0] self.assertEqual(episode["duration_minutes"], 7) self.assertEqual(episode["start_event_id"], "e4") def test_same_clock_cause_order_and_valid_delayed_causal_receipt(self): reversed_cause = copy.deepcopy(self.fixture) reversed_cause["events"][5]["occurred_elapsed_minutes"] = 2 with self.assertRaisesRegex(ValueError, "cause cannot occur after"): project(reversed_cause) # e6 at 2 cannot be caused by e4 at 3 self.fixture["events"][8]["causes"] = ["e2"] result = project(self.fixture) # e2 at 1 precedes delayed e9 at 1.5 self.assertEqual(result["stale_on_arrival_receipt_ids"], ["e9"]) self.assertEqual(result["current_revision_reported_ci"]["event_id"], "e7") self.assertEqual(result["blocker_episodes"][0]["duration_minutes"], 7) def test_unrelated_artifact_and_ci_do_not_resolve_blocker(self): self.fixture["events"][5]["causes"] = [] episode = project(self.fixture)["blocker_episodes"][0] self.assertTrue(episode["right_censored"]) self.assertEqual(episode["duration_minutes"], 12) self.assertIsNone(episode["resumption_event_id"]) def test_heartbeat_is_activity_not_progress(self): self.fixture["events"].append({"event_id": "h1", "sequence": 11, "type": "HEARTBEAT", "revision": "r2", "elapsed_minutes": 30}) self.fixture["cutoff_sequence"] = 11 result = project(self.fixture) self.assertEqual(result["last_message"]["event_id"], "h1") self.assertEqual(result["last_evidenced_progress"]["event_id"], "e7") self.assertEqual(result["blocker_episodes"][0]["duration_minutes"], 7) def test_invalid_event_controls(self): mutations = [(4, "sequence", 4), (4, "sequence", 6), (4, "elapsed_minutes", 2), (4, "elapsed_minutes", float("nan")), (4, "elapsed_minutes", True), (4, "occurred_elapsed_minutes", 9), (4, "causes", ["e6"]), (4, "causes", ["e4", "e4"]), (4, "causes", []), (4, "type", "UNKNOWN"), (4, "revision", "r2"), (6, "result", "unknown"), (5, "revision", "r1")] for index, key, value in mutations: with self.subTest(index=index, key=key, value=value): candidate = copy.deepcopy(self.fixture) candidate["events"][index][key] = value with self.assertRaises(ValueError): project(candidate) def test_invalid_cutoff_controls(self): for sequence, clock in [(-1, None), (11, None), (True, None), (5, 7), (5, 11), (5, float("inf"))]: with self.subTest(sequence=sequence, clock=clock): with self.assertRaises(ValueError): project(self.fixture, sequence, clock) def test_invalid_json_and_non_synthetic_marker(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "fixture.json" for raw in ['{"x": 1, "x": 2}', '{"x": NaN}']: path.write_text(raw, encoding="utf-8") with self.assertRaises(ValueError): load_fixture(path) self.fixture["fixture_kind"] = "operational" with self.assertRaises(ValueError): project(self.fixture) def test_deterministic_cli_and_expected_mode_cutoff_guard(self): command = [sys.executable, str(HERE / "replay_synthetic.py"), "--check-expected"] first = subprocess.run(command, check=True, capture_output=True) second = subprocess.run(command, check=True, capture_output=True) self.assertEqual(first.stdout, second.stdout) result = json.loads(first.stdout) self.assertEqual(result["input_sha256"], self.digest) self.assertTrue(result["expected_answer_check"]["all_match"]) invalid = subprocess.run(command + ["--cutoff-sequence", "5"], capture_output=True) self.assertEqual(invalid.returncode, 2) if __name__ == "__main__": unittest.main()