08f3a70ce6
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 1s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 4m41s
CD Pipeline / build-and-deploy (push) Successful in 15m36s
CD Pipeline / post-deploy-checks (push) Successful in 4m19s
354 lines
11 KiB
Python
354 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
# ruff: noqa: E402
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
os.environ.setdefault(
|
|
"DATABASE_URL",
|
|
"postgresql+asyncpg://test:test@localhost/test",
|
|
)
|
|
|
|
from src.services.db_bounded_executor import (
|
|
CANONICAL_ASSET,
|
|
COMMAND_ID,
|
|
CREATE_INDEX_SQL,
|
|
MIGRATION_FILE,
|
|
MIGRATION_SHA256,
|
|
NORMALIZED_SQL,
|
|
NORMALIZED_SQL_SHA256,
|
|
BoundedExecutorContractError,
|
|
evaluate_catalog_snapshot,
|
|
execute_bounded_db_command,
|
|
validate_execution_contract,
|
|
)
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SOURCE_REVISION = "a" * 40
|
|
IDENTITY = {
|
|
"trace_id": "trace-AIA-SRE-013-db-index",
|
|
"run_id": "run-AIA-SRE-013-db-index",
|
|
"work_item_id": "AIA-SRE-013",
|
|
}
|
|
|
|
|
|
def _normalized_migration_sql() -> str:
|
|
source = (ROOT / "migrations" / MIGRATION_FILE).read_text()
|
|
sql_only = " ".join(
|
|
line for line in source.splitlines() if not line.lstrip().startswith("--")
|
|
)
|
|
return " ".join(sql_only.split())
|
|
|
|
|
|
def _absent_catalog() -> dict:
|
|
return {
|
|
"table_exists": True,
|
|
"columns_complete": True,
|
|
"table_kind_supported": True,
|
|
"primary_database": True,
|
|
"database_writable": True,
|
|
"schema_usage": True,
|
|
"schema_create": True,
|
|
"table_owner_privilege_inherited": True,
|
|
"table_select": True,
|
|
"index_count": 0,
|
|
"index_ready": False,
|
|
"index_valid": False,
|
|
"index_non_unique": False,
|
|
"key_count_exact": False,
|
|
"total_attribute_count_exact": False,
|
|
"access_method_exact": False,
|
|
"key_1": "",
|
|
"key_2": "",
|
|
"include_1": "",
|
|
"predicate": "",
|
|
}
|
|
|
|
|
|
def _exact_catalog() -> dict:
|
|
return {
|
|
**_absent_catalog(),
|
|
"index_count": 1,
|
|
"index_ready": True,
|
|
"index_valid": True,
|
|
"index_non_unique": True,
|
|
"key_count_exact": True,
|
|
"total_attribute_count_exact": True,
|
|
"access_method_exact": True,
|
|
"key_1": "project_id",
|
|
"key_2": "queued_at DESC",
|
|
"include_1": "send_status",
|
|
"predicate": """
|
|
((channel_type = 'telegram'::text) AND
|
|
(((source_envelope #>> '{callback_reply,action}'::text[]) =
|
|
'controlled_apply_result'::text) OR
|
|
((source_envelope #>> '{alert_notification,status}'::text[]) =
|
|
'alert_notification_sent'::text)))
|
|
""",
|
|
}
|
|
|
|
|
|
def _contract(mode: str = "check") -> dict:
|
|
return {
|
|
"mode": mode,
|
|
"command_id": COMMAND_ID,
|
|
"canonical_asset": CANONICAL_ASSET,
|
|
"migration_sha256": MIGRATION_SHA256,
|
|
"normalized_sql_sha256": NORMALIZED_SQL_SHA256,
|
|
**IDENTITY,
|
|
"source_revision": SOURCE_REVISION,
|
|
"runtime_source_revision": SOURCE_REVISION,
|
|
"pod_name": "awoooi-api-7bc9867b77-alpha1",
|
|
}
|
|
|
|
|
|
def test_fixed_migration_and_embedded_sql_digests_match() -> None:
|
|
migration = (ROOT / "migrations" / MIGRATION_FILE).read_bytes()
|
|
|
|
assert hashlib.sha256(migration).hexdigest() == MIGRATION_SHA256
|
|
assert _normalized_migration_sql() == NORMALIZED_SQL
|
|
assert hashlib.sha256(NORMALIZED_SQL.encode()).hexdigest() == NORMALIZED_SQL_SHA256
|
|
|
|
|
|
def test_embedded_sql_is_one_fixed_transactionless_create_only() -> None:
|
|
normalized = CREATE_INDEX_SQL.upper()
|
|
|
|
assert normalized.count("CREATE INDEX CONCURRENTLY IF NOT EXISTS") == 1
|
|
assert normalized.count(";") == 1
|
|
for forbidden in (
|
|
"DROP ",
|
|
"TRUNCATE ",
|
|
"DELETE ",
|
|
"UPDATE ",
|
|
"INSERT ",
|
|
"ALTER ",
|
|
"BEGIN",
|
|
"COMMIT",
|
|
):
|
|
assert forbidden not in normalized
|
|
|
|
|
|
def test_precheck_requires_immediately_effective_owner_and_plain_table() -> None:
|
|
source = (ROOT / "src/services/db_bounded_executor.py").read_text()
|
|
|
|
assert "pg_has_role(current_user, t.table_owner_oid, 'USAGE')" in source
|
|
assert "pg_has_role(current_user, t.table_owner_oid, 'MEMBER')" not in source
|
|
assert "COALESCE(t.table_relkind = 'r', false)" in source
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "value", "error"),
|
|
(
|
|
("command_id", "arbitrary_sql_v1", "command_id_contract_mismatch"),
|
|
("canonical_asset", "public.users", "canonical_asset_contract_mismatch"),
|
|
("migration_sha256", "0" * 64, "migration_sha256_contract_mismatch"),
|
|
(
|
|
"normalized_sql_sha256",
|
|
"f" * 64,
|
|
"normalized_sql_sha256_contract_mismatch",
|
|
),
|
|
("mode", "rollback", "mode_not_allowlisted"),
|
|
("pod_name", "awoooi-worker-1", "runtime_not_allowlisted_api_pod"),
|
|
(
|
|
"runtime_source_revision",
|
|
"b" * 40,
|
|
"runtime_source_revision_mismatch",
|
|
),
|
|
),
|
|
)
|
|
def test_contract_rejects_any_scope_or_runtime_drift(
|
|
field: str,
|
|
value: str,
|
|
error: str,
|
|
) -> None:
|
|
payload = _contract()
|
|
payload[field] = value
|
|
|
|
with pytest.raises(BoundedExecutorContractError, match=error):
|
|
validate_execution_contract(**payload)
|
|
|
|
|
|
def test_catalog_precheck_allows_absent_index_but_rejects_drift() -> None:
|
|
absent = evaluate_catalog_snapshot(_absent_catalog())
|
|
exact = evaluate_catalog_snapshot(_exact_catalog())
|
|
drift_row = {**_exact_catalog(), "key_2": "queued_at"}
|
|
drift = evaluate_catalog_snapshot(drift_row)
|
|
non_owner = evaluate_catalog_snapshot(
|
|
{**_absent_catalog(), "table_owner_privilege_inherited": False}
|
|
)
|
|
select_revoked = evaluate_catalog_snapshot(
|
|
{**_absent_catalog(), "table_select": False}
|
|
)
|
|
partitioned_parent = evaluate_catalog_snapshot(
|
|
{**_absent_catalog(), "table_kind_supported": False}
|
|
)
|
|
|
|
assert absent["apply_ready"] is True
|
|
assert absent["index_exists"] is False
|
|
assert exact["index_shape_exact"] is True
|
|
assert exact["apply_ready"] is True
|
|
assert drift["catalog_drift"] is True
|
|
assert drift["apply_ready"] is False
|
|
assert non_owner["apply_ready"] is False
|
|
# CREATE INDEX requires table ownership, not a separate SELECT grant.
|
|
assert select_revoked["apply_ready"] is True
|
|
assert partitioned_parent["apply_ready"] is False
|
|
|
|
|
|
class _Result:
|
|
def __init__(self, *, row: dict | None = None, scalar: bool | None = None):
|
|
self.row = row
|
|
self.scalar = scalar
|
|
|
|
def mappings(self):
|
|
return self
|
|
|
|
def one(self):
|
|
assert self.row is not None
|
|
return self.row
|
|
|
|
def scalar_one(self):
|
|
return self.scalar
|
|
|
|
|
|
class _Connection:
|
|
def __init__(self, catalog_rows: list[dict]):
|
|
self.catalog_rows = list(catalog_rows)
|
|
self.statements: list[str] = []
|
|
self.isolation_level = ""
|
|
|
|
async def execution_options(self, *, isolation_level: str):
|
|
self.isolation_level = isolation_level
|
|
return self
|
|
|
|
async def execute(self, statement):
|
|
sql = str(statement)
|
|
self.statements.append(sql)
|
|
if "WITH target AS" in sql:
|
|
return _Result(row=self.catalog_rows.pop(0))
|
|
if sql.startswith("SET "):
|
|
return _Result()
|
|
if "pg_try_advisory_lock" in sql:
|
|
return _Result(scalar=True)
|
|
if "pg_advisory_unlock" in sql:
|
|
return _Result(scalar=True)
|
|
if "CREATE INDEX CONCURRENTLY IF NOT EXISTS" in sql:
|
|
return _Result()
|
|
raise AssertionError(f"unexpected SQL: {sql}")
|
|
|
|
|
|
class _ConnectionContext:
|
|
def __init__(self, connection: _Connection):
|
|
self.connection = connection
|
|
|
|
async def __aenter__(self):
|
|
return self.connection
|
|
|
|
async def __aexit__(self, *_args):
|
|
return False
|
|
|
|
|
|
class _Engine:
|
|
def __init__(self, connection: _Connection):
|
|
self.connection = connection
|
|
self.connect_count = 0
|
|
|
|
def connect(self):
|
|
self.connect_count += 1
|
|
return _ConnectionContext(self.connection)
|
|
|
|
|
|
class _CreateFailureConnection(_Connection):
|
|
async def execute(self, statement):
|
|
sql = str(statement)
|
|
if "CREATE INDEX CONCURRENTLY IF NOT EXISTS" in sql:
|
|
self.statements.append(sql)
|
|
raise SQLAlchemyError("sensitive backend detail must not escape")
|
|
return await super().execute(statement)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_uses_autocommit_fixed_sql_and_same_session_lock_release() -> None:
|
|
connection = _Connection([_absent_catalog(), _absent_catalog(), _exact_catalog()])
|
|
engine = _Engine(connection)
|
|
|
|
receipt = await execute_bounded_db_command(
|
|
**_contract("apply"),
|
|
engine=engine,
|
|
)
|
|
|
|
assert receipt["terminal"] == "apply_verified_local"
|
|
assert receipt["writes_performed"] is True
|
|
assert receipt["domain"] == "database"
|
|
assert receipt["cross_domain_fallback_allowed"] is False
|
|
assert receipt["secret_value_read"] is False
|
|
assert connection.isolation_level == "AUTOCOMMIT"
|
|
assert engine.connect_count == 1
|
|
assert connection.statements[:3] == [
|
|
"SET search_path TO pg_catalog, public",
|
|
"SET lock_timeout TO '5s'",
|
|
"SET statement_timeout TO '10min'",
|
|
]
|
|
assert sum("CREATE INDEX CONCURRENTLY" in sql for sql in connection.statements) == 1
|
|
assert connection.statements.index(CREATE_INDEX_SQL) < next(
|
|
index
|
|
for index, sql in enumerate(connection.statements)
|
|
if "pg_advisory_unlock" in sql
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_is_idempotent_when_exact_index_already_exists() -> None:
|
|
connection = _Connection([_exact_catalog(), _exact_catalog(), _exact_catalog()])
|
|
engine = _Engine(connection)
|
|
|
|
receipt = await execute_bounded_db_command(
|
|
**_contract("apply"),
|
|
engine=engine,
|
|
)
|
|
|
|
assert receipt["terminal"] == "apply_verified_local"
|
|
assert receipt["writes_performed"] is False
|
|
assert all("CREATE INDEX CONCURRENTLY" not in sql for sql in connection.statements)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_is_catalog_only_and_never_takes_the_apply_lock() -> None:
|
|
connection = _Connection([_exact_catalog()])
|
|
engine = _Engine(connection)
|
|
|
|
receipt = await execute_bounded_db_command(
|
|
**_contract("verify"),
|
|
engine=engine,
|
|
)
|
|
|
|
assert receipt["terminal"] == "verified_healthy"
|
|
assert receipt["executor_role"] == "independent_catalog_verifier"
|
|
assert receipt["writes_performed"] is False
|
|
assert all("advisory_lock" not in sql for sql in connection.statements)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_failed_create_reports_unknown_write_state_without_backend_detail() -> (
|
|
None
|
|
):
|
|
connection = _CreateFailureConnection([_absent_catalog(), _absent_catalog()])
|
|
engine = _Engine(connection)
|
|
|
|
receipt = await execute_bounded_db_command(
|
|
**_contract("apply"),
|
|
engine=engine,
|
|
)
|
|
|
|
assert receipt["terminal"] == "failed"
|
|
assert receipt["write_attempted"] is True
|
|
assert receipt["writes_performed"] is None
|
|
assert receipt["error_code"] == "database_operation_failed"
|
|
assert "sensitive backend detail" not in json.dumps(receipt)
|
|
assert any("pg_advisory_unlock" in sql for sql in connection.statements)
|