| 1 | """Tests for scripts/lib/fanout.run_competitor_fanout.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import io |
| 6 | import threading |
| 7 | import time |
| 8 | import unittest |
| 9 | from contextlib import redirect_stderr |
| 10 | from unittest import mock |
| 11 | |
| 12 | from lib import fanout |
| 13 | |
| 14 | |
| 15 | def _fake_report(topic: str): |
| 16 | """Build a lightweight Report stand-in. Tests only check identity.""" |
| 17 | class _R: |
| 18 | pass |
| 19 | |
| 20 | r = _R() |
| 21 | r.topic = topic |
| 22 | return r |
| 23 | |
| 24 | |
| 25 | class FanoutOrchestratorTests(unittest.TestCase): |
| 26 | def test_main_plus_two_competitors_all_succeed(self): |
| 27 | def main_runner(): |
| 28 | return _fake_report("OpenAI") |
| 29 | |
| 30 | def comp_runner(entity): |
| 31 | return _fake_report(entity) |
| 32 | |
| 33 | err = io.StringIO() |
| 34 | with redirect_stderr(err): |
| 35 | results = fanout.run_competitor_fanout( |
| 36 | main_topic="OpenAI", |
| 37 | main_runner=main_runner, |
| 38 | competitors=["Anthropic", "xAI"], |
| 39 | competitor_runner=comp_runner, |
| 40 | ) |
| 41 | labels = [label for label, _ in results] |
| 42 | self.assertEqual(labels, ["OpenAI", "Anthropic", "xAI"]) |
| 43 | self.assertEqual(results[0][1].topic, "OpenAI") |
| 44 | self.assertEqual(results[1][1].topic, "Anthropic") |
| 45 | |
| 46 | def test_one_competitor_failure_degrades_gracefully(self): |
| 47 | def main_runner(): |
| 48 | return _fake_report("OpenAI") |
| 49 | |
| 50 | def comp_runner(entity): |
| 51 | if entity == "BrokenCo": |
| 52 | raise RuntimeError("upstream offline") |
| 53 | return _fake_report(entity) |
| 54 | |
| 55 | err = io.StringIO() |
| 56 | with redirect_stderr(err): |
| 57 | results = fanout.run_competitor_fanout( |
| 58 | main_topic="OpenAI", |
| 59 | main_runner=main_runner, |
| 60 | competitors=["Anthropic", "BrokenCo", "xAI"], |
| 61 | competitor_runner=comp_runner, |
| 62 | ) |
| 63 | labels = [label for label, _ in results] |
| 64 | self.assertEqual(labels, ["OpenAI", "Anthropic", "xAI"]) |
| 65 | self.assertIn("BrokenCo", err.getvalue()) |
| 66 | self.assertIn("upstream offline", err.getvalue()) |
| 67 | |
| 68 | def test_main_topic_failure_leaves_only_competitors(self): |
| 69 | def main_runner(): |
| 70 | raise RuntimeError("main exploded") |
| 71 | |
| 72 | def comp_runner(entity): |
| 73 | return _fake_report(entity) |
| 74 | |
| 75 | err = io.StringIO() |
| 76 | with redirect_stderr(err): |
| 77 | results = fanout.run_competitor_fanout( |
| 78 | main_topic="OpenAI", |
| 79 | main_runner=main_runner, |
| 80 | competitors=["Anthropic", "xAI"], |
| 81 | competitor_runner=comp_runner, |
| 82 | ) |
| 83 | labels = [label for label, _ in results] |
| 84 | self.assertEqual(labels, ["Anthropic", "xAI"]) |
| 85 | self.assertIn("main exploded", err.getvalue()) |
| 86 | |
| 87 | def test_empty_competitor_list_runs_only_main(self): |
| 88 | def main_runner(): |
| 89 | return _fake_report("OpenAI") |
| 90 | |
| 91 | def comp_runner(_entity): |
| 92 | raise AssertionError("should not be called when competitors=[]") |
| 93 | |
| 94 | err = io.StringIO() |
| 95 | with redirect_stderr(err): |
| 96 | results = fanout.run_competitor_fanout( |
| 97 | main_topic="OpenAI", |
| 98 | main_runner=main_runner, |
| 99 | competitors=[], |
| 100 | competitor_runner=comp_runner, |
| 101 | ) |
| 102 | self.assertEqual([label for label, _ in results], ["OpenAI"]) |
| 103 | |
| 104 | def test_fanout_clears_youtube_search_cache_before_parallel_work(self): |
| 105 | """Comparison mode must not inherit a prior run's ytsearch cache.""" |
| 106 | from lib import youtube_yt |
| 107 | |
| 108 | youtube_yt._search_cache[("prior", 8, "2026-01-01")] = {"items": []} |
| 109 | cleared = [] |
| 110 | |
| 111 | def main_runner(): |
| 112 | cleared.append("main" in youtube_yt._search_cache or len(youtube_yt._search_cache) == 0) |
| 113 | return _fake_report("OpenAI") |
| 114 | |
| 115 | with mock.patch.object( |
| 116 | youtube_yt, "reset_search_cache", wraps=youtube_yt.reset_search_cache |
| 117 | ) as reset_mock, redirect_stderr(io.StringIO()): |
| 118 | fanout.run_competitor_fanout( |
| 119 | main_topic="OpenAI", |
| 120 | main_runner=main_runner, |
| 121 | competitors=["Anthropic"], |
| 122 | competitor_runner=lambda e: _fake_report(e), |
| 123 | ) |
| 124 | reset_mock.assert_called() |
| 125 | self.assertNotIn(("prior", 8, "2026-01-01"), youtube_yt._search_cache) |
| 126 | |
| 127 | def test_sub_runs_execute_in_parallel(self): |
| 128 | """Wall clock should be closer to max(latency) than sum(latency).""" |
| 129 | delay = 0.2 |
| 130 | call_count = 3 # main + 2 competitors |
| 131 | |
| 132 | def make_runner(_label): |
| 133 | def runner(): |
| 134 | time.sleep(delay) |
| 135 | return _fake_report(_label) |
| 136 | return runner |
| 137 | |
| 138 | def comp_runner(entity): |
| 139 | return make_runner(entity)() |
| 140 | |
| 141 | start = time.monotonic() |
| 142 | with redirect_stderr(io.StringIO()): |
| 143 | results = fanout.run_competitor_fanout( |
| 144 | main_topic="OpenAI", |
| 145 | main_runner=make_runner("OpenAI"), |
| 146 | competitors=["Anthropic", "xAI"], |
| 147 | competitor_runner=comp_runner, |
| 148 | ) |
| 149 | elapsed = time.monotonic() - start |
| 150 | self.assertEqual(len(results), 3) |
| 151 | # Generous margin: parallel execution should finish well under |
| 152 | # sum(call_count * delay) == 0.6s. We accept anything under 0.5s. |
| 153 | self.assertLess( |
| 154 | elapsed, delay * call_count, |
| 155 | f"Expected parallel execution < {delay * call_count:.2f}s, " |
| 156 | f"got {elapsed:.2f}s (sub-runs likely serialized)", |
| 157 | ) |
| 158 | |
| 159 | def test_all_competitors_fail_leaves_main_only(self): |
| 160 | def main_runner(): |
| 161 | return _fake_report("OpenAI") |
| 162 | |
| 163 | def comp_runner(_entity): |
| 164 | raise RuntimeError("all offline") |
| 165 | |
| 166 | with redirect_stderr(io.StringIO()): |
| 167 | results = fanout.run_competitor_fanout( |
| 168 | main_topic="OpenAI", |
| 169 | main_runner=main_runner, |
| 170 | competitors=["A", "B", "C"], |
| 171 | competitor_runner=comp_runner, |
| 172 | ) |
| 173 | self.assertEqual([label for label, _ in results], ["OpenAI"]) |
| 174 | |
| 175 | if __name__ == "__main__": |
| 176 | unittest.main() |
| 177 |