homelab/hosts/mars/gitea-pr-comment-filter-test.py

"""Contract test for gitea-pr-comment-filter.py.

Hermes treats "[SILENT]"/empty/nonzero-exit as ignore, a JSON object as a
payload replacement, and ANY OTHER stdout text as allow-with-script_output.
So each case asserts on the exact stdout discipline, not just the decision.
"""
import json, subprocess, sys, pathlib

SCRIPT = str(pathlib.Path(__file__).with_name("gitea-pr-comment-filter.py"))

def payload(action="created", author="darman", body="please fix the typo",
            previous=None, is_pull=True, cid=42, number=7):
    p = {"action": action, "is_pull": is_pull,
         "comment": {"id": cid, "body": body, "user": {"login": author},
                     "html_url": "https://git.mgaction.town/darman/homelab/pulls/7#issuecomment-42"},
         "issue": {"number": number, "title": "some PR"},
         "repository": {"full_name": "darman/homelab"},
         "sender": {"login": author}}
    if previous is not None:
        p["changes"] = {"body": {"from": previous}}
    return p

def run(p):
    r = subprocess.run([sys.executable, SCRIPT], input=json.dumps(p),
                       capture_output=True, text=True)
    return r.returncode, r.stdout, r.stderr

def classify(rc, out):
    """Replicate Hermes's own interpretation of the script result."""
    if rc != 0 or out.strip() == "" or out.strip() == "[SILENT]":
        return "IGNORED"
    try:
        v = json.loads(out)
        return "ALLOWED" if isinstance(v, dict) else "ALLOWED(script_output)"
    except ValueError:
        return "ALLOWED(script_output)"

fails = []
def check(name, p, expect):
    rc, out, err = run(p)
    got = classify(rc, out)
    ok = got == expect
    print(f"{'PASS' if ok else 'FAIL'}  {name:<52} {got}")
    if not ok:
        fails.append(name); print(f"      expected {expect}; stdout={out!r} stderr={err.strip()!r}")
    return out

# --- the loop guard, the whole reason this exists ---
check("luna's own comment is dropped (LOOP GUARD)", payload(author="luna"), "IGNORED")
check("luna in different case is dropped",          payload(author="LUNA"), "IGNORED")

# --- action handling ---
check("created by human is allowed",                payload(), "ALLOWED")
check("deleted is dropped",                         payload(action="deleted"), "IGNORED")
check("edited with changed body is allowed",
      payload(action="edited", body="new text", previous="old text"), "ALLOWED")
check("edited with unchanged body is dropped",
      payload(action="edited", body="same", previous="same"), "IGNORED")
check("unknown action is dropped",                  payload(action="reopened"), "IGNORED")

# --- misc guards ---
check("issue comment (is_pull=false) is dropped",   payload(is_pull=False), "IGNORED")
check("empty body is dropped",                      payload(body="   "), "IGNORED")
check("missing comment object is dropped",          {"action": "created"}, "IGNORED")
check("malformed payload is dropped",               "not-a-dict", "IGNORED")

# --- normalisation: the prompt's {changes.body.from} must always resolve ---
out = check("created event still allowed", payload(), "ALLOWED")
norm = json.loads(out)
c1 = norm.get("changes", {}).get("body", {}).get("from")
print(f"{'PASS' if c1 == '' else 'FAIL'}  {'created: changes.body.from normalised to empty':<52} {c1!r}")
if c1 != "": fails.append("normalise-created")

out = check("edited event still allowed", payload(action="edited", body="new", previous="old"), "ALLOWED")
c2 = json.loads(out).get("changes", {}).get("body", {}).get("from")
print(f"{'PASS' if c2 == 'old' else 'FAIL'}  {'edited: changes.body.from preserved':<52} {c2!r}")
if c2 != "old": fails.append("normalise-edited")

# --- payload passthrough: prompt paths must survive the transform ---
norm = json.loads(run(payload())[1])
for path in [("comment","id"), ("comment","body"), ("comment","user","login"),
             ("comment","html_url"), ("issue","number"), ("issue","title"),
             ("repository","full_name"), ("action",)]:
    cur, ok = norm, True
    for k in path:
        if isinstance(cur, dict) and k in cur: cur = cur[k]
        else: ok = False; break
    label = ".".join(path)
    print(f"{'PASS' if ok else 'FAIL'}  {'prompt path survives: {' + label + '}':<52} {cur if ok else 'MISSING'}")
    if not ok: fails.append(f"path-{label}")

# --- drop contract: nonzero exit + empty stdout + reason on stderr ---
# Nonzero is what gets the reason into the gateway log (Hermes logs
# "script ignored webhook path=... code=... stderr=..." only on that path).
rc, out, err = run(payload(author="luna"))
print(f"{'PASS' if rc == 3 else 'FAIL'}  {'drop exits 3 (not 0, so Hermes logs it)':<52} rc={rc}")
if rc != 3: fails.append("drop-exit-code")
print(f"{'PASS' if out == '' else 'FAIL'}  {'drop writes nothing to stdout':<52} {out!r}")
if out != "": fails.append("drop-stdout-empty")
print(f"{'PASS' if 'luna' in err else 'FAIL'}  {'drop names the rule on stderr':<52} {err.strip()[-44:]!r}")
if "luna" not in err: fails.append("stderr-reason")

# a crash must stay distinguishable from a deliberate drop
rc, out, err = run("not-a-dict")
print(f"{'PASS' if rc == 3 else 'FAIL'}  {'malformed payload is a drop (3), not a crash':<52} rc={rc}")
if rc != 3: fails.append("malformed-exit-code")

print()
print("ALL PASSED" if not fails else "FAILURES: " + ", ".join(fails))
sys.exit(1 if fails else 0)