"""Tests for scoring.py (pure scorer), validate.py, and the fixture-driven Q1 -> Q2 end-to-end flow through the ledger and scorecard renderers.""" import json import os import sys import tempfile import unittest sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import ledger as ledger_mod import scorecard import scoring import validate FIXTURES = os.path.join(os.path.dirname(os.path.abspath(__file__)), "fixtures") WEIGHTS = { "profitabilityKpi": 30, "otherKpi": 20, "forecastIntegrity": 10, "qualCategoryMax": 5, "redFlagCap": 15, "kpiCreditFloor": 0.5, "droppedKpiPenalty": 2, "droppedKpiMax": 3, "evidenceFullCredit": 400, "singleSourceFlagFactor": 0.5, } PINNED_CASH = [{"kpi": "cash_balance", "target": 12.0, "unit": "$M", "direction": "gte", "profitability": True}] def _fixture(name): with open(os.path.join(FIXTURES, name), encoding="utf-8") as f: return json.load(f) def _kpi(canonical, actual, direction="gte", prof=False, tid=None, name=None, unit=""): return {"name": name or canonical, "canonical_name": canonical, "actual": actual, "unit": unit, "period": None, "direction": direction, "profitability": prof, "target_in_deck": tid, "source": "slide 1", "notes": ""} def _ft(canonical, target, direction="gte", target_period="2026-Q2", prof=False): return {"name": canonical, "canonical_name": canonical, "target": target, "unit": "", "target_period": target_period, "direction": direction, "profitability": prof, "source": "slide 9"} def _extraction(kpis=None, forward=None, flags=None, period="2026-Q2"): return {"schema_version": 1, "deck": {"period": period}, "kpis": kpis or [], "forward_targets": forward or [], "red_flag_candidates": flags or [], "narrative": {"summary": "test deck", "asks": []}} def _grade(grader="grader-a", score=3, quote_chars=0, red_flags=None, overrides=None): cats = [] for cid in "ABCDEFGH": s, qc = score, quote_chars if overrides and cid in overrides: s, qc = overrides[cid] ev = [{"quote": "q" * qc, "location": "slide 1"}] if qc else [] cats.append({"id": cid, "score": s, "evidence": ev, "rationale": f"cat {cid}"}) return {"schema_version": 1, "grader": grader, "categories": cats, "red_flags": red_flags or [], "overall_comment": "ok"} def _meta(period="2026-Q2", deck_id="d1"): return {"company": "acme", "period": period, "deck_id": deck_id, "job_id": "job-1", "graded_at": "2026-07-06T12:00:00Z", "panel": [{"rid": "grader-a", "model": "grader-a", "valid": True}], "artifacts": {"extraction": "extraction.json"}} def _score(extraction, grades=None, pinned=None, prior=None, aliases=None, meta=None): return scoring.score_deck(extraction, grades if grades is not None else [_grade()], pinned or [], prior or [], aliases or {}, WEIGHTS, meta or _meta()) def _flag(rec, code): return [f for f in rec["penalties"]["flags"] if f["code"] == code] class TestMatchKpi(unittest.TestCase): def test_exact(self): cand, via = scoring.match_kpi("arr", [{"canonical_name": "arr", "name": "ARR"}], {}) self.assertEqual(via, "exact") self.assertEqual(cand["name"], "ARR") def test_alias_forward_and_reverse(self): aliases = {"arr": ["Annual Recurring Revenue", "run_rate_arr"]} cand, via = scoring.match_kpi( "arr", [{"canonical_name": "revenue_annualized", "name": "Annual Recurring Revenue"}], aliases) self.assertEqual(via, "alias") cand, via = scoring.match_kpi( "run_rate_arr", [{"canonical_name": "arr", "name": "ARR"}], aliases) self.assertEqual(via, "alias") def test_fuzzy(self): cand, via = scoring.match_kpi( "ebitda_margin", [{"canonical_name": "ebitda_margins", "name": "x"}], {}) self.assertEqual(via, "fuzzy") def test_no_match(self): self.assertEqual( scoring.match_kpi("arr", [{"canonical_name": "cash_balance", "name": "Cash"}], {}), (None, None)) self.assertEqual(scoring.match_kpi("", [{"canonical_name": "arr"}], {}), (None, None)) class TestCredit(unittest.TestCase): def test_lte_credit(self): rec = _score(_extraction([_kpi("churn_rate", 6.0, "lte", tid=5.0)])) self.assertAlmostEqual(rec["kpi_results"][0]["credit"], 0.6667, places=4) rec = _score(_extraction([_kpi("churn_rate", 4.0, "lte", tid=5.0)])) self.assertEqual(rec["kpi_results"][0]["credit"], 1.0) def test_floor(self): rec = _score(_extraction([_kpi("arr", 4.0, tid=10.0)])) # r=0.4 < floor self.assertEqual(rec["kpi_results"][0]["credit"], 0.0) rec = _score(_extraction([_kpi("arr", 7.5, tid=10.0)])) # r=0.75 -> 0.5 self.assertAlmostEqual(rec["kpi_results"][0]["credit"], 0.5, places=4) def test_guards(self): self.assertEqual(scoring._credit(5, 0, "gte", 0.5), 1.0) # zero target, passes self.assertEqual(scoring._credit(-5, 0, "gte", 0.5), 0.0) # zero target, fails self.assertEqual(scoring._credit(-1, 1, "gte", 0.5), 0.0) # sign mismatch, fails self.assertEqual(scoring._credit(1, -1, "gte", 0.5), 1.0) # sign mismatch, passes self.assertEqual(scoring._credit(0, 5, "lte", 0.5), 1.0) # lte zero actual def test_negative_targets(self): # EBITDA margin: target -2, actual -3 -> two thirds of the way -> 0.3333 self.assertAlmostEqual(scoring._credit(-3, -2, "gte", 0.5), 1 / 3, places=4) self.assertEqual(scoring._credit(-1, -2, "gte", 0.5), 1.0) class TestQuantBuckets(unittest.TestCase): def test_first_deck_renormalization(self): # No prior targets -> forecast NA -> its 10 points redistribute 36/24. rec = _score(_extraction([_kpi("ebitda_margin", 5.0, prof=True, tid=5.0), _kpi("arr", 10.0, tid=10.0)])) q = rec["quant"] self.assertTrue(q["forecast_integrity"]["na"]) self.assertAlmostEqual(q["profitability"]["weight"], 36.0) self.assertAlmostEqual(q["profitability"]["score"], 36.0) self.assertAlmostEqual(q["other"]["weight"], 24.0) self.assertAlmostEqual(q["other"]["score"], 24.0) self.assertAlmostEqual(q["score"], 60.0) self.assertEqual(rec["penalties"]["flags"], []) self.assertAlmostEqual(rec["composite"], 84.0) # 60 quant + 24 qual (all 3s) def test_forecast_integrity_second_deck(self): prior = [_ft("arr", 12.0), _ft("churn_rate", 4.0, "lte")] rec = _score(_extraction([_kpi("arr", 11.0), _kpi("churn_rate", 3.5, "lte"), _kpi("fcf", 1.0, prof=True, tid=1.0)]), prior=prior) fi = rec["quant"]["forecast_integrity"] self.assertFalse(fi["na"]) self.assertEqual(fi["weight"], 10.0) self.assertEqual(fi["kpi_count"], 2) accs = {f["canonical_name"]: f["accuracy"] for f in rec["forecast_results"]} self.assertAlmostEqual(accs["arr"], 0.9167, places=4) # 1/12 undershoot self.assertAlmostEqual(accs["churn_rate"], 0.9375, places=4) # overshoot halved self.assertAlmostEqual(fi["score"], (0.9167 + 0.9375) / 2 * 10, places=3) def test_no_profitability_flag_and_redistribution(self): rec = _score(_extraction([_kpi("arr", 10.0, tid=10.0)])) q = rec["quant"] self.assertTrue(q["profitability"]["na"]) self.assertTrue(q["forecast_integrity"]["na"]) self.assertAlmostEqual(q["other"]["weight"], 60.0) self.assertAlmostEqual(q["score"], 60.0) flags = _flag(rec, "no_profitability_visibility") self.assertEqual(len(flags), 1) self.assertEqual(flags[0]["points"], 3.0) # scoring flags never damped self.assertEqual(flags[0]["sources"], ["scoring"]) def test_profitability_kpis_without_targets_na_no_flag(self): rec = _score(_extraction([_kpi("ebitda_margin", -5.0, prof=True), _kpi("arr", 10.0, tid=10.0)])) self.assertTrue(rec["quant"]["profitability"]["na"]) self.assertEqual(_flag(rec, "no_profitability_visibility"), []) def test_all_quant_na_scales_qual(self): rec = _score(_extraction([])) # qual 24 (all 3s) scaled to 60, minus no_profitability(3) + no_quantitative(4) self.assertTrue(all(rec["quant"][b]["na"] for b in ("profitability", "other", "forecast_integrity"))) self.assertEqual(len(_flag(rec, "no_quantitative_kpis")), 1) self.assertAlmostEqual(rec["composite"], 53.0) class TestTargetPrecedence(unittest.TestCase): def test_pinned_beats_extracted_beats_in_deck(self): kpis = [_kpi("arr", 11.0, tid=9.0)] pinned = [{"kpi": "arr", "target": 10.0, "unit": "$M", "direction": "gte", "profitability": False}] prior = [_ft("arr", 12.0)] r = _score(_extraction(kpis), pinned=pinned, prior=prior)["kpi_results"][0] self.assertEqual((r["target"], r["target_source"], r["matched_via"]), (10.0, "pinned", "exact")) self.assertEqual(r["credit"], 1.0) r = _score(_extraction(kpis), prior=prior)["kpi_results"][0] self.assertEqual((r["target"], r["target_source"]), (12.0, "extracted")) self.assertAlmostEqual(r["credit"], 0.8333, places=4) r = _score(_extraction(kpis))["kpi_results"][0] self.assertEqual((r["target"], r["target_source"], r["matched_via"]), (9.0, "in_deck", None)) def test_untargeted_kpi_reported_with_none(self): r = _score(_extraction([_kpi("nps", 40.0)]))["kpi_results"][0] self.assertIsNone(r["target"]) self.assertIsNone(r["credit"]) self.assertIsNone(r["target_source"]) class TestQualitative(unittest.TestCase): def test_evidence_regression_both_directions(self): # Median 5 with no quotes regresses to 3; so does median 1. rec = _score(_extraction([]), grades=[_grade(score=5, quote_chars=0)]) self.assertEqual(rec["qual"]["categories"]["A"]["adjusted"], 3.0) rec = _score(_extraction([]), grades=[_grade(score=1, quote_chars=0)]) self.assertEqual(rec["qual"]["categories"]["A"]["adjusted"], 3.0) self.assertAlmostEqual(rec["qual"]["score"], 24.0) def test_full_evidence_keeps_extreme_scores(self): # Per-quote chars cap at 200, so full credit (400) needs two quotes. g = _grade(score=5, quote_chars=200) for cat in g["categories"]: cat["evidence"].append({"quote": "q" * 200, "location": "slide 2"}) rec = _score(_extraction([]), grades=[g]) cat = rec["qual"]["categories"]["A"] self.assertEqual(cat["evidence_quality"], 1.0) self.assertEqual(cat["adjusted"], 5.0) self.assertEqual(cat["points"], 5.0) def test_quote_chars_capped_at_200_each(self): # One 1000-char quote counts as 200 -> e = 0.5 -> adjusted 4. rec = _score(_extraction([]), grades=[_grade(score=5, quote_chars=1000)]) self.assertEqual(rec["qual"]["categories"]["A"]["evidence_quality"], 0.5) self.assertEqual(rec["qual"]["categories"]["A"]["adjusted"], 4.0) def test_panel_median_and_rationales(self): grades = [_grade("g1", score=4, quote_chars=400), _grade("g2", score=4, quote_chars=400), _grade("g3", score=2, quote_chars=400)] rec = _score(_extraction([]), grades=grades) cat = rec["qual"]["categories"]["B"] self.assertEqual(cat["panel_scores"], [4, 4, 2]) self.assertEqual(cat["median"], 4.0) self.assertEqual(len(cat["rationales"]), 3) self.assertEqual(cat["rationales"][0]["grader"], "g1") class TestPenalties(unittest.TestCase): def test_single_source_damping(self): rec = _score(_extraction([], flags=[{"code": "adjusted_metrics", "description": "d", "severity": 4}])) f = _flag(rec, "adjusted_metrics")[0] self.assertEqual(f["points"], 2.0) self.assertEqual(f["sources"], ["extractor"]) def test_two_sources_full_severity_max_wins(self): grades = [_grade("g1", red_flags=[{"code": "governance_gap", "description": "weak", "severity": 2}]), _grade("g2", red_flags=[{"code": "governance_gap", "description": "worse", "severity": 3}])] rec = _score(_extraction([]), grades=grades) f = _flag(rec, "governance_gap")[0] self.assertEqual(f["severity"], 3) self.assertEqual(f["points"], 3.0) self.assertEqual(f["sources"], ["g1", "g2"]) def test_penalty_cap(self): codes = ["related_party", "channel_stuffing_risk", "suppressed_dissent", "metric_redefinition"] flags = [{"code": c, "description": c, "severity": 5} for c in codes] rec = _score(_extraction([_kpi("fcf", 1.0, prof=True, tid=1.0)], flags=flags), grades=[_grade("g1", red_flags=flags)]) self.assertEqual(rec["penalties"]["total"], 15.0) # 4x5=20 capped def test_dropped_kpi_flags_capped(self): prior = [_ft(c, 1.0) for c in ("alpha_metric", "beta_metric", "gamma_metric", "delta_metric", "epsilon_metric")] rec = _score(_extraction([]), prior=prior) dropped = _flag(rec, "kpi_dropped") self.assertEqual(len(dropped), 3) # droppedKpiMax for f in dropped: self.assertEqual(f["points"], 2.0) # droppedKpiPenalty, never damped class TestValidate(unittest.TestCase): def test_parse_json_text(self): self.assertEqual(validate.parse_json_text('{"a": 1}'), {"a": 1}) salvaged = validate.parse_json_text( 'Sure! Here is the JSON:\n```json\n{"a": {"b": "}"}}\n```\ntrailing prose') self.assertEqual(salvaged, {"a": {"b": "}"}}) self.assertIsNone(validate.parse_json_text("no json here")) self.assertIsNone(validate.parse_json_text("[1, 2, 3]")) self.assertIsNone(validate.parse_json_text("")) def test_schemas_load_and_fixtures_validate(self): self.assertIn("properties", validate.load_schema("extraction")) self.assertIn("properties", validate.load_schema("grades")) for name, schema in (("extraction_q1.json", "extraction"), ("extraction_q2.json", "extraction"), ("grade_a.json", "grades"), ("grade_b.json", "grades"), ("grade_c.json", "grades")): err = validate.validate_obj(_fixture(name), schema) self.assertIsNone(err, f"{name}: {err}") def test_validate_obj_rejects_bad(self): self.assertIsNotNone(validate.validate_obj({"schema_version": 1}, "grades")) def test_validate_file(self): obj, err = validate.validate_file(os.path.join(FIXTURES, "grade_a.json"), "grades") self.assertIsNone(err) self.assertEqual(obj["grader"], "grader-a") obj, err = validate.validate_file("/nonexistent.json", "grades") self.assertIsNone(obj) self.assertIsNotNone(err) class TestEndToEnd(unittest.TestCase): """Fixture-driven Q1 -> Q2 flow: score, ledger round-trip, rendering.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.ledger = ledger_mod.Ledger(os.path.join(self._tmp.name, "ledger")) self.grades = [_fixture("grade_a.json"), _fixture("grade_b.json"), _fixture("grade_c.json")] self.q1 = _fixture("extraction_q1.json") self.q2 = _fixture("extraction_q2.json") def tearDown(self): self._tmp.cleanup() def _score_q1(self): return scoring.score_deck(self.q1, self.grades, PINNED_CASH, [], {}, WEIGHTS, _meta("2026-Q1", "2026-Q1")) def test_q1_first_deck(self): rec = self._score_q1() q = rec["quant"] self.assertTrue(q["forecast_integrity"]["na"]) self.assertAlmostEqual(q["score"], 60.0) # every KPI at/above target # qual: A-G 3.5 pts each (median 4, evidence 0.5), H 2.6667 self.assertAlmostEqual(rec["qual"]["score"], 27.1667, places=3) self.assertAlmostEqual(rec["qual"]["categories"]["H"]["points"], 2.6667, places=3) # hockey_stick (extractor only, sev 3 -> 1.5) + governance_gap (2 graders -> 3) self.assertAlmostEqual(rec["penalties"]["total"], 4.5) self.assertAlmostEqual(rec["composite"], 82.7) cash = next(k for k in rec["kpi_results"] if k["canonical_name"] == "cash_balance") self.assertEqual(cash["target_source"], "pinned") def test_q2_against_q1_targets(self): rec1 = self._score_q1() self.ledger.record_deck("acme", rec1, self.q1["forward_targets"]) prior = self.ledger.prior_targets("acme", "2026-Q2") self.assertEqual(len(prior), 4) rec2 = scoring.score_deck(self.q2, self.grades, PINNED_CASH, prior, {}, WEIGHTS, _meta("2026-Q2", "2026-Q2")) by_name = {k["canonical_name"]: k for k in rec2["kpi_results"]} self.assertAlmostEqual(by_name["arr"]["credit"], 0.8333, places=4) self.assertEqual(by_name["churn_rate"]["credit"], 1.0) self.assertAlmostEqual(by_name["ebitda_margin"]["credit"], 0.3333, places=4) self.assertEqual(by_name["cash_balance"]["target_source"], "pinned") self.assertEqual(by_name["cash_balance"]["credit"], 1.0) q = rec2["quant"] self.assertAlmostEqual(q["profitability"]["score"], 20.0, places=2) self.assertAlmostEqual(q["other"]["score"], 18.333, places=2) self.assertAlmostEqual(q["forecast_integrity"]["score"], 7.847, places=2) self.assertEqual(len(rec2["forecast_results"]), 3) # qualified_pipeline guided in Q1 but not reported in Q2 -> dropped flag dropped = _flag(rec2, "kpi_dropped") self.assertEqual(len(dropped), 1) self.assertIn("qualified_pipeline", dropped[0]["description"]) # adjusted_metrics 1.0 + governance_gap 3.0 + kpi_dropped 2.0 self.assertAlmostEqual(rec2["penalties"]["total"], 6.0) self.assertAlmostEqual(rec2["composite"], 67.3) self.assertAlmostEqual( rec2["composite"], round(q["score"] + rec2["qual"]["score"] - rec2["penalties"]["total"], 1)) # ledger round-trip + rendering self.ledger.record_deck("acme", rec2, self.q2["forward_targets"]) records = self.ledger.deck_records("acme") self.assertEqual([r["period"] for r in records], ["2026-Q1", "2026-Q2"]) report = scorecard.render_deck_report(rec2, self.q2, adjudication_md="Chair memo.") self.assertIn("67.3", report) self.assertIn("pinned", report) self.assertIn("## Panel adjudication", report) self.assertIn("Chair memo.", report) self.assertIn("kpi_dropped", report) card = scorecard.render_scorecard(self.ledger.get_company("acme"), records) self.assertIn("2026-Q1", card) self.assertIn("2026-Q2", card) self.assertIn("↓", card) # composite fell Q1 -> Q2 self.assertIn("KPI hit-rate", card) self.assertIn("arr", card) def test_meta_passthrough_and_record_shape(self): rec = self._score_q1() self.assertEqual(rec["company"], "acme") self.assertEqual(rec["deck_id"], "2026-Q1") self.assertEqual(rec["job_id"], "job-1") self.assertEqual(rec["panel"][0]["rid"], "grader-a") self.assertEqual(rec["artifacts"], {"extraction": "extraction.json"}) self.assertEqual(rec["schema_version"], 1) self.assertIn("summary", rec["narrative"]) for key in ("composite", "quant", "qual", "penalties", "kpi_results", "forecast_results"): self.assertIn(key, rec) # the record must be JSON-serializable as produced json.dumps(rec) if __name__ == "__main__": unittest.main()