57 lines
2.1 KiB
Python
57 lines
2.1 KiB
Python
"""앱의 런타임 준비 계약과 infra SQL SSOT 사이의 회귀 검사."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from .runtime_schema import NOTIFICATION_SCHEMA_CONTRACT, REVIEW_SCHEMA_CONTRACT
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[3]
|
|
INFRA_SQL = "\n".join(
|
|
path.read_text(encoding="utf-8")
|
|
for path in sorted((REPO_ROOT / "infra" / "db" / "init").glob("*.sql"))
|
|
)
|
|
|
|
|
|
class RuntimeSchemaSsotTest(unittest.TestCase):
|
|
def test_runtime_contract_objects_are_owned_by_infra_sql(self) -> None:
|
|
for contract in (REVIEW_SCHEMA_CONTRACT, NOTIFICATION_SCHEMA_CONTRACT):
|
|
for relation in contract.relations:
|
|
schema, table = relation.split(".")
|
|
pattern = rf"CREATE TABLE IF NOT EXISTS\s+{re.escape(schema)}\.{re.escape(table)}\b"
|
|
self.assertRegex(
|
|
INFRA_SQL, pattern, msg=f"infra SQL missing relation: {relation}"
|
|
)
|
|
|
|
for qualified_name in contract.columns:
|
|
_, table, column = qualified_name.split(".")
|
|
table_mentions = [
|
|
block
|
|
for block in re.split(
|
|
r"(?=CREATE TABLE IF NOT EXISTS|ALTER TABLE)", INFRA_SQL
|
|
)
|
|
if re.search(rf"\bapp\.{re.escape(table)}\b", block)
|
|
]
|
|
self.assertTrue(
|
|
any(
|
|
re.search(rf"\b{re.escape(column)}\b", block)
|
|
for block in table_mentions
|
|
),
|
|
msg=f"infra SQL missing column: {qualified_name}",
|
|
)
|
|
|
|
for qualified_name in contract.policies:
|
|
_, table, policy = qualified_name.split(".")
|
|
pattern = rf"CREATE POLICY\s+{re.escape(policy)}\s+ON\s+app\.{re.escape(table)}\b"
|
|
self.assertRegex(
|
|
INFRA_SQL,
|
|
pattern,
|
|
msg=f"infra SQL missing policy: {qualified_name}",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|