ewoooc/tests/test_migration_metadata_coverage.py

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_ai_calls_provider_check_accepts_ollama_other_telemetry_bucket():
    migration = (ROOT / "migrations" / "043_allow_ollama_other_ai_calls_provider.sql").read_text(encoding="utf-8")

    assert "DROP CONSTRAINT IF EXISTS chk_ai_calls_provider" in migration
    assert "ADD CONSTRAINT chk_ai_calls_provider" in migration
    assert "'ollama_other'" in migration
    assert "NOT VALID" in migration


def test_rag_embedding_signature_migration_covers_query_and_learning_tables():
    migration = (ROOT / "migrations" / "034_add_embedding_signature_to_rag_tables.sql").read_text(encoding="utf-8")

    expected_snippets = [
        "ALTER TABLE IF EXISTS rag_query_log",
        "ADD COLUMN IF NOT EXISTS embedding_signature VARCHAR(64)",
        "idx_rag_query_log_embedding_signature",
        "WHERE query_embedding IS NOT NULL AND embedding_signature IS NOT NULL",
        "ALTER TABLE IF EXISTS learning_episodes",
        "idx_le_embedding_signature",
        "WHERE embedding IS NOT NULL AND embedding_signature IS NOT NULL",
    ]
    for snippet in expected_snippets:
        assert snippet in migration


def test_incidents_dual_column_migration_backfills_legacy_and_current_columns():
    migration = (ROOT / "migrations" / "036_normalize_incidents_dual_columns.sql").read_text(encoding="utf-8")

    expected_snippets = [
        "ADD COLUMN IF NOT EXISTS error_traceback TEXT",
        "ADD COLUMN IF NOT EXISTS traceback_str TEXT",
        "ADD COLUMN IF NOT EXISTS playbook_id INTEGER",
        "ADD COLUMN IF NOT EXISTS matched_playbook_id INTEGER",
        "SET traceback_str = error_traceback",
        "SET error_traceback = traceback_str",
        "SET matched_playbook_id = playbook_id",
        "SET playbook_id = matched_playbook_id",
        "incidents_playbook_id_fkey",
        "incidents_matched_playbook_id_fkey",
    ]
    for snippet in expected_snippets:
        assert snippet in migration


def test_action_plans_guardrail_migration_keeps_source_and_status_checks():
    migration = (ROOT / "migrations" / "037_add_action_plans_guardrails.sql").read_text(encoding="utf-8")

    expected_constraints = [
        "chk_action_plans_source_marker",
        "chk_action_plans_action_type",
        "chk_action_plans_created_by",
        "chk_action_plans_status",
    ]
    for constraint in expected_constraints:
        assert constraint in migration

    assert "CHECK (action_type IS NOT NULL OR created_by IS NOT NULL)" in migration
    assert "'code_review_fix'" in migration
    assert "'openclaw_recommendation'" in migration
    assert "'pending_review'" in migration
    assert migration.count("NOT VALID") >= 4


def test_dashboard_hot_path_index_migration_keeps_reboot_fix_durable():
    migration = (ROOT / "migrations" / "040_dashboard_hot_path_indexes.sql").read_text(encoding="utf-8")

    expected_snippets = [
        "idx_comp_match_attempts_source_sku_attempted_at",
        "ON competitor_match_attempts (source, sku, attempted_at DESC)",
        "INCLUDE (attempt_status)",
        "idx_price_records_product_timestamp_id_desc",
        "ON price_records (product_id, timestamp DESC, id DESC)",
        "INCLUDE (price)",
        "idx_products_status_id_icode",
        "ON products (status, id, i_code)",
        "ANALYZE products",
        "ANALYZE price_records",
        "ANALYZE competitor_match_attempts",
    ]
    for snippet in expected_snippets:
        assert snippet in migration

    assert "DROP " not in migration.upper()
    assert "DELETE " not in migration.upper()
    assert "TRUNCATE " not in migration.upper()


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()