"""Tests for harness/orchestrator.run_day with fully faked I/O.""" import json from dataclasses import dataclass from datetime import datetime, timedelta, timezone from harness import orchestrator from harness.genome import VideoArtifact, MANIFEST_FILENAME from harness.scoreboard import ScoreRow def _recent_date(days_ago: int = 5) -> str: return (datetime.now(timezone.utc) - timedelta(days=days_ago)).date().isoformat() @dataclass class FakePub: video_id: str upload_date: str def _make_variants(tmp_path, ids): root = tmp_path / "variants" for vid in ids: d = root / vid d.mkdir(parents=True) (d / MANIFEST_FILENAME).write_text(json.dumps({"variant_id": vid, "genome": {"k": vid}}), encoding="utf-8") return root def _artifact(i=1): return VideoArtifact(video_path=f"/tmp/v{i}.mp4", title=f"title {i}", description="desc", music_name="m", music_attribution="attr") def test_run_day_allocates_produces_publishes_attributes(tmp_path): root = _make_variants(tmp_path, ["A", "B"]) rows = [ScoreRow("v1", _recent_date(5), "A", "h", "seed", 90, 0.9, 9.0)] # A scored, B newborn recorded = [] produce_calls = [] def produce(manifest, n): produce_calls.append((manifest.variant_id, n)) return [_artifact(i) for i in range(n)] def record(**kw): recorded.append(kw) counter = {"n": 0} def publish(manifest, artifact): counter["n"] += 1 return FakePub(video_id=f"vid{counter['n']}", upload_date="2026-06-18") report = orchestrator.run_day( scoreboard_rows=rows, produce=produce, publish=publish, record_attribution=record, budget=3, variants_dir=root, ) assert report.slots == {"A": 2, "B": 1} # A scored → soaks taper; B newborn → floor of 1 assert report.published_count == 3 assert ("A", 2) in produce_calls and ("B", 1) in produce_calls # every published video was attributed to a real variant + its genome hash assert len(recorded) == 3 assert all(r["variant_id"] in {"A", "B"} and r["genome_hash"] for r in recorded) def test_run_day_isolates_publish_failure(tmp_path): root = _make_variants(tmp_path, ["A"]) rows = [ScoreRow("v1", _recent_date(5), "A", "h", "seed", 90, 0.9, 9.0)] def publish(manifest, artifact): raise RuntimeError("youtube down") report = orchestrator.run_day( scoreboard_rows=rows, produce=lambda m, n: [_artifact(i) for i in range(n)], publish=publish, record_attribution=lambda **kw: None, budget=2, variants_dir=root, ) assert report.published_count == 0 assert any("publish" in e for e in report.errors) def test_run_day_isolates_produce_failure(tmp_path): root = _make_variants(tmp_path, ["A", "B"]) rows = [] # both newborns → floor 1 each def produce(manifest, n): if manifest.variant_id == "A": raise RuntimeError("render exploded") return [_artifact(1)] report = orchestrator.run_day( scoreboard_rows=rows, produce=produce, publish=lambda m, a: FakePub("vidB", "2026-06-18"), record_attribution=lambda **kw: None, budget=2, variants_dir=root, ) # A's failure is isolated; B still publishes. assert report.published_count == 1 assert any("A: produce" in e for e in report.errors) def test_run_day_empty_produce_is_reported(tmp_path): root = _make_variants(tmp_path, ["A"]) report = orchestrator.run_day( scoreboard_rows=[], produce=lambda m, n: [], publish=lambda m, a: FakePub("x", "y"), record_attribution=lambda **kw: None, budget=2, variants_dir=root, ) assert report.published_count == 0 assert any("no videos rendered" in e for e in report.errors) def test_run_day_parallel_produce_publishes_all(tmp_path): import threading root = _make_variants(tmp_path, ["A", "B"]) seen: list[str] = [] lock = threading.Lock() def produce(manifest, n): with lock: seen.append(manifest.variant_id) return [_artifact(1)] report = orchestrator.run_day( scoreboard_rows=[], # both newborns → 1 slot each produce=produce, publish=lambda m, a: FakePub("vid", "2026-06-18"), record_attribution=lambda **kw: None, budget=2, variants_dir=root, max_workers=2, ) assert set(seen) == {"A", "B"} # both variants produced assert report.published_count == 2 # publishing stays correct under parallel produce def test_run_day_tracks_extinction_streak(tmp_path): root = _make_variants(tmp_path, ["A", "B"]) rows = [ScoreRow("v1", _recent_date(5), "A", "h", "seed", 90, 0.9, 9.0)] # A scored, B newborn saved: dict[str, int] = {} def save(streak): saved.clear() saved.update(streak) report = orchestrator.run_day( scoreboard_rows=rows, produce=lambda m, n: [_artifact(1)], publish=lambda m, a: FakePub("vid", "2026-06-18"), record_attribution=lambda **kw: None, budget=1, variants_dir=root, # budget 1 → newborn B takes the floor slot, A starves load_streak=lambda: {"A": 11}, save_streak=save, extinction_k=12, ) assert report.slots == {"A": 0, "B": 1} # A got 0 slots this run assert saved["A"] == 12 and saved["B"] == 0 # A's zero-streak ticks to 12; B resets assert report.extinct == ["A"] # crossed the threshold → flagged class _Readability: def __init__(self, score, issues=()): self.score = score self.issues = list(issues) @property def assessed(self): return self.score >= 0.0 def test_run_day_skips_unreadable_and_records_score(tmp_path): root = _make_variants(tmp_path, ["A"]) recorded = [] report = orchestrator.run_day( scoreboard_rows=[], produce=lambda m, n: [_artifact(i) for i in range(n)], # 1 newborn → 1 slot → 1 video publish=lambda m, a: FakePub("vidA", "2026-06-18"), record_attribution=lambda **kw: recorded.append(kw), budget=1, variants_dir=root, assess_readability=lambda art: _Readability(0.1, ["text clipped"]), readability_floor=0.35, ) assert report.published_count == 0 # skipped: 0.1 < floor 0.35 assert report.unreadable and "readability 0.10" in report.unreadable[0] assert recorded == [] # nothing published → nothing attributed def test_run_day_publishes_readable_and_carries_readability(tmp_path): root = _make_variants(tmp_path, ["A"]) recorded = [] report = orchestrator.run_day( scoreboard_rows=[], produce=lambda m, n: [_artifact(1)], publish=lambda m, a: FakePub("vidA", "2026-06-18"), record_attribution=lambda **kw: recorded.append(kw), budget=1, variants_dir=root, assess_readability=lambda art: _Readability(0.9, []), readability_floor=0.35, ) assert report.published_count == 1 assert recorded[0]["readability"] == 0.9 # score rode into attribution assert report.published[0].readability == 0.9 def test_run_day_readability_fails_open_on_not_assessed(tmp_path): root = _make_variants(tmp_path, ["A"]) report = orchestrator.run_day( scoreboard_rows=[], produce=lambda m, n: [_artifact(1)], publish=lambda m, a: FakePub("vidA", "2026-06-18"), record_attribution=lambda **kw: None, budget=1, variants_dir=root, assess_readability=lambda art: _Readability(-1.0), # not assessed → must still publish readability_floor=0.99, ) assert report.published_count == 1 # non-assessment never blocks def test_run_day_no_variants(tmp_path): root = tmp_path / "empty" root.mkdir() report = orchestrator.run_day( scoreboard_rows=[], produce=lambda m, n: [], publish=lambda m, a: FakePub("", ""), record_attribution=lambda **kw: None, budget=3, variants_dir=root, ) assert report.errors == ["no living variants"]