1import io2import json3import sys4import tempfile5import time6import unittest7from pathlib import Path89from cross_client_sync_gate import SCENARIOS, run_gate101112class CrossClientSyncGateTest(unittest.TestCase):13 def test_records_start_finish_and_summary_for_every_scenario(self) -> None:14 lines = [f" ✓ {scenario.id} {scenario.name} (0.0s)" for scenario in SCENARIOS]1516 result, summary, events = self.run_fake(lines)1718 self.assertEqual(0, result)19 self.assertEqual("pass", summary["overallStatus"])20 self.assertEqual(["pass"] * len(SCENARIOS), [row["status"] for row in summary["scenarios"]])21 self.assertEqual(len(SCENARIOS), sum(event["event"] == "scenario_start" for event in events))22 self.assertEqual(len(SCENARIOS), sum(event["event"] == "scenario_finish" for event in events))23 self.assertEqual("summary", events[-1]["event"])2425 def test_timeout_terminates_runner_and_records_every_scenario(self) -> None:26 command = [sys.executable, "-u", "-c", "import time; time.sleep(5)"]2728 started = time.monotonic()29 result, summary, events = self.run_command(command, scenario_timeout=0.05)3031 self.assertLess(time.monotonic() - started, 1.0)32 self.assertEqual(1, result)33 self.assertEqual("fail", summary["overallStatus"])34 self.assertEqual("timeout", summary["scenarios"][0]["status"])35 self.assertEqual(36 ["fail"] * (len(SCENARIOS) - 1),37 [row["status"] for row in summary["scenarios"][1:]],38 )39 self.assertEqual("scenario_timeout", summary["scenarios"][0]["reason"])40 self.assertEqual("summary", events[-1]["event"])4142 def test_nonzero_exit_marks_unreported_rows_failed(self) -> None:43 first = SCENARIOS[0]44 code = f"print({f' ✓ {first.id} {first.name} (0.0s)'!r}, flush=True); raise SystemExit(9)"4546 result, summary, _ = self.run_command([sys.executable, "-u", "-c", code])4748 self.assertEqual(1, result)49 self.assertEqual("pass", summary["scenarios"][0]["status"])50 self.assertTrue(all(row["status"] == "fail" for row in summary["scenarios"][1:]))51 self.assertEqual("runner_exit_9", summary["scenarios"][1]["reason"])5253 def run_fake(self, lines: list[str]) -> tuple[int, dict, list[dict]]:54 code = "import json, sys; [print(line, flush=True) for line in json.loads(sys.argv[1])]"55 return self.run_command([sys.executable, "-u", "-c", code, json.dumps(lines)])5657 def run_command(58 self,59 command: list[str],60 scenario_timeout: float = 1.0,61 ) -> tuple[int, dict, list[dict]]:62 events: list[dict] = []63 with tempfile.TemporaryDirectory() as directory:64 summary_path = Path(directory) / "summary.json"65 result = run_gate(66 command,67 summary_path,68 scenario_timeout_seconds=scenario_timeout,69 total_timeout_seconds=2.0,70 emit=events.append,71 raw_output=io.StringIO(),72 )73 summary = json.loads(summary_path.read_text(encoding="utf-8"))74 return result, summary, events757677if __name__ == "__main__":78 unittest.main()