134 lines
5.5 KiB
Python
134 lines
5.5 KiB
Python
"""Verify W1-017 behavior and preservation against the archived baseline."""
|
|
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / 'src'))
|
|
|
|
from opportunity_intelligence.collectors import lrs, usac # noqa: E402
|
|
|
|
|
|
def load_baseline(name, path):
|
|
"""Load an archived stdlib-only collector without restoring old paths."""
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def require(condition, message):
|
|
"""Fail explicitly even when Python runs with assertions disabled."""
|
|
if not condition:
|
|
raise RuntimeError(message)
|
|
|
|
|
|
def main():
|
|
"""Run offline commands, compare old/new results, and save a new report."""
|
|
started = time.monotonic()
|
|
archive = ROOT / 'runs/2026-09-17/w1-017'
|
|
output = archive / 'verification'
|
|
output.mkdir(exist_ok=False)
|
|
run = ROOT / 'runs/2026-09-17/lrs-162640'
|
|
report = {'success': False, 'commands': []}
|
|
try:
|
|
commands = [
|
|
['-m', 'unittest', 'discover', '-s', 'tests', '-v'],
|
|
['w1.py', '--help'],
|
|
['w1.py', 'usac', '--help'],
|
|
['w1.py', 'lrs', '--help'],
|
|
['w1.py', 'qualify', '--help'],
|
|
['tools/inspect_lrs.py'],
|
|
[
|
|
'w1.py', 'qualify', '--run', str(run),
|
|
'--output', str(output / 'qualification'),
|
|
],
|
|
]
|
|
for index, arguments in enumerate(commands):
|
|
command_started = time.monotonic()
|
|
result = subprocess.run(
|
|
[sys.executable, *arguments], cwd=ROOT,
|
|
capture_output=True, text=True, encoding='utf-8',
|
|
errors='replace', timeout=120,
|
|
env={**os.environ, 'PYTHONIOENCODING': 'utf-8'},
|
|
)
|
|
log = f'command-{index + 1}.txt'
|
|
(output / log).write_text(
|
|
result.stdout + result.stderr, encoding='utf-8'
|
|
)
|
|
report['commands'].append({
|
|
'arguments': arguments, 'exit_code': result.returncode,
|
|
'seconds': round(time.monotonic() - command_started, 3),
|
|
'log': log,
|
|
})
|
|
require(result.returncode == 0, f'Command failed; inspect {log}')
|
|
|
|
old_lrs = load_baseline('baseline_lrs', archive / 'before/lrs_discovery.py')
|
|
old_usac = load_baseline('baseline_usac', archive / 'before/discovery.py')
|
|
listing = (run / 'listing.html').read_text(encoding='utf-8')
|
|
require(lrs.discover(listing) == old_lrs.discover(listing),
|
|
'LRS discovery changed')
|
|
records = json.loads((run / 'lrs.json').read_text(encoding='utf-8'))
|
|
for record in records:
|
|
page = (run / (record['source_id'] + '.html')).read_text(
|
|
encoding='utf-8'
|
|
)
|
|
arguments = (page, record['url'], record['checked'])
|
|
require(lrs.extract(*arguments) == old_lrs.extract(*arguments),
|
|
f"LRS extraction changed: {record['source_id']}")
|
|
report['equivalent_lrs_records'] = len(records)
|
|
for name in ('usac-links.html', 'usac-dates.html'):
|
|
page = (ROOT / 'tests/fixtures' / name).read_text(encoding='utf-8')
|
|
require(usac.collect(page) == old_usac.collect(page),
|
|
f'USAC parsing changed: {name}')
|
|
report['equivalent_usac_fixtures'] = 2
|
|
|
|
previous = run / 'qualification-W1-016'
|
|
current = output / 'qualification'
|
|
for name in ('brief.md', 'shortlist.json', 'audit.json'):
|
|
require((previous / name).read_bytes() == (current / name).read_bytes(),
|
|
f'Qualification artifact changed: {name}')
|
|
old_summary = json.loads((previous / 'summary.json').read_text())
|
|
new_summary = json.loads((current / 'summary.json').read_text())
|
|
# The source fingerprint must change after a code refactor. Every
|
|
# business result, input hash and pipeline hash must still agree.
|
|
old_summary.pop('rules_sha256')
|
|
new_summary.pop('rules_sha256')
|
|
require(old_summary == new_summary, 'Qualification summary changed')
|
|
report['qualification_counts'] = new_summary['counts']
|
|
report['qualification_equivalent'] = True
|
|
|
|
historical = json.loads(
|
|
(archive / 'historical-hashes.json').read_text(encoding='utf-8-sig')
|
|
)
|
|
for item in historical:
|
|
path = ROOT / item['path']
|
|
require(path.is_file(), f'Historical evidence missing: {path}')
|
|
digest = hashlib.sha256(path.read_bytes()).hexdigest()
|
|
require(digest == item['sha256'], f'Historical file changed: {path}')
|
|
report['unchanged_historical_files'] = len(historical)
|
|
pipeline = (ROOT / 'data/opportunities.json').read_bytes()
|
|
require(pipeline == (archive / 'before/opportunities.json').read_bytes(),
|
|
'Reviewed pipeline bytes changed')
|
|
report['pipeline_sha256'] = hashlib.sha256(pipeline).hexdigest()
|
|
report['success'] = True
|
|
except Exception as error:
|
|
report['error'] = str(error)
|
|
finally:
|
|
report['machine_seconds'] = round(time.monotonic() - started, 3)
|
|
lrs.write_json(output / 'report.json', report)
|
|
print(json.dumps(report, indent=2))
|
|
if not report['success']:
|
|
raise SystemExit(1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|