progress.py reads the whole graded ledger and answers two questions the
per-deck scorecard can't: is the company actually progressing (composite/
BDEF-category/KPI trajectories, recurring vs resolved flags), and is the
material good enough to judge them by — a deterministic gap engine spots
what the decks are NOT showing (no profitability visibility, untargeted
KPIs, no forward guidance, broken forecast chain, silently dropped KPIs,
thin-evidence BDEF categories, no board asks) and renders each gap as a
concrete, paste-ready request for the next deck.
Served live from the ledger (no GPU) at /api/companies/{slug}/progress(.md),
written to /data/ledger/<slug>/PROGRESS.md + /data/reports/latest-progress.md
after each graded deck, and viewable/downloadable from the dashboard company
card ("View progress review"). 18 new tests.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
193 lines
8.8 KiB
Python
193 lines
8.8 KiB
Python
"""Tests for progress.py: trajectory, KPI/category trends, and the gap engine."""
|
|
import os
|
|
import sys
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
import progress
|
|
|
|
|
|
def _cat(adjusted, evidence=0.8):
|
|
return {"adjusted": adjusted, "evidence_quality": evidence}
|
|
|
|
|
|
def _kpi(canonical, credit, profitability=False, source="pinned"):
|
|
return {"canonical_name": canonical, "name": canonical, "credit": credit,
|
|
"profitability": profitability,
|
|
"target_source": None if credit is None else source}
|
|
|
|
|
|
def _record(period, composite, cats=None, kpis=None, flags=None,
|
|
prof_na=False, forecast_na=False, asks=None):
|
|
return {
|
|
"period": period, "composite": composite, "deck_id": period.lower(),
|
|
"quant": {"score": composite * 0.6,
|
|
"profitability": {"na": prof_na},
|
|
"other": {"na": False},
|
|
"forecast_integrity": {"na": forecast_na}},
|
|
"qual": {"score": composite * 0.4, "categories": cats or {}},
|
|
"penalties": {"total": 0.0, "flags": flags or []},
|
|
"kpi_results": kpis or [],
|
|
"narrative": {"asks": asks if asks is not None else ["approve budget"]},
|
|
}
|
|
|
|
|
|
def _company(extracted_targets=None):
|
|
return {"slug": "acme", "name": "Acme", "pinned_targets": [],
|
|
"extracted_targets": extracted_targets or {}}
|
|
|
|
|
|
class TestTrend(unittest.TestCase):
|
|
def test_improving(self):
|
|
recs = [_record("2025-Q4", 40.0), _record("2026-Q1", 42.0),
|
|
_record("2026-Q2", 45.0)]
|
|
a = progress.analyze(_company(), recs)
|
|
self.assertEqual(a["trend"]["direction"], "improving")
|
|
self.assertAlmostEqual(a["trend"]["net_change"], 5.0)
|
|
self.assertEqual(a["trend"]["best"]["period"], "2026-Q2")
|
|
self.assertEqual(a["trend"]["series"][1]["delta"], 2.0)
|
|
|
|
def test_flat_and_declining(self):
|
|
flat = progress.analyze(_company(), [_record("2026-Q1", 50.0),
|
|
_record("2026-Q2", 50.5)])
|
|
self.assertEqual(flat["trend"]["direction"], "flat")
|
|
down = progress.analyze(_company(), [_record("2026-Q1", 50.0),
|
|
_record("2026-Q2", 44.0)])
|
|
self.assertEqual(down["trend"]["direction"], "declining")
|
|
|
|
def test_single_deck_insufficient(self):
|
|
a = progress.analyze(_company(), [_record("2026-Q1", 50.0)])
|
|
self.assertEqual(a["trend"]["direction"], "insufficient")
|
|
|
|
|
|
class TestCategoryAndKpiTrends(unittest.TestCase):
|
|
def test_category_direction(self):
|
|
recs = [_record("2026-Q1", 50, cats={"A": _cat(2.5), "B": _cat(4.0)}),
|
|
_record("2026-Q2", 51, cats={"A": _cat(3.5), "B": _cat(3.0)})]
|
|
a = progress.analyze(_company(), recs)
|
|
self.assertEqual(a["categories"]["A"]["direction"], "improving")
|
|
self.assertEqual(a["categories"]["B"]["direction"], "declining")
|
|
self.assertAlmostEqual(a["categories"]["A"]["net_change"], 1.0)
|
|
|
|
def test_kpi_statuses(self):
|
|
recs = [
|
|
_record("2026-Q1", 50, kpis=[_kpi("arr", 1.0), _kpi("churn", 1.0),
|
|
_kpi("nps", None)]),
|
|
_record("2026-Q2", 51, kpis=[_kpi("arr", 0.4), _kpi("nps", None)]),
|
|
]
|
|
a = progress.analyze(_company(), recs)
|
|
self.assertEqual(a["kpis"]["arr"]["status"], "slipped")
|
|
self.assertEqual(a["kpis"]["churn"]["status"], "not-reported")
|
|
self.assertEqual(a["kpis"]["nps"]["status"], "untargeted")
|
|
|
|
def test_kpi_recovered(self):
|
|
recs = [_record("2026-Q1", 50, kpis=[_kpi("arr", 0.2)]),
|
|
_record("2026-Q2", 51, kpis=[_kpi("arr", 1.0)])]
|
|
a = progress.analyze(_company(), recs)
|
|
self.assertEqual(a["kpis"]["arr"]["status"], "recovered")
|
|
|
|
|
|
class TestFlagHistory(unittest.TestCase):
|
|
def test_recurring_new_resolved(self):
|
|
f = lambda code: {"code": code, "description": code, "severity": 2,
|
|
"points": 2.0, "sources": ["extractor"]}
|
|
recs = [_record("2026-Q1", 50, flags=[f("adjusted_metrics"), f("related_party")]),
|
|
_record("2026-Q2", 51, flags=[f("adjusted_metrics"), f("governance_gap")])]
|
|
a = progress.analyze(_company(), recs)
|
|
self.assertEqual([x["code"] for x in a["flags"]["recurring"]], ["adjusted_metrics"])
|
|
self.assertEqual([x["code"] for x in a["flags"]["new_in_latest"]], ["governance_gap"])
|
|
self.assertEqual([x["code"] for x in a["flags"]["resolved"]], ["related_party"])
|
|
|
|
|
|
class TestGapEngine(unittest.TestCase):
|
|
def codes(self, a):
|
|
return {g["code"] for g in a["gaps"]}
|
|
|
|
def test_no_profitability_gap(self):
|
|
a = progress.analyze(_company(), [_record("2026-Q2", 50, prof_na=True)])
|
|
self.assertIn("no_profitability_kpis", self.codes(a))
|
|
|
|
def test_untargeted_kpis_gap(self):
|
|
a = progress.analyze(_company(), [_record("2026-Q2", 50,
|
|
kpis=[_kpi("nps", None)])])
|
|
self.assertIn("untargeted_kpis", self.codes(a))
|
|
|
|
def test_forward_guidance(self):
|
|
recs = [_record("2026-Q2", 50)]
|
|
bare = progress.analyze(_company(), recs)
|
|
self.assertIn("no_forward_guidance", self.codes(bare))
|
|
guided = progress.analyze(
|
|
_company(extracted_targets={"2026-Q3": {"from_deck": "2026-Q2",
|
|
"targets": [{}]}}), recs)
|
|
self.assertNotIn("no_forward_guidance", self.codes(guided))
|
|
|
|
def test_forecast_chain_broken_needs_history(self):
|
|
one = progress.analyze(_company(), [_record("2026-Q2", 50, forecast_na=True)])
|
|
self.assertNotIn("forecast_chain_broken", self.codes(one))
|
|
two = progress.analyze(_company(), [_record("2026-Q1", 50),
|
|
_record("2026-Q2", 50, forecast_na=True)])
|
|
self.assertIn("forecast_chain_broken", self.codes(two))
|
|
|
|
def test_dropped_kpis_gap(self):
|
|
recs = [_record("2026-Q1", 50, kpis=[_kpi("churn", 1.0)]),
|
|
_record("2026-Q2", 51, kpis=[])]
|
|
a = progress.analyze(_company(), recs)
|
|
self.assertIn("dropped_kpis", self.codes(a))
|
|
|
|
def test_thin_evidence_gap_names_material(self):
|
|
a = progress.analyze(_company(),
|
|
[_record("2026-Q2", 50, cats={"E": _cat(3.0, 0.05)})])
|
|
gap = next(g for g in a["gaps"] if g["code"] == "thin_evidence_E")
|
|
self.assertIn("retention/churn", gap["ask"])
|
|
|
|
def test_no_board_asks_gap(self):
|
|
a = progress.analyze(_company(), [_record("2026-Q2", 50, asks=[])])
|
|
self.assertIn("no_board_asks", self.codes(a))
|
|
b = progress.analyze(_company(), [_record("2026-Q2", 50)])
|
|
self.assertNotIn("no_board_asks", self.codes(b))
|
|
|
|
def test_recurring_flag_gap(self):
|
|
f = {"code": "cash_runway_silence", "description": "no runway shown",
|
|
"severity": 3, "points": 3.0, "sources": ["extractor"]}
|
|
recs = [_record("2026-Q1", 50, flags=[f]), _record("2026-Q2", 51, flags=[f])]
|
|
a = progress.analyze(_company(), recs)
|
|
self.assertIn("recurring_cash_runway_silence", self.codes(a))
|
|
|
|
def test_recurring_flag_resolved_in_latest_no_gap(self):
|
|
f = {"code": "cash_runway_silence", "description": "no runway shown",
|
|
"severity": 3, "points": 3.0, "sources": ["extractor"]}
|
|
recs = [_record("2026-Q1", 50, flags=[f]), _record("2026-Q2", 51, flags=[f]),
|
|
_record("2026-Q3", 52, flags=[])]
|
|
a = progress.analyze(_company(), recs)
|
|
self.assertNotIn("recurring_cash_runway_silence", self.codes(a))
|
|
self.assertIn("cash_runway_silence",
|
|
[x["code"] for x in a["flags"]["resolved"]])
|
|
|
|
|
|
class TestRender(unittest.TestCase):
|
|
def test_render_full(self):
|
|
recs = [
|
|
_record("2026-Q1", 42.0, cats={"A": _cat(2.5), "E": _cat(3.0, 0.1)},
|
|
kpis=[_kpi("arr", 1.0, profitability=True), _kpi("churn", 0.5)]),
|
|
_record("2026-Q2", 45.0, cats={"A": _cat(3.4), "E": _cat(3.0, 0.1)},
|
|
kpis=[_kpi("arr", 1.0, profitability=True), _kpi("churn", 0.3)]),
|
|
]
|
|
md = progress.render_progress_md(progress.analyze(_company(), recs))
|
|
for section in ("## Verdict", "## Composite progress", "## What's moving",
|
|
"## KPI trajectory", "## Flag history",
|
|
"## What the materials aren't showing",
|
|
"## Requests for the next deck"):
|
|
self.assertIn(section, md)
|
|
self.assertIn("improving", md)
|
|
self.assertIn("✓ ✓", md) # arr hit both periods
|
|
self.assertIn("missing repeatedly", md) # churn missed twice
|
|
|
|
def test_render_empty(self):
|
|
md = progress.render_progress_md(progress.analyze(_company(), []))
|
|
self.assertIn("No graded decks yet.", md)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|