| 1 | """Tests for shared run-context schema and fan-in validation contract. |
| 2 | |
| 3 | Covers acceptance criteria from issue #333: |
| 4 | - Shared run-context schema prevents independent wall-clock computation |
| 5 | - Fan-in validation contract: schema/checksum/window consistency, |
| 6 | deterministic ordering, duplicate handling, stale cache rejection, |
| 7 | source status metadata, required-vs-optional failure behavior |
| 8 | - Fixture-based byte-stability checks (same inputs → same output) |
| 9 | """ |
| 10 | |
| 11 | from __future__ import annotations |
| 12 | |
| 13 | import hashlib |
| 14 | import json |
| 15 | from datetime import UTC, datetime |
| 16 | from typing import Any |
| 17 | |
| 18 | from scripts.fan_in_validator import ( |
| 19 | detect_duplicate_repos, |
| 20 | detect_duplicate_urls, |
| 21 | run_full_validation, |
| 22 | validate_artifact_schema, |
| 23 | validate_checksum_integrity, |
| 24 | validate_deterministic_ordering, |
| 25 | validate_source_status, |
| 26 | validate_stale_artifacts, |
| 27 | validate_window_consistency, |
| 28 | verify_byte_stability, |
| 29 | ) |
| 30 | from scripts.run_context import ( |
| 31 | SCHEMA_VERSION, |
| 32 | RunContext, |
| 33 | build_run_context, |
| 34 | contexts_compatible, |
| 35 | validate_run_context, |
| 36 | ) |
| 37 | |
| 38 | # --- Fixtures --- |
| 39 | |
| 40 | WEEK = "2026-W24" |
| 41 | SINCE = datetime(2026, 6, 8, 0, 0, 0, tzinfo=UTC) |
| 42 | UNTIL = datetime(2026, 6, 15, 0, 0, 0, tzinfo=UTC) |
| 43 | NOW = datetime(2026, 6, 14, 12, 0, 0, tzinfo=UTC) |
| 44 | SOURCE_CHECKSUM = "a" * 64 |
| 45 | TOPIC_CHECKSUM = "b" * 64 |
| 46 | CODE_SHA = "c" * 64 |
| 47 | |
| 48 | |
| 49 | def _make_run_context(**overrides: Any) -> RunContext: |
| 50 | defaults = dict( |
| 51 | week=WEEK, |
| 52 | since=SINCE, |
| 53 | until=UNTIL, |
| 54 | source_config_checksum=SOURCE_CHECKSUM, |
| 55 | topic_config_checksum=TOPIC_CHECKSUM, |
| 56 | code_sha=CODE_SHA, |
| 57 | created_at=NOW, |
| 58 | ) |
| 59 | defaults.update(overrides) |
| 60 | return build_run_context(**defaults) |
| 61 | |
| 62 | |
| 63 | def _make_rss_artifact( |
| 64 | source_id: str = "techcrunch", |
| 65 | articles: list[dict[str, Any]] | None = None, |
| 66 | run_context: RunContext | None = None, |
| 67 | crawled_at: str | None = None, |
| 68 | status_success: bool = True, |
| 69 | ) -> dict[str, Any]: |
| 70 | ctx = run_context or _make_run_context() |
| 71 | arts = articles or [ |
| 72 | {"url": f"https://example.com/{source_id}/1", "title": "Article 1", "source": source_id}, |
| 73 | {"url": f"https://example.com/{source_id}/2", "title": "Article 2", "source": source_id}, |
| 74 | ] |
| 75 | artifact = { |
| 76 | "source_artifact_schema_version": "1", |
| 77 | "source_id": source_id, |
| 78 | "crawled_at": crawled_at or NOW.strftime("%Y-%m-%dT%H:%M:%SZ"), |
| 79 | "run_context": { |
| 80 | "week": ctx.week, |
| 81 | "since": ctx.since, |
| 82 | "until": ctx.until, |
| 83 | "crawl_window": {"since": ctx.since, "until": ctx.until}, |
| 84 | "source_config_checksum": ctx.source_config_checksum, |
| 85 | "schema_checksum": "1", |
| 86 | }, |
| 87 | "status": {"success": status_success, "error_message": "" if status_success else "timeout"}, |
| 88 | "metrics": { |
| 89 | "total_articles": len(arts), |
| 90 | "relevant_articles": len(arts), |
| 91 | "content_checksum": hashlib.sha256( |
| 92 | json.dumps(arts, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode( |
| 93 | "utf-8" |
| 94 | ) |
| 95 | ).hexdigest(), |
| 96 | }, |
| 97 | "articles": arts, |
| 98 | } |
| 99 | # Compute artifact_checksum |
| 100 | payload = { |
| 101 | "source_id": artifact["source_id"], |
| 102 | "run_context": artifact["run_context"], |
| 103 | "articles": artifact["articles"], |
| 104 | } |
| 105 | serialized = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False) |
| 106 | artifact["artifact_checksum"] = hashlib.sha256(serialized.encode("utf-8")).hexdigest() |
| 107 | return artifact |
| 108 | |
| 109 | |
| 110 | # --- Run Context Tests --- |
| 111 | |
| 112 | |
| 113 | class TestRunContext: |
| 114 | def test_build_run_context_deterministic_run_id(self): |
| 115 | ctx1 = _make_run_context() |
| 116 | ctx2 = _make_run_context() |
| 117 | assert ctx1.run_id == ctx2.run_id |
| 118 | |
| 119 | def test_build_run_context_different_inputs_different_id(self): |
| 120 | ctx1 = _make_run_context() |
| 121 | ctx2 = _make_run_context(source_config_checksum="d" * 64) |
| 122 | assert ctx1.run_id != ctx2.run_id |
| 123 | |
| 124 | def test_schema_version_is_set(self): |
| 125 | ctx = _make_run_context() |
| 126 | assert ctx.schema_version == SCHEMA_VERSION |
| 127 | |
| 128 | def test_run_context_serialization_roundtrip(self): |
| 129 | ctx = _make_run_context() |
| 130 | json_str = ctx.to_json() |
| 131 | restored = RunContext.from_json(json_str) |
| 132 | assert ctx == restored |
| 133 | |
| 134 | def test_validate_run_context_valid(self): |
| 135 | ctx = _make_run_context() |
| 136 | errors = validate_run_context(ctx) |
| 137 | assert errors == [] |
| 138 | |
| 139 | def test_validate_run_context_missing_field(self): |
| 140 | ctx = _make_run_context() |
| 141 | data = ctx.to_dict() |
| 142 | del data["week"] |
| 143 | errors = validate_run_context(data) |
| 144 | assert any("week" in e for e in errors) |
| 145 | |
| 146 | def test_validate_run_context_bad_week_format(self): |
| 147 | ctx = _make_run_context() |
| 148 | data = ctx.to_dict() |
| 149 | data["week"] = "2026-24" # Missing W prefix |
| 150 | errors = validate_run_context(data) |
| 151 | assert any("week" in e for e in errors) |
| 152 | |
| 153 | def test_validate_run_context_bad_timestamp(self): |
| 154 | ctx = _make_run_context() |
| 155 | data = ctx.to_dict() |
| 156 | data["since"] = "not-a-timestamp" |
| 157 | errors = validate_run_context(data) |
| 158 | assert any("since" in e for e in errors) |
| 159 | |
| 160 | def test_contexts_compatible_identical(self): |
| 161 | ctx = _make_run_context() |
| 162 | assert contexts_compatible(ctx, ctx) == [] |
| 163 | |
| 164 | def test_contexts_compatible_different_week(self): |
| 165 | ctx1 = _make_run_context() |
| 166 | ctx2 = _make_run_context(week="2026-W25") |
| 167 | mismatches = contexts_compatible(ctx1, ctx2) |
| 168 | assert any("week" in m for m in mismatches) |
| 169 | |
| 170 | def test_contexts_compatible_different_checksum(self): |
| 171 | ctx1 = _make_run_context() |
| 172 | ctx2 = _make_run_context(source_config_checksum="f" * 64) |
| 173 | mismatches = contexts_compatible(ctx1, ctx2) |
| 174 | assert len(mismatches) > 0 |
| 175 | |
| 176 | def test_prevents_wall_clock_computation(self): |
| 177 | """Run context enforces that legs cannot compute their own window.""" |
| 178 | ctx = _make_run_context() |
| 179 | # The since/until are fixed at build time, not computed from wall clock |
| 180 | assert ctx.since == "2026-06-08T00:00:00Z" |
| 181 | assert ctx.until == "2026-06-15T00:00:00Z" |
| 182 | # Even if built at a different time, same inputs yield same window |
| 183 | ctx2 = build_run_context( |
| 184 | week=WEEK, |
| 185 | since=SINCE, |
| 186 | until=UNTIL, |
| 187 | source_config_checksum=SOURCE_CHECKSUM, |
| 188 | topic_config_checksum=TOPIC_CHECKSUM, |
| 189 | code_sha=CODE_SHA, |
| 190 | created_at=datetime(2026, 6, 20, 0, 0, 0, tzinfo=UTC), |
| 191 | ) |
| 192 | assert ctx.since == ctx2.since |
| 193 | assert ctx.until == ctx2.until |
| 194 | |
| 195 | |
| 196 | # --- Fan-In Validation Tests --- |
| 197 | |
| 198 | |
| 199 | class TestFanInValidation: |
| 200 | def test_validate_artifact_schema_valid(self): |
| 201 | artifact = _make_rss_artifact() |
| 202 | errors = validate_artifact_schema(artifact, "1") |
| 203 | assert errors == [] |
| 204 | |
| 205 | def test_validate_artifact_schema_mismatch(self): |
| 206 | artifact = _make_rss_artifact() |
| 207 | errors = validate_artifact_schema(artifact, "2") |
| 208 | assert any("mismatch" in e for e in errors) |
| 209 | |
| 210 | def test_validate_checksum_integrity_valid(self): |
| 211 | artifact = _make_rss_artifact() |
| 212 | errors = validate_checksum_integrity(artifact) |
| 213 | assert errors == [] |
| 214 | |
| 215 | def test_validate_checksum_integrity_tampered(self): |
| 216 | artifact = _make_rss_artifact() |
| 217 | artifact["artifact_checksum"] = "0" * 64 |
| 218 | errors = validate_checksum_integrity(artifact) |
| 219 | assert len(errors) > 0 |
| 220 | |
| 221 | def test_window_consistency_valid(self): |
| 222 | ctx = _make_run_context() |
| 223 | artifacts = [_make_rss_artifact(run_context=ctx)] |
| 224 | errors = validate_window_consistency(artifacts, ctx) |
| 225 | assert errors == [] |
| 226 | |
| 227 | def test_window_consistency_mismatch(self): |
| 228 | ctx = _make_run_context() |
| 229 | artifact = _make_rss_artifact(run_context=ctx) |
| 230 | artifact["run_context"]["week"] = "2026-W99" |
| 231 | errors = validate_window_consistency([artifact], ctx) |
| 232 | assert len(errors) > 0 |
| 233 | |
| 234 | def test_deterministic_ordering_sorted(self): |
| 235 | articles = [ |
| 236 | {"source": "a", "url": "https://a.com/1"}, |
| 237 | {"source": "a", "url": "https://a.com/2"}, |
| 238 | {"source": "b", "url": "https://b.com/1"}, |
| 239 | ] |
| 240 | errors = validate_deterministic_ordering(articles) |
| 241 | assert errors == [] |
| 242 | |
| 243 | def test_deterministic_ordering_unsorted(self): |
| 244 | articles = [ |
| 245 | {"source": "b", "url": "https://b.com/1"}, |
| 246 | {"source": "a", "url": "https://a.com/1"}, |
| 247 | ] |
| 248 | errors = validate_deterministic_ordering(articles) |
| 249 | assert len(errors) > 0 |
| 250 | |
| 251 | def test_detect_duplicate_urls(self): |
| 252 | articles = [ |
| 253 | {"url": "https://example.com/1"}, |
| 254 | {"url": "https://example.com/1"}, |
| 255 | {"url": "https://example.com/2"}, |
| 256 | ] |
| 257 | dupes = detect_duplicate_urls(articles) |
| 258 | assert len(dupes) == 1 |
| 259 | |
| 260 | def test_detect_duplicate_urls_normalized(self): |
| 261 | articles = [ |
| 262 | {"url": "https://example.com/path?query=1"}, |
| 263 | {"url": "https://example.com/path?query=2"}, |
| 264 | ] |
| 265 | dupes = detect_duplicate_urls(articles) |
| 266 | # Same path, different query → treated as same after normalization |
| 267 | assert len(dupes) == 1 |
| 268 | |
| 269 | def test_detect_duplicate_repos(self): |
| 270 | repos = [ |
| 271 | {"full_name": "owner/repo1"}, |
| 272 | {"full_name": "owner/repo1"}, |
| 273 | {"full_name": "owner/repo2"}, |
| 274 | ] |
| 275 | dupes = detect_duplicate_repos(repos) |
| 276 | assert dupes == ["owner/repo1"] |
| 277 | |
| 278 | def test_stale_artifact_rejected(self): |
| 279 | artifact = _make_rss_artifact( |
| 280 | crawled_at="2026-06-12T00:00:00Z" # >24h before NOW |
| 281 | ) |
| 282 | stale = validate_stale_artifacts([artifact], reference_time=NOW) |
| 283 | assert len(stale) > 0 |
| 284 | |
| 285 | def test_fresh_artifact_accepted(self): |
| 286 | artifact = _make_rss_artifact(crawled_at=NOW.strftime("%Y-%m-%dT%H:%M:%SZ")) |
| 287 | stale = validate_stale_artifacts([artifact], reference_time=NOW) |
| 288 | assert stale == [] |
| 289 | |
| 290 | def test_source_status_required_missing(self): |
| 291 | artifacts = [_make_rss_artifact(source_id="techcrunch")] |
| 292 | errors, warnings = validate_source_status( |
| 293 | artifacts, required_sources=["techcrunch", "nvidia_blog"] |
| 294 | ) |
| 295 | assert any("nvidia_blog" in e for e in errors) |
| 296 | |
| 297 | def test_source_status_required_failed(self): |
| 298 | artifact = _make_rss_artifact(source_id="techcrunch", status_success=False) |
| 299 | errors, warnings = validate_source_status([artifact], required_sources=["techcrunch"]) |
| 300 | assert any("techcrunch" in e for e in errors) |
| 301 | |
| 302 | def test_source_status_optional_missing_is_warning(self): |
| 303 | artifacts = [_make_rss_artifact(source_id="techcrunch")] |
| 304 | errors, warnings = validate_source_status( |
| 305 | artifacts, |
| 306 | required_sources=["techcrunch"], |
| 307 | optional_sources=["huggingface"], |
| 308 | ) |
| 309 | assert errors == [] |
| 310 | assert any("huggingface" in w for w in warnings) |
| 311 | |
| 312 | |
| 313 | # --- Byte Stability Tests --- |
| 314 | |
| 315 | |
| 316 | class TestByteStability: |
| 317 | def test_same_inputs_same_output(self): |
| 318 | """Fixture check: same inputs produce byte-identical output.""" |
| 319 | output1 = { |
| 320 | "articles": [ |
| 321 | {"url": "https://a.com/1", "source": "a", "title": "A1"}, |
| 322 | {"url": "https://b.com/1", "source": "b", "title": "B1"}, |
| 323 | ], |
| 324 | "merged_at": "2026-06-14T12:00:00Z", |
| 325 | "checksum": "abc", |
| 326 | } |
| 327 | output2 = { |
| 328 | "articles": [ |
| 329 | {"url": "https://a.com/1", "source": "a", "title": "A1"}, |
| 330 | {"url": "https://b.com/1", "source": "b", "title": "B1"}, |
| 331 | ], |
| 332 | "merged_at": "2026-06-14T13:00:00Z", # Different timestamp |
| 333 | "checksum": "abc", |
| 334 | } |
| 335 | errors = verify_byte_stability(output1, output2) |
| 336 | assert errors == [] # Timestamps excluded |
| 337 | |
| 338 | def test_different_content_detected(self): |
| 339 | output1 = {"articles": [{"url": "https://a.com/1"}], "merged_at": "t1"} |
| 340 | output2 = {"articles": [{"url": "https://b.com/1"}], "merged_at": "t1"} |
| 341 | errors = verify_byte_stability(output1, output2) |
| 342 | assert len(errors) > 0 |
| 343 | |
| 344 | def test_deterministic_merge_fixture(self): |
| 345 | """Prove that merging the same artifacts twice yields identical output.""" |
| 346 | ctx = _make_run_context() |
| 347 | a1 = _make_rss_artifact(source_id="alpha", run_context=ctx) |
| 348 | a2 = _make_rss_artifact(source_id="beta", run_context=ctx) |
| 349 | |
| 350 | def merge(artifacts: list[dict[str, Any]]) -> dict[str, Any]: |
| 351 | sorted_arts = sorted(artifacts, key=lambda a: a["source_id"]) |
| 352 | all_articles = [] |
| 353 | for art in sorted_arts: |
| 354 | all_articles.extend(art.get("articles", [])) |
| 355 | # Deterministic sort |
| 356 | all_articles.sort(key=lambda a: (a.get("source", ""), a.get("url", ""))) |
| 357 | return { |
| 358 | "articles": all_articles, |
| 359 | "sources": [a["source_id"] for a in sorted_arts], |
| 360 | "checksum": hashlib.sha256( |
| 361 | json.dumps(all_articles, sort_keys=True).encode() |
| 362 | ).hexdigest(), |
| 363 | } |
| 364 | |
| 365 | result1 = merge([a1, a2]) |
| 366 | result2 = merge([a2, a1]) # Different input order |
| 367 | errors = verify_byte_stability(result1, result2) |
| 368 | assert errors == [], "Same artifacts in different order must produce identical output" |
| 369 | |
| 370 | |
| 371 | # --- Full Validation Integration Tests --- |
| 372 | |
| 373 | |
| 374 | class TestFullValidation: |
| 375 | def test_valid_artifacts_pass(self): |
| 376 | ctx = _make_run_context() |
| 377 | artifacts = [ |
| 378 | _make_rss_artifact(source_id="techcrunch", run_context=ctx), |
| 379 | _make_rss_artifact(source_id="nvidia_blog", run_context=ctx), |
| 380 | ] |
| 381 | result = run_full_validation( |
| 382 | artifacts, |
| 383 | ctx, |
| 384 | required_sources=["techcrunch", "nvidia_blog"], |
| 385 | expected_schema_version="1", |
| 386 | reference_time=NOW, |
| 387 | ) |
| 388 | assert result.valid |
| 389 | assert result.errors == [] |
| 390 | |
| 391 | def test_empty_artifacts_fail(self): |
| 392 | ctx = _make_run_context() |
| 393 | result = run_full_validation([], ctx) |
| 394 | assert not result.valid |
| 395 | assert "no artifacts" in result.errors[0] |
| 396 | |
| 397 | def test_schema_mismatch_fails(self): |
| 398 | ctx = _make_run_context() |
| 399 | artifact = _make_rss_artifact(run_context=ctx) |
| 400 | result = run_full_validation([artifact], ctx, expected_schema_version="99") |
| 401 | assert not result.valid |
| 402 | |
| 403 | def test_missing_required_source_fails(self): |
| 404 | ctx = _make_run_context() |
| 405 | artifact = _make_rss_artifact(source_id="techcrunch", run_context=ctx) |
| 406 | result = run_full_validation( |
| 407 | [artifact], |
| 408 | ctx, |
| 409 | required_sources=["techcrunch", "missing_source"], |
| 410 | expected_schema_version="1", |
| 411 | reference_time=NOW, |
| 412 | ) |
| 413 | assert not result.valid |
| 414 | assert "missing_source" in str(result.errors) |
| 415 | |
| 416 | def test_optional_missing_source_warns(self): |
| 417 | ctx = _make_run_context() |
| 418 | artifact = _make_rss_artifact(source_id="techcrunch", run_context=ctx) |
| 419 | result = run_full_validation( |
| 420 | [artifact], |
| 421 | ctx, |
| 422 | required_sources=["techcrunch"], |
| 423 | optional_sources=["huggingface"], |
| 424 | expected_schema_version="1", |
| 425 | reference_time=NOW, |
| 426 | ) |
| 427 | assert result.valid # Optional missing doesn't fail |
| 428 | assert "huggingface" in str(result.warnings) |