claudetools/.claude/skills/errorlog-dream/scripts/selftest.py

#!/usr/bin/env python
"""selftest for errorlog-dream: run the analyzer against a synthetic errorlog
in a temp dir and assert each detector fires, then assert --apply-archive
moves exactly the old entries and leaves the marker, recent entries, and
unparsed legacy blocks intact."""

import io
import os
import shutil
import sys
import tempfile
from datetime import datetime, timedelta, timezone

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import errorlog_dream as ed


def build_fixture(root, today):
    d = lambda n: (today - timedelta(days=n)).strftime("%Y-%m-%d")
    old1 = d(95)
    old2 = d(65)
    recent = d(3)
    lines = [
        "# Error Log",
        "",
        ed.MARKER,
        "",
        # noise cluster: same machine+skill+shape on 3 distinct days
        "%s | BOX-A | widget/api | HTTP 500 on id 12345 [ctx: cmd=list]" % d(1),
        "",
        "%s | BOX-A | widget/api | HTTP 500 on id 99881 [ctx: cmd=list] (x3)" % d(2),
        "",
        "%s | BOX-A | widget/api | HTTP 500 on id 40404 [ctx: cmd=list]" % recent,
        "",
        # repeat ref (2x) citing an existing memory
        "%s | BOX-B | bash/env | [friction] used /tmp again [ctx: ref=fix_tmp_rule]" % d(4),
        "",
        "%s | BOX-B | bash/env | [friction] used /tmp AGAIN again [ctx: ref=fix_tmp_rule]" % recent,
        "",
        # machine-name case drift
        "%s | box-b | coord | HTTP 0 talking to coord" % d(5),
        "",
        # resolved entry
        "%s | BOX-A | sync/submodules | checkout aborted. [RESOLVED %s] fixed in sync.sh" % (d(6), d(5)),
        "",
        # correction
        "%s | BOX-A | client/foo | [correction] assumed X; correct is Y" % d(7),
        "",
        # old entries -> archive candidates (two months)
        "%s | BOX-A | oldskill | ancient failure one" % old1,
        "",
        "%s | BOX-B | oldskill | ancient failure two" % old2,
        "",
        # legacy multi-line unparsed block (no pipes on first line)
        "Some legacy hand-written note",
        "spanning two lines with no format.",
        "",
    ]
    log = os.path.join(root, "errorlog.md")
    with io.open(log, "w", encoding="utf-8", newline="\n") as fh:
        fh.write("\n".join(lines))
    mem = os.path.join(root, ".claude", "memory")
    os.makedirs(mem)
    with io.open(os.path.join(mem, "fix_tmp_rule.md"), "w", encoding="utf-8") as fh:
        fh.write("---\nname: fix_tmp_rule\n---\nrule body\n")
    return log


def main():
    tmp = tempfile.mkdtemp(prefix="eldream-test-")
    failures = []
    ok = lambda cond, name: failures.append(name) if not cond else None
    try:
        today = datetime.now(timezone.utc)
        log = build_fixture(tmp, today)
        header, entries, unparsed = ed.parse_log(log)

        ok(len(entries) == 10, "parse: 10 entries (got %d)" % len(entries))
        ok(len(unparsed) == 1, "parse: 1 unparsed block")
        ok(any(e.count == 3 for e in entries), "parse: (x3) counter read")

        r = ed.analyze(entries, unparsed, tmp, 60, today)
        ok(r["weighted"] == 12, "weighted count incl x3 (got %d)" % r["weighted"])
        ok(r["by_type"].get("friction") == 2 and r["by_type"].get("correction") == 1,
           "type tally")
        ok(any(ref == "fix_tmp_rule" and n == 2 and exists
               for ref, n, exists, _ in r["repeat_refs"]),
           "repeat-ref detector (existing memory)")
        ok(any(skill == "widget/api" and w == 5 for _, skill, _, w, _ in r["noise"]),
           "noise-cluster detector")
        ok(len(r["resolved"]) == 1, "resolved detector")
        ok("box-b" in r["machine_drift"], "machine-drift detector")
        ok(len(r["archive"]) == 2, "archive candidates (got %d)" % len(r["archive"]))

        report = ed.render(r, 60)
        for token in ("STRENGTHEN?", "SUPPRESS?", "ARCHIVE?", "MACHINE-NAME DRIFT"):
            ok(token in report, "report contains %s" % token)

        # --apply-archive: moves the 2 old entries, keeps everything else
        rc = ed.main(["--log", log, "--root", tmp, "--no-file", "--apply-archive"])
        ok(rc == 0, "apply-archive exit 0")
        h2, e2, u2 = ed.parse_log(log)
        ok(len(e2) == 8, "post-archive: 8 entries kept (got %d)" % len(e2))
        ok(len(u2) == 1, "post-archive: unparsed block untouched")
        with io.open(log, encoding="utf-8") as fh:
            body = fh.read()
        ok(ed.MARKER in body, "post-archive: marker intact")
        ok("ancient failure" not in body, "post-archive: old entries removed")
        arch = os.path.join(tmp, "errorlog-archive")
        months = sorted(f for f in os.listdir(arch) if f.endswith(".md"))
        ok(len(months) == 2, "archive split by month (got %s)" % months)
        joined = ""
        for f in months:
            with io.open(os.path.join(arch, f), encoding="utf-8") as fh:
                joined += fh.read()
        ok("ancient failure one" in joined and "ancient failure two" in joined,
           "archived content present")

        # idempotent: second run archives nothing
        rc = ed.main(["--log", log, "--root", tmp, "--no-file", "--apply-archive"])
        _, e3, _ = ed.parse_log(log)
        ok(rc == 0 and len(e3) == 8, "second apply-archive is a no-op")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)

    if failures:
        print("[ERROR] selftest FAILED:")
        for f in failures:
            print("  - " + f)
        return 1
    print("[OK] errorlog-dream selftest: all assertions passed")
    return 0


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