import textwrap
from pathlib import Path

ROOT = Path(__file__).resolve().parents[3]
MIGRATION = (
    ROOT
    / "apps"
    / "api"
    / "migrations"
    / "asset_inventory_canonical_integrity_2026-07-14.sql"
)
CD_WORKFLOW = ROOT / ".gitea" / "workflows" / "cd.yaml"


def test_migration_quarantines_duplicates_without_deleting_history() -> None:
    sql = MIGRATION.read_text(encoding="utf-8")
    upper = sql.upper()

    assert "SET lock_timeout = '10s'" in sql
    assert "SET statement_timeout = '90s'" in sql
    assert "SET enable_indexscan = 'off'" in sql
    assert "SET enable_indexonlyscan = 'off'" in sql
    assert "SET enable_bitmapscan = 'off'" in sql
    assert "ROW_NUMBER() OVER" in sql
    assert "repair_row_count > 100" in sql
    assert "__integrity_quarantine__" in sql
    assert "lifecycle_state = 'deprecated'" in sql
    assert "REINDEX INDEX asset_inventory_asset_key_key" in sql
    assert "HAVING COUNT(*) > 1" in sql
    assert "ON CONFLICT (asset_key) DO UPDATE" in sql
    assert "gitea_cd_asset_integrity_reconciler" in sql
    assert '"work_item_id":"AIA-P0-006"' in sql
    assert "'data_rows_deleted', 0" in sql
    assert "DELETE FROM" not in upper
    assert "DROP TABLE" not in upper
    assert "TRUNCATE" not in upper


def test_cd_runs_bounded_integrity_repair_and_independent_verifier() -> None:
    workflow = CD_WORKFLOW.read_text(encoding="utf-8")
    step = workflow.split(
        "- name: Repair Asset Inventory Canonical Integrity", 1
    )[1].split("# 2026-03-31 ogt: 移除中間通知", 1)[0]
    script = step.split("run: |", 1)[1]

    assert "MIGRATION_DATABASE_URL: ${{ secrets.MIGRATION_DATABASE_URL }}" in step
    assert "DATABASE_URL: ${{ secrets.DATABASE_URL }}" in step
    assert "${{ secrets." not in script
    assert "--network bridge" in script
    assert "--network host" not in script
    assert "asset_inventory_canonical_integrity_2026-07-14.sql" in script
    assert "SET enable_indexscan TO off" in script
    assert "SET enable_indexonlyscan TO off" in script
    assert "SET enable_bitmapscan TO off" in script
    assert "asset_inventory_integrity_migration_applied=" in script
    assert "duplicate_group_count=" in script
    assert "quarantined_row_count=" in script
    assert "unique_index_healthy=" in script
    assert "durable_integrity_receipt_present=" in script
    assert "canonical_integrity_verified=" in script
    assert "asset_inventory_canonical_integrity_verifier_failed" in script


def test_cd_embedded_integrity_migration_python_compiles() -> None:
    workflow = CD_WORKFLOW.read_text(encoding="utf-8")
    step = workflow.split(
        "- name: Repair Asset Inventory Canonical Integrity", 1
    )[1]
    source = step.split("python - <<'PY'", 1)[1].split("          PY", 1)[0]

    compile(textwrap.dedent(source), "<asset-inventory-integrity-migration>", "exec")