108 lines
3.8 KiB
Python
108 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
D2 — compare_schema.py
|
|
Vergleicht zwei Schema-Modelle (Soll = kanonischer L0-Contract, Ist = tatsächlich).
|
|
|
|
Semantische Diff-Dimensionen (§12):
|
|
fehlende/extras Tabellen, Spalten, Typ-Mismatches, Nullable-Mismatches,
|
|
Default-Mismatches, PK, UNIQUE, FK, CHECK, Index-Mismatches.
|
|
|
|
Fail-closed: exit-Nonzero bei beliebigem semantischem Mismatch.
|
|
Acceptance: alle Mismatch-Counts == 0.
|
|
"""
|
|
import sys
|
|
import json
|
|
import argparse
|
|
from canonical_schema import (canonical_columns, canonical_constraints,
|
|
canonical_fks, canonical_checks, canonical_indexes)
|
|
|
|
|
|
def diff_counts(actual, expected):
|
|
"""Liefert dict mit Mismatch-Counts. 0 = keine."""
|
|
counts = {
|
|
"missing_tables": 0, "extra_tables": 0,
|
|
"missing_columns": 0, "extra_columns": 0,
|
|
"type_mismatch": 0, "nullable_mismatch": 0, "default_mismatch": 0,
|
|
"pk_mismatch": 0, "unique_mismatch": 0, "fk_mismatch": 0,
|
|
"check_mismatch": 0, "index_mismatch": 0,
|
|
}
|
|
# Tabellen
|
|
a_tbl = set((actual.get("tables") or []))
|
|
e_tbl = set((expected.get("tables") or []))
|
|
counts["missing_tables"] = len(e_tbl - a_tbl)
|
|
counts["extra_tables"] = len(a_tbl - e_tbl)
|
|
|
|
# Spalten je Tabelle
|
|
a_cols = {}
|
|
for c in actual.get("columns") or []:
|
|
a_cols.setdefault(c["table"], []).append(c)
|
|
e_cols = {}
|
|
for c in expected.get("columns") or []:
|
|
e_cols.setdefault(c["table"], []).append(c)
|
|
for tbl in e_tbl:
|
|
ec = {c["column"]: c for c in e_cols.get(tbl, [])}
|
|
ac = {c["column"]: c for c in a_cols.get(tbl, [])}
|
|
for cname, cexp in ec.items():
|
|
if cname not in ac:
|
|
counts["missing_columns"] += 1
|
|
continue
|
|
cact = ac[cname]
|
|
if _n(cact.get("type")) != _n(cexp.get("type")):
|
|
counts["type_mismatch"] += 1
|
|
if bool(cact.get("nullable")) != bool(cexp.get("nullable")):
|
|
counts["nullable_mismatch"] += 1
|
|
if _n(cact.get("default")) != _n(cexp.get("default")):
|
|
counts["default_mismatch"] += 1
|
|
for cname in ac:
|
|
if cname not in ec:
|
|
counts["extra_columns"] += 1
|
|
|
|
# Constraints
|
|
if canonical_constraints(actual.get("primary_keys") or [], "PK") != \
|
|
canonical_constraints(expected.get("primary_keys") or [], "PK"):
|
|
counts["pk_mismatch"] += 1
|
|
if canonical_constraints(actual.get("unique_constraints") or [], "UNIQUE") != \
|
|
canonical_constraints(expected.get("unique_constraints") or [], "UNIQUE"):
|
|
counts["unique_mismatch"] += 1
|
|
if canonical_fks(actual.get("foreign_keys") or []) != \
|
|
canonical_fks(expected.get("foreign_keys") or []):
|
|
counts["fk_mismatch"] += 1
|
|
if canonical_checks(actual.get("check_constraints") or []) != \
|
|
canonical_checks(expected.get("check_constraints") or []):
|
|
counts["check_mismatch"] += 1
|
|
if canonical_indexes(actual.get("indexes") or []) != \
|
|
canonical_indexes(expected.get("indexes") or []):
|
|
counts["index_mismatch"] += 1
|
|
|
|
return counts
|
|
|
|
|
|
def _n(x):
|
|
return None if x is None else " ".join(str(x).split())
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("expected", help="Kanonisches L0-Contract-Modell")
|
|
ap.add_argument("actual", help="Tatsächlich gecaptures Modell")
|
|
args = ap.parse_args()
|
|
|
|
with open(args.expected) as f:
|
|
expected = json.load(f)
|
|
with open(args.actual) as f:
|
|
actual = json.load(f)
|
|
|
|
counts = diff_counts(actual, expected)
|
|
total = sum(counts.values())
|
|
print(json.dumps(counts, indent=2))
|
|
print(f"TOTAL_MISMATCH={total}")
|
|
|
|
if total == 0:
|
|
print("PASS: Semantic diff = 0")
|
|
return 0
|
|
print("FAIL: semantic drift detected", file=sys.stderr)
|
|
return 1 # FAIL CLOSED
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|