tests/test_search_and_health.py

#!/usr/bin/env python3 """ test_search_and_health.py - the daily command, and the check that guards it. Every case here is written the same way: do the thing, then BREAK the thing and require the failure. A search test that passes on a store where everything matches proves nothing, and a health check that has never been watched fail is not a check. 1. A question typed the way a person types it finds the row. This is the one that matters most: the old behaviour joined every word with AND, so an ordinary sentence returned "no matches" against a store that held the answer - and the user then re-explained it all by hand, which is the cost this tool exists to remove. 2. Widening is DECLARED. Quietly turning "all your words" into "any of them" would make the results impossible to judge. 3. Session summaries are searchable. They were indexed, populated, documented as searchable, and no command ever read the index. 4. An id from search opens the row it came from. Prompts, summaries and observations all number from 1, so a bare id was ambiguous and `show` silently returned a different row. 5. health FAILS on a dead search index - proven by killing the index. Neither count(*) nor FTS5's own integrity-check can see this. 6. health FAILS when the Stop hook is missing - the case where nothing is ever captured while the light stays green. python3 tests/test_search_and_health.py """ from __future__ import annotations import io import json import os import contextlib import sqlite3 import sys import tempfile import time from pathlib import Path BIN = Path(__file__).resolve().parent.parent / "bin" sys.path.insert(0, str(BIN)) FAILURES: list[str] = [] def check(ok: bool, label: str, detail: str = "") -> None: print((" ok " if ok else " FAIL ") + label + (f" {detail}" if detail else "")) if not ok: FAILURES.append(label) class Args: def __init__(self, **kw): self.query = ""; self.limit = 25; self.json = False; self.prompts = False self.project = None; self.since = None; self.verbose = False; self.quiet = False self.strict = False; self.ids = [] self.__dict__.update(kw) def run(fn, **kw) -> str: buf = io.StringIO() with contextlib.redirect_stdout(buf): fn(Args(**kw)) return buf.getvalue() def seed(db: Path) -> None: import store con = store.connect(db) now = int(time.time() * 1000) rows = [ ("proj-a", "bugfix", "Fixed the cron timer that stopped firing on the build box", "the schedule was never loaded", "The timer was never running at all."), ("proj-a", "change", "Rewrote the deploy script", "idempotent now", "No double apply."), ("proj-b", "change", "Rewrote the deploy script", "idempotent now", "No double apply."), ] for i, (proj, typ, title, sub, body) in enumerate(rows): con.execute( """INSERT INTO observations (session_uid,project,type,title,subtitle,body,created_at, created_at_epoch,redactions) VALUES (?,?,?,?,?,?,?,?,0)""", (f"s{i}", proj, typ, title, sub, body, "2026-08-20T10:00:00Z", now - i * 1000)) con.execute( """INSERT INTO summaries (session_uid,project,request,learned,created_at,created_at_epoch,redactions) VALUES ('s9','proj-b','Make the nightly backup reliable', 'the volume filled and the writer still exited zero', '2026-08-20T12:00:00Z',?,0)""", (now,)) con.commit() con.close() def main() -> int: tmp = Path(tempfile.mkdtemp(prefix="recall-sh-")) db = tmp / "recall.db" # Isolate the DATA ROOT as well as the database. Setting only the db path left # hook.log and the failure report pointing at the real install, so this test read # someone else's state and could also leave litter in it. os.environ["FRIDAY_RECALL_DB"] = str(db) os.environ["FRIDAY_RECALL_ROOT"] = str(tmp) import store store.DB_PATH = db seed(db) import recall recall.store.DB_PATH = db # 1 + 2 - a natural question, and the widening declared out = run(recall.cmd_search, query="why did the cron timer stop firing") check("cron timer" in out, "a question in plain words finds the row", "'why did the cron timer stop firing'") check("no row had all of your words" in out, "widening the query is declared, not silent") # the self-check: the strict reading really does miss, so the pass above # is doing work rather than being true by accident con = sqlite3.connect(f"file:{db}?mode=ro", uri=True) strict = con.execute( "SELECT count(*) FROM observations_fts WHERE observations_fts MATCH ?", (" ".join('"%s"' % w for w in "why did the cron timer stop firing".split()),) ).fetchone()[0] check(strict == 0, " (self-check) the un-widened query really does return nothing") con.close() # 3 - summaries reachable out = run(recall.cmd_search, query="nightly backup") check("s1" in out, "session summaries are searchable", "id s1 returned") # 4 - a tagged id opens the row it came from out = run(recall.cmd_show, ids=["s1"]) check("session summary" in out and "nightly backup" in out.lower(), "`show s1` opens the summary, not observation 1") out = run(recall.cmd_show, ids=["1"]) check("cron timer" in out, "`show 1` still opens observation 1") # 4b - --json carries the tagged id, not the raw one out = run(recall.cmd_search, query="nightly backup", json=True) payload = json.loads(out) tagged = [r for r in payload if r.get("kind") == "summary"] check(bool(tagged) and tagged[0]["id"] == "s1", "--json reports the tagged id", f"got {tagged[0]['id']!r}" if tagged else "no summary row") check(bool(tagged) and tagged[0].get("row_id") == 1, " --json still carries the raw row_id for consumers that want it") # 5 - dedupe out = run(recall.cmd_search, query="deploy script") check(out.count("Rewrote the deploy script") == 1, "the same title captured twice is shown once") # 6 - health fails on a dead search index def health_rc() -> int: buf = io.StringIO() with contextlib.redirect_stdout(buf), contextlib.redirect_stderr(buf): rc = recall.cmd_health(Args()) return rc before = health_rc() con = sqlite3.connect(db) con.execute("INSERT INTO observations_fts(observations_fts) VALUES('delete-all')") con.commit() still_counts = con.execute("SELECT count(*) FROM observations_fts").fetchone()[0] con.close() after = health_rc() check(still_counts > 0, " (self-check) count(*) still reports rows on a dead index", f"{still_counts}") check(after != 0, "health FAILS when the search index is dead") check(before != after, " (self-check) that check can also pass", f"{before} -> {after}") # 7 - health fails when the Stop hook is gone home = tmp / "home" (home / ".claude").mkdir(parents=True) (home / ".claude" / "settings.json").write_text(json.dumps({"hooks": { "SessionStart": [{"hooks": [{"type": "command", "command": f'{sys.executable} "{BIN}/hook.py" session-start'}]}]}})) # Redirect EVERY variable a platform might use to answer "where is home". # # Setting HOME alone is a POSIX habit. On Windows Path.home() reads USERPROFILE # and ignores HOME completely, so this check was reading the REAL user's # ~/.claude/settings.json instead of the one it had just written - and then # reporting a failure about the tester's own machine rather than about the code. # Same mistake, three separate places in this suite. If a test needs a home # directory, it has to move all of them. _home_vars = ("HOME", "USERPROFILE", "HOMEDRIVE", "HOMEPATH") _saved = {k: os.environ.get(k) for k in _home_vars} os.environ["HOME"] = os.environ["USERPROFILE"] = os.environ["HOMEPATH"] = str(home) os.environ["HOMEDRIVE"] = "" try: notes = recall._hook_health() finally: for k, v in _saved.items(): if v is None: os.environ.pop(k, None) else: os.environ[k] = v check(any("Stop is not wired" in n for n in notes), "health names the MISSING Stop hook", "nothing would ever be captured") check(all(not n.startswith("HOOK") or "Stop" in n or "SessionStart" in n for n in notes), " (self-check) the wired SessionStart hook is not reported missing") # 8 - the probe must not invent a failure on a title that is not English con2 = sqlite3.connect(db) con2.execute("""INSERT INTO observations(session_uid,project,type,title,subtitle, body,created_at,created_at_epoch,redactions) VALUES('s9','proj-c','change', ?,'sub','body','2026-08-20T10:00:00Z',?,0)""", ("\u00dcberpr\u00fcfung der Datenbank", int(time.time() * 1000) + 5000)) con2.commit(); con2.close() _c = sqlite3.connect(f"file:{db}?mode=ro", uri=True) _c.row_factory = sqlite3.Row probe = recall._search_probe(_c) check(probe is None, "a non-English title does not trigger a false SEARCH IS BROKEN", repr(probe)) # 9 - a clean capture must not erase a DIFFERENT lane's failures log = Path(recall._hook_log()); log.parent.mkdir(parents=True, exist_ok=True) log.write_text("\n".join([ "session-start FAILED: boom", "stop: abc123 prompts=4 obs=2 redactions=0", "session-start FAILED: boom again", "stop: abc124 prompts=4 obs=2 redactions=0", ]) + "\n") streak, _ = recall._hook_failure_streak() check(streak == 2, "a broken session-start survives clean captures", f"streak={streak}") # and a clean capture DOES still clear capture failures log.write_text("\n".join([ "stop: capture rc=1 boom", "stop: abc125 prompts=4 obs=2 redactions=0", ]) + "\n") streak2, _ = recall._hook_failure_streak() check(streak2 == 0, " a clean capture still clears CAPTURE failures", f"streak={streak2}") log.unlink() # 10 - health must not crash on a settings file of an unexpected shape shapes = ['{"hooks":[1,2]}', '{"hooks":{"Stop":"nope"}}', '{"hooks":{"Stop":[{"hooks":"x"}]}}', '{"hooks":{"Stop":["notadict"]}}'] crashed = [] for i, shape in enumerate(shapes): h = tmp / f"home{i}" / ".claude" h.mkdir(parents=True, exist_ok=True) (h / "settings.json").write_text(shape) # All of them again, for the same reason as above: on Windows, HOME is not # the one that is read. os.environ["HOME"] = os.environ["USERPROFILE"] = str(h.parent) os.environ["HOMEPATH"] = str(h.parent) os.environ["HOMEDRIVE"] = "" try: recall._hook_health() except Exception as e: crashed.append(f"{shape} -> {type(e).__name__}") for k, v in _saved.items(): if v is None: os.environ.pop(k, None) else: os.environ[k] = v check(not crashed, "health survives a settings.json of any shape", "; ".join(crashed)) # 11 - the build refuses to package its own output, by CONTENT not just by name sys.path.insert(0, str(BIN.parent)) import build shipped = [f.name for f in build.sources(BIN.parent)] bad = [n for n in shipped if n.endswith((".tar.gz", ".sha256", ".zip", ".pyc"))] check(not bad, "no build output or checksum file is packaged as source", str(bad)) print() if FAILURES: print(f"SEARCH/HEALTH TEST FAILED ({len(FAILURES)}): " + "; ".join(FAILURES), file=sys.stderr) return 1 print("search/health test OK - plain questions land, summaries are reachable,") print(" ids are unambiguous, and health can see a dead index and a missing hook") return 0 if __name__ == "__main__": sys.exit(main())