erhan/abap-llm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Object type mix: INTF, TABL, STRU, MSAG, exception tasks (harness G2, mutants, generator notes), balanced generator, kind-deficit job order, dashboard mix card

Browse Source 
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Kral
2026-10-05 18:52:29 +02:00
parent 454db6626c
commit b119f1afac
39 changed files with 2329 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
harness/mutation.py
View File

@@ -148,6 +148,72 @@ def mutants(src, otype, seed, n=MAX_MUTANTS):
return out
def decl_mutants(src, otype, seed, n=MAX_MUTANTS):
"""Mutants of declarations that carry the behavior of DDIC objects and interfaces (no executable code):
TABL / STRU: field length, decimals, data type, key flag. INTF: constant values, type lengths."""
sites = [] # (kind, start, end, new, line, old)
def add(kind, m, new):
sites.append((kind, m.start(), m.end(), new, src.count("\n", 0, m.start()) + 1, m.group(0)))
if otype in ("TABL", "STRU"):
for m in re.finditer(r"abap\.(char|numc|dec|curr|quan|lang|cuky|unit)\((\d+)(?:,\s*(\d+))?\)", src, re.I):
kind_, ln, dec = m.group(1).lower(), int(m.group(2)), m.group(3)
if kind_ in ("char", "numc") and ln > 1:
add("length", m, f"abap.{kind_}({ln - 1})")
if dec is not None and int(dec) < ln - 1:
add("decimals", m, f"abap.{kind_}({ln},{int(dec) + 1})")
if kind_ == "char" and ln > 1:
add("type", m, f"abap.numc({ln})")
for m in re.finditer(r"abap\.(int4|int8|timestamp|dats|tims)\b", src, re.I):
add("type", m, {"int4": "abap.int8", "int8": "abap.int4", "timestamp": "abap.dats",
"dats": "abap.tims", "tims": "abap.dats"}[m.group(1).lower()])
for m in re.finditer(r"^(\s*)key(\s+)(?!client\b)(\w+\s*:)", src, re.I | re.M):
add("key", m, f"{m.group(1)}{m.group(3)}")
if otype == "INTF":
for m in re.finditer(r"(\bVALUE\s+)(\d+)(?=\s*\.)", src, re.I):
add("const", m, f"{m.group(1)}{int(m.group(2)) + 1}")
for m in re.finditer(r"(\bVALUE\s+)'([^'\n]{1,20})'", src, re.I):
v = m.group(2)
add("lit", m, f"{m.group(1)}'" + ("Z" if v[0] != "Z" else "Y") + v[1:] + "'")
for m in re.finditer(r"(\bVALUE\s+)(abap_true|abap_false)\b", src, re.I):
add("bool", m, m.group(1) + ("abap_false" if m.group(2).lower() == "abap_true" else "abap_true"))
for m in re.finditer(r"\bLENGTH\s+(\d+)", src, re.I):
if int(m.group(1)) > 1:
add("length", m, f"LENGTH {int(m.group(1)) - 1}")
rnd = random.Random(seed)
rnd.shuffle(sites)
chosen, kinds = [], set()
for prefer_new in (True, False):
for st in sites:
if len(chosen) >= n:
break
if st in chosen or (prefer_new and st[0] in kinds):
continue
chosen.append(st)
kinds.add(st[0])
return [(f"line {line}: {old.strip()} -> {new.strip()} ({kind})", src[:a] + new + src[b:])
for kind, a, b, new, line, old in sorted(chosen, key=lambda x: x[1])]
def msag_mutants(messages, seed, n=MAX_MUTANTS):
"""Mutants of a message class: changed text, changed placeholder, a message moved to another number."""
out = []
for i, m in enumerate(messages):
t = m.get("text", "")
mut = [dict(x) for x in messages]
mut[i]["text"] = t + " x" if len(t) < 70 else t[:-1]
out.append((f"message {m['msgno']}: text + ' x' (text)", mut))
if "&1" in t:
mut = [dict(x) for x in messages]
mut[i]["text"] = t.replace("&1", "&2", 1)
out.append((f"message {m['msgno']}: &1 -> &2 (placeholder)", mut))
if len(messages) > 1:
mut = [dict(x) for x in messages]
mut[i]["msgno"] = str(int(m["msgno"]) + 50).zfill(3)
out.append((f"message {m['msgno']}: number + 50 (number)", mut))
random.Random(seed).shuffle(out)
return out[:n]
def check_task(pool_root, task_id, run_base, n=MAX_MUTANTS, keep=False):
"""Run the mutants of one task. Returns a summary dict; writes it to <task>/mutation.json."""
task_dir = os.path.join(pool_root, task_id)
@@ -156,14 +222,22 @@ def check_task(pool_root, task_id, run_base, n=MAX_MUTANTS, keep=False):
def is_test(o):
return o["type"] == "CLAS" and re.search(r"^\s*CLASS\s+\S+\s+DEFINITION[^.]*FOR\s+TESTING",
open(os.path.join(task_dir, o["file"])).read(), re.I | re.M)
targets = [o for o in meta["reference"] if o.get("file") and o["type"] in ("CLAS", "FUNC", "PROG", "DDLS")
targets = [o for o in meta["reference"] if o.get("file") and o["type"] in ("CLAS", "FUNC", "PROG", "DDLS", "INTF",
"TABL", "STRU")
and not is_test(o)]
targets += [o for o in meta["reference"] if o["type"] == "MSAG" and o.get("messages")]
work = os.path.join(ROOT, "runs", "gen", "mut")
os.makedirs(work, exist_ok=True)
runner = Runner(os.path.join(work, "pool"), os.path.join(work, "runs"))
# candidates per object, then round robin: objects without mutation sites (exception classes) give their share
cand = [[(o, d, m) for d, m in mutants(open(os.path.join(task_dir, o["file"])).read(), o["type"],
f"{task_id}:{o['name']}", n)] for o in targets]
def _muts(o):
if o["type"] == "MSAG":
return msag_mutants(o["messages"], f"{task_id}:{o['name']}", n)
src = open(os.path.join(task_dir, o["file"])).read()
if o["type"] in ("TABL", "STRU", "INTF"):
return decl_mutants(src, o["type"], f"{task_id}:{o['name']}", n)
return mutants(src, o["type"], f"{task_id}:{o['name']}", n)
cand = [[(o, d, m) for d, m in _muts(o)] for o in targets]
plan = []
while len(plan) < n and any(cand):
for c in cand:
@@ -174,7 +248,14 @@ def check_task(pool_root, task_id, run_base, n=MAX_MUTANTS, keep=False):
mdir = os.path.join(work, "pool", task_id)
shutil.rmtree(mdir, ignore_errors=True)
shutil.copytree(task_dir, mdir)
open(os.path.join(mdir, o["file"]), "w").write(msrc)
if o["type"] == "MSAG": # the mutant changes the messages of the reference in task.json
tj = json.load(open(os.path.join(mdir, "task.json")))
for r in tj["reference"]:
if r["name"] == o["name"]:
r["messages"] = msrc
json.dump(tj, open(os.path.join(mdir, "task.json"), "w"), indent=2)
else:
open(os.path.join(mdir, o["file"]), "w").write(msrc)
rep, _ = runner.run(task_id, OracleAgent(), run_base + k)
h = rep.get("hidden_tests") or {}
g = rep.get("gates") or {}
@@ -187,7 +268,7 @@ def check_task(pool_root, task_id, run_base, n=MAX_MUTANTS, keep=False):
results.append({"object": o["name"], "mutant": desc, "status": status,
"hidden": f"{h.get('passed')}/{h.get('total')}",
"failed_tests": [d["method"] for d in h.get("detail", []) if not d["ok"]]})
if keep and status == "killed": # candidate faulty reference for own-test scoring
if keep and status == "killed" and o["type"] != "MSAG": # candidate faulty reference for own-test scoring
fdir = os.path.join(task_dir, "faulty")
os.makedirs(fdir, exist_ok=True)
open(os.path.join(fdir, f"m{k}_{os.path.basename(o['file'])}"), "w").write(msrc)