"""Tests for the local run viewer (strix.viewer) and its path helpers.""" from __future__ import annotations import json import os import urllib.error import urllib.request from typing import TYPE_CHECKING from strix.core.paths import latest_run_dir, runs_base_dir from strix.viewer.server import serve from strix.viewer.transcript import ( build_run_state, read_report_markdown, read_run_summary, read_vulnerabilities, ) if TYPE_CHECKING: from collections.abc import Mapping from pathlib import Path import pytest def _make_run(base: Path, name: str, *, status: str, end_time: str | None) -> Path: run_dir = base / "strix_runs" / name state_dir = run_dir / ".state" state_dir.mkdir(parents=True) record = {"run_name": name, "status": status, "end_time": end_time} (run_dir / "run.json").write_text(json.dumps(record), encoding="utf-8") agents = { "statuses": {"root": "completed", "child": "running"}, "names": {"root": "strix", "child": "recon"}, "parent_of": {"root": None, "child": "root"}, } (state_dir / "agents.json").write_text(json.dumps(agents), encoding="utf-8") return run_dir def test_latest_run_dir_none_when_no_runs(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.chdir(tmp_path) assert latest_run_dir() is None assert runs_base_dir() == tmp_path / "strix_runs" def test_latest_run_dir_picks_newest_by_record_mtime( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) older = _make_run(tmp_path, "old", status="completed", end_time="2026-01-01T00:00:00Z") newer = _make_run(tmp_path, "new", status="running", end_time=None) # Force a newer mtime on the second run's record. os.utime(newer / "run.json", (2_000_000_000, 2_000_000_000)) os.utime(older / "run.json", (1_000_000_000, 1_000_000_000)) assert latest_run_dir() == newer def test_read_run_summary_finished_flag(tmp_path: Path) -> None: finished = _make_run(tmp_path, "done", status="completed", end_time="2026-01-01T00:00:00Z") live = _make_run(tmp_path, "live", status="running", end_time=None) assert read_run_summary(finished)["finished"] is True assert read_run_summary(live)["finished"] is False # A terminal status without an end_time is not "finished". partial = _make_run(tmp_path, "partial", status="failed", end_time=None) assert read_run_summary(partial)["finished"] is False def test_read_missing_artifacts_return_defaults(tmp_path: Path) -> None: run_dir = _make_run(tmp_path, "empty", status="running", end_time=None) assert read_vulnerabilities(run_dir) == [] assert read_report_markdown(run_dir) == "" def test_build_run_state_from_agents_json(tmp_path: Path) -> None: run_dir = _make_run(tmp_path, "graph", status="running", end_time=None) state = build_run_state(run_dir) ids = {a["id"] for a in state["agents"]} assert ids == {"root", "child"} child = next(a for a in state["agents"] if a["id"] == "child") assert child["parent_id"] == "root" assert child["name"] == "recon" # No agents.db, so no message/tool events. assert state["events"] == [] def _get(url: str) -> tuple[int, str, bytes]: with urllib.request.urlopen(url) as resp: # noqa: S310 - localhost test server return resp.status, resp.headers.get("Content-Type", ""), resp.read() def test_server_serves_api_and_static(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: run_dir = _make_run(tmp_path, "served", status="completed", end_time="2026-01-01T00:00:00Z") assets = tmp_path / "bundle" (assets / "assets").mkdir(parents=True) (assets / "index.html").write_text("
", encoding="utf-8") (assets / "assets" / "app.js").write_text("console.log(1)", encoding="utf-8") monkeypatch.setattr("strix.viewer.server.bundle_dir", lambda: assets) httpd, url = serve(run_dir, open_browser=False) try: status, ctype, body = _get(f"{url}/api/run") assert status == 200 assert "application/json" in ctype assert json.loads(body)["finished"] is True status, _, body = _get(f"{url}/api/transcript") assert {a["id"] for a in json.loads(body)["agents"]} == {"root", "child"} # Real asset is served. status, ctype, _ = _get(f"{url}/assets/app.js") assert status == 200 # Unknown non-API route falls back to index.html (SPA routing). status, ctype, body = _get(f"{url}/agents/root") assert status == 200 assert b"
" in body finally: httpd.shutdown() httpd.server_close() def test_server_event_endpoint_forwards_cta( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: run_dir = _make_run(tmp_path, "evt", status="running", end_time=None) assets = tmp_path / "bundle" assets.mkdir() (assets / "index.html").write_text("x", encoding="utf-8") monkeypatch.setattr("strix.viewer.server.bundle_dir", lambda: assets) seen: list[tuple[str, str | None]] = [] monkeypatch.setattr( "strix.telemetry.posthog.viewer_cta_clicked", lambda cta, surface=None: seen.append((cta, surface)), ) httpd, url = serve(run_dir, open_browser=False) try: body = json.dumps( {"event": "cta_clicked", "cta": "PR reviews", "surface": "sidebar_nav"} ).encode() req = urllib.request.Request( # noqa: S310 - localhost test server f"{url}/api/event", data=body, headers={"Content-Type": "application/json"} ) with urllib.request.urlopen(req) as resp: # noqa: S310 assert resp.status == 204 assert seen == [("PR reviews", "sidebar_nav")] finally: httpd.shutdown() httpd.server_close() def test_server_event_endpoint_forwards_email_funnel( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: run_dir = _make_run(tmp_path, "evt2", status="running", end_time=None) assets = tmp_path / "bundle" assets.mkdir() (assets / "index.html").write_text("x", encoding="utf-8") monkeypatch.setattr("strix.viewer.server.bundle_dir", lambda: assets) seen: list[tuple[str, str | None]] = [] monkeypatch.setattr( "strix.telemetry.posthog.viewer_email_event", lambda step, purpose=None: seen.append((step, purpose)), ) httpd, url = serve(run_dir, open_browser=False) try: # A whitelisted funnel event is forwarded; an unknown event is ignored. for payload, expected in ( ({"event": "email_verified", "purpose": "report"}, [("email_verified", "report")]), ({"event": "not_a_real_event"}, [("email_verified", "report")]), ): req = urllib.request.Request( # noqa: S310 - localhost test server f"{url}/api/event", data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"}, ) with urllib.request.urlopen(req) as resp: # noqa: S310 assert resp.status == 204 assert seen == expected finally: httpd.shutdown() httpd.server_close() def _post( url: str, path: str, payload: Mapping[str, object], *, cookie: str | None = None ) -> tuple[int, bytes]: headers = {"Content-Type": "application/json"} if cookie: headers["Cookie"] = cookie req = urllib.request.Request( # noqa: S310 - localhost test server url + path, data=json.dumps(payload).encode(), headers=headers, method="POST" ) try: with urllib.request.urlopen(req) as resp: # noqa: S310 return resp.status, resp.read() except urllib.error.HTTPError as exc: return exc.code, exc.read() def _session_cookie(url: str) -> str: """Fetch index.html and return its ``name=value`` session cookie.""" with urllib.request.urlopen(url + "/") as resp: # noqa: S310 - localhost test server raw = str(resp.headers.get("Set-Cookie", "")) return raw.split(";", 1)[0] def _get_status(url: str) -> int: try: with urllib.request.urlopen(url) as resp: # noqa: S310 - localhost test server return int(resp.status) except urllib.error.HTTPError as exc: return int(exc.code) def _bundle(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: assets = tmp_path / "bundle" assets.mkdir() (assets / "index.html").write_text("
", encoding="utf-8") monkeypatch.setattr("strix.viewer.server.bundle_dir", lambda: assets) def test_index_sets_session_cookie_but_assets_do_not( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: run_dir = _make_run(tmp_path, "cookie", status="running", end_time=None) assets = tmp_path / "bundle" (assets / "assets").mkdir(parents=True) (assets / "index.html").write_text("index", encoding="utf-8") (assets / "assets" / "app.js").write_text("1", encoding="utf-8") monkeypatch.setattr("strix.viewer.server.bundle_dir", lambda: assets) httpd, url = serve(run_dir, open_browser=False) try: with urllib.request.urlopen(url + "/") as resp: # noqa: S310 cookie = resp.headers.get("Set-Cookie", "") assert "strix_viewer_session=" in cookie assert "HttpOnly" in cookie and "SameSite=Strict" in cookie with urllib.request.urlopen(url + "/assets/app.js") as resp: # noqa: S310 assert resp.headers.get("Set-Cookie") is None finally: httpd.shutdown() httpd.server_close() def test_steer_requires_session_cookie(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: run_dir = _make_run(tmp_path, "steer", status="running", end_time=None) _bundle(tmp_path, monkeypatch) delivered: list[tuple[str, str]] = [] def handler(agent_id: str, message: str) -> bool: delivered.append((agent_id, message)) return True httpd, url = serve(run_dir, open_browser=False, steer_handler=handler) try: body = {"agent_id": "root", "message": "focus on auth"} # No cookie: rejected before reaching the live coordinator. status, _ = _post(url, "/api/agents/steer", body) assert status == 403 assert delivered == [] # With the session cookie the message is delivered. status, raw = _post(url, "/api/agents/steer", body, cookie=_session_cookie(url)) assert status == 200 assert json.loads(raw)["ok"] is True assert delivered == [("root", "focus on auth")] finally: httpd.shutdown() httpd.server_close() def test_report_send_requires_session_cookie( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: run_dir = _make_run(tmp_path, "report", status="completed", end_time="2026-01-01T00:00:00Z") _bundle(tmp_path, monkeypatch) # A verified machine token exists, but that alone must not authorize a caller. monkeypatch.setattr("strix.viewer.auth.read_auth", lambda: {"email": "a@b.com", "token": "t"}) httpd, url = serve(run_dir, open_browser=False) try: # No cookie: forbidden before the machine token is ever consulted. status, _ = _post(url, "/api/report/send", {}) assert status == 403 # With the cookie the request clears the session gate; it then reaches # the run resolver, so an unknown run is a 404 rather than a 403. status, _ = _post( url, "/api/report/send", {"run": "does-not-exist"}, cookie=_session_cookie(url) ) assert status == 404 finally: httpd.shutdown() httpd.server_close() def test_historical_run_data_requires_verification( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: launched = _make_run(tmp_path, "launched", status="completed", end_time="2026-01-01T00:00:00Z") _make_run(tmp_path, "other", status="completed", end_time="2026-01-01T00:00:00Z") _bundle(tmp_path, monkeypatch) verified = {"value": False} monkeypatch.setattr("strix.viewer.auth.is_verified", lambda: verified["value"]) httpd, url = serve(launched, open_browser=False) try: # The launched run is always viewable, no verification required. status, _, _ = _get(f"{url}/api/run") assert status == 200 # A different run's data is gated behind verification. assert _get_status(f"{url}/api/run?run=other") == 401 # Once verified, the historical run resolves. verified["value"] = True status, _, _ = _get(f"{url}/api/run?run=other") assert status == 200 finally: httpd.shutdown() httpd.server_close() def test_server_rejects_path_traversal(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: run_dir = _make_run(tmp_path, "guard", status="completed", end_time="2026-01-01T00:00:00Z") secret = tmp_path / "secret.txt" secret.write_text("top secret", encoding="utf-8") assets = tmp_path / "bundle" assets.mkdir() (assets / "index.html").write_text("index", encoding="utf-8") monkeypatch.setattr("strix.viewer.server.bundle_dir", lambda: assets) httpd, url = serve(run_dir, open_browser=False) try: # A traversal target must never leak the file; it falls back to index.html. _, _, body = _get(f"{url}/..%2f..%2fsecret.txt") assert b"top secret" not in body finally: httpd.shutdown() httpd.server_close()