33 lines
2.0 KiB
Python
33 lines
2.0 KiB
Python
#!/usr/bin/env python3
|
|
import hashlib,json,os,pathlib,re,signal,subprocess,sys
|
|
root=pathlib.Path('/mnt/storage/src/mosaic-stack-rocko-49')
|
|
out=pathlib.Path(__file__).resolve().parent/'mutants'
|
|
paths=['packages/bus/schema.sql','packages/bus/src/migrations/3b-3c.sql','packages/bus/src/store.mjs']
|
|
original={path:(root/path).read_bytes() for path in paths}
|
|
results=[]
|
|
for id,pattern in [
|
|
('19-append-only-cursor-evidence',r'CREATE TRIGGER task_comment_cursors_no_(?:update|delete|replace)\b.*?END;\n'),
|
|
('20-monotonic-cursor',r'CREATE TRIGGER task_comment_cursors_forward\b.*?END;\n'),
|
|
]:
|
|
try:
|
|
for path in paths[:2]:
|
|
text=original[path].decode();text,n=re.subn(pattern,'',text,flags=re.S)
|
|
assert n==(3 if id.startswith('19') else 1)
|
|
(root/path).write_text(text)
|
|
sha=hashlib.sha256((root/paths[0]).read_bytes()).hexdigest()
|
|
(root/paths[2]).write_text(re.sub(r"export const SCHEMA_SHA256 = '[a-f0-9]+';",f"export const SCHEMA_SHA256 = '{sha}';",original[paths[2]].decode()))
|
|
with (out/(id+'.log')).open('w') as log:
|
|
p=subprocess.Popen(['node','--test','packages/bus/tests/comment-cursors.test.mjs'],cwd=root,
|
|
env={**os.environ,'DOCKER_HOST':'unix:///nonexistent.sock'},stdout=log,stderr=subprocess.STDOUT,start_new_session=True)
|
|
try:rc=p.wait(timeout=90)
|
|
except subprocess.TimeoutExpired:os.killpg(p.pid,signal.SIGKILL);p.wait();rc='timeout'
|
|
content=(out/(id+'.log')).read_text()
|
|
invalid=any(x in content for x in ['SyntaxError:', 'ERR_MODULE_NOT_FOUND','schema-source-mismatch','migration-target-mismatch'])
|
|
verdict='INVALID' if invalid else 'TIMEOUT' if rc=='timeout' else 'SURVIVED' if rc==0 else 'KILLED'
|
|
results.append(dict(id=id,path='packages/bus/schema.sql + matching migration DDL',test='packages/bus/tests/comment-cursors.test.mjs',result=verdict,exit=rc))
|
|
print(id,verdict,rc,flush=True)
|
|
finally:
|
|
for path,data in original.items():(root/path).write_bytes(data)
|
|
(out/'sql-results.json').write_text(json.dumps(results,indent=2)+'\n')
|
|
sys.exit(0 if all(r['result']=='KILLED' for r in results) else 1)
|