"""Tests for ``arborist.substrate.fork_score`` — pure-function ForkScore computation over (parent, child) BatteryResult bundles (ticket #000012 Phase 1a). Covers: - bench_result_to_metrics adapter (BatteryResult → nested dict) - fork_score over the three rate-deltas (5S, 5T, 5F) and the five non-bench inputs - hard-reject paths: NEG_INF_REGRESSION, per-battery HARD_REGRESSION_FLOOR - verdict band: ACCEPT (≥SIGNAL_FLOOR), MARGINAL ([0, SIGNAL_FLOOR)), REJECT (<0 or hard-reject) - empty inputs (zero parent, zero child) → score 0, verdict MARGINAL - ScoredFork.to_dict serializability - breakdown sum identity (score ≡ Σ breakdown) """ from __future__ import annotations import math import pytest from arborist.substrate.fork_score import ( HARD_REGRESSION_FLOOR, INFINITE_BONUS_CAP, SIGNAL_FLOOR, ScoredFork, bench_result_to_metrics, fork_score, ) from arborist.substrate.weights import DEFAULT_WEIGHTS, WeightSet # --- bench_result_to_metrics adapter -------------------------------- def test_bench_result_to_metrics_groups_by_battery(): """Adapter from runner JSON to nested-dict shape.""" payload = { "results": [ {"battery": "5s", "sub_battery": "syntax", "metrics": {"parse_pass_rate": 0.9}}, {"battery": "5s", "sub_battery": "semantics", "metrics": {"equivalence_recovery_rate": 0.85}}, {"battery": "5t", "sub_battery": "transfer-learning", "metrics": {"transfer_learning_success_rate": 0.7}}, {"battery": "5f", "sub_battery": "function", "metrics": {"function_pass_rate": 0.95}}, ], } out = bench_result_to_metrics(payload) assert set(out.keys()) == {"5s", "5t", "5f"} assert out["5s"]["syntax"]["parse_pass_rate"] == 0.9 assert out["5s"]["semantics"]["equivalence_recovery_rate"] == 0.85 assert out["5t"]["transfer-learning"]["transfer_learning_success_rate"] == 0.7 assert out["5f"]["function"]["function_pass_rate"] == 0.95 def test_bench_result_to_metrics_skips_unknown_battery(): """Battery keys outside {5s, 5t, 5f} get dropped.""" payload = { "results": [ {"battery": "5r", "sub_battery": "rho", "metrics": {"rho_rate": 0.5}}, {"battery": "5s", "sub_battery": "syntax", "metrics": {"parse_pass_rate": 0.8}}, ], } out = bench_result_to_metrics(payload) assert "5r" not in out assert out["5s"]["syntax"]["parse_pass_rate"] == 0.8 def test_bench_result_to_metrics_empty_payload(): """No `results` key → empty nested dicts (still keyed by battery).""" out = bench_result_to_metrics({}) assert out == {"5s": {}, "5t": {}, "5f": {}} # --- fork_score happy path ------------------------------------------ def _bench_dict(rates_by_battery_sub: dict[str, dict[str, dict[str, float]]]): """Helper: build a nested metrics dict directly.""" out = {"5s": {}, "5t": {}, "5f": {}} for battery, subs in rates_by_battery_sub.items(): out[battery] = {sub: dict(metrics) for sub, metrics in subs.items()} return out def test_fork_score_pure_improvement_accepts(): """Child improves on every 5S sub-battery by 10pp → score ≥ SIGNAL_FLOOR → ACCEPT verdict.""" parent = _bench_dict({ "5s": { "syntax": {"parse_pass_rate": 0.5}, "semantics": {"equivalence_recovery_rate": 0.5}, "syllogism": {"step_validity_rate": 0.5}, "synthesis": {"derivation_pass_rate": 0.5}, "semiotics": {"invariance_under_swap": 0.5}, }, }) child = _bench_dict({ "5s": { "syntax": {"parse_pass_rate": 0.6}, "semantics": {"equivalence_recovery_rate": 0.6}, "syllogism": {"step_validity_rate": 0.6}, "synthesis": {"derivation_pass_rate": 0.6}, "semiotics": {"invariance_under_swap": 0.6}, }, }) r = fork_score(parent, child) assert r.verdict == "ACCEPT" assert r.score >= SIGNAL_FLOOR def test_fork_score_marginal_band(): """Child improves by less than SIGNAL_FLOOR's mapped weight → MARGINAL band (score in [0, SIGNAL_FLOOR)).""" # Tiny improvement — 1pp on one sub-battery — well below the # signal floor when averaged. parent = _bench_dict({ "5s": { "syntax": {"parse_pass_rate": 0.50}, "semantics": {"equivalence_recovery_rate": 0.50}, "syllogism": {"step_validity_rate": 0.50}, "synthesis": {"derivation_pass_rate": 0.50}, "semiotics": {"invariance_under_swap": 0.50}, }, }) child = _bench_dict({ "5s": { "syntax": {"parse_pass_rate": 0.51}, "semantics": {"equivalence_recovery_rate": 0.50}, "syllogism": {"step_validity_rate": 0.50}, "synthesis": {"derivation_pass_rate": 0.50}, "semiotics": {"invariance_under_swap": 0.50}, }, }) r = fork_score(parent, child) assert r.verdict == "MARGINAL" assert 0 <= r.score < SIGNAL_FLOOR def test_fork_score_zero_zero_yields_zero(): """Zero parent + zero child → score 0 (no terms fire) → MARGINAL (score in [0, SIGNAL_FLOOR)).""" r = fork_score({"5s": {}, "5t": {}, "5f": {}}, {"5s": {}, "5t": {}, "5f": {}}) assert r.score == pytest.approx(0.0, abs=1e-12) assert r.verdict == "MARGINAL" # --- regression / hard-reject paths --------------------------------- def test_fork_score_hard_regression_5s_rejects(): """Drop ≥HARD_REGRESSION_FLOOR on a 5S sub → REJECT regardless of overall positive score.""" parent = _bench_dict({ "5s": { "syntax": {"parse_pass_rate": 0.9}, "semantics": {"equivalence_recovery_rate": 0.5}, "syllogism": {"step_validity_rate": 0.5}, "synthesis": {"derivation_pass_rate": 0.5}, "semiotics": {"invariance_under_swap": 0.5}, }, }) # syntax drops by 0.20 (>= 0.05 hard floor); other subs improve. child = _bench_dict({ "5s": { "syntax": {"parse_pass_rate": 0.7}, "semantics": {"equivalence_recovery_rate": 0.95}, "syllogism": {"step_validity_rate": 0.95}, "synthesis": {"derivation_pass_rate": 0.95}, "semiotics": {"invariance_under_swap": 0.95}, }, }) r = fork_score(parent, child) assert r.verdict == "REJECT" assert any("REGRESSION_5S" in f and "syntax" in f for f in r.flags) def test_fork_score_negative_score_rejects(): """Score < 0 (more decreases than increases) → REJECT.""" parent = _bench_dict({ "5s": { "syntax": {"parse_pass_rate": 0.8}, "semantics": {"equivalence_recovery_rate": 0.8}, "syllogism": {"step_validity_rate": 0.8}, "synthesis": {"derivation_pass_rate": 0.8}, "semiotics": {"invariance_under_swap": 0.8}, }, }) # All decrease by 0.04 (less than HARD_REGRESSION_FLOOR=0.05), # so no per-sub regression flag — but mean delta is negative # → score < 0 → REJECT child = _bench_dict({ "5s": { "syntax": {"parse_pass_rate": 0.76}, "semantics": {"equivalence_recovery_rate": 0.76}, "syllogism": {"step_validity_rate": 0.76}, "synthesis": {"derivation_pass_rate": 0.76}, "semiotics": {"invariance_under_swap": 0.76}, }, }) r = fork_score(parent, child) assert r.verdict == "REJECT" assert r.score < 0 def test_fork_score_neg_infinite_count_in_5f_hard_rejects(): """`adaptation_efficiency_neg_infinite_count > 0` on child → auto-REJECT regardless of other terms.""" parent = _bench_dict({ "5f": { "function": {"function_pass_rate": 0.5}, "finetuning": {"adaptation_improvement_rate": 0.5}, "falsification": {"error_detection_rate": 0.5}, "formulate": {"structural_match_rate": 0.5}, "feedback-loop": {"integration_coverage_rate": 0.5}, }, }) child = _bench_dict({ "5f": { "function": {"function_pass_rate": 0.95}, "finetuning": { "adaptation_improvement_rate": 0.95, "adaptation_efficiency_neg_infinite_count": 1, }, "falsification": {"error_detection_rate": 0.95}, "formulate": {"structural_match_rate": 0.95}, "feedback-loop": {"integration_coverage_rate": 0.95}, }, }) r = fork_score(parent, child) assert r.verdict == "REJECT" assert any("NEG_INF_REGRESSION" in f for f in r.flags) # --- non-bench input terms ------------------------------------------ def test_fork_score_capital_cost_penalty_subtracts(): """capital_delta > 0 reduces the score (cost penalty).""" parent = _bench_dict({"5s": {}, "5t": {}, "5f": {}}) child = _bench_dict({"5s": {}, "5t": {}, "5f": {}}) no_penalty = fork_score(parent, child, capital_delta=0.0) with_penalty = fork_score(parent, child, capital_delta=10.0) # capital_delta term is -theta * capital_delta — so positive # capital_delta should REDUCE the score assert with_penalty.score < no_penalty.score def test_fork_score_audit_completeness_adds(): """audit_completeness > 0 increases the score (positive term).""" parent = _bench_dict({"5s": {}, "5t": {}, "5f": {}}) child = _bench_dict({"5s": {}, "5t": {}, "5f": {}}) base = fork_score(parent, child) bonus = fork_score(parent, child, audit_completeness=1.0) assert bonus.score > base.score def test_fork_score_security_risk_subtracts_when_iota_positive(): """security_risk reduces the score when the iota weight is non-zero. Default WeightSet sets iota=0 (security risk is opt- in for the validator), so we test with an explicit iota>0.""" parent = _bench_dict({"5s": {}, "5t": {}, "5f": {}}) child = _bench_dict({"5s": {}, "5t": {}, "5f": {}}) iota_on = WeightSet( alpha=DEFAULT_WEIGHTS.alpha, beta=DEFAULT_WEIGHTS.beta, gamma=DEFAULT_WEIGHTS.gamma, delta=DEFAULT_WEIGHTS.delta, epsilon=DEFAULT_WEIGHTS.epsilon, zeta=DEFAULT_WEIGHTS.zeta, eta=DEFAULT_WEIGHTS.eta, theta=DEFAULT_WEIGHTS.theta, iota=1.0, # turn on kappa=DEFAULT_WEIGHTS.kappa, lambda_=DEFAULT_WEIGHTS.lambda_, ) base = fork_score(parent, child, weights=iota_on) risky = fork_score(parent, child, weights=iota_on, security_risk=1.0) assert risky.score < base.score def test_fork_score_security_risk_inert_under_default_weights(): """Honest documentation of the default behavior: with iota=0 (the default), passing security_risk does NOT change the score. This is by design — operators must opt-in to the security-risk penalty by setting iota>0.""" parent = _bench_dict({"5s": {}, "5t": {}, "5f": {}}) child = _bench_dict({"5s": {}, "5t": {}, "5f": {}}) assert DEFAULT_WEIGHTS.iota == 0.0, ( "test assumes iota=0 default; update if WeightSet defaults change" ) base = fork_score(parent, child) with_risk = fork_score(parent, child, security_risk=1.0) assert with_risk.score == pytest.approx(base.score, abs=1e-12) # --- weights customization ----------------------------------------- def test_fork_score_weights_recorded_in_output(): """Custom WeightSet flows through to the output's weights dict.""" custom = WeightSet(alpha=0.5, beta=0.5, gamma=0.5, delta=0.1, epsilon=0.1, zeta=0.1, eta=1.0, theta=1.0, iota=1.0, kappa=1.0, lambda_=1.0) r = fork_score({}, {}, weights=custom) assert r.weights["alpha"] == pytest.approx(0.5) assert r.weights["lambda_"] == pytest.approx(1.0) def test_fork_score_default_weights_used_when_unspecified(): r = fork_score({}, {}) assert r.weights == DEFAULT_WEIGHTS.as_dict() # --- ScoredFork API ------------------------------------------------- def test_scored_fork_to_dict_serializable(): """ScoredFork.to_dict() returns JSON-serializable types.""" r = fork_score({}, {}) d = r.to_dict() import json json.dumps(d) # must not raise assert "score" in d assert "verdict" in d assert "breakdown" in d assert "flags" in d assert "weights" in d # --- score = sum(breakdown) closure --------------------------------- def test_score_equals_sum_of_breakdown(): """Closure check: score ≡ Σ breakdown values. Multiple parent/child configurations to widen the cone.""" configs = [ # Plain improvement on 5S only. ( {"5s": {"syntax": {"parse_pass_rate": 0.5}}}, {"5s": {"syntax": {"parse_pass_rate": 0.7}}}, ), # Mixed deltas across all three batteries. ( _bench_dict({ "5s": {"syntax": {"parse_pass_rate": 0.6}}, "5t": {"transfer-learning": {"transfer_learning_success_rate": 0.6}}, "5f": {"function": {"function_pass_rate": 0.6}}, }), _bench_dict({ "5s": {"syntax": {"parse_pass_rate": 0.7}}, "5t": {"transfer-learning": {"transfer_learning_success_rate": 0.65}}, "5f": {"function": {"function_pass_rate": 0.55}}, }), ), # All-zero inputs → score 0. ({"5s": {}, "5t": {}, "5f": {}}, {"5s": {}, "5t": {}, "5f": {}}), ] for parent, child in configs: r = fork_score(parent, child) assert r.score == pytest.approx(sum(r.breakdown.values()), abs=1e-9), ( f"score {r.score} ≠ Σbreakdown {sum(r.breakdown.values())} on " f"parent={parent!r}, child={child!r}" ) # --- constants honored from arborist.substrate.fork_score ----------- def test_signal_floor_honored(): """Score exactly at SIGNAL_FLOOR is ACCEPT (≥, not >).""" # We construct a child whose 5S delta * alpha == SIGNAL_FLOOR. # alpha defaults to 0.30 per WeightSet (see arborist.substrate.weights). # Solve for delta: delta = SIGNAL_FLOOR / alpha = 0.05 / 0.30 ≈ 0.1667. # Average of 5 sub-deltas; set each sub to 0.1667 to hit it. parent = _bench_dict({ "5s": { "syntax": {"parse_pass_rate": 0.5}, "semantics": {"equivalence_recovery_rate": 0.5}, "syllogism": {"step_validity_rate": 0.5}, "synthesis": {"derivation_pass_rate": 0.5}, "semiotics": {"invariance_under_swap": 0.5}, }, }) target_delta = SIGNAL_FLOOR / DEFAULT_WEIGHTS.alpha child = _bench_dict({ "5s": { "syntax": {"parse_pass_rate": 0.5 + target_delta}, "semantics": {"equivalence_recovery_rate": 0.5 + target_delta}, "syllogism": {"step_validity_rate": 0.5 + target_delta}, "synthesis": {"derivation_pass_rate": 0.5 + target_delta}, "semiotics": {"invariance_under_swap": 0.5 + target_delta}, }, }) r = fork_score(parent, child) # Score = alpha · target_delta = SIGNAL_FLOOR exactly. Verdict ACCEPT. assert r.score == pytest.approx(SIGNAL_FLOOR, abs=1e-9) assert r.verdict == "ACCEPT" # --------------------------------------------------------------------- # #000047 — delta_aggregator (mean / max / sum) # --------------------------------------------------------------------- def test_delta_aggregator_default_is_mean(): assert DEFAULT_WEIGHTS.delta_aggregator == "mean" def test_aggregate_helper(): from arborist.substrate.fork_score import _aggregate assert _aggregate([], "mean") == 0.0 assert _aggregate([], "max") == 0.0 assert _aggregate([], "sum") == 0.0 assert _aggregate([0.1, 0.2, 0.3, 0.0, 0.0], "mean") == pytest.approx(0.12) assert _aggregate([0.6, 0.0, 0.0, 0.0, 0.0], "max") == pytest.approx(0.6) assert _aggregate([-0.1, -0.2, 0.0], "max") == 0.0 # floored at 0 assert _aggregate([0.05] * 5, "sum") == pytest.approx(0.25) with pytest.raises(ValueError, match="unknown delta aggregator"): _aggregate([0.1], "bogus") def test_weightset_rejects_bad_aggregator(): with pytest.raises(ValueError, match="delta_aggregator must be one of"): WeightSet(delta_aggregator="median") def test_weights_from_dict_aggregator(): from arborist.substrate.weights import from_dict assert from_dict({}).delta_aggregator == "mean" assert from_dict({"delta_aggregator": "sum"}).delta_aggregator == "sum" assert from_dict({"alpha": 2.0, "delta_aggregator": "max"}).delta_aggregator == "max" assert from_dict({"alpha": 2.0, "delta_aggregator": "max"}).alpha == 2.0 def test_fork_score_aggregator_changes_5f_term_for_single_sub_gain(): """A child that lifts ONE 5f sub by +0.6 (the rest flat): mean dilutes it to 0.12, max/sum weigh it at face value 0.6.""" parent = _bench_dict({"5f": {"falsification": {"error_detection_rate": 0.4}}}) child = _bench_dict({"5f": {"falsification": {"error_detection_rate": 1.0}}}) r_mean = fork_score(parent, child) # default mean r_max = fork_score(parent, child, weights=WeightSet(delta_aggregator="max")) r_sum = fork_score(parent, child, weights=WeightSet(delta_aggregator="sum")) assert r_mean.breakdown["gamma_x_delta_5f"] == pytest.approx(0.6 / 5) assert r_max.breakdown["gamma_x_delta_5f"] == pytest.approx(0.6) assert r_sum.breakdown["gamma_x_delta_5f"] == pytest.approx(0.6) def test_fork_score_sum_vs_max_diverge_for_broad_gain(): """Two 5f subs lifted by +0.6 each: mean 0.24, max 0.6, sum 1.2 — three distinct values, so sum ≠ max once the improvement is broad.""" parent = _bench_dict({"5f": { "falsification": {"error_detection_rate": 0.4}, "formulate": {"structural_match_rate": 0.4}, }}) child = _bench_dict({"5f": { "falsification": {"error_detection_rate": 1.0}, "formulate": {"structural_match_rate": 1.0}, }}) assert fork_score(parent, child).breakdown["gamma_x_delta_5f"] == pytest.approx(1.2 / 5) assert fork_score(parent, child, weights=WeightSet(delta_aggregator="max")).breakdown["gamma_x_delta_5f"] == pytest.approx(0.6) assert fork_score(parent, child, weights=WeightSet(delta_aggregator="sum")).breakdown["gamma_x_delta_5f"] == pytest.approx(1.2) def test_fork_score_records_aggregator_in_weights(): parent = _bench_dict({"5f": {"falsification": {"error_detection_rate": 0.4}}}) child = _bench_dict({"5f": {"falsification": {"error_detection_rate": 0.5}}}) assert fork_score(parent, child).weights["delta_aggregator"] == "mean" r = fork_score(parent, child, weights=WeightSet(delta_aggregator="max")) assert r.weights["delta_aggregator"] == "max" def test_hard_regression_flag_independent_of_aggregator(): """A single-sub regression below HARD_REGRESSION_FLOOR forces REJECT under every aggregator — the per-sub flag is computed before aggregation.""" parent = _bench_dict({"5f": {"falsification": {"error_detection_rate": 0.5}}}) child = _bench_dict({"5f": {"falsification": {"error_detection_rate": 0.4}}}) # -0.1 for agg in ("mean", "max", "sum"): r = fork_score(parent, child, weights=WeightSet(delta_aggregator=agg)) assert r.verdict == "REJECT", agg assert any(f.startswith("REGRESSION_5F:") for f in r.flags), agg # --------------------------------------------------------------------- # Phase 1c — branch-set persistence (#000012 §7 Phase 1c) # --------------------------------------------------------------------- def _scored(): """Build a representative ScoredFork via the real fork_score().""" parent = _bench_dict({ "5s": {"syntax": {"parse_pass_rate": 0.5}}, }) child = _bench_dict({ "5s": {"syntax": {"parse_pass_rate": 0.7}}, }) return fork_score(parent, child) def test_phase1c_migration_creates_table(tmp_path): """Opening a connection runs the Phase 1c migration; the table + both indexes exist.""" from arborist.store import connect, invalidate_migration_cache db = tmp_path / "shard.db" invalidate_migration_cache(db) conn = connect(db) try: row = conn.execute( "SELECT name FROM sqlite_master " "WHERE type='table' AND name='fork_score_branches'" ).fetchone() assert row is not None idx_names = { r[0] for r in conn.execute( "SELECT name FROM sqlite_master WHERE type='index' " "AND tbl_name='fork_score_branches'" ).fetchall() } assert "idx_fork_score_branches_set" in idx_names assert "idx_fork_score_branches_parent" in idx_names finally: conn.close() invalidate_migration_cache(db) def test_phase1c_persist_branch_score_writes_one_row(tmp_path): from arborist.store import connect, invalidate_migration_cache, transaction from arborist.substrate.fork_score import ( ESTIMATOR_VERSION, persist_branch_score, ) db = tmp_path / "shard.db" invalidate_migration_cache(db) conn = connect(db) try: with transaction(conn): persist_branch_score( conn, branch_set_id="cp-1", branch_id="b-A", parent_root="parent-root-aaaa", child_root="child-root-A", scored=_scored(), weights_id="default", ) rows = conn.execute( "SELECT branch_set_id, branch_id, parent_root, child_root, " " verdict, weights_id, estimator_version " "FROM fork_score_branches" ).fetchall() assert len(rows) == 1 r = rows[0] assert r["branch_set_id"] == "cp-1" assert r["branch_id"] == "b-A" assert r["parent_root"] == "parent-root-aaaa" assert r["child_root"] == "child-root-A" assert r["verdict"] in ("ACCEPT", "MARGINAL", "REJECT") assert r["estimator_version"] == ESTIMATOR_VERSION finally: conn.close() invalidate_migration_cache(db) def test_phase1c_persist_upserts_on_pk(tmp_path): """Re-scoring the same (branch_set_id, branch_id) is an upsert, not a duplicate row. Fields refresh.""" from arborist.store import connect, invalidate_migration_cache, transaction from arborist.substrate.fork_score import persist_branch_score db = tmp_path / "shard.db" invalidate_migration_cache(db) conn = connect(db) try: s1 = _scored() with transaction(conn): persist_branch_score( conn, branch_set_id="cp-up", branch_id="b-up", parent_root="p1", child_root="c1", scored=s1, weights_id="w1", ts=1700000000, ) with transaction(conn): persist_branch_score( conn, branch_set_id="cp-up", branch_id="b-up", parent_root="p1", child_root="c2-new", scored=s1, weights_id="w2-new", ts=1700000999, ) rows = conn.execute( "SELECT child_root, weights_id, recorded_at " "FROM fork_score_branches WHERE branch_set_id = 'cp-up'" ).fetchall() assert len(rows) == 1 assert rows[0]["child_root"] == "c2-new" assert rows[0]["weights_id"] == "w2-new" assert rows[0]["recorded_at"] == 1700000999 finally: conn.close() invalidate_migration_cache(db) def test_phase1c_branch_set_density_counts_branches(tmp_path): """branch_set_density returns 0 / N for the queried checkpoint; rows under other checkpoints don't leak.""" from arborist.store import connect, invalidate_migration_cache, transaction from arborist.substrate.fork_score import ( branch_set_density, persist_branch_score, ) db = tmp_path / "shard.db" invalidate_migration_cache(db) conn = connect(db) try: assert branch_set_density(conn, "missing") == 0 s = _scored() with transaction(conn): for i in range(4): persist_branch_score( conn, branch_set_id="cp-A", branch_id=f"branch-{i}", parent_root="parent-root", child_root=f"child-root-{i}", scored=s, ) persist_branch_score( conn, branch_set_id="cp-B", branch_id="lone", parent_root="parent-root", child_root="child-root-Z", scored=s, ) # #000037 §12 Trigger 1 satisfied: ≥4 branches at "cp-A". assert branch_set_density(conn, "cp-A") == 4 assert branch_set_density(conn, "cp-B") == 1 assert branch_set_density(conn, "cp-missing") == 0 finally: conn.close() invalidate_migration_cache(db) def test_phase1c_breakdown_blob_round_trips_as_json(tmp_path): """breakdown_blob stores the per-term breakdown losslessly so a downstream reader can replay the verdict.""" import json as _json from arborist.store import connect, invalidate_migration_cache, transaction from arborist.substrate.fork_score import persist_branch_score db = tmp_path / "shard.db" invalidate_migration_cache(db) conn = connect(db) try: scored = _scored() with transaction(conn): persist_branch_score( conn, branch_set_id="cp-blob", branch_id="b-blob", parent_root="p", child_root="c", scored=scored, ) row = conn.execute( "SELECT breakdown_blob FROM fork_score_branches WHERE " "branch_set_id='cp-blob'" ).fetchone() recovered = _json.loads(row["breakdown_blob"]) assert sum(recovered.values()) == pytest.approx( scored.score, abs=1e-9 ) assert set(recovered.keys()) == set(scored.breakdown.keys()) finally: conn.close() invalidate_migration_cache(db)