| 1 | """Tests for scripts/hindsight_validation.py""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | from datetime import datetime, timedelta |
| 7 | from pathlib import Path |
| 8 | from typing import Any |
| 9 | |
| 10 | import pytest |
| 11 | |
| 12 | from scripts.hindsight_validation import ( |
| 13 | build_repo_set, |
| 14 | build_repo_stars, |
| 15 | current_iso_week, |
| 16 | generate_scorecard, |
| 17 | is_old_enough, |
| 18 | iso_week_to_date, |
| 19 | load_predictions, |
| 20 | run_validation, |
| 21 | save_predictions, |
| 22 | validate_breakout_candidate, |
| 23 | validate_declining_signal, |
| 24 | validate_momentum_shift, |
| 25 | validate_prediction, |
| 26 | validate_rising_star, |
| 27 | week_offset, |
| 28 | ) |
| 29 | |
| 30 | |
| 31 | def _old_week(weeks_back: int = 5) -> str: |
| 32 | """Return a week string N weeks in the past.""" |
| 33 | dt = datetime.now() - timedelta(weeks=weeks_back) |
| 34 | cal = dt.isocalendar() |
| 35 | return f"{cal[0]}-W{cal[1]:02d}" |
| 36 | |
| 37 | |
| 38 | def _make_raw_data(repos: list[dict[str, Any]], section: str = "trending_repos") -> dict[str, Any]: |
| 39 | return { |
| 40 | "new_repos": repos if section == "new_repos" else [], |
| 41 | "trending_repos": repos if section == "trending_repos" else [], |
| 42 | } |
| 43 | |
| 44 | |
| 45 | def _write_raw(raw_dir: Path, week: str, data: dict[str, Any]) -> None: |
| 46 | raw_dir.mkdir(parents=True, exist_ok=True) |
| 47 | with open(raw_dir / f"{week}.json", "w") as f: |
| 48 | json.dump(data, f) |
| 49 | |
| 50 | |
| 51 | def _write_predictions(metrics_dir: Path, predictions: list[dict[str, Any]]) -> None: |
| 52 | metrics_dir.mkdir(parents=True, exist_ok=True) |
| 53 | with open(metrics_dir / "predictions.jsonl", "w") as f: |
| 54 | for p in predictions: |
| 55 | f.write(json.dumps(p) + "\n") |
| 56 | |
| 57 | |
| 58 | class TestIsoWeekConversion: |
| 59 | def test_round_trip(self): |
| 60 | week = "2026-W21" |
| 61 | dt = iso_week_to_date(week) |
| 62 | assert dt.year == 2026 |
| 63 | |
| 64 | def test_week_offset_forward(self): |
| 65 | assert week_offset("2026-W10", 4) == "2026-W14" |
| 66 | |
| 67 | def test_week_offset_backward(self): |
| 68 | assert week_offset("2026-W10", -2) == "2026-W08" |
| 69 | |
| 70 | def test_current_iso_week_format(self): |
| 71 | week = current_iso_week() |
| 72 | assert len(week.split("-W")) == 2 |
| 73 | |
| 74 | |
| 75 | class TestLoadSavePredictions: |
| 76 | def test_load_missing_file(self, tmp_path: Path): |
| 77 | result = load_predictions(tmp_path / "nonexistent.jsonl") |
| 78 | assert result == [] |
| 79 | |
| 80 | def test_round_trip(self, tmp_path: Path): |
| 81 | preds = [ |
| 82 | {"week": "2026-W21", "repo": "org/a", "prediction": "rising_star", |
| 83 | "confidence": 0.7, "reason": "test", "validated": None}, |
| 84 | ] |
| 85 | path = tmp_path / "predictions.jsonl" |
| 86 | save_predictions(preds, path) |
| 87 | loaded = load_predictions(path) |
| 88 | assert loaded == preds |
| 89 | |
| 90 | def test_empty_lines_ignored(self, tmp_path: Path): |
| 91 | path = tmp_path / "predictions.jsonl" |
| 92 | path.write_text('{"week":"2026-W01","repo":"x/y","prediction":"rising_star","confidence":0.5,"reason":"r","validated":null}\n\n') |
| 93 | loaded = load_predictions(path) |
| 94 | assert len(loaded) == 1 |
| 95 | |
| 96 | |
| 97 | class TestBuildRepoHelpers: |
| 98 | def test_build_repo_stars(self): |
| 99 | raw = {"trending_repos": [{"full_name": "a/b", "stars": 100}], "new_repos": []} |
| 100 | assert build_repo_stars(raw) == {"a/b": 100} |
| 101 | |
| 102 | def test_build_repo_set(self): |
| 103 | raw = {"trending_repos": [{"full_name": "a/b", "stars": 100}], |
| 104 | "new_repos": [{"full_name": "c/d", "stars": 50}]} |
| 105 | assert build_repo_set(raw) == {"a/b", "c/d"} |
| 106 | |
| 107 | |
| 108 | class TestValidateRisingStar: |
| 109 | def test_growth_above_threshold(self, tmp_path: Path): |
| 110 | pred_week = _old_week(5) |
| 111 | later_week = week_offset(pred_week, 4) |
| 112 | _write_raw(tmp_path, later_week, _make_raw_data( |
| 113 | [{"full_name": "org/repo", "stars": 150}])) |
| 114 | result = validate_rising_star("org/repo", 100, tmp_path, pred_week, 4) |
| 115 | assert result is True |
| 116 | |
| 117 | def test_growth_below_threshold(self, tmp_path: Path): |
| 118 | pred_week = _old_week(5) |
| 119 | later_week = week_offset(pred_week, 4) |
| 120 | _write_raw(tmp_path, later_week, _make_raw_data( |
| 121 | [{"full_name": "org/repo", "stars": 110}])) |
| 122 | result = validate_rising_star("org/repo", 100, tmp_path, pred_week, 4) |
| 123 | assert result is False |
| 124 | |
| 125 | def test_no_data_returns_none(self, tmp_path: Path): |
| 126 | pred_week = _old_week(5) |
| 127 | result = validate_rising_star("org/repo", 100, tmp_path, pred_week, 4) |
| 128 | assert result is None |
| 129 | |
| 130 | def test_zero_stars_baseline(self, tmp_path: Path): |
| 131 | pred_week = _old_week(5) |
| 132 | later_week = week_offset(pred_week, 4) |
| 133 | _write_raw(tmp_path, later_week, _make_raw_data( |
| 134 | [{"full_name": "org/repo", "stars": 10}])) |
| 135 | result = validate_rising_star("org/repo", 0, tmp_path, pred_week, 4) |
| 136 | assert result is True |
| 137 | |
| 138 | |
| 139 | class TestValidateBreakoutCandidate: |
| 140 | def test_found_in_trending(self, tmp_path: Path): |
| 141 | pred_week = _old_week(5) |
| 142 | w2 = week_offset(pred_week, 2) |
| 143 | _write_raw(tmp_path, w2, _make_raw_data( |
| 144 | [{"full_name": "org/repo", "stars": 200}], "trending_repos")) |
| 145 | result = validate_breakout_candidate("org/repo", tmp_path, pred_week, 4) |
| 146 | assert result is True |
| 147 | |
| 148 | def test_not_found(self, tmp_path: Path): |
| 149 | pred_week = _old_week(5) |
| 150 | w1 = week_offset(pred_week, 1) |
| 151 | _write_raw(tmp_path, w1, _make_raw_data( |
| 152 | [{"full_name": "other/repo", "stars": 200}], "trending_repos")) |
| 153 | result = validate_breakout_candidate("org/repo", tmp_path, pred_week, 4) |
| 154 | assert result is False |
| 155 | |
| 156 | |
| 157 | class TestValidateMomentumShift: |
| 158 | def test_continued_growth(self, tmp_path: Path): |
| 159 | pred_week = _old_week(5) |
| 160 | w2 = week_offset(pred_week, 2) |
| 161 | _write_raw(tmp_path, w2, _make_raw_data( |
| 162 | [{"full_name": "org/repo", "stars": 1200}])) |
| 163 | result = validate_momentum_shift("org/repo", 1000, tmp_path, pred_week, 4) |
| 164 | assert result is True |
| 165 | |
| 166 | def test_declined(self, tmp_path: Path): |
| 167 | pred_week = _old_week(5) |
| 168 | w2 = week_offset(pred_week, 2) |
| 169 | _write_raw(tmp_path, w2, _make_raw_data( |
| 170 | [{"full_name": "org/repo", "stars": 800}])) |
| 171 | result = validate_momentum_shift("org/repo", 1000, tmp_path, pred_week, 4) |
| 172 | assert result is False |
| 173 | |
| 174 | def test_no_data(self, tmp_path: Path): |
| 175 | pred_week = _old_week(5) |
| 176 | result = validate_momentum_shift("org/repo", 1000, tmp_path, pred_week, 4) |
| 177 | assert result is None |
| 178 | |
| 179 | |
| 180 | class TestValidateDecliningSignal: |
| 181 | def test_repo_disappeared(self, tmp_path: Path): |
| 182 | pred_week = _old_week(5) |
| 183 | for i in range(1, 5): |
| 184 | w = week_offset(pred_week, i) |
| 185 | _write_raw(tmp_path, w, _make_raw_data( |
| 186 | [{"full_name": "other/repo", "stars": 50}])) |
| 187 | result = validate_declining_signal("org/repo", tmp_path, pred_week, 4) |
| 188 | assert result is True |
| 189 | |
| 190 | def test_repo_still_present(self, tmp_path: Path): |
| 191 | pred_week = _old_week(5) |
| 192 | for i in range(1, 5): |
| 193 | w = week_offset(pred_week, i) |
| 194 | _write_raw(tmp_path, w, _make_raw_data( |
| 195 | [{"full_name": "org/repo", "stars": 50}])) |
| 196 | result = validate_declining_signal("org/repo", tmp_path, pred_week, 4) |
| 197 | assert result is False |
| 198 | |
| 199 | def test_no_data(self, tmp_path: Path): |
| 200 | pred_week = _old_week(5) |
| 201 | result = validate_declining_signal("org/repo", tmp_path, pred_week, 4) |
| 202 | assert result is None |
| 203 | |
| 204 | |
| 205 | class TestIsOldEnough: |
| 206 | def test_old_prediction(self): |
| 207 | week = _old_week(6) |
| 208 | assert is_old_enough(week, 4) is True |
| 209 | |
| 210 | def test_recent_prediction(self): |
| 211 | week = current_iso_week() |
| 212 | assert is_old_enough(week, 4) is False |
| 213 | |
| 214 | def test_invalid_week(self): |
| 215 | assert is_old_enough("", 4) is False |
| 216 | assert is_old_enough("not-a-week", 4) is False |
| 217 | |
| 218 | |
| 219 | class TestGenerateScorecard: |
| 220 | def test_empty(self): |
| 221 | sc = generate_scorecard([]) |
| 222 | assert sc["total_validated"] == 0 |
| 223 | assert sc["accuracy"] == 0.0 |
| 224 | |
| 225 | def test_mixed_results(self): |
| 226 | preds = [ |
| 227 | {"prediction": "rising_star", "validated": True}, |
| 228 | {"prediction": "rising_star", "validated": False}, |
| 229 | {"prediction": "momentum_shift", "validated": True}, |
| 230 | {"prediction": "rising_star", "validated": None}, |
| 231 | ] |
| 232 | sc = generate_scorecard(preds) |
| 233 | assert sc["total_validated"] == 3 |
| 234 | assert sc["correct"] == 2 |
| 235 | assert sc["incorrect"] == 1 |
| 236 | assert sc["accuracy"] == pytest.approx(2 / 3, abs=0.001) |
| 237 | assert sc["by_type"]["rising_star"]["total"] == 2 |
| 238 | assert sc["by_type"]["rising_star"]["correct"] == 1 |
| 239 | assert sc["by_type"]["momentum_shift"]["total"] == 1 |
| 240 | |
| 241 | |
| 242 | class TestValidatePrediction: |
| 243 | def test_unknown_type(self, tmp_path: Path): |
| 244 | pred = {"week": _old_week(5), "repo": "org/repo", "prediction": "unknown_type"} |
| 245 | result = validate_prediction(pred, tmp_path, 4) |
| 246 | assert result is None |
| 247 | |
| 248 | def test_missing_repo(self, tmp_path: Path): |
| 249 | pred = {"week": _old_week(5), "repo": "", "prediction": "rising_star"} |
| 250 | result = validate_prediction(pred, tmp_path, 4) |
| 251 | assert result is None |
| 252 | |
| 253 | def test_emerging_topic_uses_declining_logic(self, tmp_path: Path): |
| 254 | pred_week = _old_week(5) |
| 255 | pred = {"week": pred_week, "repo": "org/repo", "prediction": "emerging_topic"} |
| 256 | for i in range(1, 5): |
| 257 | w = week_offset(pred_week, i) |
| 258 | _write_raw(tmp_path, w, _make_raw_data( |
| 259 | [{"full_name": "other/repo", "stars": 50}])) |
| 260 | result = validate_prediction(pred, tmp_path, 4) |
| 261 | assert result is True |
| 262 | |
| 263 | |
| 264 | class TestRunValidation: |
| 265 | def test_full_run(self, tmp_path: Path): |
| 266 | pred_week = _old_week(5) |
| 267 | raw = tmp_path / "raw" |
| 268 | metrics = tmp_path / "metrics" |
| 269 | |
| 270 | _write_raw(raw, pred_week, _make_raw_data( |
| 271 | [{"full_name": "org/star", "stars": 100}])) |
| 272 | later = week_offset(pred_week, 4) |
| 273 | _write_raw(raw, later, _make_raw_data( |
| 274 | [{"full_name": "org/star", "stars": 200}])) |
| 275 | |
| 276 | preds = [ |
| 277 | {"week": pred_week, "repo": "org/star", "prediction": "rising_star", |
| 278 | "confidence": 0.7, "reason": "test", "validated": None}, |
| 279 | ] |
| 280 | _write_predictions(metrics, preds) |
| 281 | |
| 282 | scorecard = run_validation(topic_id=None, weeks_ago=4, data_dir=str(tmp_path)) |
| 283 | |
| 284 | assert scorecard["total_validated"] == 1 |
| 285 | assert scorecard["correct"] == 1 |
| 286 | assert scorecard["accuracy"] == 1.0 |
| 287 | |
| 288 | updated = load_predictions(metrics / "predictions.jsonl") |
| 289 | assert updated[0]["validated"] is True |
| 290 | |
| 291 | scorecards = list((metrics / "scorecards").glob("*-scorecard.json")) |
| 292 | assert len(scorecards) == 1 |
| 293 | |
| 294 | def test_skips_recent_predictions(self, tmp_path: Path): |
| 295 | raw = tmp_path / "raw" |
| 296 | metrics = tmp_path / "metrics" |
| 297 | raw.mkdir(parents=True, exist_ok=True) |
| 298 | |
| 299 | recent_week = current_iso_week() |
| 300 | preds = [ |
| 301 | {"week": recent_week, "repo": "org/new", "prediction": "rising_star", |
| 302 | "confidence": 0.7, "reason": "too new", "validated": None}, |
| 303 | ] |
| 304 | _write_predictions(metrics, preds) |
| 305 | |
| 306 | scorecard = run_validation(topic_id=None, weeks_ago=4, data_dir=str(tmp_path)) |
| 307 | assert scorecard["total_validated"] == 0 |
| 308 | |
| 309 | updated = load_predictions(metrics / "predictions.jsonl") |
| 310 | assert updated[0]["validated"] is None |
| 311 | |
| 312 | def test_empty_predictions(self, tmp_path: Path): |
| 313 | metrics = tmp_path / "metrics" |
| 314 | metrics.mkdir(parents=True, exist_ok=True) |
| 315 | scorecard = run_validation(topic_id=None, weeks_ago=4, data_dir=str(tmp_path)) |
| 316 | assert scorecard["total_validated"] == 0 |
| 317 | |
| 318 | def test_topic_path_resolution(self, tmp_path: Path): |
| 319 | pred_week = _old_week(5) |
| 320 | topic = "ai-ml" |
| 321 | raw = tmp_path / "raw" / topic |
| 322 | metrics = tmp_path / "metrics" / topic |
| 323 | |
| 324 | _write_raw(raw, pred_week, _make_raw_data( |
| 325 | [{"full_name": "org/ai", "stars": 100}])) |
| 326 | later = week_offset(pred_week, 4) |
| 327 | _write_raw(raw, later, _make_raw_data( |
| 328 | [{"full_name": "org/ai", "stars": 130}])) |
| 329 | |
| 330 | preds = [ |
| 331 | {"week": pred_week, "repo": "org/ai", "prediction": "rising_star", |
| 332 | "confidence": 0.6, "reason": "test", "validated": None}, |
| 333 | ] |
| 334 | _write_predictions(metrics, preds) |
| 335 | |
| 336 | scorecard = run_validation(topic_id=topic, weeks_ago=4, data_dir=str(tmp_path)) |
| 337 | assert scorecard["total_validated"] == 1 |
| 338 | assert scorecard["correct"] == 1 |
| 339 | |
| 340 | def test_already_validated_skipped(self, tmp_path: Path): |
| 341 | raw = tmp_path / "raw" |
| 342 | metrics = tmp_path / "metrics" |
| 343 | raw.mkdir(parents=True, exist_ok=True) |
| 344 | |
| 345 | pred_week = _old_week(5) |
| 346 | preds = [ |
| 347 | {"week": pred_week, "repo": "org/done", "prediction": "rising_star", |
| 348 | "confidence": 0.7, "reason": "already done", "validated": True}, |
| 349 | ] |
| 350 | _write_predictions(metrics, preds) |
| 351 | |
| 352 | scorecard = run_validation(topic_id=None, weeks_ago=4, data_dir=str(tmp_path)) |
| 353 | assert scorecard["total_validated"] == 1 |
| 354 | assert scorecard["correct"] == 1 |