cesnimda
92 lines
5.5 KiB
Python
92 lines
5.5 KiB
Python
"""Stage fresh PDF evidence and run the unchanged oracle after producers stop."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[4]
|
|
ORACLE = ROOT / 'scripts/verify-cv-fidelity.py'
|
|
|
|
def digest(path):
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
def merged_measurements(paths, expected):
|
|
records = [json.loads(path.read_text(encoding='utf-8-sig')) for path in paths]
|
|
versions = {record['browserVersion'] for record in records}
|
|
fixtures = [fixture for record in records for fixture in record['fixtures']]
|
|
names = [fixture['fixture'] for fixture in fixtures]
|
|
if len(versions) != 1 or len(fixtures) != expected or len(set(names)) != expected:
|
|
raise ValueError('Unexpected browser identity, fixture count or duplicate fixture names.')
|
|
if any(Path(name).name != name or name in ('.', '..') for name in names):
|
|
raise ValueError('Fixture name must identify one child directory.')
|
|
return {'browserVersion': versions.pop(), 'fixtures': fixtures}
|
|
|
|
def stage(source, destination, measurements, hashes):
|
|
destination.mkdir()
|
|
for fixture in measurements['fixtures']:
|
|
name = fixture['fixture']
|
|
files = [source / name / 'export.pdf', source / name / 'render.html']
|
|
files += [source / name / f'preview-{page}.png' for page in range(1, fixture['previewPages'] + 1)]
|
|
(destination / name).mkdir()
|
|
for path in files:
|
|
hashes[str(path.resolve())] = digest(path)
|
|
shutil.copy2(path, destination / name / path.name)
|
|
(destination / 'measurements.json').write_text(json.dumps(measurements, ensure_ascii=False, indent=2), encoding='utf-8')
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--platform', choices=['windows', 'linux'], required=True)
|
|
parser.add_argument('--fidelity-root', type=Path, required=True)
|
|
parser.add_argument('--settings-root', type=Path, required=True)
|
|
parser.add_argument('--output', type=Path, required=True, help='New directory; never overwrites an earlier run.')
|
|
parser.add_argument('--record', type=Path, required=True, help='New native-pdf.json or linux-pdf.json record.')
|
|
args = parser.parse_args()
|
|
if args.output.exists() or args.record.exists():
|
|
parser.error('Output directory and result record must both be new.')
|
|
inputs = [args.fidelity_root / 'measurements.json']
|
|
settings = [args.settings_root / f'{theme}-{language}-measurements.json'
|
|
for theme in ['light', 'dark'] for language in ['en', 'nb']]
|
|
corpus = merged_measurements(inputs, 16)
|
|
setting_corpus = merged_measurements(settings, 8)
|
|
expected_settings = {f'{theme}-{language}-{phase}' for theme in ['light', 'dark']
|
|
for language in ['en', 'nb'] for phase in ['before', 'after']}
|
|
if {fixture['fixture'] for fixture in setting_corpus['fixtures']} != expected_settings:
|
|
raise ValueError('Settings fixture membership differs from all eight required states.')
|
|
hashes = {str(path.resolve()): digest(path) for path in [ORACLE, *inputs, *settings]}
|
|
args.output.mkdir(parents=True)
|
|
record = {'status': 'FAIL', 'platform': args.platform, 'oracle': str(ORACLE), 'inputs': hashes, 'lanes': []}
|
|
try:
|
|
for name, source, measurements, count, pages in [
|
|
('fidelity', args.fidelity_root, corpus, 16, 36),
|
|
('settings-fidelity', args.settings_root, setting_corpus, 8, 24)]:
|
|
destination = args.output / name
|
|
stage(source, destination, measurements, hashes)
|
|
command = [sys.executable, '-B', str(ORACLE), str(destination)]
|
|
process = subprocess.run(command, capture_output=True, text=True, encoding='utf-8', errors='replace')
|
|
(destination / 'oracle.log').write_text(process.stdout + process.stderr, encoding='utf-8')
|
|
report_path = destination / 'fidelity-results.json'
|
|
report = json.loads(report_path.read_text(encoding='utf-8')) if report_path.exists() else {}
|
|
native = report.get('fixtures', [])
|
|
passed = (process.returncode == 0 and report.get('passed') == count and report.get('total') == count
|
|
and len(native) == count and {row['fixture'] for row in native} == {row['fixture'] for row in measurements['fixtures']}
|
|
and all(row['pass'] and not row['failures'] for row in native)
|
|
and sum(row['pdfPages'] for row in native) == pages)
|
|
record['lanes'].append({'lane': name, 'status': 'PASS' if passed else 'FAIL', 'command': command,
|
|
'exitCode': process.returncode, 'report': str(report_path),
|
|
'reportSha256': digest(report_path) if report_path.exists() else None,
|
|
'fixtures': len(native), 'pdfPages': sum(row['pdfPages'] for row in native)})
|
|
record['inputDrift'] = [path for path, expected in hashes.items() if digest(Path(path)) != expected]
|
|
if not record['inputDrift'] and all(lane['status'] == 'PASS' for lane in record['lanes']):
|
|
record['status'] = 'PASS'
|
|
except Exception as error:
|
|
record['error'] = f'{type(error).__name__}: {error}'
|
|
args.record.write_text(json.dumps(record, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
|
|
print(json.dumps({'status': record['status'], 'record': str(args.record)}))
|
|
raise SystemExit(0 if record['status'] == 'PASS' else 1)
|
|
|
|
if __name__ == '__main__':
|
|
main()
|