48 lines
1.9 KiB
Python
48 lines
1.9 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import re
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import pymysql
|
||
|
|
|
||
|
|
|
||
|
|
ROOT = Path(__file__).resolve().parents[1]
|
||
|
|
SQL_FILE = ROOT / "alembic" / "baseline_generated.sql"
|
||
|
|
|
||
|
|
|
||
|
|
def expected_tables() -> set[str]:
|
||
|
|
text = SQL_FILE.read_text(encoding="utf-8")
|
||
|
|
return set(re.findall(r"CREATE TABLE `?([A-Za-z0-9_]+)`?", text))
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> None:
|
||
|
|
connection = pymysql.connect(host="127.0.0.1", port=3306, user="root", password="123456", database="jr")
|
||
|
|
with connection.cursor() as cursor:
|
||
|
|
cursor.execute("SELECT table_name FROM information_schema.tables WHERE table_schema=%s", ("jr",))
|
||
|
|
actual = {row[0] for row in cursor.fetchall()} - {"alembic_version"}
|
||
|
|
expected = expected_tables() | {
|
||
|
|
"agent_run", "domain_event_outbox", "request_idempotency", "config_release",
|
||
|
|
"platform_config_item", "model_endpoint_config", "model_routing_rule",
|
||
|
|
"model_routing_fallback", "prompt_template_version",
|
||
|
|
"outbox_delivery",
|
||
|
|
"svc_conversation_session",
|
||
|
|
"api_request_receipt",
|
||
|
|
}
|
||
|
|
missing = expected - actual
|
||
|
|
unexpected = actual - expected
|
||
|
|
if missing or unexpected:
|
||
|
|
raise SystemExit(f"table mismatch missing={sorted(missing)} unexpected={sorted(unexpected)}")
|
||
|
|
cursor.execute("SELECT table_name, column_name FROM information_schema.columns WHERE table_schema=%s", ("jr",))
|
||
|
|
actual_columns: dict[str, set[str]] = {}
|
||
|
|
for table, column in cursor.fetchall():
|
||
|
|
actual_columns.setdefault(table, set()).add(column)
|
||
|
|
for table in sorted(expected):
|
||
|
|
if not actual_columns.get(table):
|
||
|
|
raise SystemExit(f"missing columns for {table}")
|
||
|
|
connection.close()
|
||
|
|
print(f"schema audit passed: {len(expected)} business tables, no missing or unexpected tables")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|