4508d19d4b
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m32s
CD Pipeline / build-and-deploy (push) Successful in 7m58s
CD Pipeline / post-deploy-checks (push) Successful in 2m25s
72 lines
2.8 KiB
Python
72 lines
2.8 KiB
Python
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")
|