71 lines
2.0 KiB
Python
71 lines
2.0 KiB
Python
"""Static safety and inventory checks for the model_operations V0.1 DDL."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
|
|
|
|
BASE = Path(__file__).parent
|
|
EXPECTED_TABLES = {
|
|
"ops_model_categories",
|
|
"ops_banks",
|
|
"ops_model_instances",
|
|
"ops_model_versions",
|
|
"ops_monitor_batches",
|
|
"ops_monitor_results",
|
|
"ops_monitor_feature_metrics",
|
|
"ops_monitor_distributions",
|
|
"ops_rule_versions",
|
|
"ops_rule_items",
|
|
"ops_monitor_evaluations",
|
|
"ops_monitor_reviews",
|
|
"ops_outbox_events",
|
|
"ops_consumer_inbox",
|
|
"ops_report_template_versions",
|
|
"ops_prompt_versions",
|
|
"ops_prompt_regression_runs",
|
|
"ops_reports",
|
|
"ops_report_revisions",
|
|
"ops_bank_report_configs",
|
|
"ops_workflows",
|
|
"ops_workflow_stages",
|
|
"ops_documents",
|
|
"ops_usage_events",
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
sql_files = sorted(BASE.glob("*.sql"))
|
|
combined = "\n".join(path.read_text(encoding="utf-8") for path in sql_files)
|
|
tables = {
|
|
match.lower()
|
|
for match in re.findall(
|
|
r"CREATE\s+TABLE\s+IF\s+NOT\s+EXISTS\s+([a-zA-Z0-9_]+)",
|
|
combined,
|
|
flags=re.IGNORECASE,
|
|
)
|
|
}
|
|
missing = EXPECTED_TABLES - tables
|
|
extra = tables - EXPECTED_TABLES
|
|
assert not missing, f"missing tables: {sorted(missing)}"
|
|
assert not extra, f"unexpected tables: {sorted(extra)}"
|
|
assert len(re.findall(r"CREATE\s+OR\s+REPLACE\s+VIEW", combined, re.I)) == 1
|
|
assert not re.search(r"\b(DROP|TRUNCATE|DELETE\s+FROM)\b", combined, re.I)
|
|
assert not re.search(
|
|
r"\b(?:FROM|JOIN|UPDATE|INTO|REFERENCES|TABLE)\s+model_platform\.",
|
|
combined,
|
|
re.I,
|
|
), "cross-database SQL object reference found"
|
|
for path in sql_files:
|
|
line_count = len(path.read_text(encoding="utf-8").splitlines())
|
|
assert line_count <= 500, f"{path.name} exceeds 500 lines"
|
|
print(
|
|
f"validated {len(sql_files)} SQL files: "
|
|
f"{len(tables)} tables, 1 view, no destructive statements"
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|