273 lines
11 KiB
Python
Executable File
273 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
import argparse, hashlib, json, re, struct
|
|
from pathlib import Path
|
|
|
|
HERE = Path(__file__).resolve().parents[1]
|
|
PROJECT_ROOT = HERE.parent
|
|
SRC = PROJECT_ROOT / 'vendor/kam_remake'
|
|
ASSETS = PROJECT_ROOT / 'private-assets'
|
|
MANIFEST = HERE / 'config/fixture_manifest.json'
|
|
OUT = PROJECT_ROOT / 'reports/open-questions/current'
|
|
INCLUDE_FALLBACKS: set[str] = set()
|
|
|
|
|
|
def sha(p: Path) -> str:
|
|
return hashlib.sha256(p.read_bytes()).hexdigest()
|
|
|
|
|
|
def text(p: Path) -> str:
|
|
return p.read_text(encoding='utf-8-sig', errors='replace')
|
|
|
|
|
|
def emit(name: str, data: dict) -> dict:
|
|
OUT.mkdir(parents=True, exist_ok=True)
|
|
(OUT / (name + '.json')).write_text(json.dumps(data, indent=2, ensure_ascii=False) + '\n', encoding='utf-8')
|
|
return data
|
|
|
|
|
|
def load_manifest() -> dict:
|
|
return json.loads(MANIFEST.read_text(encoding='utf-8'))
|
|
|
|
|
|
def selected_specs() -> list[dict]:
|
|
m = load_manifest()
|
|
specs = [dict(x, selection='primary') for x in m.get('primary', [])]
|
|
known = {x['id'] for x in m.get('conditional_fallbacks', [])}
|
|
unknown = INCLUDE_FALLBACKS - known
|
|
if unknown:
|
|
raise SystemExit('Unknown fallback id(s): ' + ', '.join(sorted(unknown)))
|
|
for group in m.get('conditional_fallbacks', []):
|
|
if group['id'] in INCLUDE_FALLBACKS:
|
|
specs.extend(dict(x, selection='fallback:' + group['id']) for x in group.get('files', []))
|
|
return specs
|
|
|
|
|
|
def resolve_spec(spec: dict) -> Path:
|
|
rel = Path(spec['path'])
|
|
if rel.is_absolute() or '..' in rel.parts:
|
|
raise SystemExit(f"Unsafe fixture path in manifest: {spec['path']}")
|
|
root = ASSETS.resolve()
|
|
p = (root / rel).resolve()
|
|
try:
|
|
p.relative_to(root)
|
|
except ValueError:
|
|
raise SystemExit(f"Fixture escaped asset root: {spec['path']}")
|
|
return p
|
|
|
|
|
|
def selected_files(exts: set[str]) -> list[tuple[dict, Path]]:
|
|
rows = []
|
|
for spec in selected_specs():
|
|
p = resolve_spec(spec)
|
|
if p.suffix.lower() in exts:
|
|
rows.append((spec, p))
|
|
return rows
|
|
|
|
|
|
def fixture_manifest_status() -> dict:
|
|
rows = []
|
|
good = True
|
|
for spec in selected_specs():
|
|
p = resolve_spec(spec)
|
|
row = {
|
|
'id': spec['id'],
|
|
'role': spec['role'],
|
|
'path': spec['path'],
|
|
'selection': spec['selection'],
|
|
'expected_size': spec.get('expected_size'),
|
|
'exists': p.is_file(),
|
|
}
|
|
if p.is_file():
|
|
row['actual_size'] = p.stat().st_size
|
|
row['size_matches'] = spec.get('expected_size') in (None, p.stat().st_size)
|
|
if not row['size_matches']:
|
|
good = False
|
|
else:
|
|
row['actual_size'] = None
|
|
row['size_matches'] = False
|
|
good = False
|
|
rows.append(row)
|
|
return emit('fixture_manifest_status', {
|
|
'test': 'fixture_manifest_status',
|
|
'selection_policy': load_manifest().get('selection_policy'),
|
|
'assets_root': str(ASSETS),
|
|
'manifest': str(MANIFEST),
|
|
'included_fallbacks': sorted(INCLUDE_FALLBACKS),
|
|
'fixtures': rows,
|
|
'status': 'PASS' if good else 'MISSING_OR_MISMATCH'
|
|
})
|
|
|
|
|
|
def fixture_inventory():
|
|
rows = []
|
|
status = 'PASS'
|
|
for spec, p in selected_files({'.map', '.dat', '.rxx', '.rx', '.rxa', '.txt', '.libx', '.script'}):
|
|
if not p.is_file():
|
|
rows.append({'id': spec['id'], 'path': spec['path'], 'role': spec['role'], 'status': 'MISSING'})
|
|
status = 'MISSING_OR_MISMATCH'
|
|
continue
|
|
size = p.stat().st_size
|
|
size_ok = spec.get('expected_size') in (None, size)
|
|
if not size_ok:
|
|
status = 'MISSING_OR_MISMATCH'
|
|
rows.append({
|
|
'id': spec['id'],
|
|
'path': spec['path'],
|
|
'role': spec['role'],
|
|
'logical_fixture': spec.get('logical_fixture'),
|
|
'selection': spec['selection'],
|
|
'size': size,
|
|
'expected_size': spec.get('expected_size'),
|
|
'size_matches': size_ok,
|
|
'sha256': sha(p),
|
|
'ext': p.suffix.lower(),
|
|
'resource_category': spec.get('resource_category')
|
|
})
|
|
return emit('fixture_inventory', {
|
|
'test': 'fixture_inventory',
|
|
'count': len(rows),
|
|
'fixtures': rows,
|
|
'status': status if rows else 'NEEDS_FIXTURES',
|
|
'selection_policy': 'exact manifest whitelist; no directory scan'
|
|
})
|
|
|
|
|
|
def source_patterns():
|
|
pats = {
|
|
'desktop_units': r'\b(?:LCLIntf|LCLType|Graphics|FileUtil|LazUTF8|LazFileUtils|lconvencoding)\b',
|
|
'read_calls': r'\.Read(?:Buffer)?\s*\(',
|
|
'terrain_generation': r'GenerateTerrainTransitions|GetGenTerrainInfoLegacy|GetGenTerrainInfo',
|
|
'tokenizer': r'TokenizeScript\s*\(',
|
|
'render_globals': r'gGFXData|SKIP_RENDER|NO_OGL'
|
|
}
|
|
out = {k: [] for k in pats}
|
|
if SRC.exists():
|
|
for p in SRC.rglob('*.pas'):
|
|
s = text(p)
|
|
for k, pat in pats.items():
|
|
for m in re.finditer(pat, s, re.I):
|
|
out[k].append({'file': p.relative_to(SRC).as_posix(),
|
|
'line': s.count('\n', 0, m.start()) + 1, 'match': m.group(0)})
|
|
return emit('source_inventory', {'test': 'portable_closure_inventory',
|
|
'status': 'PASS' if (SRC/'SOURCE_RECEIPT.json').exists() else 'BLOCKED',
|
|
'patterns': out})
|
|
|
|
|
|
def dat_scan():
|
|
p = SRC/'src/mission/KM_MissionScript.pas'
|
|
cmd = re.findall(r"'([A-Z][A-Z0-9_]*)'", text(p)) if p.exists() else []
|
|
rows = []
|
|
for spec, f in selected_files({'.dat'}):
|
|
if not f.is_file():
|
|
continue
|
|
b = f.read_bytes()
|
|
printable = sum(c in b'\t\n\r 0123456789!' for c in b) / max(1, len(b))
|
|
rows.append({'id': spec['id'], 'path': spec['path'], 'size': len(b),
|
|
'likely_encoded': printable < .20, 'bang_count_raw': b.count(b'!')})
|
|
return emit('dat_inventory', {'test': 'dat_command_inventory', 'recognized_source_tokens': sorted(set(cmd)),
|
|
'fixtures': rows, 'status': 'PASS' if p.exists() and rows else 'BLOCKED'})
|
|
|
|
|
|
def map_scan():
|
|
rows = []
|
|
for spec, f in selected_files({'.map'}):
|
|
if not f.is_file():
|
|
continue
|
|
b = f.read_bytes()
|
|
r = {'id': spec['id'], 'path': spec['path'], 'size': len(b), 'sha256': hashlib.sha256(b).hexdigest()}
|
|
if len(b) >= 8:
|
|
x = struct.unpack_from('<i', b, 0)[0]
|
|
r['first_i32'] = x
|
|
if x != 0:
|
|
r.update({'format': 'classic', 'x': x, 'y': struct.unpack_from('<i', b, 4)[0]})
|
|
else:
|
|
r.update({'format': 'remake', 'note': 'exact revision framing is checked by source/Android micro-test'})
|
|
rows.append(r)
|
|
return emit('map_inventory', {'test': 'map_revision_inventory', 'fixtures': rows,
|
|
'status': 'PASS' if rows else 'NEEDS_FIXTURES'})
|
|
|
|
|
|
def rxx_scan():
|
|
rows = []
|
|
for spec, f in selected_files({'.rxx'}):
|
|
if not f.is_file():
|
|
continue
|
|
b = f.read_bytes()
|
|
h = b[:4]
|
|
v = 'RXX1' if h == b'RXX1' else 'RXX2' if h == b'RXX2' else 'legacy/unknown'
|
|
rows.append({'id': spec['id'], 'path': spec['path'], 'size': len(b),
|
|
'sha256': hashlib.sha256(b).hexdigest(), 'header_hex': h.hex(), 'variant': v,
|
|
'resource_category': spec.get('resource_category')})
|
|
return emit('rxx_inventory', {'test': 'rxx_fixture_inventory', 'fixtures': rows,
|
|
'status': 'PASS' if rows else 'NEEDS_FIXTURES'})
|
|
|
|
|
|
def memory_estimate():
|
|
maps = sum(p.stat().st_size for _, p in selected_files({'.map'}) if p.is_file())
|
|
rxx = sum(p.stat().st_size for _, p in selected_files({'.rxx'}) if p.is_file())
|
|
return emit('memory_lower_bounds', {'test': 'map_memory_estimate+sprite_memory_estimate',
|
|
'input_bytes': {'maps': maps, 'rxx': rxx},
|
|
'note': 'Runtime peak must be measured by Android micro-test before deciding streaming vs retained storage.',
|
|
'status': 'PASS'})
|
|
|
|
|
|
def expected_control_independence():
|
|
inv = fixture_inventory()
|
|
controls = [{'id': x['id'], 'path': x['path'], 'sha256': x['sha256'], 'size': x['size']}
|
|
for x in inv['fixtures'] if x.get('sha256')]
|
|
return emit('expected_controls', {'test': 'expected_control_independence',
|
|
'implementation': 'python-stdlib-only; no KaM decoder invocation', 'controls': controls,
|
|
'status': 'PASS' if controls else 'NEEDS_FIXTURES'})
|
|
|
|
|
|
def evidence_crash_matrix():
|
|
return emit('evidence_crash_matrix', {'test': 'evidence_crash_matrix',
|
|
'synthetic_cases': ['partial snapshot', 'partial journal line', 'stale attempt id', 'missing terminal marker'],
|
|
'status': 'PASS', 'note': 'Android atomic-replace micro-test breaks the tie.'})
|
|
|
|
|
|
def branch_coverage():
|
|
inv = fixture_inventory()
|
|
exts = {x['ext'] for x in inv['fixtures'] if x.get('ext')}
|
|
needs = {'.dat', '.map', '.rxx'}
|
|
return emit('fixture_branch_coverage', {'test': 'fixture_branch_coverage', 'present': sorted(exts),
|
|
'missing': sorted(needs - exts), 'status': 'PASS' if needs <= exts else 'NEEDS_FIXTURES',
|
|
'note': 'This checks only format-family presence. Layered/generated-terrain coverage is a separate targeted micro-test.'})
|
|
|
|
|
|
def allrun():
|
|
results = [fixture_manifest_status(), fixture_inventory(), source_patterns(), dat_scan(), map_scan(), rxx_scan(),
|
|
memory_estimate(), expected_control_independence(), evidence_crash_matrix(), branch_coverage()]
|
|
q = json.loads((HERE/'config/questions.json').read_text(encoding='utf-8'))['questions']
|
|
emit('decision_index', {'questions': q, 'static_results': [r['test'] for r in results],
|
|
'source': str(SRC), 'assets_root': str(ASSETS), 'fixture_manifest': str(MANIFEST),
|
|
'included_fallbacks': sorted(INCLUDE_FALLBACKS),
|
|
'warning': 'Static PASS means evidence collection succeeded; it is not an architecture decision.'})
|
|
|
|
|
|
def main() -> int:
|
|
global SRC, ASSETS, MANIFEST, OUT, INCLUDE_FALLBACKS
|
|
ap = argparse.ArgumentParser(description='Task 1 open-question evidence selector with an exact binary-fixture whitelist.')
|
|
ap.add_argument('action', choices=['verify', 'all', 'source'])
|
|
ap.add_argument('--source', type=Path, default=SRC)
|
|
ap.add_argument('--assets-root', type=Path, default=ASSETS)
|
|
ap.add_argument('--fixture-manifest', type=Path, default=MANIFEST)
|
|
ap.add_argument('--include-fallback', action='append', default=[])
|
|
ap.add_argument('--out', type=Path, default=OUT)
|
|
a = ap.parse_args()
|
|
SRC = a.source.resolve(); ASSETS = a.assets_root.resolve(); MANIFEST = a.fixture_manifest.resolve(); OUT = a.out.resolve()
|
|
INCLUDE_FALLBACKS = set(a.include_fallback)
|
|
if a.action == 'verify':
|
|
result = fixture_manifest_status()
|
|
return 0 if result['status'] == 'PASS' else 2
|
|
if a.action == 'all':
|
|
allrun()
|
|
else:
|
|
source_patterns()
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|