chorus/eval/test_patch_semantics.py

#!/usr/bin/env python3
"""test_patch_semantics.py — H4: prove the hi-fi mock models REAL PATCH semantics.

These are exactly the behaviors the shipped naive mock cannot reproduce (it appends
everything at EOF), so without this the section-replace / section-append / missing-
heading / frontmatter-replace paths are untested. Drives the real chorus.py against
mock_olrapi_hifi.py.

Run:  python test_patch_semantics.py [--port 8820]
"""
from __future__ import annotations

import argparse
import os
import subprocess
import sys
import tempfile
import time
import urllib.request
from pathlib import Path

HERE = Path(__file__).resolve().parent
CHORUS = HERE.parent / "chorus-memory.plugin.src" / "skills" / "chorus-memory" / "scripts" / "chorus.py"
KEY = "test-key-not-a-real-secret"

failures = []


def check(name, cond, detail=""):
    print(f"{'ok  ' if cond else 'FAIL'} {name}" + (f"  -- {detail}" if not cond else ""))
    if not cond:
        failures.append(name)


def http(method, url, body=None, headers=None):
    data = body.encode() if isinstance(body, str) else body
    req = urllib.request.Request(url, data=data, method=method,
                                 headers={"Authorization": f"Bearer {KEY}", **(headers or {})})
    try:
        with urllib.request.urlopen(req, timeout=10) as r:
            return r.status, r.read().decode("utf-8", "replace")
    except Exception as e:  # noqa: BLE001
        return getattr(e, "code", 0), ""


def tmp(content):
    f = tempfile.NamedTemporaryFile("w", suffix=".md", delete=False, encoding="utf-8")
    f.write(content)
    f.close()
    return f.name


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--port", type=int, default=8820)
    a = ap.parse_args()
    base = f"http://127.0.0.1:{a.port}"
    srv = subprocess.Popen([sys.executable, str(HERE / "mock_olrapi_hifi.py"), "--port", str(a.port)],
                           stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)

    def chorus(*args):
        env = dict(os.environ, CHORUS_BASE=base, CHORUS_KEY=KEY, CHORUS_VERIFY="0")
        return subprocess.run([sys.executable, str(CHORUS), *args], capture_output=True, text=True, env=env)

    def ground(path):
        _, body = http("GET", f"{base}/__debug__?path={path}")
        return None if body == "<<MISSING>>" else body

    try:
        for _ in range(50):
            try:
                urllib.request.urlopen(f"{base}/__debug__reset", data=b"", timeout=1)
                break
            except Exception:
                time.sleep(0.1)

        doc = "doc.md"
        seed = ("---\ntype: note\nupdated: 2026-01-01\n---\n\n"
                "# Doc\n\n## Scope\nold scope\n\n## Other\nkeep me\n")

        # replace: Scope body becomes new; Other untouched.
        http("PUT", f"{base}/vault/{doc}", seed)
        chorus("patch", doc, "replace", "heading", "Doc::Scope", tmp("new scope"))
        g = ground(doc) or ""
        check("heading replace swaps section body", "new scope" in g and "old scope" not in g, g)
        check("heading replace leaves sibling section intact", "## Other\nkeep me" in g, g)

        # append: lands INSIDE the section (before '## Other'), not at EOF.
        chorus("patch", doc, "append", "heading", "Doc::Scope", tmp("appended line"))
        g = ground(doc) or ""
        check("heading append inserts within the section (not EOF)",
              "appended line" in g and g.index("appended line") < g.index("## Other"), g)

        # prepend: lands at the TOP of the section body.
        chorus("patch", doc, "prepend", "heading", "Doc::Scope", tmp("first line"))
        g = ground(doc) or ""
        check("heading prepend inserts at section top",
              "first line" in g and g.index("first line") < g.index("new scope"), g)

        # missing heading -> 400 -> chorus.py exits non-zero (the silent-loss guard).
        r = chorus("patch", doc, "append", "heading", "Doc::Nope", tmp("lost?"))
        check("missing heading fails loud (non-zero exit)", r.returncode != 0, r.stderr)
        check("missing heading does not write", "lost?" not in (ground(doc) or ""))

        # frontmatter replace on an existing field updates it.
        chorus("fm", doc, "updated", "2026-06-22")
        g = ground(doc) or ""
        check("frontmatter replace updates existing field", "updated: 2026-06-22" in g, g)

        # fm on a MISSING field: the server still 400s the raw PATCH (matches real API),
        # but cmd_fm (v1.5) falls back to a surgical create — the key is inserted and
        # every other line is preserved. (The old contract failed loud on exactly the
        # notes that needed repair.)
        before = ground(doc) or ""
        r = chorus("fm", doc, "nonexistent_field", "x")
        check("fm creates a missing field (create-or-replace)", r.returncode == 0,
              r.stderr or r.stdout)
        g = ground(doc) or ""
        check("fm-created key holds the value", "nonexistent_field: x" in g, g)
        check("fm create preserves every existing line",
              all(line in g for line in before.splitlines() if line.strip()), g)

        # raw `patch replace frontmatter` (no fallback) still fails loud on a missing key.
        r = chorus("patch", doc, "replace", "frontmatter", "still_missing", tmp('"x"'))
        check("raw frontmatter PATCH of a missing field fails loud", r.returncode != 0, r.stderr)

        print(f"\n{len(failures)} failure(s)" if failures else "\nall PATCH-semantics tests passed")
        return 1 if failures else 0
    finally:
        srv.terminate()


if __name__ == "__main__":
    raise SystemExit(main())