"""Tests for the LOCKED scorer (harness/fitness.py): leash, HALT, formula, aggregation.""" from datetime import datetime, timedelta, timezone import pytest from harness import fitness from harness.fitness import VideoAnalytics, ChannelHalt, compute_fitness from harness.scoreboard import ( ScoreRow, CHANNEL_TERMINATED, CHANNEL_SUSPENDED, CHANNEL_ACTIVE, ) NOW = datetime(2026, 6, 18, tzinfo=timezone.utc) def _date(days_ago: int) -> str: return (NOW - timedelta(days=days_ago)).date().isoformat() # ── fitness formula ────────────────────────────────────────────────────────── def test_fitness_bounds(): assert compute_fitness(0, 0) == 0.0 assert abs(compute_fitness(100, 1.0) - 10.0) < 1e-9 def test_fitness_weights_apv_and_vsa(): # (0.6*0.8 + 0.4*0.5) * 10 = 6.8 assert abs(compute_fitness(80.0, 0.5) - 6.8) < 1e-9 def test_fitness_clamps_out_of_range(): assert compute_fitness(200, 5) == 10.0 # clamped to max assert compute_fitness(-50, -1) == 0.0 # clamped to min def test_fitness_monotonic_in_apv(): assert compute_fitness(90, 0.5) > compute_fitness(50, 0.5) # ── the 3-day leash ─────────────────────────────────────────────────────────── @pytest.mark.parametrize("days_ago,expected", [(0, False), (1, False), (2, False), (3, True), (10, True)]) def test_leash_threshold(days_ago, expected): assert fitness._within_leash(_date(days_ago), NOW) is expected def test_leash_rejects_garbage_date(): assert fitness._within_leash("not-a-date", NOW) is False # ── refresh_scoreboard: leash filtering, context passthrough, HALT ───────────── def _analytics(video_id, days_ago, **kw): base = dict(variant_id="variant_1", genome_hash="abc", parent_genome="seed", apv=80.0, vsa=0.5) base.update(kw) return VideoAnalytics(video_id=video_id, upload_date=_date(days_ago), **base) def test_refresh_filters_fresh_videos(monkeypatch): captured = {} def fake_upsert(rows): captured["rows"] = list(rows) return len(captured["rows"]) monkeypatch.setattr(fitness, "upsert_rows", fake_upsert) n = fitness.refresh_scoreboard( lab_channel_id="LAB", get_channel_status=lambda c: CHANNEL_ACTIVE, get_channel_analytics=lambda c: [_analytics("old", 5), _analytics("fresh", 1)], now=NOW, ) rows = captured["rows"] assert n == 1 assert [r.video_id for r in rows] == ["old"] # fresh one excluded by leash assert rows[0].channel_status == CHANNEL_ACTIVE def test_refresh_passes_context_metrics(monkeypatch): captured = {} monkeypatch.setattr(fitness, "upsert_rows", lambda rows: captured.setdefault("rows", list(rows)) or 1) fitness.refresh_scoreboard( lab_channel_id="LAB", get_channel_status=lambda c: CHANNEL_ACTIVE, get_channel_analytics=lambda c: [_analytics("v", 4, likes=12, shares=3, views=900, subscribers_gained=2)], now=NOW, ) row: ScoreRow = captured["rows"][0] assert row.likes == 12 and row.shares == 3 and row.views == 900 and row.subscribers_gained == 2 assert abs(row.fitness - 6.8) < 1e-9 # fitness still only APV/VSA def test_refresh_carries_readability_context(monkeypatch): captured = {} monkeypatch.setattr(fitness, "upsert_rows", lambda rows: captured.setdefault("rows", list(rows)) or 1) fitness.refresh_scoreboard( lab_channel_id="LAB", get_channel_status=lambda c: CHANNEL_ACTIVE, get_channel_analytics=lambda c: [_analytics("v", 4, readability=0.72)], now=NOW, ) row: ScoreRow = captured["rows"][0] assert abs(row.readability - 0.72) < 1e-9 # vision score rides into the scoreboard row assert abs(row.fitness - 6.8) < 1e-9 # but is NOT part of the fitness scalar @pytest.mark.parametrize("dead", [CHANNEL_TERMINATED, CHANNEL_SUSPENDED]) def test_dead_channel_halts_not_scores(monkeypatch, dead): monkeypatch.setattr(fitness, "upsert_rows", lambda rows: pytest.fail("must not write rows for a dead channel")) with pytest.raises(ChannelHalt): fitness.refresh_scoreboard( lab_channel_id="LAB", get_channel_status=lambda c: dead, get_channel_analytics=lambda c: [_analytics("v", 5)], now=NOW, ) # ── trailing-window aggregation ──────────────────────────────────────────────── def test_fitness_by_variant_windowing_raw_mean(): # prior_pseudo_count=0 → plain mean (the original behaviour): only the in-window row counts. rows = [ ScoreRow("v1", _date(2), "A", "h", "seed", 80, 0.5, 6.8), ScoreRow("v2", _date(100), "A", "h", "seed", 0, 0, 0.0), # outside 60d window ScoreRow("v3", _date(5), "B", "h", "seed", 50, 0.5, compute_fitness(50, 0.5)), ] agg = fitness.fitness_by_variant(rows, window_days=60, now=NOW, prior_pseudo_count=0.0) assert set(agg) == {"A", "B"} assert abs(agg["A"] - 6.8) < 1e-9 assert abs(agg["B"] - 5.0) < 1e-9 def test_fitness_by_variant_shrinkage_pulls_small_samples_toward_prior(): # A: one 6.8 video. B: one 5.0 video. Population mean = 5.9. With pseudo-count 2 each 1-video # mean is pulled most of the way to 5.9 — A: (6.8+2*5.9)/3 = 6.2, B: (5.0+2*5.9)/3 = 5.6. rows = [ ScoreRow("v1", _date(2), "A", "h", "seed", 80, 0.5, 6.8), ScoreRow("v3", _date(5), "B", "h", "seed", 50, 0.5, 5.0), ] agg = fitness.fitness_by_variant(rows, window_days=60, now=NOW, prior_pseudo_count=2.0) assert abs(agg["A"] - 6.2) < 1e-6 assert abs(agg["B"] - 5.6) < 1e-6 assert agg["A"] > agg["B"] # ordering preserved assert agg["A"] < 6.8 and agg["B"] > 5.0 # both shrunk toward the prior def test_fitness_by_variant_shrinkage_barely_moves_large_samples(): # A has 8 videos all at 8.0; B has 1 video at 2.0. Population mean ≈ 7.33 (A dominates it). # Shrinkage barely touches A's 8-sample mean but pulls B's single sample far toward the prior. rows = [ScoreRow(f"a{i}", _date(2), "A", "h", "seed", 0, 0, 8.0) for i in range(8)] rows.append(ScoreRow("b1", _date(2), "B", "h", "seed", 0, 0, 2.0)) agg = fitness.fitness_by_variant(rows, window_days=60, now=NOW, prior_pseudo_count=2.0) a_moved = abs(8.0 - agg["A"]) b_moved = abs(2.0 - agg["B"]) assert agg["A"] > 7.5 # 8 samples → barely shrunk (stays near 8.0) assert b_moved > 3.0 # 1 sample → pulled hard toward the prior assert b_moved > 10 * a_moved # the small sample moves far more than the large one