diff --git a/README.md b/README.md index 0c51132..e94a336 100644 --- a/README.md +++ b/README.md @@ -155,9 +155,10 @@ an edge table only when hop depth or corpus size demands it. - No incremental indexing (tracked in the issues). Packaging exists since v0.3.0 (`pip install sqlite-graph-memory`). There IS an eval suite now — see [Measure before you swap](#measure-before-you-swap--eval) — but it scores retrieval on wikilink-derived gold, which is weak supervision and says so out loud. Test coverage is uneven: - `index_notes.py`, `brain_ask.py` and `mcp_server.py` have unit tests, `turnstate_hook.py` - has none. Run `pytest -q` for the current count rather than trusting a number written in - prose here — that is how this file ended up claiming "no tests" while seven were passing. + `index_notes.py`, `brain_ask.py` and `mcp_server.py` have unit tests, and + `turnstate_hook.py` has write-path coverage. Run `pytest -q` for the current count rather + than trusting a number written in prose here — that is how this file ended up claiming + "no tests" while seven were passing. The status line above says **pilot** on purpose. ## Quickstart diff --git a/tests/test_turnstate_hook.py b/tests/test_turnstate_hook.py new file mode 100644 index 0000000..b306eec --- /dev/null +++ b/tests/test_turnstate_hook.py @@ -0,0 +1,126 @@ +"""Tests for the Stop hook's per-turn SQLite ledger write path.""" +import io +import json +import os +import sqlite3 +import subprocess +import sys +import tempfile +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "src")) + +from sqlite_graph_memory import turnstate_hook # noqa: E402 + + +class _Stdin: + def __init__(self, payload): + self.buffer = io.BytesIO(payload) + + +def _event(kind, content): + return {"type": kind, "message": {"content": content}} + + +def _run_hook(monkeypatch, payload): + monkeypatch.setattr(turnstate_hook.sys, "stdin", _Stdin(json.dumps(payload).encode())) + turnstate_hook.main() + + +def test_main_writes_extracted_turn_to_ledger(tmp_path, monkeypatch): + db_path = tmp_path / "turns.db" + cwd = tmp_path / "my-project" + cwd.mkdir() + transcript_path = tmp_path / "transcript.jsonl" + events = [ + _event("user", "How should we test the ledger write path?"), + _event("assistant", [ + {"type": "text", "text": "Decision: we will write one pytest file."}, + {"type": "tool_use", "name": "Write", + "input": {"file_path": str(cwd / "test_turnstate_hook.py")}}, + {"type": "tool_use", "name": "Bash", + "input": {"command": "python -m pytest -q"}}, + ]), + ] + transcript_path.write_text( + "\n".join(json.dumps(event) for event in events) + "\n", encoding="utf-8") + + monkeypatch.setenv("TURNSTATE_DB", str(db_path)) + monkeypatch.setattr(tempfile, "tempdir", str(tmp_path)) + _run_hook(monkeypatch, { + "session_id": "test-session", + "transcript_path": str(transcript_path), + "cwd": str(cwd), + }) + + with sqlite3.connect(db_path) as con: + row = con.execute( + "SELECT ask, files, tools, commands, decisions, project FROM turns" + ).fetchone() + + assert row is not None + ask, files, tools, commands, decisions, project = row + assert ask == "How should we test the ledger write path?" + assert json.loads(files) == [str(cwd / "test_turnstate_hook.py")] + assert json.loads(tools) == ["Write", "Bash"] + assert json.loads(commands) == ["python -m pytest -q"] + assert json.loads(decisions) == ["Decision: we will write one pytest file."] + assert project == "my-project" + + +def test_main_does_not_write_again_without_new_transcript_bytes(tmp_path, monkeypatch): + db_path = tmp_path / "turns.db" + transcript_path = tmp_path / "transcript.jsonl" + transcript_path.write_text( + json.dumps(_event("user", "Record this turn in the ledger.")) + "\n", + encoding="utf-8") + payload = { + "session_id": "repeat-session", + "transcript_path": str(transcript_path), + "cwd": str(tmp_path), + } + monkeypatch.setenv("TURNSTATE_DB", str(db_path)) + monkeypatch.setattr(tempfile, "tempdir", str(tmp_path)) + + _run_hook(monkeypatch, payload) + _run_hook(monkeypatch, payload) + + with sqlite3.connect(db_path) as con: + count = con.execute("SELECT COUNT(*) FROM turns").fetchone()[0] + assert count == 1 + + +def test_module_wrapper_exits_successfully_on_bad_input(tmp_path): + env = os.environ.copy() + env["PYTHONPATH"] = str(ROOT / "src") + env["TURNSTATE_DB"] = str(tmp_path / "turns.db") + result = subprocess.run( + [sys.executable, "-m", "sqlite_graph_memory.turnstate_hook"], + input=b"not json", + capture_output=True, + cwd=ROOT, + env=env, + check=False, + ) + assert result.returncode == 0 + + +def test_module_wrapper_exits_successfully_when_transcript_is_missing(tmp_path): + env = os.environ.copy() + env["PYTHONPATH"] = str(ROOT / "src") + env["TURNSTATE_DB"] = str(tmp_path / "turns.db") + payload = { + "session_id": "missing-session", + "transcript_path": str(tmp_path / "missing.jsonl"), + "cwd": str(tmp_path), + } + result = subprocess.run( + [sys.executable, "-m", "sqlite_graph_memory.turnstate_hook"], + input=json.dumps(payload).encode(), + capture_output=True, + cwd=ROOT, + env=env, + check=False, + ) + assert result.returncode == 0