railiance-platform/tests/test_t02_attended_window.py

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()