FabledScribe/tests/test_after_write_hook.py

"""The PostToolUse after-write hook (#2901): code written through Bash — sed,
heredocs, scripts — gets the same prior-art / ledger checks as a Write/Edit.

Runs the real shell against a temp git repo and a throwaway HTTP sink, like
the pre-write hook's end-to-end tests. Skips where the hook's tools are
missing; asserts on content where they are present."""
from __future__ import annotations

import http.server
import json
import os
import shutil
import subprocess
import threading
import urllib.parse
from pathlib import Path

import pytest

PLUGIN = Path(__file__).resolve().parents[1] / "plugin"
HOOK = PLUGIN / "hooks" / "scribe_after_write.sh"


def _env(tmp_path, url="http://127.0.0.1:9"):
    for tool in ("git", "jq", "curl", "bash"):
        if shutil.which(tool) is None:
            pytest.skip(f"hook runtime tool {tool!r} not installed")
    return {"PATH": os.environ["PATH"], "SCRIBE_URL": url, "SCRIBE_TOKEN": "t",
            "TMPDIR": str(tmp_path), "HOME": str(tmp_path),
            "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@x", 
            "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@x"}


def _repo(tmp_path, env):
    repo = tmp_path / "repo"
    repo.mkdir()
    subprocess.run(["git", "init", "-q"], cwd=repo, check=True, env=env)
    (repo / "b.py").write_text("def one():\n    return 1\n")
    (repo / "c.py").write_text("def slug(t):\n    return t.lower()\n")
    subprocess.run(["git", "add", "."], cwd=repo, check=True, env=env)
    subprocess.run(["git", "commit", "-q", "-m", "base"], cwd=repo, check=True, env=env)
    return repo


def _run(repo, env, session="s-after-1", tool="Bash"):
    out = subprocess.run(
        ["bash", str(HOOK)],
        input=json.dumps({"session_id": session, "cwd": str(repo), "tool_name": tool,
                          "tool_input": {"command": "cat > x"}, "tool_response": {}}),
        capture_output=True, text=True, env=env,
    )
    assert out.returncode == 0, out.stderr
    return out.stdout


class _Sink(http.server.BaseHTTPRequestHandler):
    seen: list[dict] = []
    reply = b'{"context":"> family named","note_ids":[],"sync_note_ids":[],"derive_keys":["dup:483a"]}'

    def do_GET(self):
        type(self).seen.append(urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query))
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(type(self).reply)

    def log_message(self, *a):
        pass


@pytest.fixture
def sink():
    _Sink.seen = []
    server = http.server.HTTPServer(("127.0.0.1", 0), _Sink)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    try:
        yield server
    finally:
        server.shutdown()


def test_after_write_names_what_bash_just_wrote_then_stays_quiet_until_the_next_change(tmp_path, sink):
    env = _env(tmp_path, url=f"http://127.0.0.1:{sink.server_port}")
    repo = _repo(tmp_path, env)
    # "A Bash call" wrote an untracked stylesheet and appended to a tracked file.
    (repo / "a.css").write_text(".log-empty {\n  color: red;\n}\n")
    (repo / "b.py").write_text("def one():\n    return 1\n\ndef slug(t):\n    return t\n")
    out = _run(repo, env)
    by_path = {q["path"][0]: q for q in _Sink.seen}
    assert set(by_path) == {"a.css", "b.py"}           # repo-relative, like the pre hook
    assert by_path["a.css"]["shapes"] == ["css:log-empty"]
    assert by_path["b.py"]["shapes"] == ["sym:slug"]
    # Added lines only for the tracked file — the existing def is not "just written".
    assert "def slug" in by_path["b.py"]["code"][0] and "def one" not in by_path["b.py"]["code"][0]
    ctx = json.loads(out)["hookSpecificOutput"]
    assert ctx["hookEventName"] == "PostToolUse"
    assert "> family named" in ctx["additionalContext"]
    # The local by-name arm rides along: `slug` already lives in c.py.
    assert "`slug` is already defined in 1 other file(s): c.py" in ctx["additionalContext"]
    # Derive keys landed on the SHARED channel the pre-write hook reads.
    state = tmp_path / "scribe-priorart" / "s-after-1.derive.ids"
    assert "dup:483a" in state.read_text().split()

    # Nothing changed → one git status, no request, no output.
    _Sink.seen = []
    assert _run(repo, env) == ""
    assert _Sink.seen == []

    # Another change → only that file, and the dedup channel goes back up.
    (repo / "a.css").write_text(".log-empty {\n  color: red;\n}\n.other {\n  margin: 0;\n}\n")
    _run(repo, env)
    assert [q["path"][0] for q in _Sink.seen] == ["a.css"]
    assert _Sink.seen[0]["exclude_derive"] == ["dup:483a"]
    assert set(_Sink.seen[0]["shapes"][0].split(",")) == {"css:log-empty", "css:other"}


def test_after_write_is_silent_where_it_has_nothing_to_say(tmp_path):
    env = _env(tmp_path)
    repo = _repo(tmp_path, env)
    # Not a Bash call → nothing (hooks.json matches Bash, the script re-checks).
    (repo / "a.css").write_text(".x {\n  color: red;\n}\n")
    assert _run(repo, env, tool="Write") == ""
    # Not a git repo → nothing.
    loose = tmp_path / "loose"
    loose.mkdir()
    (loose / "a.css").write_text(".x {\n  color: red;\n}\n")
    assert _run(loose, env, session="s-loose") == ""
    # A change that defines nothing (prose, a call-site edit) → nothing, even
    # with the server unreachable (port 9 refuses): no definitions, no arms.
    (repo / "README.md").write_text("# notes\n")
    (repo / "b.py").write_text("def one():\n    return one_more()\n")
    assert _run(repo, env, session="s-quiet") == ""


def test_after_write_local_arm_and_record_nudge_work_without_a_server(tmp_path):
    """The local by-name arm needs no instance (#2280) and the record nudge
    (#2664) fails open with it — a refused connection stands in for the
    instance."""
    env = _env(tmp_path)
    repo = _repo(tmp_path, env)
    (repo / "d.py").write_text("def slug(t):\n    return t.lower()\n")
    out = _run(repo, env, session="s-local")
    ctx = json.loads(out)["hookSpecificOutput"]["additionalContext"]
    assert "`slug` is already defined in 1 other file(s): c.py" in ctx
    assert "create_snippet" in ctx