G: EPOD acceptance tests; harness: hide and clean other runs' mid-name objects (proxy, teardown, sweep); lock leak cause (Eclipse restart during a write)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Kral
174
epod_tests/run_tests.py
Normal file
174
epod_tests/run_tests.py
Normal file
@@ -0,0 +1,174 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user