| 1 | from __future__ import annotations |
| 2 | |
| 3 | import json |
| 4 | from copy import deepcopy |
| 5 | from unittest import mock |
| 6 | |
| 7 | from . import harness |
| 8 | |
| 9 | |
| 10 | def test_fixture_matrix_covers_required_topic_archetypes(): |
| 11 | fixtures = harness.load_fixtures() |
| 12 | archetypes = {fixture.manifest["archetype"] for fixture in fixtures} |
| 13 | |
| 14 | assert 6 <= len(fixtures) <= 8 |
| 15 | assert { |
| 16 | "tech-product", |
| 17 | "person", |
| 18 | "comparison", |
| 19 | "breaking-event", |
| 20 | "niche", |
| 21 | "non-english-cjk", |
| 22 | } <= archetypes |
| 23 | |
| 24 | |
| 25 | def test_research_quality_scores_meet_committed_baselines(): |
| 26 | results = harness.evaluate_all() |
| 27 | print(harness.format_score_table(results)) |
| 28 | |
| 29 | failures = harness.baseline_failures(harness.aggregate_scores(results)) |
| 30 | failures += harness.per_fixture_failures(results) |
| 31 | assert not failures, "\n".join(failures) |
| 32 | |
| 33 | |
| 34 | def test_per_fixture_floor_catches_single_broken_archetype(): |
| 35 | results = harness.evaluate_all() |
| 36 | # Simulate a total clustering failure on one clustered fixture: the |
| 37 | # average stays above the aggregate floor but the per-fixture floor fires. |
| 38 | broken = None |
| 39 | for result in results: |
| 40 | if result.fixture.manifest.get("expects_clusters"): |
| 41 | result.scores["cluster_coherence"] = 0.0 |
| 42 | broken = result.fixture.name |
| 43 | break |
| 44 | assert broken is not None |
| 45 | aggregate_ok = not harness.baseline_failures(harness.aggregate_scores(results)) |
| 46 | per_fixture = harness.per_fixture_failures(results) |
| 47 | assert any(f.startswith(f"{broken}/cluster_coherence") for f in per_fixture) |
| 48 | # Document why the per-fixture layer exists: with 7 fixtures the aggregate |
| 49 | # can absorb one zero. |
| 50 | if aggregate_ok: |
| 51 | assert per_fixture |
| 52 | |
| 53 | |
| 54 | def test_entity_overlap_predicate_pinned(): |
| 55 | # The coherence metric shares extract_text_entities/entity_overlap with |
| 56 | # production clustering. Pin the predicate on fixed inputs so a |
| 57 | # too-permissive drift is caught independently of the (circular) metric. |
| 58 | from lib import entity_extract |
| 59 | |
| 60 | same = entity_extract.entity_overlap( |
| 61 | entity_extract.extract_text_entities("OpenAI ships GPT-6 to enterprise customers"), |
| 62 | entity_extract.extract_text_entities("Enterprise customers get GPT-6 from OpenAI"), |
| 63 | ) |
| 64 | unrelated = entity_extract.entity_overlap( |
| 65 | entity_extract.extract_text_entities("OpenAI ships GPT-6 to enterprise customers"), |
| 66 | entity_extract.extract_text_entities("Best sourdough starter recipes for beginners"), |
| 67 | ) |
| 68 | assert same >= harness.ENTITY_OVERLAP_FLOOR, f"related pair fell below floor: {same}" |
| 69 | assert unrelated < harness.ENTITY_OVERLAP_FLOOR, ( |
| 70 | f"unrelated pair passed the overlap floor ({unrelated}); the shared " |
| 71 | "predicate got too permissive and the coherence metric is now blind" |
| 72 | ) |
| 73 | |
| 74 | |
| 75 | def test_replay_uses_manifest_source_availability(tmp_path): |
| 76 | fixture_path = tmp_path / "cli-sources" |
| 77 | fixture_path.mkdir() |
| 78 | (fixture_path / "http.json").write_text( |
| 79 | json.dumps( |
| 80 | { |
| 81 | "format": "last30days-http-fixture/v1", |
| 82 | "exchanges": [], |
| 83 | "source_exchanges": [], |
| 84 | } |
| 85 | ), |
| 86 | encoding="utf-8", |
| 87 | ) |
| 88 | fixture = harness.EvalFixture( |
| 89 | name="cli-sources", |
| 90 | path=fixture_path, |
| 91 | manifest={ |
| 92 | "topic": "fixture topic", |
| 93 | "as_of_date": "2026-07-10", |
| 94 | "fixture_sources": ["digg", "arxiv", "techmeme", "trustpilot"], |
| 95 | "plan": {}, |
| 96 | }, |
| 97 | input_urls=frozenset(), |
| 98 | ) |
| 99 | |
| 100 | def observe_availability(**_kwargs): |
| 101 | return harness.pipeline.available_sources({}, fixture.manifest["fixture_sources"]) |
| 102 | |
| 103 | with mock.patch.object(harness.pipeline, "run", side_effect=observe_availability), \ |
| 104 | mock.patch.object(harness.pipeline, "which", return_value=None): |
| 105 | available = harness._run_once(fixture) |
| 106 | |
| 107 | assert available == fixture.manifest["fixture_sources"] |
| 108 | |
| 109 | |
| 110 | def test_intentional_out_of_window_regression_fails_recency_floor(): |
| 111 | fixture = harness.load_fixtures()[0] |
| 112 | result = harness.evaluate_fixture(fixture) |
| 113 | regressed = deepcopy(result.report) |
| 114 | primary = regressed.ranked_candidates[0].source_items[0] |
| 115 | primary.published_at = "2025-01-01" |
| 116 | |
| 117 | scores = harness.score_report(regressed, fixture, deterministic=True) |
| 118 | failures = harness.baseline_failures(scores) |
| 119 | |
| 120 | assert scores["recency_compliance"] < 1.0 |
| 121 | assert any(failure.startswith("recency_compliance:") for failure in failures) |
| 122 | |
| 123 | |
| 124 | def test_coherence_fails_when_expected_clusters_vanish(): |
| 125 | fixtures = {f.name: f for f in harness.load_fixtures()} |
| 126 | clustered = fixtures["breaking-event"] |
| 127 | assert clustered.manifest["expects_clusters"] is True |
| 128 | report = harness._run_once(clustered) |
| 129 | # Simulate cluster formation regressing to singletons. |
| 130 | report.clusters = [] |
| 131 | assert harness._cluster_coherence(report, clustered) == 0.0 |
| 132 | |
| 133 | |
| 134 | def test_coherence_allows_singletons_for_sparse_fixtures(): |
| 135 | fixtures = {f.name: f for f in harness.load_fixtures()} |
| 136 | sparse = fixtures["niche"] |
| 137 | assert sparse.manifest.get("expects_clusters") is False |
| 138 | report = harness._run_once(sparse) |
| 139 | report.clusters = [] |
| 140 | assert harness._cluster_coherence(report, sparse) == 1.0 |
| 141 | |
| 142 | |
| 143 | def test_enrichment_replay_merges_metadata_without_replacing_items(): |
| 144 | import sys |
| 145 | sys.path.insert(0, "skills/last30days/scripts") |
| 146 | from lib import pipeline, schema |
| 147 | |
| 148 | fresh = schema.SourceItem( |
| 149 | item_id="yt-1", |
| 150 | source="youtube", |
| 151 | title="Fresh title from current normalization", |
| 152 | body="fresh body", |
| 153 | url="https://youtube.com/watch?v=1", |
| 154 | published_at="2026-07-01", |
| 155 | snippet="fresh snippet", |
| 156 | engagement={"views": 10}, |
| 157 | metadata={"channel": "fresh-channel"}, |
| 158 | ) |
| 159 | replayed = [{ |
| 160 | "item_id": "yt-1", |
| 161 | "title": "STALE fixture title", |
| 162 | "snippet": "STALE snippet", |
| 163 | "metadata": {"transcript_snippet": "recorded transcript"}, |
| 164 | }] |
| 165 | merged = pipeline._merge_replayed_enrichment([fresh], replayed) |
| 166 | assert merged[0].title == "Fresh title from current normalization" |
| 167 | assert merged[0].snippet == "fresh snippet" |
| 168 | assert merged[0].metadata["transcript_snippet"] == "recorded transcript" |
| 169 | assert merged[0].metadata["channel"] == "fresh-channel" |
| 170 | |
| 171 | |
| 172 | def test_star_enrichment_apply_map_offline(): |
| 173 | import sys |
| 174 | sys.path.insert(0, "skills/last30days/scripts") |
| 175 | from lib import github, schema |
| 176 | |
| 177 | candidate = schema.Candidate( |
| 178 | candidate_id="c-gh", |
| 179 | item_id="gh-1", |
| 180 | source="github", |
| 181 | title="repo mvanhorn/last30days-skill discussion", |
| 182 | url="https://github.com/mvanhorn/last30days-skill", |
| 183 | snippet="s", |
| 184 | subquery_labels=["primary"], |
| 185 | native_ranks={"primary:github": 1}, |
| 186 | local_relevance=0.9, |
| 187 | freshness=90, |
| 188 | engagement=10, |
| 189 | source_quality=0.5, |
| 190 | rrf_score=0.1, |
| 191 | final_score=90, |
| 192 | cluster_id="cl", |
| 193 | source_items=[], |
| 194 | metadata={}, |
| 195 | ) |
| 196 | enriched = github.apply_star_map( |
| 197 | [candidate], {"mvanhorn/last30days-skill": 51436} |
| 198 | ) |
| 199 | assert enriched == 1 |
| 200 | assert candidate.metadata["github_stars"]["mvanhorn/last30days-skill"] == 51436 |
| 201 |