| 1 | """U5: doctor cache — results persist across invocations with a TTL. |
| 2 | |
| 3 | Covers the plan's U5 scenarios (R5, R7, KTD 8): |
| 4 | 1. Fresh cache within TTL -> `--cached` returns the stored result and |
| 5 | spawns zero probes (probe layer spied, never called). |
| 6 | 2. Stale or missing cache -> live run executes and rewrites the cache. |
| 7 | 3. Explicit `doctor` (no `--cached`) -> always live; cache refreshed |
| 8 | even when a fresh cache exists. |
| 9 | 4. TTL env override (LAST30DAYS_DOCTOR_TTL, seconds) respected; |
| 10 | malformed/corrupt cache treated as absent — never a crash. |
| 11 | 5. No secret values in the cache file under seeded fake credentials. |
| 12 | 6. SKILL.md contract (test_onboarding_contract.py pattern): doctor |
| 13 | trigger phrases and the cached standing rule are present, so future |
| 14 | SKILL.md edits cannot silently erode the integration. |
| 15 | """ |
| 16 | |
| 17 | import datetime |
| 18 | import io |
| 19 | import json |
| 20 | import os |
| 21 | import sys |
| 22 | import tempfile |
| 23 | import unittest |
| 24 | from contextlib import redirect_stderr, redirect_stdout |
| 25 | from pathlib import Path |
| 26 | from unittest import mock |
| 27 | |
| 28 | import last30days as cli |
| 29 | from lib import doctor, health |
| 30 | |
| 31 | ROOT = Path(__file__).resolve().parents[1] |
| 32 | SKILL_MD = ROOT / "skills" / "last30days" / "SKILL.md" |
| 33 | |
| 34 | BIRD_STATUS_OFF = { |
| 35 | "installed": False, |
| 36 | "authenticated": False, |
| 37 | "username": None, |
| 38 | "can_install": True, |
| 39 | } |
| 40 | |
| 41 | # Obvious dummies only (repo security hygiene). |
| 42 | FAKE_SECRETS = { |
| 43 | "SCRAPECREATORS_API_KEY": "dummy-sc-secret-000", |
| 44 | "XAI_API_KEY": "dummy-xai-secret-000", |
| 45 | "BRAVE_API_KEY": "dummy-brave-secret-000", |
| 46 | "AUTH_TOKEN": "dummy-auth-token-secret-000", |
| 47 | "CT0": "dummy-ct0-secret-000", |
| 48 | "BSKY_HANDLE": "dummy.example.social", |
| 49 | "BSKY_APP_PASSWORD": "dummy-bsky-secret-000", |
| 50 | "TRUTHSOCIAL_TOKEN": "dummy-truth-secret-000", |
| 51 | "GITHUB_TOKEN": "dummy-github-secret-000", |
| 52 | } |
| 53 | |
| 54 | |
| 55 | def _iso_utc(seconds_ago: float = 0.0) -> str: |
| 56 | return ( |
| 57 | datetime.datetime.now(datetime.timezone.utc) |
| 58 | - datetime.timedelta(seconds=seconds_ago) |
| 59 | ).isoformat() |
| 60 | |
| 61 | |
| 62 | def _fake_probe(name, timeout=health.PROBE_TIMEOUT): |
| 63 | return health.DependencyProbe( |
| 64 | name=name, |
| 65 | status=health.MISSING, |
| 66 | detail=f"{name} probe simulated missing", |
| 67 | prescription=f"install {name}", |
| 68 | owner_pkg_manager="brew", |
| 69 | ) |
| 70 | |
| 71 | |
| 72 | class _Hermetic: |
| 73 | """Machine-independent doctor runs (mirrors tests/test_doctor.py).""" |
| 74 | |
| 75 | def __init__(self, probe=None): |
| 76 | self.probe_spy = mock.Mock(side_effect=probe or _fake_probe) |
| 77 | self._patches = [ |
| 78 | mock.patch("lib.health.probe_dependency", self.probe_spy), |
| 79 | mock.patch("lib.bird_x.is_bird_installed", return_value=False), |
| 80 | mock.patch("lib.bird_x.set_credentials", lambda *a, **k: None), |
| 81 | mock.patch("lib.bird_x.get_bird_status", return_value=dict(BIRD_STATUS_OFF)), |
| 82 | # Doctor path is local-only for xurl: the live `xurl whoami` |
| 83 | # network check must never run (no-network guarantee). |
| 84 | mock.patch( |
| 85 | "lib.xurl_x.is_available", |
| 86 | side_effect=AssertionError( |
| 87 | "doctor path ran the live `xurl whoami` network check" |
| 88 | ), |
| 89 | ), |
| 90 | mock.patch("lib.xurl_x.has_stored_auth", return_value=False), |
| 91 | mock.patch( |
| 92 | "lib.xurl_x.stored_auth_status", |
| 93 | return_value=("missing", "no token store at ~/.xurl"), |
| 94 | ), |
| 95 | mock.patch("lib.backends.which", lambda name: None), |
| 96 | ] |
| 97 | |
| 98 | def __enter__(self): |
| 99 | for p in self._patches: |
| 100 | p.start() |
| 101 | return self |
| 102 | |
| 103 | def __exit__(self, *exc): |
| 104 | for p in reversed(self._patches): |
| 105 | p.stop() |
| 106 | return False |
| 107 | |
| 108 | |
| 109 | class _CacheDirCase(unittest.TestCase): |
| 110 | """Base: isolated CONFIG_DIR + clean TTL env for every test.""" |
| 111 | |
| 112 | def setUp(self): |
| 113 | self._tmp = tempfile.TemporaryDirectory() |
| 114 | self.addCleanup(self._tmp.cleanup) |
| 115 | self.config_dir = Path(self._tmp.name) |
| 116 | patcher = mock.patch.object(cli.env, "CONFIG_DIR", self.config_dir) |
| 117 | patcher.start() |
| 118 | self.addCleanup(patcher.stop) |
| 119 | env_patcher = mock.patch.dict(os.environ, {}, clear=False) |
| 120 | env_patcher.start() |
| 121 | self.addCleanup(env_patcher.stop) |
| 122 | os.environ.pop("LAST30DAYS_DOCTOR_TTL", None) |
| 123 | |
| 124 | @property |
| 125 | def cache_file(self) -> Path: |
| 126 | return self.config_dir / doctor.CACHE_FILENAME |
| 127 | |
| 128 | @staticmethod |
| 129 | def valid_report(marker="cached-sentinel-report"): |
| 130 | """A report satisfying the render contract, carrying a marker.""" |
| 131 | return { |
| 132 | "engine_version": marker, |
| 133 | "config": {"global_env": None, "config_source": None}, |
| 134 | "setup": {"setup_complete": False, "keys_present": {}}, |
| 135 | "permissions": {"status": "ok"}, |
| 136 | "sources": { |
| 137 | "hackernews": { |
| 138 | "tier": "ok", "status": "ok", "mode": "single", |
| 139 | "backends": None, "active_backend": None, "fix": "", |
| 140 | "requires": "none", "note": marker, "detail": "", |
| 141 | "pin_var": None, "pin_flag": None, "pinned": False, |
| 142 | }, |
| 143 | }, |
| 144 | } |
| 145 | |
| 146 | def write_cache( |
| 147 | self, |
| 148 | *, |
| 149 | seconds_ago=0.0, |
| 150 | marker="cached-sentinel-report", |
| 151 | config=None, |
| 152 | schema=None, |
| 153 | fingerprint=None, |
| 154 | report=None, |
| 155 | ): |
| 156 | """Seed a valid cached payload (schema + fingerprint stamped).""" |
| 157 | if report is None: |
| 158 | report = self.valid_report(marker) |
| 159 | payload = { |
| 160 | "schema": doctor.DOCTOR_CACHE_SCHEMA_VERSION if schema is None else schema, |
| 161 | "fingerprint": ( |
| 162 | doctor._config_fingerprint(dict(config or {})) |
| 163 | if fingerprint is None |
| 164 | else fingerprint |
| 165 | ), |
| 166 | "timestamp": _iso_utc(seconds_ago), |
| 167 | "report": report, |
| 168 | } |
| 169 | self.cache_file.write_text(json.dumps(payload), encoding="utf-8") |
| 170 | return report |
| 171 | |
| 172 | def run_doctor(self, config=None, *, cached, emit_json=True): |
| 173 | with _Hermetic() as h: |
| 174 | stdout = io.StringIO() |
| 175 | with redirect_stdout(stdout): |
| 176 | rc = doctor.run(dict(config or {}), emit_json=emit_json, cached=cached) |
| 177 | return rc, stdout.getvalue(), h.probe_spy |
| 178 | |
| 179 | |
| 180 | class FreshCacheServed(_CacheDirCase): |
| 181 | """Scenario 1: fresh cache within TTL -> stored result, zero probes.""" |
| 182 | |
| 183 | def test_cached_returns_stored_report(self): |
| 184 | self.write_cache(seconds_ago=1) |
| 185 | rc, out, _ = self.run_doctor(cached=True) |
| 186 | self.assertEqual(0, rc) |
| 187 | self.assertIn("cached-sentinel-report", out) |
| 188 | payload = json.loads(out) |
| 189 | self.assertIn("sources", payload) |
| 190 | |
| 191 | def test_cached_spawns_zero_probes_and_no_live_aggregation(self): |
| 192 | self.write_cache(seconds_ago=1) |
| 193 | with mock.patch( |
| 194 | "lib.doctor.build_report", |
| 195 | side_effect=AssertionError("live aggregation must not run"), |
| 196 | ) as build: |
| 197 | rc, out, probe_spy = self.run_doctor(cached=True) |
| 198 | self.assertEqual(0, rc) |
| 199 | self.assertFalse(probe_spy.called, "probe layer must not be touched") |
| 200 | self.assertFalse(build.called) |
| 201 | |
| 202 | def test_cached_hit_does_not_rewrite_cache(self): |
| 203 | self.write_cache(seconds_ago=1) |
| 204 | before = self.cache_file.read_text(encoding="utf-8") |
| 205 | self.run_doctor(cached=True) |
| 206 | self.assertEqual(before, self.cache_file.read_text(encoding="utf-8")) |
| 207 | |
| 208 | def test_cached_text_render_from_cache(self): |
| 209 | self.write_cache(seconds_ago=1) |
| 210 | rc, out, probe_spy = self.run_doctor(cached=True, emit_json=False) |
| 211 | self.assertEqual(0, rc) |
| 212 | self.assertIn("last30days doctor", out) |
| 213 | self.assertFalse(probe_spy.called) |
| 214 | |
| 215 | |
| 216 | class StaleOrMissingCacheRunsLive(_CacheDirCase): |
| 217 | """Scenario 2: stale or missing cache -> live run rewrites the cache.""" |
| 218 | |
| 219 | def test_stale_cache_falls_through_to_live_run(self): |
| 220 | self.write_cache(seconds_ago=doctor.DEFAULT_CACHE_TTL_SECONDS * 2) |
| 221 | rc, out, probe_spy = self.run_doctor(cached=True) |
| 222 | self.assertEqual(0, rc) |
| 223 | self.assertNotIn("cached-sentinel-report", out) |
| 224 | self.assertTrue(probe_spy.called, "stale cache must trigger live probes") |
| 225 | |
| 226 | def test_stale_cache_is_rewritten(self): |
| 227 | self.write_cache(seconds_ago=doctor.DEFAULT_CACHE_TTL_SECONDS * 2) |
| 228 | self.run_doctor(cached=True) |
| 229 | payload = json.loads(self.cache_file.read_text(encoding="utf-8")) |
| 230 | self.assertNotIn("cached-sentinel-report", json.dumps(payload)) |
| 231 | self.assertIn("report", payload) |
| 232 | self.assertIn("sources", payload["report"]) |
| 233 | # New timestamp is fresh. |
| 234 | age = datetime.datetime.now(datetime.timezone.utc) - datetime.datetime.fromisoformat( |
| 235 | payload["timestamp"] |
| 236 | ) |
| 237 | self.assertLess(age.total_seconds(), 60) |
| 238 | |
| 239 | def test_missing_cache_runs_live_and_creates_file(self): |
| 240 | self.assertFalse(self.cache_file.exists()) |
| 241 | rc, out, probe_spy = self.run_doctor(cached=True) |
| 242 | self.assertEqual(0, rc) |
| 243 | self.assertTrue(probe_spy.called) |
| 244 | self.assertTrue(self.cache_file.exists()) |
| 245 | payload = json.loads(self.cache_file.read_text(encoding="utf-8")) |
| 246 | self.assertEqual( |
| 247 | set(doctor.SOURCE_ORDER), set(payload["report"]["sources"].keys()) |
| 248 | ) |
| 249 | |
| 250 | |
| 251 | class ExplicitDoctorAlwaysLive(_CacheDirCase): |
| 252 | """Scenario 3: no --cached -> live run even when a fresh cache exists.""" |
| 253 | |
| 254 | def test_fresh_cache_ignored_without_cached_flag(self): |
| 255 | self.write_cache(seconds_ago=1) |
| 256 | rc, out, probe_spy = self.run_doctor(cached=False) |
| 257 | self.assertEqual(0, rc) |
| 258 | self.assertNotIn("cached-sentinel-report", out) |
| 259 | self.assertTrue(probe_spy.called) |
| 260 | |
| 261 | def test_live_run_refreshes_fresh_cache(self): |
| 262 | self.write_cache(seconds_ago=1) |
| 263 | self.run_doctor(cached=False) |
| 264 | raw = self.cache_file.read_text(encoding="utf-8") |
| 265 | self.assertNotIn("cached-sentinel-report", raw) |
| 266 | self.assertIn("sources", json.loads(raw)["report"]) |
| 267 | |
| 268 | |
| 269 | class TtlOverrideAndCorruptCache(_CacheDirCase): |
| 270 | """Scenario 4: TTL env override respected; corrupt cache = absent.""" |
| 271 | |
| 272 | def test_ttl_env_override_shrinks_window(self): |
| 273 | self.write_cache(seconds_ago=10) |
| 274 | os.environ["LAST30DAYS_DOCTOR_TTL"] = "1" |
| 275 | rc, out, probe_spy = self.run_doctor(cached=True) |
| 276 | self.assertEqual(0, rc) |
| 277 | self.assertNotIn("cached-sentinel-report", out) |
| 278 | self.assertTrue(probe_spy.called) |
| 279 | |
| 280 | def test_ttl_env_override_widens_window(self): |
| 281 | self.write_cache(seconds_ago=doctor.DEFAULT_CACHE_TTL_SECONDS * 2) |
| 282 | os.environ["LAST30DAYS_DOCTOR_TTL"] = str(doctor.DEFAULT_CACHE_TTL_SECONDS * 10) |
| 283 | rc, out, probe_spy = self.run_doctor(cached=True) |
| 284 | self.assertIn("cached-sentinel-report", out) |
| 285 | self.assertFalse(probe_spy.called) |
| 286 | |
| 287 | def test_ttl_from_config_layer(self): |
| 288 | # Registered env key: a .env-set value reaches doctor via config. |
| 289 | self.write_cache(seconds_ago=10) |
| 290 | rc, out, probe_spy = self.run_doctor({"LAST30DAYS_DOCTOR_TTL": "1"}, cached=True) |
| 291 | self.assertNotIn("cached-sentinel-report", out) |
| 292 | self.assertTrue(probe_spy.called) |
| 293 | |
| 294 | def test_ttl_zero_disables_cache_reuse(self): |
| 295 | self.write_cache(seconds_ago=0) |
| 296 | os.environ["LAST30DAYS_DOCTOR_TTL"] = "0" |
| 297 | rc, out, probe_spy = self.run_doctor(cached=True) |
| 298 | self.assertNotIn("cached-sentinel-report", out) |
| 299 | self.assertTrue(probe_spy.called) |
| 300 | |
| 301 | def test_garbage_ttl_falls_back_to_default(self): |
| 302 | self.write_cache(seconds_ago=1) |
| 303 | os.environ["LAST30DAYS_DOCTOR_TTL"] = "not-a-number" |
| 304 | rc, out, probe_spy = self.run_doctor(cached=True) |
| 305 | self.assertIn("cached-sentinel-report", out) |
| 306 | self.assertFalse(probe_spy.called) |
| 307 | |
| 308 | def test_corrupt_cache_files_treated_as_absent(self): |
| 309 | for corrupt in ( |
| 310 | "not json at all {", |
| 311 | json.dumps(["a", "list"]), |
| 312 | json.dumps({"timestamp": _iso_utc(), "report": "not-a-dict"}), |
| 313 | json.dumps({"timestamp": "garbage-timestamp", "report": {"sources": {}}}), |
| 314 | json.dumps({"report": {"sources": {}}}), # no timestamp |
| 315 | json.dumps({"timestamp": _iso_utc(), "report": {}}), # no sources |
| 316 | ): |
| 317 | with self.subTest(corrupt=corrupt[:40]): |
| 318 | self.cache_file.write_text(corrupt, encoding="utf-8") |
| 319 | rc, out, probe_spy = self.run_doctor(cached=True) |
| 320 | self.assertEqual(0, rc, "corrupt cache must never crash") |
| 321 | self.assertTrue(probe_spy.called, "corrupt cache must fall through") |
| 322 | # And the corrupt file is replaced by a valid one. |
| 323 | payload = json.loads(self.cache_file.read_text(encoding="utf-8")) |
| 324 | self.assertIn("sources", payload["report"]) |
| 325 | |
| 326 | def test_no_config_dir_runs_live_without_crash(self): |
| 327 | with mock.patch.object(cli.env, "CONFIG_DIR", None): |
| 328 | rc, out, probe_spy = self.run_doctor(cached=True) |
| 329 | self.assertEqual(0, rc) |
| 330 | self.assertTrue(probe_spy.called) |
| 331 | |
| 332 | def test_ttl_env_var_is_registered(self): |
| 333 | # The repo foot-gun: unregistered keys are silently swallowed from |
| 334 | # .env. LAST30DAYS_DOCTOR_TTL must be in env.py's get_config registry. |
| 335 | import inspect |
| 336 | |
| 337 | from lib import env as env_mod |
| 338 | |
| 339 | self.assertIn("LAST30DAYS_DOCTOR_TTL", inspect.getsource(env_mod.get_config)) |
| 340 | |
| 341 | |
| 342 | class DriftedCacheShapes(_CacheDirCase): |
| 343 | """F3: cached reports that drift from the render contract fall through |
| 344 | to a live run — never a KeyError crash — in text mode too.""" |
| 345 | |
| 346 | def _write_payload(self, report, *, with_envelope=True, seconds_ago=1.0): |
| 347 | payload = {"timestamp": _iso_utc(seconds_ago), "report": report} |
| 348 | if with_envelope: |
| 349 | payload["schema"] = doctor.DOCTOR_CACHE_SCHEMA_VERSION |
| 350 | payload["fingerprint"] = doctor._config_fingerprint({}) |
| 351 | self.cache_file.write_text(json.dumps(payload), encoding="utf-8") |
| 352 | |
| 353 | def test_fresh_drifted_report_text_mode_no_crash(self): |
| 354 | # Exact F3 repro: fresh timestamp, report missing engine_version / |
| 355 | # config / setup / permissions, record missing tier. |
| 356 | self._write_payload({"sources": {"hackernews": {"status": "ok"}}}) |
| 357 | rc, out, probe_spy = self.run_doctor(cached=True, emit_json=False) |
| 358 | self.assertEqual(0, rc, "drifted cache must never crash") |
| 359 | self.assertTrue(probe_spy.called, "drifted cache must fall through live") |
| 360 | self.assertIn("last30days doctor", out) |
| 361 | |
| 362 | def test_fresh_drifted_report_pre_schema_envelope_no_crash(self): |
| 363 | # The original repro shape (no schema stamp at all). |
| 364 | self._write_payload( |
| 365 | {"sources": {"hackernews": {"status": "ok"}}}, with_envelope=False |
| 366 | ) |
| 367 | rc, out, probe_spy = self.run_doctor(cached=True, emit_json=False) |
| 368 | self.assertEqual(0, rc) |
| 369 | self.assertTrue(probe_spy.called) |
| 370 | |
| 371 | def test_shape_validator_rejects_each_drift(self): |
| 372 | good = self.valid_report() |
| 373 | drifted = [] |
| 374 | for key in ("engine_version", "config", "setup", "permissions"): |
| 375 | broken = json.loads(json.dumps(good)) |
| 376 | del broken[key] |
| 377 | drifted.append((f"missing {key}", broken)) |
| 378 | for key in ("config", "setup", "permissions"): |
| 379 | broken = json.loads(json.dumps(good)) |
| 380 | broken[key] = "not-a-dict" |
| 381 | drifted.append((f"{key} not a dict", broken)) |
| 382 | broken = json.loads(json.dumps(good)) |
| 383 | broken["sources"]["hackernews"] = "not-a-record" |
| 384 | drifted.append(("record not a dict", broken)) |
| 385 | broken = json.loads(json.dumps(good)) |
| 386 | broken["sources"]["hackernews"]["tier"] = "sideways" |
| 387 | drifted.append(("unknown tier", broken)) |
| 388 | broken = json.loads(json.dumps(good)) |
| 389 | del broken["sources"]["hackernews"]["tier"] |
| 390 | drifted.append(("missing tier", broken)) |
| 391 | broken = json.loads(json.dumps(good)) |
| 392 | broken["sources"]["hackernews"]["status"] = 7 |
| 393 | drifted.append(("non-str status", broken)) |
| 394 | for label, report in drifted: |
| 395 | with self.subTest(drift=label): |
| 396 | self._write_payload(report) |
| 397 | rc, out, probe_spy = self.run_doctor(cached=True, emit_json=False) |
| 398 | self.assertEqual(0, rc, f"{label}: must never crash") |
| 399 | self.assertTrue(probe_spy.called, f"{label}: must fall through") |
| 400 | |
| 401 | def test_run_survives_shapes_the_validator_misses(self): |
| 402 | # Even if a bad shape slips past read_cached_report, run()'s |
| 403 | # try/except falls through to a live build (never-crash contract). |
| 404 | bad = self.valid_report() |
| 405 | with mock.patch( |
| 406 | "lib.doctor.read_cached_report", |
| 407 | return_value={"sources": bad["sources"]}, # renders would KeyError |
| 408 | ): |
| 409 | rc, out, probe_spy = self.run_doctor(cached=True, emit_json=False) |
| 410 | self.assertEqual(0, rc) |
| 411 | self.assertTrue(probe_spy.called) |
| 412 | self.assertIn("last30days doctor", out) |
| 413 | |
| 414 | |
| 415 | class SchemaStamp(_CacheDirCase): |
| 416 | """F8: payloads without the current schema stamp are treated as absent.""" |
| 417 | |
| 418 | def test_schema_mismatch_runs_live(self): |
| 419 | self.write_cache(seconds_ago=1, schema="last30days-doctor-cache/v0") |
| 420 | rc, out, probe_spy = self.run_doctor(cached=True) |
| 421 | self.assertEqual(0, rc) |
| 422 | self.assertNotIn("cached-sentinel-report", out) |
| 423 | self.assertTrue(probe_spy.called) |
| 424 | |
| 425 | def test_absent_schema_runs_live(self): |
| 426 | payload = { |
| 427 | "fingerprint": doctor._config_fingerprint({}), |
| 428 | "timestamp": _iso_utc(1), |
| 429 | "report": self.valid_report(), |
| 430 | } |
| 431 | self.cache_file.write_text(json.dumps(payload), encoding="utf-8") |
| 432 | rc, out, probe_spy = self.run_doctor(cached=True) |
| 433 | self.assertEqual(0, rc) |
| 434 | self.assertNotIn("cached-sentinel-report", out) |
| 435 | self.assertTrue(probe_spy.called) |
| 436 | |
| 437 | def test_live_run_stamps_schema_and_fingerprint(self): |
| 438 | self.run_doctor(cached=False) |
| 439 | payload = json.loads(self.cache_file.read_text(encoding="utf-8")) |
| 440 | self.assertEqual(doctor.DOCTOR_CACHE_SCHEMA_VERSION, payload["schema"]) |
| 441 | self.assertEqual(doctor._config_fingerprint({}), payload["fingerprint"]) |
| 442 | |
| 443 | |
| 444 | class FingerprintInvalidation(_CacheDirCase): |
| 445 | """F12a: credential/pin/opt-in changes invalidate the cache.""" |
| 446 | |
| 447 | def test_key_added_invalidates(self): |
| 448 | self.write_cache(seconds_ago=1) # fingerprint for empty config |
| 449 | rc, out, probe_spy = self.run_doctor( |
| 450 | {"SCRAPECREATORS_API_KEY": "dummy-sc-secret-000"}, cached=True |
| 451 | ) |
| 452 | self.assertEqual(0, rc) |
| 453 | self.assertNotIn("cached-sentinel-report", out) |
| 454 | self.assertTrue(probe_spy.called, "new credential must invalidate cache") |
| 455 | |
| 456 | def test_key_removed_invalidates(self): |
| 457 | cfg = {"SCRAPECREATORS_API_KEY": "dummy-sc-secret-000"} |
| 458 | self.write_cache(seconds_ago=1, config=cfg) |
| 459 | rc, out, probe_spy = self.run_doctor({}, cached=True) |
| 460 | self.assertNotIn("cached-sentinel-report", out) |
| 461 | self.assertTrue(probe_spy.called, "removed credential must invalidate cache") |
| 462 | |
| 463 | def test_pin_change_invalidates(self): |
| 464 | self.write_cache(seconds_ago=1) |
| 465 | rc, out, probe_spy = self.run_doctor( |
| 466 | {"LAST30DAYS_X_BACKEND": "bird"}, cached=True |
| 467 | ) |
| 468 | self.assertNotIn("cached-sentinel-report", out) |
| 469 | self.assertTrue(probe_spy.called, "pin change must invalidate cache") |
| 470 | |
| 471 | def test_include_sources_change_invalidates(self): |
| 472 | self.write_cache(seconds_ago=1) |
| 473 | rc, out, probe_spy = self.run_doctor({"INCLUDE_SOURCES": "linkedin"}, cached=True) |
| 474 | self.assertNotIn("cached-sentinel-report", out) |
| 475 | self.assertTrue(probe_spy.called) |
| 476 | |
| 477 | def test_fingerprint_ignores_non_signal_config(self): |
| 478 | # TTL knob is not a fingerprint signal; same-fingerprint serve holds. |
| 479 | self.write_cache(seconds_ago=1) |
| 480 | rc, out, probe_spy = self.run_doctor( |
| 481 | {"LAST30DAYS_DOCTOR_TTL": str(doctor.DEFAULT_CACHE_TTL_SECONDS)}, |
| 482 | cached=True, |
| 483 | ) |
| 484 | self.assertIn("cached-sentinel-report", out) |
| 485 | self.assertFalse(probe_spy.called) |
| 486 | |
| 487 | |
| 488 | class StalenessSignals(_CacheDirCase): |
| 489 | """F12b: from_cache + generated_at surfaced on every doctor report.""" |
| 490 | |
| 491 | def test_matching_fingerprint_serves_cache_with_signals(self): |
| 492 | self.write_cache(seconds_ago=120) |
| 493 | original_ts = json.loads(self.cache_file.read_text(encoding="utf-8"))["timestamp"] |
| 494 | rc, out, probe_spy = self.run_doctor(cached=True) |
| 495 | self.assertEqual(0, rc) |
| 496 | self.assertFalse(probe_spy.called) |
| 497 | data = json.loads(out) |
| 498 | self.assertTrue(data["from_cache"]) |
| 499 | self.assertEqual(original_ts, data["generated_at"]) |
| 500 | |
| 501 | def test_live_run_marks_from_cache_false_with_fresh_generated_at(self): |
| 502 | rc, out, probe_spy = self.run_doctor(cached=False) |
| 503 | self.assertTrue(probe_spy.called) |
| 504 | data = json.loads(out) |
| 505 | self.assertFalse(data["from_cache"]) |
| 506 | age = datetime.datetime.now(datetime.timezone.utc) - datetime.datetime.fromisoformat( |
| 507 | data["generated_at"] |
| 508 | ) |
| 509 | self.assertLess(age.total_seconds(), 60) |
| 510 | |
| 511 | def test_text_mode_prints_cache_status_line(self): |
| 512 | self.write_cache(seconds_ago=1) |
| 513 | rc, out, probe_spy = self.run_doctor(cached=True, emit_json=False) |
| 514 | self.assertFalse(probe_spy.called) |
| 515 | self.assertIn("generated:", out) |
| 516 | self.assertIn("(cached)", out) |
| 517 | |
| 518 | def test_text_mode_live_status_line(self): |
| 519 | rc, out, _ = self.run_doctor(cached=False, emit_json=False) |
| 520 | self.assertIn("generated:", out) |
| 521 | self.assertIn("(live)", out) |
| 522 | |
| 523 | |
| 524 | class CacheWriteFailureWarns(_CacheDirCase): |
| 525 | """F11: a failing cache write warns on stderr and stays non-fatal.""" |
| 526 | |
| 527 | def test_write_failure_warns_and_exits_zero(self): |
| 528 | with _Hermetic() as h: |
| 529 | stdout, stderr = io.StringIO(), io.StringIO() |
| 530 | with mock.patch.object( |
| 531 | Path, "write_text", side_effect=OSError("disk full") |
| 532 | ), redirect_stdout(stdout), redirect_stderr(stderr): |
| 533 | rc = doctor.run({}, emit_json=True, cached=False) |
| 534 | self.assertEqual(0, rc, "cache write failure must never be fatal") |
| 535 | self.assertTrue(h.probe_spy.called) |
| 536 | err = stderr.getvalue() |
| 537 | self.assertIn("WARNING", err) |
| 538 | self.assertIn("doctor cache", err) |
| 539 | self.assertIn("disk full", err) |
| 540 | # The report itself still rendered. |
| 541 | self.assertIn("sources", json.loads(stdout.getvalue())) |
| 542 | |
| 543 | |
| 544 | class NoSecretsInCacheFile(_CacheDirCase): |
| 545 | """Scenario 5: seeded fake credentials never land in the cache file.""" |
| 546 | |
| 547 | def test_cache_file_has_no_secret_values(self): |
| 548 | rc, out, _ = self.run_doctor(dict(FAKE_SECRETS), cached=False) |
| 549 | self.assertEqual(0, rc) |
| 550 | raw = self.cache_file.read_text(encoding="utf-8") |
| 551 | for var, secret in FAKE_SECRETS.items(): |
| 552 | if var == "BSKY_HANDLE": |
| 553 | continue # a handle is an identifier, not a credential |
| 554 | self.assertNotIn(secret, raw, var) |
| 555 | |
| 556 | def test_fingerprint_field_carries_no_secret_values(self): |
| 557 | rc, out, _ = self.run_doctor(dict(FAKE_SECRETS), cached=False) |
| 558 | self.assertEqual(0, rc) |
| 559 | payload = json.loads(self.cache_file.read_text(encoding="utf-8")) |
| 560 | fingerprint = payload["fingerprint"] |
| 561 | # An opaque sha256 hex digest only — no raw values of any kind. |
| 562 | self.assertRegex(fingerprint, r"^[0-9a-f]{64}$") |
| 563 | for var, secret in FAKE_SECRETS.items(): |
| 564 | self.assertNotIn(secret, fingerprint, var) |
| 565 | |
| 566 | |
| 567 | class CliCachedPassthrough(_CacheDirCase): |
| 568 | """`doctor --cached` is an accepted passthrough flag wired to doctor.run.""" |
| 569 | |
| 570 | def _cli(self, argv): |
| 571 | with mock.patch("lib.doctor.run", return_value=0) as run, \ |
| 572 | mock.patch.object(cli.env, "get_config", return_value={}), \ |
| 573 | mock.patch.object(sys, "argv", ["last30days.py"] + argv): |
| 574 | stdout, stderr = io.StringIO(), io.StringIO() |
| 575 | with redirect_stdout(stdout), redirect_stderr(stderr): |
| 576 | rc = cli.main() |
| 577 | return rc, run |
| 578 | |
| 579 | def test_cached_flag_passes_through(self): |
| 580 | rc, run = self._cli(["doctor", "--cached"]) |
| 581 | self.assertEqual(0, rc) |
| 582 | self.assertTrue(run.call_args.kwargs.get("cached")) |
| 583 | |
| 584 | def test_cached_json_combination(self): |
| 585 | rc, run = self._cli(["doctor", "--cached", "--json"]) |
| 586 | self.assertEqual(0, rc) |
| 587 | self.assertTrue(run.call_args.kwargs.get("cached")) |
| 588 | self.assertTrue(run.call_args.kwargs.get("emit_json")) |
| 589 | |
| 590 | def test_plain_doctor_is_live(self): |
| 591 | rc, run = self._cli(["doctor"]) |
| 592 | self.assertEqual(0, rc) |
| 593 | self.assertFalse(run.call_args.kwargs.get("cached")) |
| 594 | |
| 595 | def test_cached_rejected_for_research_topics(self): |
| 596 | with mock.patch.object(sys, "argv", ["last30days.py", "some", "topic", "--cached"]): |
| 597 | stderr = io.StringIO() |
| 598 | with redirect_stderr(stderr), self.assertRaises(SystemExit) as exc: |
| 599 | cli.main() |
| 600 | self.assertEqual(2, exc.exception.code) |
| 601 | self.assertIn("--cached", stderr.getvalue()) |
| 602 | |
| 603 | |
| 604 | class DoctorSkillContract(unittest.TestCase): |
| 605 | """Scenario 6: SKILL.md integration locked against silent erosion |
| 606 | (same read-the-runtime-contract pattern as test_onboarding_contract.py).""" |
| 607 | |
| 608 | @classmethod |
| 609 | def setUpClass(cls): |
| 610 | cls.text = SKILL_MD.read_text(encoding="utf-8") |
| 611 | |
| 612 | def test_doctor_trigger_phrases_present(self): |
| 613 | for phrase in ( |
| 614 | "health check", |
| 615 | "is X working", |
| 616 | "why is a source missing", |
| 617 | "what's broken", |
| 618 | ): |
| 619 | self.assertIn(phrase, self.text, f"missing doctor trigger phrase: {phrase!r}") |
| 620 | |
| 621 | def test_cached_standing_rule_present(self): |
| 622 | self.assertIn("doctor --cached --json", self.text) |
| 623 | self.assertIn("login-backed", self.text) |
| 624 | self.assertIn("doctor-cache.json", self.text) |
| 625 | self.assertIn("LAST30DAYS_DOCTOR_TTL", self.text) |
| 626 | |
| 627 | def test_rerun_live_only_when_stale_or_degraded(self): |
| 628 | self.assertIn("stale", self.text) |
| 629 | self.assertIn("degraded login-backed source", self.text) |
| 630 | |
| 631 | def test_standing_rule_is_marked_mandatory(self): |
| 632 | # The pre-research cache consult is a must-fire rule; it uses the |
| 633 | # same bold MANDATORY marker style as Step 0.45 so it cannot read |
| 634 | # as soft advisory prose (F16b). |
| 635 | self.assertIn("**MANDATORY standing rule.**", self.text) |
| 636 | |
| 637 | def test_frontmatter_description_carries_health_check_keywords(self): |
| 638 | # Cold-start prompts like "is my last30days X search broken?" can |
| 639 | # only load the skill if the machine-parsed frontmatter description |
| 640 | # mentions the health surface (F16a). It must stay ONE line. |
| 641 | lines = self.text.splitlines() |
| 642 | self.assertEqual("---", lines[0]) |
| 643 | closing = lines[1:].index("---") + 1 |
| 644 | frontmatter = lines[1:closing] |
| 645 | desc_lines = [l for l in frontmatter if l.startswith("description:")] |
| 646 | self.assertEqual(1, len(desc_lines), "description must be one line") |
| 647 | desc = desc_lines[0].lower() |
| 648 | for needle in ("doctor", "health check", "diagnose", "broken"): |
| 649 | self.assertIn(needle, desc, f"description missing keyword: {needle!r}") |
| 650 | |
| 651 | def test_predicted_backend_announcement(self): |
| 652 | self.assertIn("active_backend", self.text) |
| 653 | |
| 654 | def test_configuration_md_documents_doctor(self): |
| 655 | config_text = (ROOT / "CONFIGURATION.md").read_text(encoding="utf-8") |
| 656 | for needle in ( |
| 657 | "doctor --json", |
| 658 | "doctor --cached", |
| 659 | "doctor-cache.json", |
| 660 | "LAST30DAYS_DOCTOR_TTL", |
| 661 | "LAST30DAYS_X_BACKEND", |
| 662 | "LAST30DAYS_REDDIT_BACKEND", |
| 663 | "--web-backend", |
| 664 | ): |
| 665 | self.assertIn(needle, config_text, f"CONFIGURATION.md missing: {needle!r}") |
| 666 | |
| 667 | |
| 668 | if __name__ == "__main__": |
| 669 | unittest.main() |
| 670 |