Files
ARCADIA-Studio/harness/experiments/analysis-source-s2c-windows-20260913/helpers/admit-analysis-s2c-gate.py
T

194 lines
9.6 KiB
Python

"""Admit only frozen S2c listing or explicit typed/raw-token focused gates.
Usage: NAME [ADMITTED_LIST_NAME SELECTOR]
Selectors: language::analysis_source or language::raw_tokens.
No compiler/test invocation. No full-suite pass or old failure waiver.
"""
from pathlib import Path
import hashlib
import json
import re
import subprocess
import sys
BASE = Path(__file__).resolve().parent.parent
ROOT = BASE / '.local/analysis-core-worktree-20260911'
FOLDER = BASE / '.local/analysis-source-gates-20260913'
SOURCE = 'f1d79df4979121335ed7b4ce4da84b2fca8a9ef4'
BASELINE = 'a2a28f8d23a46ba1696b6a8410d4d3b5c1808f49'
BASE_NAME = 'list-a2a28f8'
BASE_HELPER_SHA = '92c03d25c8f5ae75a5c1e76bd4fa24cbca7aeb10cf7ee5847bc8a324cae3ec06'
BASE_PINS = {
'.result.json': '86db5467964d3712510f867379ae32e0797b4d9f46cbde762119ed9055fd3a62',
'.inputs.json': '08b6dabbffbd06f7abea717f9aa34cd910c41c4a8c650247006a0f6858561c84',
'.log': '8be8bd7ae75a690f0e36c8b0b39dabc4c5bd5654cf1e653bd43e3c0cddf560df',
'.admission.json': 'f6b08e4d0eac3a0f17a582f9d9a51fb0cf84fc3c30d94d8fac71df52a4442bf1',
}
PREFIX = ['cargo', 'test', '--locked', '-p', 'oa-core', '--lib']
SCOPES = ['Cargo.toml', 'Cargo.lock', 'rust-toolchain.toml', '.cargo', 'build',
'crates', 'assets', 'docs', 'examples', 'harness/fixtures']
SELECTORS = {'language::analysis_source': 44, 'language::raw_tokens': 1}
NEW_NAMES = {'language::analysis_source::write::tests::' + name for name in [
'typed_oa_sa_canonical_roundtrips_match_independent_projects',
'typed_declaration_order_and_required_empty_values_are_canonical',
'all_expr_variants_and_precedence_preserve_exact_trees',
'canonical_calls_use_callee_parameter_order_and_qualified_names',
'subtree_output_respects_parent_context_and_retained_delimiters',
'canonical_strings_and_exact_literals_use_existing_encodings',
'writer_limits_are_aggregate_exact_and_latched_without_partial_results',
'writer_lookup_scope_shape_and_depth_failures_are_model_located',
'negate_override_roundtrip_preserves_depth128_within_frame256',
'iterative_writer_preserves_width1024_and_empty_call_topology',
]}
assert len(NEW_NAMES) == 10
def digest(data):
return hashlib.sha256(data).hexdigest()
def git(*arguments, data=None):
return subprocess.check_output(['git', *arguments], cwd=ROOT, input=data)
watched = {}
def read(path):
path = Path(path)
data = path.read_bytes()
assert watched.setdefault(path, data) == data, 'Changed input during admission: ' + str(path)
return data
def raw(name, suffix):
assert re.fullmatch(r'[a-zA-Z0-9-]+', name)
return read(FOLDER / (name + suffix))
def document(name, suffix):
return json.loads(raw(name, suffix))
def closed(record, source):
assert type(record['exit']) is int and record['exit'] == 0
assert record['source'] == record['head_after'] == source and record['changed_inputs'] == []
assert record['start'] and record['end'] and record['start'] <= record['end']
def verify_admission(name, admission):
for key, suffix in [('result_sha256', '.result.json'), ('inputs_sha256', '.inputs.json'), ('log_sha256', '.log')]:
assert admission[key] == digest(raw(name, suffix)), (name, key)
def listing_names(record, log, count):
assert record['command'] == PREFIX + ['--', '--list']
names = re.findall(r'^([^\r\n]+): test\r?$', log, re.M)
assert names == record['listing_members'] and len(names) == len(set(names)) == record['listing_count'] == count
assert re.findall(r'^(\d+) tests, (\d+) benchmarks\s*$', log, re.M) == [(str(count), '0')]
return names
assert len(sys.argv) in (2, 4), 'NAME [ADMITTED_LIST_NAME SELECTOR]'
name = sys.argv[1]
assert re.fullmatch(r'[a-zA-Z0-9-]+', name)
output = FOLDER / (name + '.admission.json')
assert not output.exists(), 'Existing admissions cannot be overwritten'
self_sha = digest(read(Path(__file__)))
for suffix, expected_sha in BASE_PINS.items():
assert digest(raw(BASE_NAME, suffix)) == expected_sha, 'Changed historical S2b listing'
assert digest(read(BASE / '.local/admit-analysis-s2b-gate.py')) == BASE_HELPER_SHA
baseline = document(BASE_NAME, '.result.json')
baseline_admission = document(BASE_NAME, '.admission.json')
closed(baseline, BASELINE)
verify_admission(BASE_NAME, baseline_admission)
assert baseline_admission['source'] == BASELINE and baseline_admission['mode'] == 'list'
assert baseline_admission['verified_inputs'] == 531 and baseline_admission['admission_helper_sha256'] == BASE_HELPER_SHA
assert len(baseline['artifacts']) == 1 and baseline_admission['artifact'] == baseline['artifacts'][0]
baseline_names = listing_names(baseline, raw(BASE_NAME, '.log').decode('utf-8'), 142)
assert baseline_names == baseline_admission['names'] and not (set(baseline_names) & NEW_NAMES)
expected_names = set(baseline_names) | NEW_NAMES
assert len(expected_names) == 152
for selector, count in SELECTORS.items():
assert len([n for n in expected_names if selector in n]) == count
assert git('rev-parse', 'HEAD').decode().strip() == SOURCE
subprocess.run(['git', 'merge-base', '--is-ancestor', BASELINE, SOURCE], cwd=ROOT, check=True)
paths = git('ls-tree', '-r', '--name-only', SOURCE, *SCOPES).decode().splitlines()
assert len(paths) == len(set(paths)) == 534
blobs = git('cat-file', '--batch', data=''.join(f'{SOURCE}:{p}\n' for p in paths).encode())
offset, expected_inputs, writer_tests = 0, [], None
for path in paths:
end = blobs.index(b'\n', offset)
_, kind, size = blobs[offset:end].split()
size = int(size)
data = blobs[end + 1:end + 1 + size]
assert kind == b'blob' and len(data) == size and read(ROOT / path) == data, path
expected_inputs.append({'path': path, 'bytes': size, 'sha256': digest(data)})
if path == 'crates/oa-core/src/language/analysis_source/write/tests.rs':
writer_tests = data.decode('utf-8')
offset = end + size + 2
assert offset == len(blobs) and writer_tests is not None
authored = {'language::analysis_source::write::tests::' + n for n in re.findall(r'#\[test\]\s*fn (\w+)\(', writer_tests)}
assert authored == NEW_NAMES
record = document(name, '.result.json')
closed(record, SOURCE)
assert record['inputs'] == name + '.inputs.json'
assert document(name, '.inputs.json') == {'source': SOURCE, 'inputs': expected_inputs}
assert len(record['artifacts']) == 1
artifact = record['artifacts'][0]
artifact_path = Path(artifact['path'])
assert artifact_path.is_absolute()
data = read(artifact_path)
assert len(data) == artifact['bytes'] and digest(data) == artifact['sha256']
log = raw(name, '.log').decode('utf-8')
executables = re.findall(r'Running unittests [^\r\n]+ \(([^\r\n]+\.exe)\)', log)
assert len(executables) == 1
executable = Path(executables[0])
if not executable.is_absolute():
executable = ROOT / executable
assert executable.resolve() == artifact_path.resolve()
if record['command'] == PREFIX + ['--', '--list']:
assert len(sys.argv) == 2
names = listing_names(record, log, 152)
assert set(names) == expected_names
mode, selector = 'list', None
else:
assert len(sys.argv) == 4, 'Full and other commands cannot receive an S2c pass admission'
listing_name, selector = sys.argv[2:]
assert selector in SELECTORS and record['command'] == PREFIX + [selector, '--', '--test-threads=1']
assert record['listing_members'] == [] and record['listing_count'] == 0
listing = document(listing_name, '.result.json')
admission = document(listing_name, '.admission.json')
closed(listing, SOURCE)
verify_admission(listing_name, admission)
assert admission['source'] == SOURCE and admission['mode'] == 'list'
assert admission['admission_helper_sha256'] == self_sha and admission['verified_inputs'] == 534
assert listing['artifacts'] == [artifact] and admission['artifact'] == artifact
assert document(listing_name, '.inputs.json') == {'source': SOURCE, 'inputs': expected_inputs}
listed = listing_names(listing, raw(listing_name, '.log').decode('utf-8'), 152)
assert listed == admission['names'] and set(listed) == expected_names
names = [n for n in listed if selector in n]
assert len(names) == SELECTORS[selector]
actual = re.findall(r'^test ([^\r\n]+) \.\.\. (ok|FAILED|ignored)(?:, [^\r\n]*)?\s*$', log, re.M)
assert actual == [(n, 'ok') for n in names], 'Every focused case must run exactly once and pass; no ignored cases'
assert re.findall(r'test result: ok\. (\d+) passed; (\d+) failed; (\d+) ignored; (\d+) measured; (\d+) filtered out;', log) == [(str(len(names)), '0', '0', '0', str(152-len(names)))]
if selector == 'language::analysis_source':
assert NEW_NAMES.issubset(names) and all((n, 'ok') in actual for n in NEW_NAMES)
mode = 'focused'
assert git('rev-parse', 'HEAD').decode().strip() == SOURCE
for path, data in watched.items():
assert path.read_bytes() == data, 'Final evidence/input readback differs: ' + str(path)
admission = {'source': SOURCE, 'mode': mode, 'selector': selector, 'artifact': artifact, 'names': names,
'result_sha256': digest(raw(name, '.result.json')), 'inputs_sha256': digest(raw(name, '.inputs.json')),
'log_sha256': digest(raw(name, '.log')), 'verified_inputs': 534, 'admission_helper_sha256': self_sha,
'baseline_source': BASELINE, 'baseline_result_sha256': BASE_PINS['.result.json'],
'baseline_admission_sha256': BASE_PINS['.admission.json'], 'baseline_members': 142,
'required_new_names': sorted(NEW_NAMES),
'boundary': 'Listing or exact focused evidence only; no full-suite/public8/native acceptance.'}
with output.open('x', encoding='utf-8', newline='\n') as stream:
stream.write(json.dumps(admission, indent=2) + '\n')
print(json.dumps({'name': name, 'mode': mode, 'members': len(names), 'verified_inputs': 534, 'source': SOURCE}))