60 lines
3.5 KiB
Python
60 lines
3.5 KiB
Python
import os, signal, subprocess, sys, shutil
|
|
W='/mnt/storage/scratch/sage-r64/wt'
|
|
OUT='/mnt/storage/scratch/sage-r64/mutants'
|
|
os.makedirs(OUT, exist_ok=True)
|
|
BUNDLE='packages/harness/src/bundle.mjs'; LAUNCH='packages/cli/src/launcher.mjs'
|
|
ROLE='packages/business/src/role.mjs'; RESOLVE='packages/business/src/resolve.mjs'
|
|
CLAUDE='adapters/claude/adapter.sh'; PI='adapters/pi/adapter.sh'
|
|
SKT=('packages/harness','tests/skills.test.mjs')
|
|
LT=('packages/cli','tests/launcher.test.mjs')
|
|
RT=('packages/business','tests/role.test.mjs')
|
|
RST=('packages/business','tests/resolve.test.mjs')
|
|
M=[
|
|
('M1-no-launcher-precheck', LAUNCH, ' readSkills(resolved.skills, skillsDir);\n', '', [LT]),
|
|
('M2-not-enabled-mapped-refused', LAUNCH, 'e.code === "skill-not-enabled" ? "skill-not-enabled" : "skill-refused"', '"skill-refused"', [LT]),
|
|
('M3-no-path-check', BUNDLE, ' if (!SKILL_PATH.test(entry)) throw', ' if (false) throw', [SKT]),
|
|
('M4-no-size-check', BUNDLE, ' if (size.bytes > SKILL_BYTES) throw', ' if (false) throw', [SKT]),
|
|
('M5-no-end-marker-check', BUNDLE, ' if (text.includes(SKILL_END)) throw', ' if (false) throw', [SKT]),
|
|
('M6-walk-follows-symlinks', BUNDLE, ' const stat = lstatSync(path);', ' const stat = statSync(path);', [SKT]),
|
|
('M7-found-follows-symlinks', BUNDLE, ' try { return lstatSync(path); }', ' try { return statSync(path); }', [SKT]),
|
|
('M8-no-newline-before-end', BUNDLE, '\\n${f.text}\\n${SKILL_END}', '\\n${f.text}${SKILL_END}', [SKT]),
|
|
('M9-role-allows-dot', ROLE, ' || name === "." || name === ".."', '', [RT]),
|
|
('M10-resolve-drops-skills', RESOLVE, ' skills: [...definition.skills],', ' skills: [],', [RST, SKT]),
|
|
('M11-claude-slash-on', CLAUDE, ' --disable-slash-commands \\\n', '', [SKT]),
|
|
('M12-claude-unrestricted', CLAUDE, ' --restricted \\\n', '', [SKT]),
|
|
('M13-pi-skills-on', PI, 'SKILLS_FLAG="--no-skills"', 'SKILLS_FLAG=""', [SKT]),
|
|
]
|
|
files=sorted({m[1] for m in M})
|
|
BK={}
|
|
for f in files:
|
|
b=os.path.join(OUT, f.replace('/','__')+'.orig'); shutil.copyfile(os.path.join(W,f), b)
|
|
if open(b,'rb').read()!=open(os.path.join(W,f),'rb').read(): sys.exit('backup failed '+f)
|
|
BK[f]=b
|
|
try:
|
|
for name,f,o,n,tests in M:
|
|
orig=open(BK[f]).read()
|
|
assert orig.count(o)==1, name
|
|
mut=orig.replace(o,n)
|
|
if 'statSync(path)' in n and 'lstatSync' not in n:
|
|
imp='import { lstatSync, mkdirSync,'
|
|
assert mut.count(imp)==1, name+' import'
|
|
mut=mut.replace(imp,'import { lstatSync, statSync, mkdirSync,')
|
|
open(os.path.join(W,f),'w').write(mut)
|
|
results=[]
|
|
for pkg,test in tests:
|
|
lp=os.path.join(OUT,f'{name}--{os.path.basename(test)}.log'); log=open(lp,'w')
|
|
p=subprocess.Popen(['node','--test',test],cwd=os.path.join(W,pkg),stdout=log,stderr=subprocess.STDOUT,start_new_session=True)
|
|
try: rc=p.wait(timeout=300)
|
|
except subprocess.TimeoutExpired:
|
|
os.killpg(p.pid,signal.SIGKILL); p.wait(); rc='timeout'
|
|
log.close()
|
|
txt=open(lp).read()
|
|
fails=[l for l in txt.splitlines() if l.startswith('✖ ') and 'failing tests' not in l]
|
|
results.append((os.path.basename(test),rc,fails[0][:100] if fails else ''))
|
|
killed=any(r[1]!=0 for r in results)
|
|
print(name,'KILLED' if killed else 'SURVIVED','|',' ; '.join(f'{t} rc={rc} {fl}' for t,rc,fl in results),flush=True)
|
|
shutil.copyfile(BK[f],os.path.join(W,f))
|
|
finally:
|
|
for f in files: shutil.copyfile(BK[f],os.path.join(W,f))
|
|
print('restored', all(open(BK[f],'rb').read()==open(os.path.join(W,f),'rb').read() for f in files), flush=True)
|