from __future__ import annotations import argparse import importlib.util import json import os import tempfile import unittest from datetime import UTC, datetime from pathlib import Path from unittest import mock ROOT = Path(__file__).resolve().parents[1] SPEC = importlib.util.spec_from_file_location("t02_attended_window", ROOT / "scripts" / "t02-attended-window.py") assert SPEC and SPEC.loader window = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(window) TEST_SPEC = importlib.util.spec_from_file_location( "test_audit_core_synthetic_sender_lane", ROOT / "tests" / "test_audit_core_synthetic_sender_lane.py") assert TEST_SPEC and TEST_SPEC.loader lane_tests = importlib.util.module_from_spec(TEST_SPEC) TEST_SPEC.loader.exec_module(lane_tests) CONTAINED = {"WARDEN_ROUTING_CATALOG": "/catalog.yaml", "BAO_ADDR": "http://127.0.0.1:18200", "PATH": os.environ.get("PATH", "")} class FakeForward: def __init__(self, log: list[str], fail: bool = False) -> None: self.log, self.fail = log, fail def start(self) -> None: self.log.append("port-forward-start") if self.fail: raise window.WindowError("audit-core port-forward exited early") def stop(self) -> None: self.log.append("port-forward-stop") class Runner: """Records the step of each reviewed child; optionally fails one step.""" def __init__(self, log: list[str], fail_on: str | None = None) -> None: self.log, self.fail_on, self.envs = log, fail_on, {} def __call__(self, command: list[str], *, label: str, env: dict | None = None) -> dict: step = "exercise" if "exercise" in command else command[2] if step == "cleanup" and "--abort-reason" in command: step = "abort-cleanup" self.log.append(step) self.envs[step] = env assert "--contained" in command if step == self.fail_on: raise window.WindowError(f"{label} failed (exit 1)") if step == "exercise": return {"procedure": window.LANE.LEASE.PROCEDURE, "approval_id": "t02-window-1", "completed_at": "2026-10-14T00:00:00Z", "secret_values_observed": False} return {"secret_values_observed": False} class AttendedWindowTests(unittest.TestCase): def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory() self.dir = Path(self.temp.name) os.chmod(self.dir, 0o700) self.now = datetime.now(UTC) document = lane_tests.receipt(self.now) self.approval = self.dir / "receipt.json" self.approval.write_text(json.dumps(document), encoding="utf-8") self.evidence = self.dir / "evidence" self.evidence.mkdir(mode=0o700) self.args = argparse.Namespace( approval=self.approval, load_driver=Path("/opt/driver"), bearer_file=self.dir / "t02.bearer", evidence_dir=self.evidence, confirm=window.LANE.confirm_token(document), remote="railiance01", port=18081, ) def tearDown(self) -> None: self.temp.cleanup() def run_window(self, runner: Runner, forward: FakeForward) -> dict: with mock.patch.dict(os.environ, CONTAINED, clear=True): return window.window(self.args, runner=runner, forward_factory=lambda remote, port: forward) def test_main_never_prints_and_records_failure_in_evidence_dir(self) -> None: import contextlib import io out, err = io.StringIO(), io.StringIO() with tempfile.TemporaryDirectory() as tmp, contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): evidence = Path(tmp) with mock.patch.object(window, "window", side_effect=window.WindowError("project failed (exit 1): x")): code = window.main( ["--approval", "/a", "--load-driver", "/d", "--bearer-file", "/b", "--evidence-dir", str(evidence), "--confirm", "c"] ) records = list(evidence.glob("window-error-*.json")) self.assertEqual(1, code) self.assertEqual(1, len(records)) self.assertEqual(0o600, records[0].stat().st_mode & 0o777) self.assertEqual("failed", json.loads(records[0].read_text())["status"]) with mock.patch.object(window, "window", return_value={"status": "ok"}): self.assertEqual(0, window.main( ["--approval", "/a", "--load-driver", "/d", "--bearer-file", "/b", "--evidence-dir", str(evidence), "--confirm", "c"] )) self.assertEqual("", out.getvalue()) self.assertEqual("", err.getvalue()) def test_result_file_is_new_private_and_suppresses_stdout(self) -> None: import contextlib import io out = io.StringIO() with tempfile.TemporaryDirectory() as tmp, contextlib.redirect_stdout(out): target = Path(tmp) / "r.json" window.LANE.emit_result({"secret_values_observed": False}, target) self.assertEqual(0o600, target.stat().st_mode & 0o777) with self.assertRaises(FileExistsError): window.LANE.emit_result({"secret_values_observed": False}, target) self.assertEqual("", out.getvalue()) def test_happy_path_order_and_evidence(self) -> None: log: list[str] = [] runner = Runner(log) summary = self.run_window(runner, FakeForward(log)) self.assertEqual( ["project", "port-forward-start", "exercise", "port-forward-stop", "cleanup", "port-forward-stop"], log) self.assertEqual("completed", summary["outcome"]) exercise = self.evidence / "t02-window-1-exercise.json" self.assertEqual(0o600, exercise.stat().st_mode & 0o777) self.assertTrue((self.evidence / "t02-window-1-window.json").exists()) env = runner.envs["exercise"] self.assertEqual("http://127.0.0.1:18081", env["AUDIT_T02_BASE_URL"]) self.assertEqual(str(self.args.bearer_file), env["AUDIT_T02_TOKEN_FILE"]) self.assertEqual("railiance-t02-synthetic", env["AUDIT_T02_SOURCE"]) self.assertEqual("tenant:trial:t02-synthetic-1", env["AUDIT_T02_TENANT"]) def test_failures_after_projection_run_abort_cleanup(self) -> None: for fail_on in ("project", "exercise", "cleanup"): with self.subTest(fail_on=fail_on): for path in self.evidence.iterdir(): path.unlink() log: list[str] = [] with self.assertRaises(window.WindowError): self.run_window(Runner(log, fail_on), FakeForward(log)) self.assertEqual("abort-cleanup", log[-2] if log[-1] == "port-forward-stop" else log[-1]) self.assertIn("port-forward-stop", log) summary = json.loads((self.evidence / "t02-window-1-window.json").read_text()) self.assertEqual("aborted", summary["outcome"]) self.assertFalse(summary["secret_values_observed"]) def test_port_forward_failure_aborts_without_exercise(self) -> None: log: list[str] = [] with self.assertRaises(window.WindowError): self.run_window(Runner(log), FakeForward(log, fail=True)) self.assertNotIn("exercise", log) self.assertIn("abort-cleanup", log) def test_refuses_outside_contained_login_before_any_step(self) -> None: log: list[str] = [] with mock.patch.dict(os.environ, {"BAO_TOKEN": "x", **CONTAINED}, clear=True): with self.assertRaises(window.LANE.LEASE.ProcedureError): window.window(self.args, runner=Runner(log), forward_factory=lambda r, p: FakeForward(log)) with mock.patch.dict(os.environ, {"PATH": "/usr/bin"}, clear=True): with self.assertRaises(window.LANE.LEASE.ProcedureError): window.window(self.args, runner=Runner(log), forward_factory=lambda r, p: FakeForward(log)) self.assertEqual([], log) def test_wrong_confirm_and_git_evidence_dir_refused_before_any_step(self) -> None: log: list[str] = [] self.args.confirm = "wrong" with self.assertRaises(window.WindowError): self.run_window(Runner(log), FakeForward(log)) self.args.confirm = window.LANE.confirm_token(json.loads(self.approval.read_text())) self.args.evidence_dir = ROOT / "docs" with self.assertRaises(window.LANE.LaneError): self.run_window(Runner(log), FakeForward(log)) self.assertEqual([], log) def test_port_forward_uses_ssh_remote_loopback_and_bounded_kubectl(self) -> None: command = window.PortForward("railiance01", 18081).command() self.assertIn("ExitOnForwardFailure=yes", command) self.assertIn("127.0.0.1:18081:127.0.0.1:18081", command) self.assertTrue(command[-1].startswith("timeout 900 kubectl -n audit-core port-forward --address 127.0.0.1")) self.assertTrue(command[-1].endswith("deploy/audit-core 18081:8080")) if __name__ == "__main__": unittest.main()