| """The public API serves only the audited user report; QA stays private; withheld runs never enter |
| history; the markdown download is the stored audited bytes.""" |
| import json |
| import pathlib |
| import sys |
| import time |
|
|
| |
| sys.path.append(str(pathlib.Path(__file__).resolve().parents[1] / "engine")) |
| sys.path.append(str(pathlib.Path(__file__).resolve().parents[1] / "engine" / "tests")) |
| sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) |
|
|
| import pytest |
|
|
| import analyst |
|
|
|
|
| @pytest.fixture() |
| def dirs(tmp_path, monkeypatch): |
| monkeypatch.setattr(analyst, "_REPORTS_DIR", tmp_path / "reports") |
| monkeypatch.setattr(analyst, "_PRIVATE_DIR", tmp_path / "private") |
| monkeypatch.setattr(analyst, "_QA_DIR", tmp_path / "private" / "qa") |
| monkeypatch.setattr(analyst, "_LEDGER", tmp_path / "reports" / "_ledger.jsonl") |
| monkeypatch.setattr(analyst, "_HF_TOKEN", "") |
| monkeypatch.setattr(analyst, "_price_now", lambda s: None) |
| (tmp_path / "reports").mkdir() |
| return tmp_path |
|
|
|
|
| def _final(tmp_path, published=True): |
| import pipeline_fixtures as F |
| run, payload, _ = F.run_pipeline() if published else F.run_pipeline(data=F.all_data(treasury_10y=[])) |
| path = tmp_path / "VST_final.json" |
| run.write(path, payload) |
| return path, payload |
|
|
|
|
| def _finish(job, path): |
| rec = analyst._read_final(path) |
| job["user"] = analyst.scrub_private(rec["user"]) |
| job["status"] = "done" if job["user"]["status"] == "PUBLISHED" else "withheld" |
| analyst._save_qa(job["ticker"], rec["qa"], job) |
| if job["status"] == "done": |
| analyst._save_report(job) |
|
|
|
|
| def test_published_run_serves_the_audited_document(dirs): |
| path, payload = _final(dirs) |
| job = {"id": "j1", "ticker": "VST", "status": "running", "events": [], "queue": None} |
| analyst._jobs["j1"] = job |
| _finish(job, path) |
| res = analyst.result("j1") |
| assert res["status"] == "PUBLISHED" and res["html"] == payload["html"] |
| assert set(res) <= set(analyst.PUBLIC_REPORT_KEYS) | set(analyst.PUBLIC_META_KEYS) |
| assert analyst.cached_markdown("VST") == payload["markdown"] |
| assert analyst.cached_report("VST")["version"] == payload["version"] |
| assert [r["ticker"] for r in analyst.recent_reports()] == ["VST"] |
| row = json.loads((dirs / "reports" / "_ledger.jsonl").read_text().splitlines()[-1]) |
| assert row["expected_return"] == round(payload["headline"]["value"] * 100, 1) and row["version"] == payload["version"] |
|
|
|
|
| def test_withheld_run_is_not_public_history(dirs): |
| path, _ = _final(dirs, published=False) |
| job = {"id": "j2", "ticker": "VST", "status": "running", "events": [], "queue": None} |
| analyst._jobs["j2"] = job |
| _finish(job, path) |
| res = analyst.result("j2") |
| assert res["status"] == "WITHHELD" and "D9" not in json.dumps(res) and "html" not in res |
| assert analyst.cached_report("VST") is None and analyst.recent_reports() == [] |
| qa = analyst.qa_record("VST") |
| assert qa["validations"][0]["blocking"] == ["D9"] and qa["user_status"] == "WITHHELD" |
|
|
|
|
| def test_legacy_records_are_not_served(dirs): |
| (dirs / "reports" / "OLD.json").write_text(json.dumps({"ticker": "OLD", "cached_at": time.time(), "rating": "BUY", |
| "expected_return": 12.0})) |
| assert analyst.cached_report("OLD") is None and analyst.recent_reports() == [] |
|
|
|
|
| def test_progress_events_carry_no_engine_text(): |
| job = {"pct": 0} |
| ev = analyst._classify(" [forecast] PASS — F1 PASS, F2 WARNING; coherence DIVERGENT", job) |
| assert ev and set(ev[0]) == {"t", "pct", "label", "step"} and "F2" not in json.dumps(ev) |
| assert analyst._classify(" → Final: +12.3% | BUY", {"pct": 99}) == [] |
| assert analyst._classify('@@CRUCIBLE_ROLE@@{"role": "bull", "reasoning": "x"}', {"pct": 0}) == [] |
|
|
|
|
| def test_dev_route_requires_the_token(dirs, monkeypatch): |
| from fastapi.testclient import TestClient |
| import app as appmod |
| c = TestClient(appmod.app) |
| monkeypatch.delenv("DEV_TOKEN", raising=False) |
| assert c.get("/api/dev/qa/VST").status_code == 404 |
| monkeypatch.setenv("DEV_TOKEN", "t0ken") |
| assert c.get("/api/dev/qa/VST", headers={"X-Dev-Token": "wrong"}).status_code == 403 |
|
|
|
|
|
|
| def test_engine_log_goes_to_private_qa_only(dirs): |
| path, payload = _final(dirs) |
| job = {"id": "j9", "ticker": "VST", "status": "running", "events": [], "queue": None, "lines": ["[data] PASS", "[story] n/a — reason"]} |
| rec = analyst._read_final(path) |
| job["user"] = analyst.scrub_private(rec["user"]) |
| qa = dict(rec["qa"]); qa["engine_log_tail"] = job["lines"] |
| analyst._save_qa("VST", qa, job) |
| assert analyst.qa_record("VST")["engine_log_tail"][-1] == "[story] n/a — reason" |
| assert "engine_log_tail" not in json.dumps(analyst.public_view(job["user"])) |
|
|
|
|
| def test_progress_events_say_who_finished_never_what_they_said(): |
| """The progress page lights up storytellers and debate seats; no figure, rating or text crosses.""" |
| import analyst |
| job = {"pct": 70} |
| evs = [] |
| assert analyst._classify("@@progress story-start", {"pct": 30}) == [{"t": "stage", "pct": 40, "label": "Building the valuation story", "step": "valuation"}] |
| ev = analyst._classify('@@progress story-start {"industry": "Power", "beta": 0.413918, "erp": 0.0409, "stage": "mature-growth", ' |
| '"segments": ["Retail", 7], "segment_build": false, "value": 193.1, "rating": "SELL", "story": "x"}', {"pct": 40}) |
| assert ev == [{"t": "facts", "facts": {"industry": "Power", "beta": 0.4139, "erp": 0.0409, "stage": "mature-growth", |
| "segments": ["Retail"], "segment_build": False}}] |
| for line in ["@@progress storyteller", " Neutral (fallback) -45.0% SELL ", |
| " Round 2 — Bear (Risk Mgr) · Bull (Growth Analyst) initial", |
| " Kimi (Bear) -35.0% STRONG SELL conf:72%", " Claude (PM) p_bear=50% p_bull=30% → -16.0%"]: |
| evs += analyst._classify(line, job) |
| seats = [e for e in evs if e["t"] == "seat"] |
| assert [(e["role"], e["phase"]) for e in seats] == [("neutral", "done"), ("bear", "case"), ("pm", "done")] |
| assert [e for e in evs if e["t"] == "story"] == [{"t": "story", "n": 1}] |
| for e in evs: |
| assert set(e) <= {"t", "pct", "label", "step", "role", "phase", "n"} |
| assert not any(tok in str(e) for tok in ("45", "35", "SELL", "p_bear", "16.0")) |
|
|
