39 lines
1.9 KiB
Python
39 lines
1.9 KiB
Python
"""Verify the separately staged forecast repair before target-runtime testing.
|
|
Only explicit Python source and requirements are copied. No models, credentials,
|
|
live databases or telemetry are needed by this offline test image.
|
|
"""
|
|
from pathlib import Path
|
|
import hashlib
|
|
import json
|
|
|
|
|
|
def source_paths(root):
|
|
root = Path(root)
|
|
paths = [root / 'requirements.txt'] + list(root.glob('*.py'))
|
|
for folder in ('methods', 'tests'):
|
|
paths.extend((root / folder).glob('*.py'))
|
|
if not (root / 'telemetry_quality.py').is_file():
|
|
raise ValueError('Forecast telemetry repair missing')
|
|
for path in paths:
|
|
if path.is_symlink() or not path.resolve().is_relative_to(root.resolve()):
|
|
raise ValueError('External forecast source is not allowed')
|
|
return sorted(paths)
|
|
|
|
|
|
def verify_forecast_bundle(bundle, source):
|
|
bundle, source = Path(bundle), Path(source)
|
|
manifest_path = bundle / 'SOURCE_MANIFEST.json'
|
|
manifest = json.loads(manifest_path.read_text())
|
|
expected_paths = {str(p.relative_to(source)) for p in source_paths(source)}
|
|
if not isinstance(manifest, dict) or not manifest or set(manifest) != expected_paths:
|
|
raise ValueError('Forecast source set changed; rebuild reviewed test bundle')
|
|
if {str(p.relative_to(bundle)) for p in source_paths(bundle)} != expected_paths:
|
|
raise ValueError('Staged forecast source set does not match development source')
|
|
for relative, expected in manifest.items():
|
|
if Path(relative).is_absolute() or '..' in Path(relative).parts:
|
|
raise ValueError('Unsafe forecast manifest path')
|
|
for path in (source / relative, bundle / relative):
|
|
if not path.is_file() or path.is_symlink() or hashlib.sha256(path.read_bytes()).hexdigest() != expected:
|
|
raise ValueError('Forecast source changed after staging; review before test')
|
|
return dict(manifest)
|