Packet commit, agents/researcher/work/queue-98 only. Secrets scan clean. Not pushed. Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
69 lines
3.4 KiB
Python
69 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
# Row 98 (#1592) mutants of scripts/webui-service.sh. Each edit must match exactly once and each runs
|
|
# alone against the test file that holds its case. Every run starts in its own process group and the
|
|
# group is killed when the run ends or times out. The script is restored from the bytes read before
|
|
# the edit, and its sha256 is checked against its start at the end.
|
|
# Usage: mutants.py <worktree> <out.txt> [timeout-seconds]
|
|
import hashlib, os, signal, subprocess, sys, time
|
|
|
|
WT, OUT = sys.argv[1], sys.argv[2]
|
|
TIMEOUT = int(sys.argv[3]) if len(sys.argv) > 3 else 300
|
|
S = 'scripts/webui-service.sh'
|
|
SVC, PIN = 'tests/service.test.mjs', 'tests/pinned-copy.test.mjs'
|
|
|
|
# (name, old, new, test file)
|
|
M = [
|
|
('1 bus check refuses only not-found', '[ "$load" = loaded ] ||', '[ "$load" != not-found ] ||', SVC),
|
|
('2 rename failure without a target is not refused', ' [ -e "$target" ] || die "could not rename the copy into $target"\n', '', PIN),
|
|
('3 mv without -T', 'mv -T "$COPY_TMP" "$target"', 'mv "$COPY_TMP" "$target"', PIN),
|
|
('4a status copy id loses both anchors', '[[ "$copy" =~ ^[0-9a-f]{40}$ ]]', '[[ "$copy" =~ [0-9a-f]{40} ]]', SVC),
|
|
('4b status copy id loses its start anchor', '[[ "$copy" =~ ^[0-9a-f]{40}$ ]]', '[[ "$copy" =~ [0-9a-f]{40}$ ]]', SVC),
|
|
('4c status copy id loses its end anchor', '[[ "$copy" =~ ^[0-9a-f]{40}$ ]]', '[[ "$copy" =~ ^[0-9a-f]{40} ]]', SVC),
|
|
('5a bus check reads only the first unit', ' for unit in $units; do\n out="$(systemctl --user show', ' for unit in $(printf \'%s\\n\' "$units" | head -n 1); do\n out="$(systemctl --user show', SVC),
|
|
('5b bus check applies the path rule to the first unit only', ' for p in $paths; do\n', ' [ "$unit" = "$(printf \'%s\\n\' "$units" | head -n 1)" ] || continue\n for p in $paths; do\n', SVC),
|
|
]
|
|
|
|
env = dict(os.environ, TMPDIR=os.environ.get('TMPDIR', '/mnt/storage/scratch/tmp'))
|
|
full = os.path.join(WT, S)
|
|
sha = lambda: hashlib.sha256(open(full, 'rb').read()).hexdigest()
|
|
start = sha()
|
|
|
|
def run(test, limit):
|
|
t0 = time.time()
|
|
p = subprocess.Popen(['node', '--test', test], cwd=os.path.join(WT, 'packages/webui'), env=env,
|
|
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, start_new_session=True, text=True)
|
|
try:
|
|
out, _ = p.communicate(timeout=limit)
|
|
code = p.returncode
|
|
except subprocess.TimeoutExpired:
|
|
code = 'timeout'
|
|
finally:
|
|
try: os.killpg(p.pid, signal.SIGKILL)
|
|
except ProcessLookupError: pass
|
|
try: p.communicate(timeout=5)
|
|
except Exception: pass
|
|
return code, round(time.time() - t0, 1)
|
|
|
|
with open(OUT, 'w') as f:
|
|
def say(s): print(s); f.write(s + '\n'); f.flush()
|
|
say(f'worktree {WT}')
|
|
say(f'start {time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}')
|
|
killed = 0
|
|
for name, old, new, test in M:
|
|
orig = open(full, 'rb').read()
|
|
src = orig.decode()
|
|
if src.count(old) != 1:
|
|
say(f'{name}: NO MATCH ({src.count(old)}), skipped'); continue
|
|
try:
|
|
open(full, 'w').write(src.replace(old, new))
|
|
code, secs = run(test, TIMEOUT)
|
|
finally:
|
|
open(full, 'wb').write(orig)
|
|
dead = code not in (0, 'timeout')
|
|
killed += dead
|
|
verdict = 'killed' if dead else 'STALLED' if code == 'timeout' else 'ALIVE'
|
|
say(f'{name}: {verdict} (exit {code}, {secs}s, {test.split("/")[-1]})')
|
|
ok = sha() == start
|
|
say(f'killed {killed} of {len(M)}; script restored byte for byte: {ok}')
|
|
sys.exit(0 if ok else 1)
|