| 1 | import json |
| 2 | import subprocess |
| 3 | import sys |
| 4 | import unittest |
| 5 | from pathlib import Path |
| 6 | |
| 7 | |
| 8 | REPO_ROOT = Path(__file__).resolve().parents[1] |
| 9 | |
| 10 | |
| 11 | def run_mock_json(topic: str) -> dict: |
| 12 | result = subprocess.run( |
| 13 | [ |
| 14 | sys.executable, |
| 15 | "skills/last30days/scripts/last30days.py", |
| 16 | topic, |
| 17 | "--mock", |
| 18 | "--emit=json", |
| 19 | "--json-profile=raw", |
| 20 | ], |
| 21 | cwd=REPO_ROOT, |
| 22 | capture_output=True, |
| 23 | text=True, |
| 24 | encoding="utf-8", |
| 25 | check=False, |
| 26 | ) |
| 27 | if result.returncode != 0: |
| 28 | raise AssertionError(f"mock CLI failed for {topic!r}: {result.stderr}") |
| 29 | return json.loads(result.stdout) |
| 30 | |
| 31 | |
| 32 | class RegressionTests(unittest.TestCase): |
| 33 | def assert_common_shape(self, payload: dict) -> None: |
| 34 | self.assertIn("topic", payload) |
| 35 | self.assertIn("query_plan", payload) |
| 36 | self.assertIn("ranked_candidates", payload) |
| 37 | self.assertIn("clusters", payload) |
| 38 | self.assertIn("items_by_source", payload) |
| 39 | |
| 40 | def assert_comparison_shape(self, payload: dict) -> None: |
| 41 | """Post-3.0.13: vs-topics produce N full passes, merged output has |
| 42 | comparison=True + entities list + per-entity report wrapper.""" |
| 43 | self.assertTrue(payload.get("comparison")) |
| 44 | self.assertIn("entities", payload) |
| 45 | self.assertIn("reports", payload) |
| 46 | self.assertEqual(len(payload["entities"]), len(payload["reports"])) |
| 47 | # Each report entry wraps a single-topic report |
| 48 | for entry in payload["reports"]: |
| 49 | self.assertIn("entity", entry) |
| 50 | self.assertIn("report", entry) |
| 51 | # Inner report still has the single-topic shape |
| 52 | inner = entry["report"] |
| 53 | self.assertIn("topic", inner) |
| 54 | self.assertIn("query_plan", inner) |
| 55 | self.assertIn("clusters", inner) |
| 56 | |
| 57 | def test_openclaw_three_way_comparison_preserves_entities(self): |
| 58 | payload = run_mock_json("openclaw vs. nanoclaw vs. ironclaw") |
| 59 | self.assert_comparison_shape(payload) |
| 60 | entities = [e.lower() for e in payload["entities"]] |
| 61 | self.assertIn("openclaw", entities) |
| 62 | self.assertIn("nanoclaw", entities) |
| 63 | self.assertIn("ironclaw", entities) |
| 64 | # No cross-entity keyword pollution in any per-entity report's plan |
| 65 | for entry in payload["reports"]: |
| 66 | plan = entry["report"]["query_plan"] |
| 67 | joined = "\n".join( |
| 68 | sq["search_query"] for sq in plan["subqueries"] |
| 69 | ).lower() |
| 70 | self.assertNotIn("corsair", joined) |
| 71 | self.assertNotIn("mouse", joined) |
| 72 | |
| 73 | def test_how_to_keeps_web_video_and_discussion_sources(self): |
| 74 | payload = run_mock_json("how to deploy on Fly.io") |
| 75 | self.assert_common_shape(payload) |
| 76 | plan = payload["query_plan"] |
| 77 | self.assertEqual("how_to", plan["intent"]) |
| 78 | sources = set(plan["subqueries"][0]["sources"]) |
| 79 | self.assertIn("youtube", sources) |
| 80 | self.assertIn("reddit", sources) |
| 81 | self.assertGreaterEqual(len(sources), 2) |
| 82 | |
| 83 | def test_breaking_news_query_keeps_expected_shape(self): |
| 84 | payload = run_mock_json("latest news about React 20") |
| 85 | self.assert_common_shape(payload) |
| 86 | plan = payload["query_plan"] |
| 87 | self.assertEqual("breaking_news", plan["intent"]) |
| 88 | joined_queries = "\n".join(subquery["search_query"] for subquery in plan["subqueries"]).lower() |
| 89 | self.assertIn("react 20", joined_queries) |
| 90 | self.assertGreaterEqual(len(plan["subqueries"][0]["sources"]), 2) |
| 91 | |
| 92 | def test_two_way_comparison_preserves_exact_strings(self): |
| 93 | payload = run_mock_json("DeepSeek R1 vs GPT-5") |
| 94 | self.assert_comparison_shape(payload) |
| 95 | entities_lower = [e.lower() for e in payload["entities"]] |
| 96 | self.assertIn("deepseek r1", entities_lower) |
| 97 | self.assertIn("gpt-5", entities_lower) |
| 98 | # Each per-entity pass has its own entity in its plan |
| 99 | topics_by_entity = { |
| 100 | entry["entity"].lower(): entry["report"]["topic"].lower() |
| 101 | for entry in payload["reports"] |
| 102 | } |
| 103 | self.assertEqual(topics_by_entity["deepseek r1"], "deepseek r1") |
| 104 | self.assertEqual(topics_by_entity["gpt-5"], "gpt-5") |
| 105 | # No cross-entity pollution |
| 106 | for entry in payload["reports"]: |
| 107 | plan = entry["report"]["query_plan"] |
| 108 | joined = "\n".join( |
| 109 | sq["search_query"] for sq in plan["subqueries"] |
| 110 | ).lower() |
| 111 | self.assertNotIn("corsair", joined) |
| 112 | |
| 113 | |
| 114 | if __name__ == "__main__": |
| 115 | unittest.main() |
| 116 |