返回 last30days-skill
test_last_run_state.py
根目录 / tests / test_last_run_state.py
1 import io
2 import json
3 import os
4 import re
5 import shutil
6 import stat
7 import subprocess
8 import sys
9 import tempfile
10 import unittest
11 from contextlib import redirect_stderr, redirect_stdout
12 from pathlib import Path
13 from unittest import mock
14
15 import last30days as cli
16 from lib import schema
17
18 REPO_ROOT = Path(__file__).resolve().parents[1]
19 LAST30DAYS_SCRIPT = REPO_ROOT / "skills" / "last30days" / "scripts" / "last30days.py"
20 SKILL_MD = REPO_ROOT / "skills" / "last30days" / "SKILL.md"
21
22
23 def run_last30days(topic: str, env: dict[str, str]) -> subprocess.CompletedProcess[str]:
24 return subprocess.run(
25 [sys.executable, str(LAST30DAYS_SCRIPT), topic, "--mock", "--emit=json"],
26 cwd=REPO_ROOT,
27 env=env,
28 capture_output=True,
29 text=True,
30 encoding="utf-8",
31 check=False,
32 )
33
34
35 def _report(topic: str) -> schema.Report:
36 return schema.Report(
37 topic=topic,
38 range_from="2026-05-01",
39 range_to="2026-05-31",
40 generated_at="2026-05-31T00:00:00+00:00",
41 provider_runtime=schema.ProviderRuntime(
42 reasoning_provider="local",
43 planner_model="mock-planner",
44 rerank_model="mock-rerank",
45 ),
46 query_plan=schema.QueryPlan(
47 intent="concept",
48 freshness_mode="balanced_recent",
49 cluster_mode="none",
50 raw_topic=topic,
51 subqueries=[
52 schema.SubQuery(
53 label="primary",
54 search_query=topic,
55 ranking_query=topic,
56 sources=["grounding"],
57 )
58 ],
59 source_weights={"grounding": 1.0},
60 ),
61 clusters=[],
62 ranked_candidates=[],
63 items_by_source={"grounding": []},
64 errors_by_source={},
65 )
66
67
68 def _diag() -> dict[str, object]:
69 return {
70 "available_sources": ["grounding"],
71 "providers": {"google": True, "openai": False, "xai": False},
72 "x_backend": None,
73 "bird_installed": True,
74 "bird_authenticated": False,
75 "bird_username": None,
76 "native_web_backend": "brave",
77 }
78
79
80 class LastRunStateTests(unittest.TestCase):
81 def test_empty_config_override_disables_last_run_write(self):
82 with tempfile.TemporaryDirectory() as tmp:
83 home = Path(tmp) / "home"
84 env = os.environ.copy()
85 env["HOME"] = str(home)
86 env["LAST30DAYS_CONFIG_DIR"] = ""
87
88 result = run_last30days("synthetic eval query", env)
89
90 self.assertEqual(result.returncode, 0, result.stderr)
91 self.assertFalse((home / ".config" / "last30days" / "last-run.json").exists())
92
93 def test_custom_config_override_writes_last_run_to_custom_dir(self):
94 with tempfile.TemporaryDirectory() as tmp:
95 config_dir = Path(tmp) / "custom-config"
96 env = os.environ.copy()
97 env["HOME"] = str(Path(tmp) / "home")
98 env["LAST30DAYS_CONFIG_DIR"] = str(config_dir)
99
100 result = run_last30days("custom config query", env)
101
102 self.assertEqual(result.returncode, 0, result.stderr)
103 payload = json.loads((config_dir / "last-run.json").read_text())
104 self.assertEqual(payload["topic"], "custom config query")
105 self.assertGreaterEqual(payload["total"], 0)
106 self.assertEqual(str(config_dir / "last-report.json"), payload["report_cache"])
107 self.assertTrue((config_dir / "last-report.json").exists())
108
109 def test_last_report_cache_round_trips_single_report(self):
110 with tempfile.TemporaryDirectory() as tmp:
111 config_dir = Path(tmp) / "config"
112 with mock.patch.object(cli.env, "CONFIG_DIR", config_dir):
113 report = _report("OpenClaw")
114 cli._write_last_run("OpenClaw", report)
115 loaded = cli._load_last_report_cache("OpenClaw")
116
117 self.assertIsNotNone(loaded)
118 cached_report, entity_reports, cache_path = loaded
119 self.assertEqual("OpenClaw", cached_report.topic)
120 self.assertIsNone(entity_reports)
121 self.assertEqual(config_dir / "last-report.json", cache_path)
122
123 def test_last_report_cache_expires_after_ttl(self):
124 with tempfile.TemporaryDirectory() as tmp:
125 config_dir = Path(tmp) / "config"
126 with mock.patch.object(cli.env, "CONFIG_DIR", config_dir):
127 cli._write_last_run("OpenClaw", _report("OpenClaw"))
128 cache_path = config_dir / "last-report.json"
129 payload = json.loads(cache_path.read_text(encoding="utf-8"))
130 payload["timestamp"] = "2026-01-01T00:00:00+00:00"
131 cache_path.write_text(json.dumps(payload), encoding="utf-8")
132 loaded = cli._load_last_report_cache("OpenClaw", ttl_seconds=3600)
133
134 self.assertIsNone(loaded)
135
136 def test_last_report_cache_ttl_zero_disables_reuse(self):
137 with tempfile.TemporaryDirectory() as tmp:
138 config_dir = Path(tmp) / "config"
139 with mock.patch.object(cli.env, "CONFIG_DIR", config_dir):
140 cli._write_last_run("OpenClaw", _report("OpenClaw"))
141 loaded = cli._load_last_report_cache("OpenClaw", ttl_seconds=0)
142
143 self.assertIsNone(loaded)
144
145 def test_partial_comparison_cache_does_not_degrade_to_single_report(self):
146 with tempfile.TemporaryDirectory() as tmp:
147 config_dir = Path(tmp) / "config"
148 reports = [("Alpha", _report("Alpha")), ("Beta", _report("Beta"))]
149 with mock.patch.object(cli.env, "CONFIG_DIR", config_dir):
150 cli._write_last_run("Alpha vs Beta", reports[0][1], entity_reports=reports)
151 cache_path = config_dir / "last-report.json"
152 payload = json.loads(cache_path.read_text(encoding="utf-8"))
153 payload["reports"] = payload["reports"][:1]
154 cache_path.write_text(json.dumps(payload), encoding="utf-8")
155 loaded = cli._load_last_report_cache("Alpha vs Beta")
156
157 self.assertIsNone(loaded)
158
159 def test_html_synthesis_reuses_cached_single_report_without_pipeline_run(self):
160 with tempfile.TemporaryDirectory() as tmp:
161 config_dir = Path(tmp) / "config"
162 synthesis_path = Path(tmp) / "synthesis.md"
163 synthesis_path.write_text("# OpenClaw\n\nCached synthesis body.", encoding="utf-8")
164 with mock.patch.object(cli.env, "CONFIG_DIR", config_dir):
165 cli._write_last_run("OpenClaw", _report("OpenClaw"))
166
167 with mock.patch.object(cli.env, "CONFIG_DIR", config_dir), \
168 mock.patch.object(cli.env, "get_config", return_value={}), \
169 mock.patch.object(cli.pipeline, "diagnose", return_value=_diag()), \
170 mock.patch.object(cli.pipeline, "run", side_effect=AssertionError("pipeline should not run")), \
171 mock.patch.object(sys, "argv", [
172 "last30days.py",
173 "OpenClaw",
174 "--emit=html",
175 "--synthesis-file",
176 str(synthesis_path),
177 ]), \
178 mock.patch.dict(os.environ, {"LAST30DAYS_SKIP_PREFLIGHT": "1"}, clear=False):
179 stdout = io.StringIO()
180 stderr = io.StringIO()
181 with redirect_stdout(stdout), redirect_stderr(stderr):
182 rc = cli.main()
183
184 self.assertEqual(0, rc)
185 self.assertIn("Cached synthesis body.", stdout.getvalue())
186 self.assertIn("Reusing cached report data", stderr.getvalue())
187
188 def test_html_synthesis_reuses_cached_comparison_reports(self):
189 with tempfile.TemporaryDirectory() as tmp:
190 config_dir = Path(tmp) / "config"
191 synthesis_path = Path(tmp) / "synthesis.md"
192 synthesis_path.write_text(
193 "# Alpha vs Beta\n\nCached comparison body.",
194 encoding="utf-8",
195 )
196 reports = [("Alpha", _report("Alpha")), ("Beta", _report("Beta"))]
197 with mock.patch.object(cli.env, "CONFIG_DIR", config_dir):
198 cli._write_last_run("Alpha vs Beta", reports[0][1], entity_reports=reports)
199
200 with mock.patch.object(cli.env, "CONFIG_DIR", config_dir), \
201 mock.patch.object(cli.env, "get_config", return_value={}), \
202 mock.patch.object(cli.pipeline, "diagnose", return_value=_diag()), \
203 mock.patch.object(cli.pipeline, "run", side_effect=AssertionError("pipeline should not run")), \
204 mock.patch.object(sys, "argv", [
205 "last30days.py",
206 "Alpha",
207 "vs",
208 "Beta",
209 "--emit=html",
210 "--synthesis-file",
211 str(synthesis_path),
212 ]), \
213 mock.patch.dict(os.environ, {"LAST30DAYS_SKIP_PREFLIGHT": "1"}, clear=False):
214 stdout = io.StringIO()
215 stderr = io.StringIO()
216 with redirect_stdout(stdout), redirect_stderr(stderr):
217 rc = cli.main()
218
219 self.assertEqual(0, rc)
220 self.assertIn("Cached comparison body.", stdout.getvalue())
221 self.assertIn("last30days · Alpha vs Beta", stdout.getvalue())
222 self.assertIn("Reusing cached report data", stderr.getvalue())
223
224 def test_html_synthesis_warns_and_falls_back_when_cache_topic_misses(self):
225 with tempfile.TemporaryDirectory() as tmp:
226 config_dir = Path(tmp) / "config"
227 synthesis_path = Path(tmp) / "synthesis.md"
228 synthesis_path.write_text("Cached synthesis body.", encoding="utf-8")
229 with mock.patch.object(cli.env, "CONFIG_DIR", config_dir):
230 cli._write_last_run("OpenClaw", _report("OpenClaw"))
231
232 fresh_report = _report("Different Topic")
233 with mock.patch.object(cli.env, "CONFIG_DIR", config_dir), \
234 mock.patch.object(cli.env, "get_config", return_value={}), \
235 mock.patch.object(cli.pipeline, "diagnose", return_value=_diag()), \
236 mock.patch.object(cli.pipeline, "run", return_value=fresh_report) as run_mock, \
237 mock.patch.object(cli.ui, "ProgressDisplay"), \
238 mock.patch.object(sys, "argv", [
239 "last30days.py",
240 "Different",
241 "Topic",
242 "--emit=html",
243 "--synthesis-file",
244 str(synthesis_path),
245 ]), \
246 mock.patch.dict(os.environ, {"LAST30DAYS_SKIP_PREFLIGHT": "1"}, clear=False):
247 stdout = io.StringIO()
248 stderr = io.StringIO()
249 with redirect_stdout(stdout), redirect_stderr(stderr):
250 rc = cli.main()
251
252 self.assertEqual(0, rc)
253 self.assertTrue(run_mock.called)
254 self.assertIn("No matching cached report data", stderr.getvalue())
255 self.assertIn("Cached synthesis body.", stdout.getvalue())
256
257 def test_html_synthesis_falls_back_when_cache_is_stale(self):
258 with tempfile.TemporaryDirectory() as tmp:
259 config_dir = Path(tmp) / "config"
260 synthesis_path = Path(tmp) / "synthesis.md"
261 synthesis_path.write_text("Cached synthesis body.", encoding="utf-8")
262 with mock.patch.object(cli.env, "CONFIG_DIR", config_dir):
263 cli._write_last_run("OpenClaw", _report("OpenClaw"))
264 cache_path = config_dir / "last-report.json"
265 payload = json.loads(cache_path.read_text(encoding="utf-8"))
266 payload["timestamp"] = "2026-01-01T00:00:00+00:00"
267 cache_path.write_text(json.dumps(payload), encoding="utf-8")
268
269 fresh_report = _report("OpenClaw")
270 with mock.patch.object(cli.env, "CONFIG_DIR", config_dir), \
271 mock.patch.object(cli.env, "get_config", return_value={}), \
272 mock.patch.object(cli.pipeline, "diagnose", return_value=_diag()), \
273 mock.patch.object(cli.pipeline, "run", return_value=fresh_report) as run_mock, \
274 mock.patch.object(cli.ui, "ProgressDisplay"), \
275 mock.patch.object(sys, "argv", [
276 "last30days.py",
277 "OpenClaw",
278 "--emit=html",
279 "--synthesis-file",
280 str(synthesis_path),
281 ]), \
282 mock.patch.dict(os.environ, {"LAST30DAYS_SKIP_PREFLIGHT": "1"}, clear=False):
283 stdout = io.StringIO()
284 stderr = io.StringIO()
285 with redirect_stdout(stdout), redirect_stderr(stderr):
286 rc = cli.main()
287
288 self.assertEqual(0, rc)
289 self.assertTrue(run_mock.called)
290 self.assertIn("No matching cached report data", stderr.getvalue())
291 self.assertIn("Cached synthesis body.", stdout.getvalue())
292
293 @unittest.skipIf(shutil.which("bash") is None, "bash not available")
294 def test_hook_reads_last_run_from_custom_config_dir(self):
295 with tempfile.TemporaryDirectory() as tmp:
296 config_dir = Path(tmp) / "custom-config"
297 config_dir.mkdir()
298 (config_dir / "last-run.json").write_text(
299 json.dumps(
300 {
301 "topic": "custom hook query",
302 "timestamp": "2026-04-30T00:00:00+00:00",
303 "sources": {"reddit": 2},
304 "total": 2,
305 }
306 )
307 )
308 env = os.environ.copy()
309 env["HOME"] = str(Path(tmp) / "home")
310 env["LAST30DAYS_CONFIG_DIR"] = str(config_dir)
311
312 result = subprocess.run(
313 ["bash", "hooks/scripts/check-config.sh"],
314 cwd=REPO_ROOT,
315 env=env,
316 capture_output=True,
317 text=True,
318 check=False,
319 )
320
321 self.assertEqual(result.returncode, 0, result.stderr)
322 self.assertIn('Last run: "custom hook query"', result.stdout)
323
324 def test_hook_exits_0_when_no_last_run(self):
325 """Script exits 0 when ScrapeCreators configured but no prior run (last-run.json absent)."""
326 with tempfile.TemporaryDirectory() as tmp:
327 env = os.environ.copy()
328 env["HOME"] = str(Path(tmp) / "home")
329 env["SETUP_COMPLETE"] = "true"
330 env["ENV_SCRAPECREATORS_API_KEY"] = "sk-test"
331
332 result = subprocess.run(
333 ["bash", "hooks/scripts/check-config.sh"],
334 cwd=REPO_ROOT,
335 env=env,
336 capture_output=True,
337 text=True,
338 check=False,
339 )
340
341 self.assertEqual(result.returncode, 0, result.stderr)
342 self.assertIn("Ready —", result.stdout)
343 self.assertNotIn("Last run:", result.stdout)
344
345 def test_hook_parses_dotenv_with_unbalanced_quote(self):
346 """Script exits 0 when .env contains an unbalanced quote in a value."""
347 with tempfile.TemporaryDirectory() as tmp:
348 home = Path(tmp) / "home"
349 config_dir = home / ".config" / "last30days"
350 config_dir.mkdir(parents=True)
351 env_file = config_dir / ".env"
352 env_file.write_text(
353 "SETUP_COMPLETE=true\n"
354 "XAI_API_KEY=xai-key-with-apostrophe's-ok\n"
355 "AUTH_TOKEN=test-auth\n"
356 "CT0=test-ct0\n"
357 )
358 env = os.environ.copy()
359 env["HOME"] = str(home)
360
361 result = subprocess.run(
362 ["bash", "hooks/scripts/check-config.sh"],
363 cwd=REPO_ROOT,
364 env=env,
365 capture_output=True,
366 text=True,
367 check=False,
368 )
369
370 self.assertEqual(result.returncode, 0, result.stderr)
371 self.assertIn("Ready —", result.stdout)
372
373 @staticmethod
374 def _extract_source_count(output: str) -> int:
375 match = re.search(r"Ready — (\d+) sources active", output)
376 if not match:
377 raise AssertionError(f"Could not find source count in: {repr(output[:200])}")
378 return int(match.group(1))
379
380 def _run_hook(self, tmp: str, env_overrides: dict[str, str]) -> subprocess.CompletedProcess[str]:
381 env = os.environ.copy()
382 env["HOME"] = str(Path(tmp) / "home")
383 env["SETUP_COMPLETE"] = "true"
384 # Strip credentials that could bleed in from the test-runner environment
385 # and corrupt source-count baseline comparisons.
386 for key in ("AUTH_TOKEN", "CT0", "XAI_API_KEY", "BSKY_HANDLE", "EXA_API_KEY", "SCRAPECREATORS_API_KEY"):
387 env.pop(key, None)
388 env.update(env_overrides)
389 return subprocess.run(
390 ["bash", "hooks/scripts/check-config.sh"],
391 cwd=REPO_ROOT,
392 env=env,
393 capture_output=True,
394 text=True,
395 check=False,
396 )
397
398 def test_x_not_counted_with_only_auth_token(self):
399 with tempfile.TemporaryDirectory() as tmp:
400 neither = self._extract_source_count(
401 self._run_hook(tmp, {}).stdout
402 )
403 only_auth = self._extract_source_count(
404 self._run_hook(tmp, {"AUTH_TOKEN": "test_auth"}).stdout
405 )
406 self.assertEqual(
407 only_auth, neither,
408 "X should not be counted when only AUTH_TOKEN is set (CT0 missing)",
409 )
410
411 def test_x_not_counted_with_only_ct0(self):
412 with tempfile.TemporaryDirectory() as tmp:
413 neither = self._extract_source_count(
414 self._run_hook(tmp, {}).stdout
415 )
416 only_ct0 = self._extract_source_count(
417 self._run_hook(tmp, {"CT0": "test_ct0"}).stdout
418 )
419 self.assertEqual(
420 only_ct0, neither,
421 "X should not be counted when only CT0 is set (AUTH_TOKEN missing)",
422 )
423
424 def test_x_counted_when_both_auth_token_and_ct0(self):
425 with tempfile.TemporaryDirectory() as tmp:
426 neither = self._extract_source_count(
427 self._run_hook(tmp, {}).stdout
428 )
429 both = self._extract_source_count(
430 self._run_hook(tmp, {"AUTH_TOKEN": "test_auth", "CT0": "test_ct0"}).stdout
431 )
432 self.assertEqual(
433 both, neither + 1,
434 "X should add 1 source when both AUTH_TOKEN and CT0 are set",
435 )
436
437 def test_hook_shows_last_run_when_json_exists(self):
438 """Script exits 0 and shows last-run summary when last-run.json exists."""
439 with tempfile.TemporaryDirectory() as tmp:
440 config_dir = Path(tmp) / "custom-config"
441 config_dir.mkdir()
442 (config_dir / "last-run.json").write_text(
443 json.dumps(
444 {
445 "topic": "prior research",
446 "timestamp": "2026-06-01T12:00:00+00:00",
447 "sources": {"reddit": 5},
448 "total": 5,
449 }
450 )
451 )
452 env = os.environ.copy()
453 env["HOME"] = str(Path(tmp) / "home")
454 env["SETUP_COMPLETE"] = "true"
455 env["ENV_SCRAPECREATORS_API_KEY"] = "sk-test"
456 env["LAST30DAYS_CONFIG_DIR"] = str(config_dir)
457
458 result = subprocess.run(
459 ["bash", "hooks/scripts/check-config.sh"],
460 cwd=REPO_ROOT,
461 env=env,
462 capture_output=True,
463 text=True,
464 check=False,
465 )
466
467 self.assertEqual(result.returncode, 0, result.stderr)
468 self.assertIn('Last run: "prior research"', result.stdout)
469
470
471 class TestSkillMdFirstRunReference(unittest.TestCase):
472 """Verifies SKILL.md references that exist in the CLI."""
473
474 def test_nux_wizard_not_referenced(self):
475 content = SKILL_MD.read_text(encoding="utf-8")
476 self.assertNotIn(
477 "nux-wizard.md", content,
478 "SKILL.md should not reference the missing nux-wizard.md file",
479 )
480
481 def test_skill_md_references_setup_command(self):
482 content = SKILL_MD.read_text(encoding="utf-8")
483 self.assertIn(
484 "last30days.py setup", content,
485 "SKILL.md should reference the Python setup subcommand",
486 )
487
488 def test_setup_subcommand_dispatches(self):
489 """topic 'setup' must reach setup_wizard, not be swallowed by argparse."""
490 with mock.patch.object(cli.env, "get_config", return_value={}), \
491 mock.patch("lib.setup_wizard.run_auto_setup", return_value={"cookies_found": {}}) as mock_setup, \
492 mock.patch("lib.setup_wizard.write_setup_config") as mock_write, \
493 mock.patch("lib.setup_wizard.get_setup_status_text", return_value="ok"), \
494 mock.patch.object(sys, "argv", ["last30days.py", "setup"]):
495 stderr = io.StringIO()
496 with redirect_stderr(stderr):
497 rc = cli.main()
498 self.assertEqual(0, rc)
499 mock_setup.assert_called_once()
500 mock_write.assert_called_once()
501
502
503 class TestCheckPermsAutoFix(unittest.TestCase):
504 """check_perms should auto-fix loose .env permissions instead of warning only."""
505
506 def test_loose_env_is_tightened_by_check_perms(self):
507 with tempfile.TemporaryDirectory() as tmp:
508 config_dir = Path(tmp) / ".config" / "last30days"
509 config_dir.mkdir(parents=True)
510 env_file = config_dir / ".env"
511 env_file.write_text("SETUP_COMPLETE=true\n")
512 os.chmod(env_file, 0o644)
513
514 env = os.environ.copy()
515 env["HOME"] = str(Path(tmp))
516 env["LAST30DAYS_CONFIG_DIR"] = str(config_dir)
517
518 result = subprocess.run(
519 ["bash", "hooks/scripts/check-config.sh"],
520 cwd=REPO_ROOT,
521 env=env,
522 capture_output=True,
523 text=True,
524 check=False,
525 )
526
527 self.assertEqual(result.returncode, 0, result.stderr)
528 self.assertIn("auto-fixed", result.stdout.lower())
529 self.assertEqual(stat.S_IMODE(os.stat(env_file).st_mode), 0o600)
530
531
532 if __name__ == "__main__":
533 unittest.main()
534
534 lines PYTHON