fix(security): repair canonical asset integrity
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
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
This commit is contained in:
@@ -1958,6 +1958,197 @@ jobs:
|
|||||||
asyncio.run(main())
|
asyncio.run(main())
|
||||||
PY
|
PY
|
||||||
|
|
||||||
|
# A valid unique index can still hide heap/index drift introduced by an
|
||||||
|
# older restore or repair. Quarantine only the non-canonical rows, rebuild
|
||||||
|
# the index transactionally, and prove the conflict-update path before
|
||||||
|
# rolling out the scanner.
|
||||||
|
- name: Repair Asset Inventory Canonical Integrity
|
||||||
|
env:
|
||||||
|
DATABASE_URL: ${{ secrets.DATABASE_URL }}
|
||||||
|
MIGRATION_DATABASE_URL: ${{ secrets.MIGRATION_DATABASE_URL }}
|
||||||
|
run: |
|
||||||
|
set -euo pipefail
|
||||||
|
if [ -z "${MIGRATION_DATABASE_URL:-}" ]; then
|
||||||
|
echo "::error::MIGRATION_DATABASE_URL secret not set in Gitea"
|
||||||
|
exit 1
|
||||||
|
fi
|
||||||
|
docker run --rm --network bridge -i \
|
||||||
|
-e DATABASE_URL \
|
||||||
|
-e MIGRATION_DATABASE_URL \
|
||||||
|
-v "$PWD/apps/api/migrations/asset_inventory_canonical_integrity_2026-07-14.sql:/tmp/asset_inventory_integrity.sql:ro" \
|
||||||
|
"${HARBOR}/awoooi/api:${{ github.sha }}" python - <<'PY'
|
||||||
|
import asyncio
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import asyncpg
|
||||||
|
|
||||||
|
MIGRATION_NAME = "asset_inventory_canonical_integrity_2026-07-14"
|
||||||
|
MIGRATION = Path("/tmp/asset_inventory_integrity.sql").read_text(
|
||||||
|
encoding="utf-8"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def normalize_url(value: str) -> str:
|
||||||
|
return value.replace("postgresql+asyncpg://", "postgresql://", 1)
|
||||||
|
|
||||||
|
|
||||||
|
def receipt_valid(receipt: dict[str, object] | None) -> bool:
|
||||||
|
return bool(
|
||||||
|
receipt
|
||||||
|
and receipt.get("duplicate_group_count_after") == 0
|
||||||
|
and receipt.get("index_rebuilt") is True
|
||||||
|
and receipt.get("canonical_upsert_verified") is True
|
||||||
|
and receipt.get("data_rows_deleted") == 0
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def integrity_status(
|
||||||
|
connection: asyncpg.Connection,
|
||||||
|
) -> tuple[int, int, bool, dict[str, object] | None]:
|
||||||
|
row = await connection.fetchrow(
|
||||||
|
"""
|
||||||
|
WITH duplicate_groups AS (
|
||||||
|
SELECT COUNT(*) AS count
|
||||||
|
FROM (
|
||||||
|
SELECT asset_key
|
||||||
|
FROM asset_inventory
|
||||||
|
GROUP BY asset_key
|
||||||
|
HAVING COUNT(*) > 1
|
||||||
|
) AS grouped_duplicates
|
||||||
|
), latest_receipt AS (
|
||||||
|
SELECT output::text AS output
|
||||||
|
FROM automation_operation_log
|
||||||
|
WHERE actor = 'gitea_cd_asset_integrity_reconciler'
|
||||||
|
AND operation_type = 'asset_discovered'
|
||||||
|
AND status = 'success'
|
||||||
|
AND input ->> 'migration' = $1
|
||||||
|
ORDER BY created_at DESC
|
||||||
|
LIMIT 1
|
||||||
|
)
|
||||||
|
SELECT
|
||||||
|
duplicate_groups.count AS duplicate_group_count,
|
||||||
|
(
|
||||||
|
SELECT COUNT(*)
|
||||||
|
FROM asset_inventory
|
||||||
|
WHERE lifecycle_state = 'deprecated'
|
||||||
|
AND metadata ->> 'integrity_state' =
|
||||||
|
'quarantined_duplicate'
|
||||||
|
) AS quarantined_row_count,
|
||||||
|
COALESCE((
|
||||||
|
SELECT
|
||||||
|
index_row.indisvalid
|
||||||
|
AND index_row.indisready
|
||||||
|
AND index_row.indislive
|
||||||
|
AND index_row.indisunique
|
||||||
|
FROM pg_index AS index_row
|
||||||
|
WHERE index_row.indexrelid =
|
||||||
|
'asset_inventory_asset_key_key'::regclass
|
||||||
|
), FALSE) AS unique_index_healthy,
|
||||||
|
(SELECT output FROM latest_receipt) AS receipt_output
|
||||||
|
FROM duplicate_groups
|
||||||
|
""",
|
||||||
|
MIGRATION_NAME,
|
||||||
|
)
|
||||||
|
if row is None:
|
||||||
|
raise RuntimeError("asset_inventory_integrity_status_missing")
|
||||||
|
receipt_text = row["receipt_output"]
|
||||||
|
receipt = json.loads(receipt_text) if receipt_text else None
|
||||||
|
return (
|
||||||
|
int(row["duplicate_group_count"] or 0),
|
||||||
|
int(row["quarantined_row_count"] or 0),
|
||||||
|
bool(row["unique_index_healthy"]),
|
||||||
|
receipt,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def reconcile(
|
||||||
|
url: str,
|
||||||
|
) -> tuple[int, int, bool, dict[str, object] | None, bool]:
|
||||||
|
connection = await asyncpg.connect(normalize_url(url), timeout=10)
|
||||||
|
applied = False
|
||||||
|
try:
|
||||||
|
# Do not let a corrupt-but-valid unique index hide duplicate
|
||||||
|
# heap rows from either the preflight or post-verifier.
|
||||||
|
await connection.execute("SET enable_indexscan TO off")
|
||||||
|
await connection.execute("SET enable_indexonlyscan TO off")
|
||||||
|
await connection.execute("SET enable_bitmapscan TO off")
|
||||||
|
duplicate_groups, _, index_healthy, receipt = (
|
||||||
|
await integrity_status(connection)
|
||||||
|
)
|
||||||
|
if (
|
||||||
|
duplicate_groups > 0
|
||||||
|
or not index_healthy
|
||||||
|
or not receipt_valid(receipt)
|
||||||
|
):
|
||||||
|
async with connection.transaction():
|
||||||
|
await connection.execute(MIGRATION)
|
||||||
|
applied = True
|
||||||
|
status = await integrity_status(connection)
|
||||||
|
return (*status, applied)
|
||||||
|
finally:
|
||||||
|
await connection.close()
|
||||||
|
|
||||||
|
|
||||||
|
async def main() -> None:
|
||||||
|
candidates = [
|
||||||
|
("migration_role", os.environ["MIGRATION_DATABASE_URL"]),
|
||||||
|
("owner_role", os.environ.get("DATABASE_URL", "")),
|
||||||
|
]
|
||||||
|
for index, (role, url) in enumerate(candidates):
|
||||||
|
if not url:
|
||||||
|
continue
|
||||||
|
try:
|
||||||
|
(
|
||||||
|
duplicate_groups,
|
||||||
|
quarantined_rows,
|
||||||
|
index_healthy,
|
||||||
|
receipt,
|
||||||
|
applied,
|
||||||
|
) = await reconcile(url)
|
||||||
|
verified = (
|
||||||
|
duplicate_groups == 0
|
||||||
|
and index_healthy
|
||||||
|
and receipt_valid(receipt)
|
||||||
|
)
|
||||||
|
print(f"asset_inventory_integrity_migration_role={role}")
|
||||||
|
print(
|
||||||
|
"asset_inventory_integrity_migration_applied="
|
||||||
|
f"{str(applied).lower()}"
|
||||||
|
)
|
||||||
|
print(f"duplicate_group_count={duplicate_groups}")
|
||||||
|
print(f"quarantined_row_count={quarantined_rows}")
|
||||||
|
print(f"unique_index_healthy={str(index_healthy).lower()}")
|
||||||
|
print(
|
||||||
|
"durable_integrity_receipt_present="
|
||||||
|
f"{str(receipt is not None).lower()}"
|
||||||
|
)
|
||||||
|
print(f"canonical_integrity_verified={str(verified).lower()}")
|
||||||
|
if not verified:
|
||||||
|
raise RuntimeError(
|
||||||
|
"asset_inventory_canonical_integrity_verifier_failed"
|
||||||
|
)
|
||||||
|
return
|
||||||
|
except asyncpg.PostgresError as exc:
|
||||||
|
permission_denied = getattr(exc, "sqlstate", "") == "42501"
|
||||||
|
has_owner_fallback = index == 0 and bool(candidates[1][1])
|
||||||
|
if permission_denied and has_owner_fallback:
|
||||||
|
print("asset_inventory_integrity_owner_fallback=true")
|
||||||
|
continue
|
||||||
|
print(
|
||||||
|
"asset_inventory_integrity_error_class="
|
||||||
|
f"{type(exc).__name__}"
|
||||||
|
)
|
||||||
|
raise
|
||||||
|
raise RuntimeError(
|
||||||
|
"asset_inventory_integrity_migration_no_usable_connection"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
asyncio.run(main())
|
||||||
|
PY
|
||||||
|
|
||||||
# 2026-03-31 ogt: 移除中間通知
|
# 2026-03-31 ogt: 移除中間通知
|
||||||
|
|
||||||
# 2026-03-31 ogt: P0-1 Secrets 自動注入 (ADR-035 強制)
|
# 2026-03-31 ogt: P0-1 Secrets 自動注入 (ADR-035 強制)
|
||||||
|
|||||||
@@ -0,0 +1,215 @@
|
|||||||
|
-- Repair duplicate canonical asset keys without deleting historical rows.
|
||||||
|
-- Safety: bounded quarantine, transactional index rebuild, independent verifier.
|
||||||
|
|
||||||
|
SET lock_timeout = '10s';
|
||||||
|
SET statement_timeout = '90s';
|
||||||
|
SET enable_indexscan = 'off';
|
||||||
|
SET enable_indexonlyscan = 'off';
|
||||||
|
SET enable_bitmapscan = 'off';
|
||||||
|
|
||||||
|
CREATE TEMP TABLE asset_inventory_integrity_repair_map
|
||||||
|
ON COMMIT DROP
|
||||||
|
AS
|
||||||
|
WITH ranked_assets AS (
|
||||||
|
SELECT
|
||||||
|
asset_id,
|
||||||
|
asset_key AS original_asset_key,
|
||||||
|
FIRST_VALUE(asset_id) OVER (
|
||||||
|
PARTITION BY asset_key
|
||||||
|
ORDER BY first_seen_at ASC, asset_id ASC
|
||||||
|
) AS canonical_asset_id,
|
||||||
|
ROW_NUMBER() OVER (
|
||||||
|
PARTITION BY asset_key
|
||||||
|
ORDER BY first_seen_at ASC, asset_id ASC
|
||||||
|
) AS duplicate_rank
|
||||||
|
FROM asset_inventory
|
||||||
|
)
|
||||||
|
SELECT
|
||||||
|
asset_id,
|
||||||
|
original_asset_key,
|
||||||
|
canonical_asset_id
|
||||||
|
FROM ranked_assets
|
||||||
|
WHERE duplicate_rank > 1;
|
||||||
|
|
||||||
|
DO $$
|
||||||
|
DECLARE
|
||||||
|
repair_row_count BIGINT;
|
||||||
|
BEGIN
|
||||||
|
SELECT COUNT(*) INTO repair_row_count
|
||||||
|
FROM asset_inventory_integrity_repair_map;
|
||||||
|
|
||||||
|
IF repair_row_count > 100 THEN
|
||||||
|
RAISE EXCEPTION
|
||||||
|
'asset_inventory_integrity_repair_bound_exceeded: % rows',
|
||||||
|
repair_row_count;
|
||||||
|
END IF;
|
||||||
|
END
|
||||||
|
$$;
|
||||||
|
|
||||||
|
UPDATE asset_inventory AS duplicate_asset
|
||||||
|
SET
|
||||||
|
asset_key = duplicate_asset.asset_key
|
||||||
|
|| '/__integrity_quarantine__/'
|
||||||
|
|| duplicate_asset.asset_id::TEXT,
|
||||||
|
lifecycle_state = 'deprecated',
|
||||||
|
decommissioned_at = COALESCE(duplicate_asset.decommissioned_at, NOW()),
|
||||||
|
metadata = COALESCE(duplicate_asset.metadata, '{}'::JSONB)
|
||||||
|
|| JSONB_BUILD_OBJECT(
|
||||||
|
'integrity_state', 'quarantined_duplicate',
|
||||||
|
'integrity_canonical_asset_id', repair.canonical_asset_id,
|
||||||
|
'integrity_repair', 'asset_inventory_canonical_integrity_2026-07-14'
|
||||||
|
),
|
||||||
|
tags = CASE
|
||||||
|
WHEN 'integrity:quarantined_duplicate' = ANY(duplicate_asset.tags)
|
||||||
|
THEN duplicate_asset.tags
|
||||||
|
ELSE ARRAY_APPEND(
|
||||||
|
duplicate_asset.tags,
|
||||||
|
'integrity:quarantined_duplicate'
|
||||||
|
)
|
||||||
|
END,
|
||||||
|
updated_at = NOW()
|
||||||
|
FROM asset_inventory_integrity_repair_map AS repair
|
||||||
|
WHERE duplicate_asset.asset_id = repair.asset_id;
|
||||||
|
|
||||||
|
-- Rebuild the unique index after every formerly unindexed duplicate has a
|
||||||
|
-- distinct quarantine key. The outer CD transaction rolls back on any error.
|
||||||
|
REINDEX INDEX asset_inventory_asset_key_key;
|
||||||
|
|
||||||
|
DO $$
|
||||||
|
DECLARE
|
||||||
|
duplicate_group_count BIGINT;
|
||||||
|
repaired_row_count BIGINT;
|
||||||
|
expected_repair_count BIGINT;
|
||||||
|
index_healthy BOOLEAN;
|
||||||
|
BEGIN
|
||||||
|
SELECT COUNT(*) INTO duplicate_group_count
|
||||||
|
FROM (
|
||||||
|
SELECT asset_key
|
||||||
|
FROM asset_inventory
|
||||||
|
GROUP BY asset_key
|
||||||
|
HAVING COUNT(*) > 1
|
||||||
|
) AS duplicate_groups;
|
||||||
|
|
||||||
|
SELECT COUNT(*) INTO expected_repair_count
|
||||||
|
FROM asset_inventory_integrity_repair_map;
|
||||||
|
|
||||||
|
SELECT COUNT(*) INTO repaired_row_count
|
||||||
|
FROM asset_inventory AS repaired
|
||||||
|
JOIN asset_inventory_integrity_repair_map AS repair
|
||||||
|
ON repair.asset_id = repaired.asset_id
|
||||||
|
WHERE repaired.lifecycle_state = 'deprecated'
|
||||||
|
AND repaired.asset_key = repair.original_asset_key
|
||||||
|
|| '/__integrity_quarantine__/'
|
||||||
|
|| repaired.asset_id::TEXT
|
||||||
|
AND repaired.metadata ->> 'integrity_state' = 'quarantined_duplicate';
|
||||||
|
|
||||||
|
SELECT
|
||||||
|
index_row.indisvalid
|
||||||
|
AND index_row.indisready
|
||||||
|
AND index_row.indislive
|
||||||
|
AND index_row.indisunique
|
||||||
|
INTO index_healthy
|
||||||
|
FROM pg_index AS index_row
|
||||||
|
WHERE index_row.indexrelid = 'asset_inventory_asset_key_key'::REGCLASS;
|
||||||
|
|
||||||
|
IF duplicate_group_count <> 0 THEN
|
||||||
|
RAISE EXCEPTION
|
||||||
|
'asset_inventory_duplicate_groups_remain: %',
|
||||||
|
duplicate_group_count;
|
||||||
|
END IF;
|
||||||
|
|
||||||
|
IF repaired_row_count <> expected_repair_count THEN
|
||||||
|
RAISE EXCEPTION
|
||||||
|
'asset_inventory_quarantine_verifier_failed: expected %, got %',
|
||||||
|
expected_repair_count,
|
||||||
|
repaired_row_count;
|
||||||
|
END IF;
|
||||||
|
|
||||||
|
IF index_healthy IS DISTINCT FROM TRUE THEN
|
||||||
|
RAISE EXCEPTION 'asset_inventory_unique_index_not_healthy';
|
||||||
|
END IF;
|
||||||
|
END
|
||||||
|
$$;
|
||||||
|
|
||||||
|
-- Exercise the exact conflict-update path that failed before the repair while
|
||||||
|
-- preserving every application value.
|
||||||
|
WITH probe AS (
|
||||||
|
SELECT
|
||||||
|
asset_key,
|
||||||
|
asset_type,
|
||||||
|
host,
|
||||||
|
namespace,
|
||||||
|
name,
|
||||||
|
metadata,
|
||||||
|
tags,
|
||||||
|
environment,
|
||||||
|
lifecycle_state,
|
||||||
|
first_seen_at,
|
||||||
|
last_seen_at
|
||||||
|
FROM asset_inventory
|
||||||
|
ORDER BY asset_id
|
||||||
|
LIMIT 1
|
||||||
|
)
|
||||||
|
INSERT INTO asset_inventory (
|
||||||
|
asset_key,
|
||||||
|
asset_type,
|
||||||
|
host,
|
||||||
|
namespace,
|
||||||
|
name,
|
||||||
|
metadata,
|
||||||
|
tags,
|
||||||
|
environment,
|
||||||
|
lifecycle_state,
|
||||||
|
first_seen_at,
|
||||||
|
last_seen_at
|
||||||
|
)
|
||||||
|
SELECT
|
||||||
|
asset_key,
|
||||||
|
asset_type,
|
||||||
|
host,
|
||||||
|
namespace,
|
||||||
|
name,
|
||||||
|
metadata,
|
||||||
|
tags,
|
||||||
|
environment,
|
||||||
|
lifecycle_state,
|
||||||
|
first_seen_at,
|
||||||
|
last_seen_at
|
||||||
|
FROM probe
|
||||||
|
ON CONFLICT (asset_key) DO UPDATE
|
||||||
|
SET
|
||||||
|
last_seen_at = asset_inventory.last_seen_at,
|
||||||
|
updated_at = asset_inventory.updated_at;
|
||||||
|
|
||||||
|
INSERT INTO automation_operation_log (
|
||||||
|
operation_type,
|
||||||
|
actor,
|
||||||
|
status,
|
||||||
|
input,
|
||||||
|
output,
|
||||||
|
tags
|
||||||
|
)
|
||||||
|
SELECT
|
||||||
|
'asset_discovered',
|
||||||
|
'gitea_cd_asset_integrity_reconciler',
|
||||||
|
'success',
|
||||||
|
'{"work_item_id":"AIA-P0-006","migration":"asset_inventory_canonical_integrity_2026-07-14"}'::JSONB,
|
||||||
|
JSONB_BUILD_OBJECT(
|
||||||
|
'duplicate_group_count_before', COUNT(DISTINCT original_asset_key),
|
||||||
|
'rows_quarantined', COUNT(*),
|
||||||
|
'duplicate_group_count_after', 0,
|
||||||
|
'index_rebuilt', TRUE,
|
||||||
|
'canonical_upsert_verified', TRUE,
|
||||||
|
'data_rows_deleted', 0
|
||||||
|
),
|
||||||
|
ARRAY[
|
||||||
|
'asset_control_plane',
|
||||||
|
'integrity_repair',
|
||||||
|
'controlled_apply',
|
||||||
|
'post_verified',
|
||||||
|
'no_delete'
|
||||||
|
]
|
||||||
|
FROM asset_inventory_integrity_repair_map;
|
||||||
|
|
||||||
|
COMMENT ON INDEX asset_inventory_asset_key_key IS
|
||||||
|
'Canonical asset identity unique index; transactionally rebuilt and verified by Gitea CD.';
|
||||||
@@ -114,6 +114,8 @@ async def scan_once(
|
|||||||
error_msg: str | None = None
|
error_msg: str | None = None
|
||||||
|
|
||||||
try:
|
try:
|
||||||
|
await _assert_asset_key_integrity()
|
||||||
|
|
||||||
# 2026-04-19 v3 擴充: 多資源類型掃描 + relationship 提取
|
# 2026-04-19 v3 擴充: 多資源類型掃描 + relationship 提取
|
||||||
# 資源類型: pods (container), deployments/statefulsets/daemonsets (k8s_workload),
|
# 資源類型: pods (container), deployments/statefulsets/daemonsets (k8s_workload),
|
||||||
# services (k8s_resource), nodes (host), configmaps (k8s_resource)
|
# services (k8s_resource), nodes (host), configmaps (k8s_resource)
|
||||||
@@ -917,6 +919,61 @@ async def _finalize_discovery_run(
|
|||||||
"err": (error or "")[:2000] if error else None,
|
"err": (error or "")[:2000] if error else None,
|
||||||
"tags": ["asset_scanner", "discovery", *scope],
|
"tags": ["asset_scanner", "discovery", *scope],
|
||||||
},
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def _assert_asset_key_integrity() -> None:
|
||||||
|
"""Fail closed before discovery when canonical asset identity is corrupt."""
|
||||||
|
from sqlalchemy import text as _sql
|
||||||
|
|
||||||
|
from src.db.base import get_db_context
|
||||||
|
|
||||||
|
async with get_db_context(_PROJECT_ID) as db:
|
||||||
|
# A corrupt-but-valid unique index can make the planner incorrectly
|
||||||
|
# assume the heap is unique. Inspect the heap before trusting it.
|
||||||
|
await db.execute(_sql("SET LOCAL enable_indexscan = 'off'"))
|
||||||
|
await db.execute(_sql("SET LOCAL enable_indexonlyscan = 'off'"))
|
||||||
|
await db.execute(_sql("SET LOCAL enable_bitmapscan = 'off'"))
|
||||||
|
row = (
|
||||||
|
await db.execute(
|
||||||
|
_sql(
|
||||||
|
"""
|
||||||
|
SELECT
|
||||||
|
(
|
||||||
|
SELECT COUNT(*)
|
||||||
|
FROM (
|
||||||
|
SELECT asset_key
|
||||||
|
FROM asset_inventory
|
||||||
|
GROUP BY asset_key
|
||||||
|
HAVING COUNT(*) > 1
|
||||||
|
) AS duplicate_groups
|
||||||
|
) AS duplicate_group_count,
|
||||||
|
COALESCE((
|
||||||
|
SELECT
|
||||||
|
index_row.indisvalid
|
||||||
|
AND index_row.indisready
|
||||||
|
AND index_row.indislive
|
||||||
|
AND index_row.indisunique
|
||||||
|
FROM pg_constraint AS constraint_row
|
||||||
|
JOIN pg_index AS index_row
|
||||||
|
ON index_row.indexrelid = constraint_row.conindid
|
||||||
|
WHERE constraint_row.conrelid =
|
||||||
|
'asset_inventory'::regclass
|
||||||
|
AND constraint_row.conname =
|
||||||
|
'asset_inventory_asset_key_key'
|
||||||
|
), FALSE) AS unique_index_healthy
|
||||||
|
"""
|
||||||
|
)
|
||||||
|
)
|
||||||
|
).one()
|
||||||
|
|
||||||
|
duplicate_group_count = int(row[0] or 0)
|
||||||
|
unique_index_healthy = bool(row[1])
|
||||||
|
if duplicate_group_count or not unique_index_healthy:
|
||||||
|
raise RuntimeError(
|
||||||
|
"asset_inventory_key_integrity_failed "
|
||||||
|
f"duplicate_groups={duplicate_group_count} "
|
||||||
|
f"unique_index_healthy={str(unique_index_healthy).lower()}"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -52,7 +52,7 @@ def test_cd_applies_and_independently_verifies_the_bounded_migration() -> None:
|
|||||||
workflow = CD_WORKFLOW.read_text(encoding="utf-8")
|
workflow = CD_WORKFLOW.read_text(encoding="utf-8")
|
||||||
step = workflow.split(
|
step = workflow.split(
|
||||||
"- name: Reconcile Asset Inventory Type Constraint", 1
|
"- name: Reconcile Asset Inventory Type Constraint", 1
|
||||||
)[1].split("# 2026-03-31 ogt: 移除中間通知", 1)[0]
|
)[1].split("- name: Repair Asset Inventory Canonical Integrity", 1)[0]
|
||||||
script = step.split("run: |", 1)[1]
|
script = step.split("run: |", 1)[1]
|
||||||
|
|
||||||
assert "MIGRATION_DATABASE_URL: ${{ secrets.MIGRATION_DATABASE_URL }}" in step
|
assert "MIGRATION_DATABASE_URL: ${{ secrets.MIGRATION_DATABASE_URL }}" in step
|
||||||
|
|||||||
71
apps/api/tests/test_asset_inventory_integrity_migration.py
Normal file
71
apps/api/tests/test_asset_inventory_integrity_migration.py
Normal file
@@ -0,0 +1,71 @@
|
|||||||
|
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")
|
||||||
@@ -142,6 +142,81 @@ def test_asset_upsert_propagates_database_failure(monkeypatch) -> None:
|
|||||||
asyncio.run(exercise())
|
asyncio.run(exercise())
|
||||||
|
|
||||||
|
|
||||||
|
def test_asset_key_integrity_preflight_accepts_unique_inventory(monkeypatch) -> None:
|
||||||
|
requested_projects: list[str | None] = []
|
||||||
|
statements: list[str] = []
|
||||||
|
|
||||||
|
class IntegrityResult:
|
||||||
|
def one(self):
|
||||||
|
return 0, True
|
||||||
|
|
||||||
|
class IntegrityDatabase:
|
||||||
|
async def execute(self, statement, parameters=None):
|
||||||
|
statements.append(str(statement))
|
||||||
|
return IntegrityResult()
|
||||||
|
|
||||||
|
class IntegrityContext:
|
||||||
|
async def __aenter__(self):
|
||||||
|
return IntegrityDatabase()
|
||||||
|
|
||||||
|
async def __aexit__(self, exc_type, exc, traceback):
|
||||||
|
return False
|
||||||
|
|
||||||
|
def fake_db_context(project_id=None):
|
||||||
|
requested_projects.append(project_id)
|
||||||
|
return IntegrityContext()
|
||||||
|
|
||||||
|
monkeypatch.setattr("src.db.base.get_db_context", fake_db_context)
|
||||||
|
|
||||||
|
asyncio.run(asset_scanner_job._assert_asset_key_integrity())
|
||||||
|
|
||||||
|
assert requested_projects == ["awoooi"]
|
||||||
|
assert any("enable_indexscan" in statement for statement in statements)
|
||||||
|
assert any("enable_indexonlyscan" in statement for statement in statements)
|
||||||
|
assert any("enable_bitmapscan" in statement for statement in statements)
|
||||||
|
assert any("HAVING COUNT(*) > 1" in statement for statement in statements)
|
||||||
|
assert any(
|
||||||
|
"asset_inventory_asset_key_key" in statement for statement in statements
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_asset_key_integrity_preflight_fails_closed_on_duplicate_key(
|
||||||
|
monkeypatch,
|
||||||
|
) -> None:
|
||||||
|
class IntegrityResult:
|
||||||
|
def one(self):
|
||||||
|
return 1, True
|
||||||
|
|
||||||
|
class IntegrityDatabase:
|
||||||
|
async def execute(self, statement, parameters=None):
|
||||||
|
return IntegrityResult()
|
||||||
|
|
||||||
|
class IntegrityContext:
|
||||||
|
async def __aenter__(self):
|
||||||
|
return IntegrityDatabase()
|
||||||
|
|
||||||
|
async def __aexit__(self, exc_type, exc, traceback):
|
||||||
|
return False
|
||||||
|
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"src.db.base.get_db_context",
|
||||||
|
lambda project_id=None: IntegrityContext(),
|
||||||
|
)
|
||||||
|
|
||||||
|
async def exercise() -> None:
|
||||||
|
try:
|
||||||
|
await asset_scanner_job._assert_asset_key_integrity()
|
||||||
|
except RuntimeError as exc:
|
||||||
|
assert str(exc) == (
|
||||||
|
"asset_inventory_key_integrity_failed "
|
||||||
|
"duplicate_groups=1 unique_index_healthy=true"
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
raise AssertionError("duplicate canonical keys must fail closed")
|
||||||
|
|
||||||
|
asyncio.run(exercise())
|
||||||
|
|
||||||
|
|
||||||
def test_deprecate_missing_assets_is_bounded_to_successful_prefixes(monkeypatch) -> None:
|
def test_deprecate_missing_assets_is_bounded_to_successful_prefixes(monkeypatch) -> None:
|
||||||
captured: dict[str, object] = {}
|
captured: dict[str, object] = {}
|
||||||
|
|
||||||
@@ -230,6 +305,9 @@ def test_scan_terminal_is_failed_when_any_collector_is_incomplete(monkeypatch) -
|
|||||||
async def fake_collect():
|
async def fake_collect():
|
||||||
return [], [], (), ("prometheus_targets",)
|
return [], [], (), ("prometheus_targets",)
|
||||||
|
|
||||||
|
async def fake_integrity():
|
||||||
|
return None
|
||||||
|
|
||||||
async def fake_upsert(*args, **kwargs):
|
async def fake_upsert(*args, **kwargs):
|
||||||
return 0, 0
|
return 0, 0
|
||||||
|
|
||||||
@@ -246,6 +324,11 @@ def test_scan_terminal_is_failed_when_any_collector_is_incomplete(monkeypatch) -
|
|||||||
terminal.update(kwargs)
|
terminal.update(kwargs)
|
||||||
|
|
||||||
monkeypatch.setattr(asset_scanner_job, "_start_discovery_run", fake_start)
|
monkeypatch.setattr(asset_scanner_job, "_start_discovery_run", fake_start)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
asset_scanner_job,
|
||||||
|
"_assert_asset_key_integrity",
|
||||||
|
fake_integrity,
|
||||||
|
)
|
||||||
monkeypatch.setattr(asset_scanner_job, "_collect_runtime_assets", fake_collect)
|
monkeypatch.setattr(asset_scanner_job, "_collect_runtime_assets", fake_collect)
|
||||||
monkeypatch.setattr(asset_scanner_job, "_upsert_assets", fake_upsert)
|
monkeypatch.setattr(asset_scanner_job, "_upsert_assets", fake_upsert)
|
||||||
monkeypatch.setattr(
|
monkeypatch.setattr(
|
||||||
|
|||||||
Reference in New Issue
Block a user