| 1 | """CLI parsing and validation for --competitors / --competitors-list.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import io |
| 6 | import sys |
| 7 | import unittest |
| 8 | from contextlib import redirect_stderr |
| 9 | from unittest import mock |
| 10 | |
| 11 | import last30days as cli |
| 12 | from lib import fanout |
| 13 | |
| 14 | |
| 15 | def _fake_report(topic: str): |
| 16 | """Duck-typed Report stand-in; the guard runs before any field is read.""" |
| 17 | return type("R", (), {"topic": topic, "warnings": []})() |
| 18 | |
| 19 | |
| 20 | def _parse(*argv: str): |
| 21 | parser = cli.build_parser() |
| 22 | args, _extra = parser.parse_known_args(argv) |
| 23 | return args |
| 24 | |
| 25 | |
| 26 | class CompetitorsCliTests(unittest.TestCase): |
| 27 | def test_flag_absent_returns_disabled(self): |
| 28 | args = _parse("Kanye West") |
| 29 | enabled, count, explicit = cli.resolve_competitors_args(args) |
| 30 | self.assertFalse(enabled) |
| 31 | self.assertEqual(count, 0) |
| 32 | self.assertEqual(explicit, []) |
| 33 | |
| 34 | def test_bare_flag_defaults_to_two(self): |
| 35 | args = _parse("Kanye West", "--competitors") |
| 36 | enabled, count, explicit = cli.resolve_competitors_args(args) |
| 37 | self.assertTrue(enabled) |
| 38 | self.assertEqual(count, 2) |
| 39 | self.assertEqual(explicit, []) |
| 40 | |
| 41 | def test_explicit_three_still_supported(self): |
| 42 | args = _parse("OpenAI", "--competitors", "3") |
| 43 | enabled, count, _explicit = cli.resolve_competitors_args(args) |
| 44 | self.assertTrue(enabled) |
| 45 | self.assertEqual(count, 3) |
| 46 | |
| 47 | def test_explicit_count(self): |
| 48 | args = _parse("OpenAI", "--competitors", "4") |
| 49 | enabled, count, explicit = cli.resolve_competitors_args(args) |
| 50 | self.assertTrue(enabled) |
| 51 | self.assertEqual(count, 4) |
| 52 | self.assertEqual(explicit, []) |
| 53 | |
| 54 | def test_explicit_list_preferred_over_discovery(self): |
| 55 | args = _parse( |
| 56 | "OpenAI", |
| 57 | "--competitors", |
| 58 | "--competitors-list", |
| 59 | "Anthropic,xAI,Google Gemini", |
| 60 | ) |
| 61 | enabled, count, explicit = cli.resolve_competitors_args(args) |
| 62 | self.assertTrue(enabled) |
| 63 | self.assertEqual(count, 3) |
| 64 | self.assertEqual(explicit, ["Anthropic", "xAI", "Google Gemini"]) |
| 65 | |
| 66 | def test_explicit_list_without_flag_implies_enabled(self): |
| 67 | args = _parse("OpenAI", "--competitors-list", "Anthropic,xAI") |
| 68 | enabled, count, explicit = cli.resolve_competitors_args(args) |
| 69 | self.assertTrue(enabled) |
| 70 | self.assertEqual(count, 2) |
| 71 | self.assertEqual(explicit, ["Anthropic", "xAI"]) |
| 72 | |
| 73 | def test_list_whitespace_normalized(self): |
| 74 | args = _parse("OpenAI", "--competitors-list", " Anthropic , xAI , Gemini ") |
| 75 | _enabled, count, explicit = cli.resolve_competitors_args(args) |
| 76 | self.assertEqual(count, 3) |
| 77 | self.assertEqual(explicit, ["Anthropic", "xAI", "Gemini"]) |
| 78 | |
| 79 | def test_zero_count_rejected(self): |
| 80 | args = _parse("Topic", "--competitors", "0") |
| 81 | with self.assertRaises(SystemExit) as cm, redirect_stderr(io.StringIO()) as err: |
| 82 | cli.resolve_competitors_args(args) |
| 83 | self.assertEqual(cm.exception.code, 2) |
| 84 | self.assertIn("--competitors must be >= 1", err.getvalue()) |
| 85 | |
| 86 | def test_negative_count_rejected(self): |
| 87 | args = _parse("Topic", "--competitors", "-1") |
| 88 | with self.assertRaises(SystemExit), redirect_stderr(io.StringIO()): |
| 89 | cli.resolve_competitors_args(args) |
| 90 | |
| 91 | def test_over_max_count_clamps_with_warning(self): |
| 92 | args = _parse("Topic", "--competitors", "99") |
| 93 | err = io.StringIO() |
| 94 | with redirect_stderr(err): |
| 95 | enabled, count, explicit = cli.resolve_competitors_args(args) |
| 96 | self.assertTrue(enabled) |
| 97 | self.assertEqual(count, cli.COMPETITORS_MAX) |
| 98 | self.assertEqual(explicit, []) |
| 99 | self.assertIn("clamping", err.getvalue()) |
| 100 | |
| 101 | def test_overlong_list_clamps_with_warning(self): |
| 102 | args = _parse( |
| 103 | "Topic", |
| 104 | "--competitors-list", |
| 105 | "A,B,C,D,E,F,G,H", |
| 106 | ) |
| 107 | err = io.StringIO() |
| 108 | with redirect_stderr(err): |
| 109 | enabled, count, explicit = cli.resolve_competitors_args(args) |
| 110 | self.assertTrue(enabled) |
| 111 | self.assertEqual(count, cli.COMPETITORS_MAX) |
| 112 | self.assertEqual(len(explicit), cli.COMPETITORS_MAX) |
| 113 | self.assertIn("clamping to", err.getvalue()) |
| 114 | |
| 115 | def test_list_count_mismatch_warns(self): |
| 116 | args = _parse( |
| 117 | "Topic", |
| 118 | "--competitors", |
| 119 | "5", |
| 120 | "--competitors-list", |
| 121 | "A,B", |
| 122 | ) |
| 123 | err = io.StringIO() |
| 124 | with redirect_stderr(err): |
| 125 | enabled, count, explicit = cli.resolve_competitors_args(args) |
| 126 | self.assertTrue(enabled) |
| 127 | self.assertEqual(count, 2) |
| 128 | self.assertEqual(explicit, ["A", "B"]) |
| 129 | self.assertIn("--competitors=5 ignored", err.getvalue()) |
| 130 | |
| 131 | def test_empty_list_rejected(self): |
| 132 | args = _parse("Topic", "--competitors-list", ",, ,") |
| 133 | with self.assertRaises(SystemExit) as cm, redirect_stderr(io.StringIO()): |
| 134 | cli.resolve_competitors_args(args) |
| 135 | self.assertEqual(cm.exception.code, 2) |
| 136 | |
| 137 | if __name__ == "__main__": |
| 138 | unittest.main() |
| 139 | |
| 140 | |
| 141 | class CompetitorMainTopicFailureTests(unittest.TestCase): |
| 142 | """The CLI layer above run_competitor_fanout. |
| 143 | |
| 144 | The fan-out drops a failed sub-run from its list, and the render takes |
| 145 | element 0 as the comparison's subject. Nothing between them checked that |
| 146 | the main topic survived, so a main run that raised while >=2 peers |
| 147 | succeeded produced a complete-looking comparison headed by a peer, saved |
| 148 | under that peer's slug, with the user's topic absent. |
| 149 | """ |
| 150 | |
| 151 | def _run(self, surviving_labels): |
| 152 | surviving = [(label, _fake_report(label)) for label in surviving_labels] |
| 153 | argv = [ |
| 154 | "last30days", "OpenAI", |
| 155 | "--competitors-list", "Anthropic,xAI", |
| 156 | "--mock", "--emit=json", |
| 157 | ] |
| 158 | err = io.StringIO() |
| 159 | # last30days imports fanout locally inside _main, so patch the |
| 160 | # source module rather than an attribute on the CLI module. |
| 161 | with mock.patch.object( |
| 162 | fanout, "run_competitor_fanout", return_value=surviving |
| 163 | ), mock.patch.object(sys, "argv", argv), redirect_stderr(err): |
| 164 | rc = cli.main() |
| 165 | return rc, err.getvalue() |
| 166 | |
| 167 | def test_main_topic_failure_is_not_a_competitor_promotion(self): |
| 168 | rc, err = self._run(["Anthropic", "xAI"]) |
| 169 | self.assertEqual(1, rc) |
| 170 | self.assertIn("main topic 'OpenAI' failed", err) |
| 171 | self.assertIn("Refusing to render a comparison", err) |
| 172 | |
| 173 | |
| 174 | class CompetitorDuplicateLabelTests(unittest.TestCase): |
| 175 | """run_competitor_fanout keys its results by label, so a peer sharing the |
| 176 | main topic's label collapsed both submissions onto one report while the |
| 177 | returned list still had two entries: a self-comparison whose failed main |
| 178 | run the survivor check could not see.""" |
| 179 | |
| 180 | def _run(self, peers): |
| 181 | seen: dict[str, list[str]] = {} |
| 182 | |
| 183 | def _capture(**kwargs): |
| 184 | seen["competitors"] = list(kwargs["competitors"]) |
| 185 | # One survivor trips the <2 guard, so nothing renders from the |
| 186 | # duck-typed report. The assertion is on what fan-out received. |
| 187 | return [(kwargs["main_topic"], _fake_report(kwargs["main_topic"]))] |
| 188 | |
| 189 | argv = [ |
| 190 | "last30days", "OpenAI", |
| 191 | "--competitors-list", peers, |
| 192 | "--mock", "--emit=json", |
| 193 | ] |
| 194 | err = io.StringIO() |
| 195 | with mock.patch.object( |
| 196 | fanout, "run_competitor_fanout", side_effect=_capture |
| 197 | ), mock.patch.object(sys, "argv", argv), redirect_stderr(err): |
| 198 | rc = cli.main() |
| 199 | return rc, err.getvalue(), seen.get("competitors") |
| 200 | |
| 201 | def test_peer_equal_to_main_topic_is_dropped(self): |
| 202 | _rc, err, competitors = self._run("OpenAI,Anthropic") |
| 203 | self.assertEqual(["Anthropic"], competitors) |
| 204 | self.assertIn("Dropping 'OpenAI'", err) |
| 205 | |
| 206 | def test_duplicate_match_ignores_case_and_surrounding_space(self): |
| 207 | _rc, err, competitors = self._run(" OPENAI ,Anthropic") |
| 208 | self.assertEqual(["Anthropic"], competitors) |
| 209 | self.assertIn("Dropping", err) |
| 210 | |
| 211 | def test_differently_worded_peer_is_kept(self): |
| 212 | # Normalization collapses whitespace runs and folds case; it does not |
| 213 | # strip spaces. "Open AI" stays a distinct entity from "OpenAI", |
| 214 | # because merging those would silently drop a real peer. |
| 215 | _rc, _err, competitors = self._run("Open AI,Anthropic") |
| 216 | self.assertEqual(["Open AI", "Anthropic"], competitors) |
| 217 | |
| 218 | def test_repeated_peer_is_dropped(self): |
| 219 | _rc, _err, competitors = self._run("Anthropic,anthropic,xAI") |
| 220 | self.assertEqual(["Anthropic", "xAI"], competitors) |
| 221 | |
| 222 | def test_all_peers_duplicate_main_topic_aborts(self): |
| 223 | rc, err, competitors = self._run("openai, OpenAI ") |
| 224 | self.assertEqual(2, rc) |
| 225 | self.assertIsNone(competitors, "fan-out must not run with no peers") |
| 226 | self.assertIn("No peer distinct from 'OpenAI'", err) |
| 227 | |
| 228 |