107 lines
6.6 KiB
Python
107 lines
6.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Run row225 mutants in disposable snapshots, only on Sage CLEAR.
|
|
|
|
Commit the tested candidate first. Never change the worktree or shared dependencies.
|
|
A timeout, import failure or process failure is not a mutation kill.
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import tarfile
|
|
import tempfile
|
|
|
|
BROWSER = 'packages/webui/tests/chat-replies-browser.test.mjs'
|
|
ROUTES = 'packages/webui/tests/library-routes.test.mjs'
|
|
CHAT = 'packages/webui/src/public/chat.js'
|
|
SERVE = 'packages/webui/src/serve.mjs'
|
|
RENDER = 'packages/library/src/render.mjs'
|
|
MUTANTS = [
|
|
('B1-forged-notice', RENDER, 'options.check == null ? notice(value, options.name) : null', 'notice(value, options.name)', BROWSER),
|
|
('N1-definite-refusal', CHAT, 'data?.sent === false || res.status >= 400 && res.status < 500', 'data?.sent === false', BROWSER),
|
|
('M1-badges', RENDER, 'badge: scope, path:', 'badge: null, path:', BROWSER),
|
|
('M1-agent-label', RENDER, 'canon: c.canon, agent,', 'canon: c.canon, agent: null,', BROWSER),
|
|
('M2-http-links', RENDER, "new URL(url).protocol === 'https:' ? url : null", 'url', BROWSER),
|
|
('M3-wrong-commit', CHAT, 'e.file !== e.last', 'cite.commit !== e.last', BROWSER),
|
|
('M3-missing-file', CHAT, "SHA.test(e?.file ?? '') && SHA.test(e?.last ?? '')", "SHA.test(e?.last ?? '')", BROWSER),
|
|
('M4-fallback', CHAT, 'Array.isArray(limits.fallback) ? limits.fallback.filter', 'Array.isArray(v.replyLimitsFallback) ? v.replyLimitsFallback.filter', BROWSER),
|
|
('M4-problem', CHAT, "if (CODE.test(limits.problem ?? '')) parts.push(`<code>${txt(limits.problem)}</code>`);", '', BROWSER),
|
|
('M5-send', SERVE, 'if (!send || !libraryRevert || !named)', 'if (!libraryRevert || !named)', ROUTES),
|
|
('M5-project-receipt', SERVE, 'result.project !== value.project || ', '', ROUTES),
|
|
('M5-commit-input', SERVE, '!COMMIT.test(value.commit)', 'false', ROUTES),
|
|
('M5-origin', SERVE, "if (one(req, 'origin') !== ownOrigin(preview)) return refuse(403, 'cross-origin request refused', 'a revert needs the Console\\'s own Origin');", '', ROUTES),
|
|
('M5-fetch-metadata', SERVE, "if (one(req, 'sec-fetch-site') !== 'same-origin') return refuse(403, 'cross-site request refused', 'a revert needs same-origin Fetch Metadata');", '', ROUTES),
|
|
('M6-duplicate-call', SERVE, 'result = await libraryRevert({ project: value.project, commit: value.commit });', 'result = await libraryRevert({ project: value.project, commit: value.commit }); await libraryRevert({ project: value.project, commit: value.commit });', ROUTES),
|
|
('M7-user-fence', SERVE, "if (path === '/api/chat' || path.startsWith('/api/chat/')) return true;", "if (path.startsWith('/api/library/')) return true; if (path === '/api/chat' || path.startsWith('/api/chat/')) return true;", ROUTES),
|
|
]
|
|
|
|
|
|
def digest(path):
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--clear', required=True, help='Sage CLEAR authorizing this run')
|
|
parser.add_argument('--output', required=True, type=Path)
|
|
parser.add_argument('--only', help='Comma-separated mutant names for a bounded review delta')
|
|
args = parser.parse_args()
|
|
names = set(args.only.split(',')) if args.only else {m[0] for m in MUTANTS}
|
|
if names - {m[0] for m in MUTANTS}:
|
|
raise SystemExit('Unknown mutant name')
|
|
selected = [m for m in MUTANTS if m[0] in names]
|
|
repo = Path(subprocess.check_output(['git', 'rev-parse', '--show-toplevel'], text=True).strip())
|
|
if subprocess.check_output(['git', 'status', '--porcelain', '--', 'packages'], text=True).strip():
|
|
raise SystemExit('Commit tested packages first: snapshot must match candidate.')
|
|
dependency = repo / 'node_modules'
|
|
if dependency.is_symlink() or not dependency.is_dir():
|
|
raise SystemExit('Install dependencies locally in the worktree; no canonical dependency link.')
|
|
head = subprocess.check_output(['git', 'rev-parse', 'HEAD'], text=True).strip()
|
|
before = {p: digest(repo / p) for p in [CHAT, SERVE, RENDER]}
|
|
args.output.mkdir(parents=True, exist_ok=True)
|
|
results = []
|
|
with tempfile.TemporaryDirectory(prefix='mosaic-row225-mutants-') as directory:
|
|
scratch = Path(directory)
|
|
archive = scratch / 'candidate.tar'
|
|
with archive.open('wb') as stream:
|
|
subprocess.run(['git', 'archive', 'HEAD', 'packages', 'roles', 'contracts', 'tasks', 'missions', 'package.json', 'RELEASE'], stdout=stream, check=True)
|
|
for name, target, old, new, test in selected:
|
|
tree = scratch / name
|
|
tree.mkdir()
|
|
with tarfile.open(archive) as bundle:
|
|
bundle.extractall(tree, filter='data')
|
|
(tree / 'node_modules').symlink_to((repo / 'node_modules').resolve(), target_is_directory=True)
|
|
path = tree / target
|
|
source = path.read_text()
|
|
hits = source.count(old)
|
|
if hits != 1:
|
|
raise SystemExit(f'{name}: expected one anchor, found {hits}')
|
|
path.write_text(source.replace(old, new, 1))
|
|
command = ['timeout', '-k', '10', '240' if test == BROWSER else '120', 'node', '--test', '--test-reporter=tap', '--test-concurrency=1', '--test-timeout=30000', test]
|
|
run = subprocess.run(command, cwd=tree, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, env=os.environ)
|
|
log = args.output / f'{name}.tap'
|
|
log.write_text(run.stdout)
|
|
path.write_text(source)
|
|
restored = digest(path) == hashlib.sha256(source.encode()).hexdigest()
|
|
worktree_clean = before == {p: digest(repo / p) for p in before}
|
|
if not restored or not worktree_clean:
|
|
raise SystemExit(f'{name}: restore/hash check failed')
|
|
killed = run.returncode == 1 and 'ERR_ASSERTION' in run.stdout and 'not ok ' in run.stdout
|
|
results.append({'mutant': name, 'anchorHits': hits, 'restored': restored, 'worktreeUnchanged': worktree_clean, 'exit': run.returncode, 'killed': killed, 'test': test, 'log': str(log)})
|
|
print(json.dumps(results[-1]), flush=True)
|
|
shutil.rmtree(tree)
|
|
if not killed:
|
|
break
|
|
after = {p: digest(repo / p) for p in before}
|
|
report = {'head': head, 'clear': args.clear, 'worktreeUnchanged': before == after, 'selected': sorted(names), 'results': results}
|
|
(args.output / 'summary.json').write_text(json.dumps(report, indent=2) + '\n')
|
|
if before != after or len(results) != len(selected) or not all(r['killed'] for r in results):
|
|
raise SystemExit(1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|