File size: 6,588 Bytes
c9fa753 348fb3e 682d244 401ffd0 f6b6138 682d244 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | """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
# the app root must win module resolution (both the app and the engine have a config module)
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 # noqa: E402
import analyst # noqa: E402
@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}}] # value/rating/story dropped
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"))
|