"""P1 behavioral-parity harness (Gate A). Proves the model is lossless WITHOUT byte-diffing: for each manifest, import it into in-memory rows and render it back out, then check that 1. every entry is field-for-field identical (engine-relevant fields only, dropping _comment and key order), in the same order, AND 2. the same entries fire in the same order for a set of machine-profile fixtures, using the same filter logic the engine uses (filters.py). Runs with no database (build_scope produces detached ORM objects whose relationship lists are read directly), so it is cheap to re-run and wrap in CI. Output is one readable line per scope. """ import json import os from .importer import build_scope from .serializer import scope_to_manifest, canonical_entry from .filters import applicable_entry_names _FIXTURES_PATH = os.path.join(os.path.dirname(__file__), 'parityfixtures.json') def load_fixtures(): with open(_FIXTURES_PATH) as handle: return json.load(handle) def check_scope(scopename, phase, original, fixtures): """Compare an original manifest against its import+export round-trip.""" rebuilt = scope_to_manifest(build_scope(scopename, phase, original)) orig_apps = original.get('Applications') or [] rebuilt_apps = rebuilt.get('Applications') or [] # Check 0: scope-level fields (Version, Site) round-trip. firstdiff = None for key in ('Version', 'Site'): if str(original.get(key) or '') != str(rebuilt.get(key) or ''): firstdiff = (f'scope field {key}: ' f'{original.get(key)!r} vs {rebuilt.get(key)!r}') break scope_ok = firstdiff is None # Check 1: field-identical, order-preserving. identical = 0 if firstdiff is None and len(orig_apps) != len(rebuilt_apps): firstdiff = (f'entry count {len(orig_apps)} vs {len(rebuilt_apps)}') for i in range(min(len(orig_apps), len(rebuilt_apps))): co = canonical_entry(orig_apps[i]) cr = canonical_entry(rebuilt_apps[i]) if co == cr: identical += 1 elif firstdiff is None: name = orig_apps[i].get('Name', f'#{i}') diffkeys = sorted( k for k in set(co) | set(cr) if co.get(k) != cr.get(k)) firstdiff = f'entry "{name}" differs on {diffkeys}' # Check 2: same entries fire in the same order for each profile. Inject the # scope phase so PCTypesStrict is honored only for preinstall (see filters). profiles_same = 0 for base_profile in fixtures: profile = {**base_profile, 'phase': phase} if (applicable_entry_names(orig_apps, profile) == applicable_entry_names(rebuilt_apps, profile)): profiles_same += 1 elif firstdiff is None: firstdiff = f'filter mismatch for profile {profile.get("label")}' passed = (scope_ok and identical == len(orig_apps) == len(rebuilt_apps) and profiles_same == len(fixtures)) return { 'scopename': scopename, 'phase': phase, 'entries_total': len(orig_apps), 'entries_identical': identical, 'profiles_total': len(fixtures), 'profiles_same': profiles_same, 'passed': passed, 'firstdiff': firstdiff, } def run_parity(manifests, fixtures=None): """Run parity over a list of (scopename, phase, manifest_dict). Returns (results, all_passed).""" fixtures = fixtures if fixtures is not None else load_fixtures() results = [check_scope(name, phase, manifest, fixtures) for name, phase, manifest in manifests] return results, all(r['passed'] for r in results) def format_result(result): """One human-readable line for a scope result.""" status = 'PASS' if result['passed'] else 'FAIL' line = (f"scope {result['scopename']:<28} " f"entries {result['entries_identical']}/{result['entries_total']} identical " f"profiles {result['profiles_same']}/{result['profiles_total']} same-fire " f"{status}") if not result['passed'] and result['firstdiff']: line += f"\n -> {result['firstdiff']}" return line