"""EPOD acceptance tests (2026-10-06). Run against an EPOD MCP server on a test system (probe objects ZEPODT_*, package $TMP).

  MCP_URL=http://127.0.0.1:3000/mcp MCP_TOKEN=... [A4H_URL=http://localhost:50000 A4H_USER=... A4H_PASSWORD=...] python3 run_tests.py [--only T1 T2 ...]

Each test prints PASS or FAIL with the reason and writes the result to epod_tests_result.json. Tests:
  T1 syntax_hint        a rejected write ('save operation failed') comes with syntax messages (line and text)      docs/epod-syntax-hint.md
  T2 no_lock_after_kill a client killed during a write leaves no lock (the next write and the deletion work)       docs/epod-lock-leak.md
  T3 parallel_reads     6 clients read/ATC/syntax-check in parallel without 'Concurrent call detected'
  T4 parallel_writes    3 clients create/write/activate in parallel without 'Concurrent call detected' (a fix: per system or per session lock)
  T5 lock_error_text    a write on a locked object names the lock owner (informational)
"""
import json
import os
import signal
import subprocess
import sys
import threading
import time

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from epod_client import Adt, Mcp  # noqa: E402

CLS = """CLASS {n} DEFINITION PUBLIC FINAL CREATE PUBLIC.
  PUBLIC SECTION.
    METHODS run RETURNING VALUE(rv) TYPE i.
ENDCLASS.
CLASS {n} IMPLEMENTATION.
  METHOD run.
    rv = 1.
  ENDMETHOD.
ENDCLASS.
"""
BAD = """CLASS {n} DEFINITION PUBLIC FINAL CREATE PUBLIC.
  PUBLIC SECTION.
    METHODS run IMPORTING iv_zone TYPE c LENGTH 4 RETURNING VALUE(rv) TYPE i.
ENDCLASS.
CLASS {n} IMPLEMENTATION.
  METHOD run.
    rv = 1.
  ENDMETHOD.
ENDCLASS.
"""
TC = """CLASS ltc DEFINITION FINAL FOR TESTING DURATION SHORT RISK LEVEL HARMLESS.
  PRIVATE SECTION.
    METHODS t1 FOR TESTING.
ENDCLASS.
CLASS ltc IMPLEMENTATION.
  METHOD t1.
    cl_abap_unit_assert=>assert_equals( act = 1 exp = 1 ).
  ENDMETHOD.
ENDCLASS.
"""
created = []
RUN = time.strftime("%H%M%S")


def new_class(m, tag):
    name = "ZEPODT_%s_%s" % (tag, RUN)
    m.call("sap_create_object", {"objectType": "CLAS", "objectName": name, "packageName": "$TMP", "description": "epod acceptance test"})
    created.append("/sap/bc/adt/oo/classes/" + name.lower())
    return name


def ok(text):
    return '"success":true' in text.replace(" ", "")


def t1_syntax_hint():
    with Mcp() as m:
        n = new_class(m, "T1")
        e, t = m.call("sap_push_source", {"objectType": "CLAS", "objectName": n, "source": BAD.format(n=n.lower())})
    low = t.lower()
    has_detail = ("syntaxcheck" in low or "syntax" in low and "line" in low) and "save operation" in low or "line 3" in low or '"line":3' in low.replace(" ", "")
    return bool(has_detail), "the failed write returned: " + t[:300]


def t2_no_lock_after_kill():
    victim = ("import sys,json,os\nsys.path.insert(0,%r)\nfrom epod_client import Mcp\na=json.loads(sys.argv[1])\nm=Mcp().open()\nprint('SENT',flush=True)\nm.call('sap_push_source',a)\n"
              % os.path.dirname(os.path.abspath(__file__)))
    bad = []
    for delay in (0.05, 0.2, 0.4, 0.7):
        with Mcp() as m:
            n = new_class(m, "T2")
            m.call("sap_push_source", {"objectType": "CLAS", "objectName": n, "source": CLS.format(n=n.lower())})
        p = subprocess.Popen([sys.executable, "-c", victim, json.dumps({"objectType": "CLAS", "objectName": n, "includeType": "testclasses", "source": TC})],
                             stdout=subprocess.PIPE, text=True)
        p.stdout.readline()
        time.sleep(delay)
        try:
            os.kill(p.pid, signal.SIGKILL)
        except ProcessLookupError:
            pass
        p.wait()
        time.sleep(3)
        with Mcp() as m:
            e, t = m.call("sap_push_source", {"objectType": "CLAS", "objectName": n, "includeType": "testclasses", "source": TC + "* again\n"})
        if not ok(t):
            bad.append((delay, t[:160]))
    return not bad, "after the kill the next write failed at: %s" % bad if bad else "4 kills (0.05 to 0.7 s), the next write always worked"


def parallel(n_clients, seconds, work):
    out, end = [], time.time() + seconds

    def run(i):
        locks = calls = 0
        with Mcp() as m:
            while time.time() < end:
                t = work(m, i)
                calls += 1
                locks += "Concurrent call detected" in t
        out.append((calls, locks))
    th = [threading.Thread(target=run, args=(i,)) for i in range(n_clients)]
    [x.start() for x in th]
    [x.join() for x in th]
    return sum(c for c, _ in out), sum(l for _, l in out)


def t3_parallel_reads():
    def work(m, i):
        return m.call("sap_search_object", {"query": "CL_ABAP_CHAR_UTIL*", "objType": "CLAS"})[1] + \
            m.call("sap_syntax_check", {"objectType": "CLAS", "objectName": "CL_ABAP_CHAR_UTILITIES"})[1]
    calls, locks = parallel(6, 15, work)
    return locks == 0, "%d rounds, %d 'Concurrent call detected'" % (calls, locks)


def t4_parallel_writes():
    cnt = {"n": 0}

    def work(m, i):
        cnt["n"] += 1
        name = "ZEPODT_P%d_%s%03d" % (i, RUN, cnt["n"] % 1000)
        t = m.call("sap_create_object", {"objectType": "CLAS", "objectName": name, "packageName": "$TMP", "description": "parallel"})[1]
        created.append("/sap/bc/adt/oo/classes/" + name.lower())
        return t + m.call("sap_push_source", {"objectType": "CLAS", "objectName": name, "source": CLS.format(n=name.lower())})[1]
    calls, locks = parallel(3, 25, work)
    return locks == 0, "%d create+write rounds with 3 clients, %d 'Concurrent call detected'" % (calls, locks)


def t5_lock_error_text():
    return True, "informational: see docs/epod-lock-leak.md (no way to create a lock on purpose)"


TESTS = [("T1", "syntax_hint", t1_syntax_hint), ("T2", "no_lock_after_kill", t2_no_lock_after_kill), ("T3", "parallel_reads", t3_parallel_reads),
         ("T4", "parallel_writes", t4_parallel_writes), ("T5", "lock_error_text", t5_lock_error_text)]


def main():
    only = set(sys.argv[sys.argv.index("--only") + 1:]) if "--only" in sys.argv else None
    result = []
    for tid, name, fn in TESTS:
        if only and tid not in only:
            continue
        t0 = time.time()
        try:
            passed, why = fn()
        except Exception as e:  # noqa: BLE001
            passed, why = False, "error: %r" % e
        print("%-4s %-22s %s  (%.0f s)  %s" % (tid, name, "PASS" if passed else "FAIL", time.time() - t0, why[:230]), flush=True)
        result.append({"id": tid, "name": name, "pass": passed, "detail": why[:600]})
    adt = Adt()
    if adt.available() and created:
        try:
            gone = adt.delete(created)
            print("cleanup: %d of %d probe objects deleted" % (sum(gone.values()), len(created)))
        except Exception as e:  # noqa: BLE001
            print("cleanup failed:", repr(e)[:200])
    elif created:
        print("delete these probe objects yourself (SE80 or ADT): ZEPODT_* in $TMP (%d objects)" % len(created))
    json.dump(result, open("epod_tests_result.json", "w"), indent=1)


if __name__ == "__main__":
    main()