70 lines
3.8 KiB
Python
70 lines
3.8 KiB
Python
"""Bind one isolated analysis source gate to committed inputs and actual output."""
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
|
|
base = Path(__file__).resolve().parent.parent
|
|
root = base / '.local/analysis-core-worktree-20260911'
|
|
source, name, mode = sys.argv[1:4]
|
|
assert re.fullmatch(r'[a-zA-Z0-9-]+', name)
|
|
assert subprocess.check_output(['git', 'rev-parse', 'HEAD'], cwd=root).decode().strip() == source
|
|
out = base / '.local/analysis-source-gates-20260913'
|
|
out.mkdir(exist_ok=True)
|
|
assert not any((out / (name + s)).exists() for s in ('.inputs.json', '.log', '.result.json'))
|
|
prefix = ['cargo', 'test', '--locked', '-p', 'oa-core', '--lib']
|
|
commands = {
|
|
'list': prefix + ['--', '--list'],
|
|
'focused': prefix + [sys.argv[4] if len(sys.argv) > 4 else 'analysis_source', '--', '--test-threads=1'],
|
|
'full': prefix + ['--', '--test-threads=1'],
|
|
'clippy': ['cargo', 'clippy', '--locked', '-p', 'oa-core', '--all-targets', '--', '-D', 'warnings'],
|
|
}
|
|
command = commands[mode]
|
|
paths = subprocess.check_output(['git', 'ls-tree', '-r', '--name-only', source, 'Cargo.toml', 'Cargo.lock', 'rust-toolchain.toml', '.cargo', 'build', 'crates', 'assets', 'docs', 'examples', 'harness/fixtures'], cwd=root).decode().splitlines()
|
|
blobs = subprocess.check_output(['git', 'cat-file', '--batch'], cwd=root,
|
|
input=''.join(f'{source}:{p}\n' for p in paths).encode())
|
|
offset, inputs = 0, []
|
|
for path in paths:
|
|
end = blobs.index(b'\n', offset)
|
|
_, kind, size = blobs[offset:end].split()
|
|
assert kind == b'blob'
|
|
size = int(size)
|
|
raw = blobs[end + 1:end + 1 + size]
|
|
assert (root / path).read_bytes() == raw, path
|
|
inputs.append({'path': path, 'bytes': size, 'sha256': hashlib.sha256(raw).hexdigest()})
|
|
offset = end + size + 2
|
|
assert offset == len(blobs)
|
|
manifest = out / (name + '.inputs.json')
|
|
manifest.write_text(json.dumps({'source': source, 'inputs': inputs}, indent=2) + '\n', encoding='utf-8', newline='\n')
|
|
started = datetime.now(timezone.utc).isoformat()
|
|
print(json.dumps({'start': started, 'command': command, 'source': source, 'inputs': len(inputs)}), flush=True)
|
|
log_path = out / (name + '.log')
|
|
with log_path.open('xb') as stream:
|
|
run = subprocess.run(command, cwd=root, stdout=stream, stderr=subprocess.STDOUT)
|
|
ended = datetime.now(timezone.utc).isoformat()
|
|
changed = [i['path'] for i in inputs if hashlib.sha256((root / i['path']).read_bytes()).hexdigest() != i['sha256']]
|
|
head_after = subprocess.check_output(['git', 'rev-parse', 'HEAD'], cwd=root).decode().strip()
|
|
log = log_path.read_text(encoding='utf-8', errors='replace')
|
|
members = re.findall(r'^([^\r\n]+): test\r?$', log, re.M) if mode == 'list' else []
|
|
artifacts = []
|
|
for match in re.findall(r'Running unittests [^\r\n]+ \(([^\r\n]+\.exe)\)', log):
|
|
path = Path(match)
|
|
if not path.is_absolute():
|
|
path = root / path
|
|
raw = path.read_bytes()
|
|
artifacts.append({'path': str(path), 'bytes': len(raw), 'sha256': hashlib.sha256(raw).hexdigest()})
|
|
record = {'source': source, 'head_after': head_after, 'start': started, 'end': ended,
|
|
'exit': run.returncode, 'command': command, 'inputs': manifest.name,
|
|
'changed_inputs': changed, 'artifacts': artifacts,
|
|
'environment': {k: os.environ.get(k) for k in ['CARGO_TARGET_DIR', 'CARGO_BUILD_JOBS', 'CARGO_HOME', 'RUSTUP_HOME']},
|
|
'listing_members': members, 'listing_count': len(members),
|
|
'listing_scope': 'Names emitted by this actual Cargo test listing; focused/full runtime results have their own receipts.'}
|
|
(out / (name + '.result.json')).write_text(json.dumps(record, indent=2) + '\n', encoding='utf-8', newline='\n')
|
|
print(log[-14000:])
|
|
print(json.dumps({k:v for k,v in record.items() if k != 'listing_members'}))
|
|
sys.exit(run.returncode or int(bool(changed) or head_after != source))
|