71 lines
2.3 KiB
Python
71 lines
2.3 KiB
Python
import re
|
||
import stat
|
||
from pathlib import Path
|
||
|
||
from database.manager import Base
|
||
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
|
||
|
||
def _migration_created_tables():
|
||
tables = set()
|
||
for path in (ROOT / "migrations").glob("*.sql"):
|
||
text = path.read_text(encoding="utf-8", errors="ignore")
|
||
for match in re.finditer(
|
||
r"CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?(?:public\.)?([\w_]+)",
|
||
text,
|
||
re.IGNORECASE,
|
||
):
|
||
tables.add(match.group(1))
|
||
return tables
|
||
|
||
|
||
def test_all_orm_metadata_tables_have_create_table_migration():
|
||
metadata_tables = set(Base.metadata.tables)
|
||
migration_tables = _migration_created_tables()
|
||
|
||
assert metadata_tables - migration_tables == set()
|
||
|
||
|
||
def test_v2_blocker_migrations_exist_and_are_runner_readable():
|
||
required = [
|
||
"031_fix_incidents_autoheal_schema.sql",
|
||
"032_market_intel_core_schema.sql",
|
||
"033_fix_host_health_probe_labels.sql",
|
||
"034_add_embedding_signature_to_rag_tables.sql",
|
||
"035_core_business_tables_baseline.sql",
|
||
"036_normalize_incidents_dual_columns.sql",
|
||
"037_add_action_plans_guardrails.sql",
|
||
]
|
||
|
||
for filename in required:
|
||
path = ROOT / "migrations" / filename
|
||
assert path.exists(), f"missing migration: {filename}"
|
||
mode = stat.S_IMODE(path.stat().st_mode)
|
||
assert mode & stat.S_IRGRP, f"{filename} is not group-readable"
|
||
assert mode & stat.S_IROTH, f"{filename} is not world-readable"
|
||
|
||
|
||
def test_host_health_probe_label_check_accepts_runtime_labels():
|
||
migration = (ROOT / "migrations" / "033_fix_host_health_probe_labels.sql").read_text(encoding="utf-8")
|
||
|
||
assert "DROP CONSTRAINT IF EXISTS chk_host_label_029" in migration
|
||
assert "ADD CONSTRAINT chk_host_label_029" in migration
|
||
assert "NOT VALID" in migration
|
||
|
||
expected_runtime_labels = [
|
||
"GCP-SSD",
|
||
"GCP-SSD-2",
|
||
"111 備援",
|
||
"GCP-SSD（via Nginx 110）",
|
||
"GCP-SSD-2（via Nginx 110）",
|
||
]
|
||
for label in expected_runtime_labels:
|
||
assert f"'{label}'" in migration
|
||
|
||
|
||
def test_legacy_zero_byte_database_decoys_do_not_return():
|
||
for filename in ["momo.db", "momo_data.db", "momo_database.db"]:
|
||
assert not (ROOT / "database" / filename).exists()
|