106 lines
5.2 KiB
Python
106 lines
5.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Row 298 mutants. Run only with Sage's explicit CLEAR in Boyd's slot.
|
|
|
|
Each mutation is applied to this worktree, runs one whole new test file,
|
|
then restores the exact original bytes, including after an interruption.
|
|
Logs go to a caller-supplied scratch directory, never the repository root.
|
|
"""
|
|
import argparse
|
|
import datetime
|
|
import pathlib
|
|
import subprocess
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parents[4]
|
|
BROKER = ROOT / 'packages/bus/src/broker.mjs'
|
|
VOCABULARY = ROOT / 'packages/business/src/vocabulary.mjs'
|
|
BUS_TEST = 'packages/bus/tests/grant-scope.test.mjs'
|
|
BUSINESS_TEST = 'packages/business/tests/grant-scope.test.mjs'
|
|
|
|
|
|
def replace_once(source, old, new):
|
|
if source.count(old) != 1:
|
|
raise RuntimeError(f'expected exactly one mutant site, found {source.count(old)}')
|
|
return source.replace(old, new, 1)
|
|
|
|
|
|
def mutate(name, source):
|
|
if name == 'M1':
|
|
return replace_once(source,
|
|
'if (!raise?.grant || !grantInScope(scope, raise.grant)) return view;',
|
|
'if (!raise?.grant) return view;')
|
|
if name == 'M2':
|
|
return replace_once(source,
|
|
'return scope === null ? { scope: GRANT_SCOPE_EMPTY, fallback: true } : { scope, fallback: false };',
|
|
'return { scope: scope ?? value, fallback: false };')
|
|
if name == 'M3':
|
|
return replace_once(source,
|
|
' #grantScopeVerb(s, verb, a) {\n this.#human(s);',
|
|
' #grantScopeVerb(s, verb, a) {')
|
|
if name == 'M4':
|
|
old = """ this.#grantEvent(s.business, {
|
|
action: 'grant.scope.used', grantDecision: d.id, decisionSeq: seq, method: raise.grant.method,
|
|
agent: raise.grant.agent, scope, revision, scopeSha256: sha256(JSON.stringify(scope)),
|
|
}, d.id);"""
|
|
return replace_once(source, old, ' // MUTANT: scope evidence omitted.')
|
|
if name == 'M5':
|
|
return replace_once(source,
|
|
"evidence.agent === grant.agent && evidence.scopeSha256 === sha256(JSON.stringify(checked.scope)) &&",
|
|
"evidence.agent === grant.agent &&")
|
|
if name == 'M6':
|
|
start = source.index(" checked = this.#store.transaction(() => {\n const consumed = this.#consumeGrant")
|
|
end = source.index('\n } catch (e) {', start)
|
|
return source[:start] + " checked = this.#store.transaction(() => this.#giteaRecheck(business, decision, this.#consumeGrant(business, decision, 'gitea')));" + source[end:]
|
|
if name == 'M7':
|
|
return replace_once(source,
|
|
'const same = (a, b) => a.length === b.length && a.every((x, i) => x === b[i]);',
|
|
'const same = (a, b) => a.every((x, i) => x === b[i]);')
|
|
if name == 'M8':
|
|
return replace_once(source,
|
|
" if (this.#closed(decision)?.via === 'scope' && !grantInScope(this.#scopeNow(business).scope, g)) fail('scope-changed');",
|
|
" // MUTANT: no spend-time scope recheck.")
|
|
if name == 'M9':
|
|
return replace_once(source,
|
|
' if (s.role !== this.#business(s.business).launch?.by) return view;',
|
|
' // MUTANT: any chief instance can use scope.')
|
|
if name == 'M10':
|
|
start = source.index(' const evidence = JSON.parse(row.body);', source.index(' #scopeUsed('))
|
|
end = source.index('\n }', start)
|
|
return source[:start] + ' return true;' + source[end:]
|
|
raise ValueError(name)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--sage-clear', required=True, help='CLEAR message key; required before any Node run')
|
|
parser.add_argument('--logs', type=pathlib.Path, required=True)
|
|
parser.add_argument('--mutant', choices=['M1', 'M2', 'M3', 'M4', 'M5', 'M6', 'M7', 'M8', 'M9', 'M10', 'all'], default='all')
|
|
parser.add_argument('--deadline', default='2026-10-23T19:38:00+00:00')
|
|
args = parser.parse_args()
|
|
deadline = datetime.datetime.fromisoformat(args.deadline)
|
|
args.logs.mkdir(parents=True, exist_ok=True)
|
|
names = ['M1', 'M2', 'M3', 'M4', 'M5', 'M6', 'M7', 'M8', 'M9', 'M10'] if args.mutant == 'all' else [args.mutant]
|
|
for name in names:
|
|
remaining = (deadline - datetime.datetime.now(datetime.timezone.utc)).total_seconds()
|
|
if remaining <= 15:
|
|
raise RuntimeError('Sage slot deadline reached; no new test run')
|
|
path, test = (VOCABULARY, BUSINESS_TEST) if name in ['M2', 'M7'] else (BROKER, BUS_TEST)
|
|
original = path.read_bytes()
|
|
mutated = mutate(name, original.decode())
|
|
log = args.logs / f'{name}.log'
|
|
try:
|
|
path.write_text(mutated)
|
|
with log.open('w') as output:
|
|
result = subprocess.run(['timeout', '-k', '10', str(int(min(180, remaining - 13))), 'node', '--test', '--test-concurrency=1', test],
|
|
cwd=ROOT, stdout=output, stderr=subprocess.STDOUT,
|
|
timeout=min(195, remaining - 1), check=False)
|
|
print(f'{name}: exit={result.returncode} log={log}', flush=True)
|
|
finally:
|
|
path.write_bytes(original)
|
|
# Exit 1 needs human inspection of the failing test names; errors are not kills.
|
|
if result.returncode == 0:
|
|
raise RuntimeError(f'{name} survived')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|