Files
CASAN/packages/casan-harness/tests/local-viewer-tests.py
T

150 lines
5.9 KiB
Python

#!/usr/bin/env python3
"""Acceptance tests for the stdlib-only CASAN Core reporting boundary."""
from __future__ import annotations
import importlib.util
import json
from pathlib import Path
import tempfile
import unittest
MODULE = Path(__file__).resolve().parents[1] / "scripts" / "python" / "local_report.py"
SPEC = importlib.util.spec_from_file_location("casan_local_report", MODULE)
REPORT = importlib.util.module_from_spec(SPEC)
assert SPEC.loader
SPEC.loader.exec_module(REPORT)
class LocalReportTests(unittest.TestCase):
def setUp(self):
self.temporary = tempfile.TemporaryDirectory()
self.root = Path(self.temporary.name)
(self.root / ".casan").mkdir()
(self.root / ".casan" / "config.json").write_text(json.dumps({
"project_id": "project-one",
"project_name": "Project One",
"edition": "core",
"enforcement_mode": "enforce",
"maturity": {"level": 4, "status": "assessed"},
}), encoding="utf-8")
for path in (
".specify/state",
".specify/logs/trace",
".specify/logs/trace-events",
".specify/logs/cost",
".specify/logs/level5",
".specify/agentops",
):
(self.root / path).mkdir(parents=True, exist_ok=True)
def tearDown(self):
self.temporary.cleanup()
def write_jsonl(self, relative, rows):
path = self.root / relative
path.write_text(
"".join(json.dumps(row, separators=(",", ":")) + "\n" for row in rows),
encoding="utf-8",
)
def write_run(self, trace_id="trace-1", quality="partial"):
trace = {
"trace_id": trace_id,
"project_id": "project-one",
"certified": True,
"certification_strength": "project_hook",
"finalized_at": "2026-07-28T10:00:07Z",
"tool_calls": 2,
"failures": 0,
"evidence": [
{"h": f"H{index}", "at": f"2026-07-28T10:00:0{index}Z", "decision": "pass", "kind": "gate", "detail": f"H{index} passed"}
for index in range(1, 8)
],
}
(self.root / f".specify/logs/trace/agentic-{trace_id}.json").write_text(
json.dumps(trace), encoding="utf-8")
metric = {
"timestamp": "2026-07-28T10:00:06Z",
"trace_id": trace_id,
"project_id": "project-one",
"step": "agentic-turn",
"status": "success",
"latency_ms": 1200,
"retry_count": 0,
"input_tokens": None,
"output_tokens": None,
"total_tokens": None,
"cost_estimate": None,
"telemetry_quality": quality,
"alerts": [],
}
self.write_jsonl(".specify/logs/cost/metrics.jsonl", [metric])
return metric
def test_core_empty_state_is_truthful(self):
registry = REPORT.runs(self.root)
h6 = REPORT.h6_report(self.root)
self.assertEqual(registry["count"], 0)
self.assertEqual(h6["verdict"], "no_data")
self.assertEqual(h6["summary"]["runs"], 0)
self.assertIsNone(h6["summary"]["tokens"]["total"])
self.assertTrue(any(item["code"] == "METRICS_MISSING" for item in h6["findings"]))
def test_run_reconstructs_all_gates_and_redacts_sensitive_fields(self):
self.write_run()
event = {
"timestamp": "2026-07-28T10:00:01Z",
"trace_id": "trace-1",
"gate_id": "H1-context",
"status": "pass",
"reason": "admitted",
"evidence": {"kind": "context", "prompt": "private customer prompt", "access_token": "secret"},
}
self.write_jsonl(".specify/logs/trace-events/trace-1.jsonl", [event])
report = REPORT.run_report(self.root, "trace-1")
self.assertEqual(report["verdict"], "certified")
self.assertEqual(report["summary"]["gates_observed"], 1)
self.assertEqual(report["gates"][0]["evidence"]["prompt"], "[redacted]")
self.assertEqual(report["gates"][0]["evidence"]["access_token"], "[redacted]")
self.assertIsNone(report["h6"]["total_tokens"])
def test_legacy_trace_reconstructs_h1_to_h7(self):
self.write_run()
graph = REPORT.trace_graph(self.root, "trace-1")
self.assertTrue(graph["terminal"])
self.assertEqual(graph["progress"], 7)
self.assertEqual([node["status"] for node in graph["nodes"]], [
"pass", "pass", "pass", "pass", "pass", "warning", "pass",
])
def test_h6_preserves_unknown_token_and_cost_as_null(self):
self.write_run()
report = REPORT.h6_report(self.root, run="trace-1")
self.assertEqual(report["summary"]["coverage"]["token_pct"], 0)
self.assertEqual(report["summary"]["coverage"]["cost_pct"], 0)
self.assertIsNone(report["summary"]["tokens"]["total"])
self.assertIsNone(report["summary"]["cost_usd"]["estimated"])
self.assertEqual(report["data_quality"]["status"], "insufficient")
self.assertIn("TELEMETRY_COVERAGE_GAP", report["verdict_reasons"])
def test_self_contained_html_escapes_evidence(self):
self.write_run()
report = REPORT.run_report(self.root, "trace-1")
report["gates"][0]["reason"] = "<script>alert(1)</script>"
html = REPORT.report_html(report, "run")
self.assertIn("<!doctype html>", html.lower())
self.assertNotIn("<script>alert(1)</script>", html)
self.assertIn("&lt;script&gt;alert(1)&lt;/script&gt;", html)
self.assertIn("Machine-auditable contract", html)
def test_unsafe_trace_id_never_reads_outside_project(self):
report = REPORT.run_report(self.root, "../../etc/passwd")
self.assertEqual(report["verdict"], "not_found")
self.assertFalse(report["source"]["trace_found"])
if __name__ == "__main__":
unittest.main()