GovOPlaN/govoplan-core
1
0
Fork 0
Code Issues 44 Pull Requests Actions Packages Projects Releases Wiki Activity

test(db): prove PostgreSQL retirement atomicity

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-22 15:25:28 +02:00
parent d487726f4d
commit f876345656
3 changed files with 453 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
docs/DEPLOYMENT_OPERATOR_GUIDE.md
View File

@@ -144,8 +144,12 @@ tools/checks/postgres-integration-check.py \
```
The integration check runs migrations and startup smoke checks across the
standard module permutations. `--reset-schema` is destructive and belongs only
on throwaway databases.
standard module permutations. It first requires the retirement atomicity proof,
using Files' real secret-owning provider and Audit's persistent recorder. That
proof uses only random, test-owned schemas and cleans them afterward; it does
not reset `public`. `--reset-schema` is destructive and belongs only on
throwaway databases. Do not pass `--skip-retirement-atomicity` when collecting
release evidence.
### Broker And Workers

13
docs/RELEASE_DEPENDENCIES.md
View File

@@ -788,7 +788,18 @@ tools/checks/postgres-integration-check.py \
The script checks migrations and `/health` startup for core-only, files-only,
mail-only, campaign-only, campaign+files, campaign+mail, and full-product
module sets. `--reset-schema` is destructive and must only be used against a
throwaway database.
throwaway database. Before those permutations, the required Core proof runs in
random, test-owned schemas without modifying `public`. It exercises Files' real
credential-owning retirement provider and proves that credential scrubbing,
non-secret audit
insertion, and table retirement commit together; database-injected audit and
DDL failures roll the entire unit back. A 500 ms PostgreSQL `lock_timeout` and
captured backend process IDs also prove that each `DROP TABLE` uses the
installer Session connection instead of waiting through a second connection.
The meta check enables the release-gate flag so missing PostgreSQL configuration
or full-stack test packages are a hard failure; ordinary Core-only test discovery
skips this integration proof. Do not pass `--skip-retirement-atomicity` when
collecting release evidence.
## Migration Baselines

435
tests/test_postgres_retirement_atomicity.py Normal file
View File

@@ -0,0 +1,435 @@
from __future__ import annotations
import json
import os
import unittest
from uuid import uuid4
from sqlalchemy import create_engine, event, inspect, text
from sqlalchemy.engine import Connection, Engine
from sqlalchemy.exc import DBAPIError
from sqlalchemy.orm import Session
from sqlalchemy.schema import CreateSchema, DropSchema
from govoplan_core.core.access import CAPABILITY_AUDIT_RECORDER
from govoplan_core.core.modules import MigrationRetirementPlan, ModuleContext
from govoplan_core.core.runtime import clear_runtime, configure_runtime, get_runtime_context
from govoplan_core.db.base import Base
_CREDENTIAL_ID = "retirement-credential"
_TENANT_ID = "retirement-tenant"
_PRIVATE_FIXTURE_VALUES = (
"opaque-password-ciphertext",
"opaque-token-ciphertext",
"vault:test-only:credential",
"private-test-metadata",
)
_ROW_COUNT_QUERIES = {
"audit_log": text("SELECT count(*) FROM audit_log"),
"core_change_sequence": text("SELECT count(*) FROM core_change_sequence"),
"retirement_scrub_probe": text("SELECT count(*) FROM retirement_scrub_probe"),
}
def _postgres_database_url() -> str:
candidate = (
os.environ.get("GOVOPLAN_POSTGRES_DATABASE_URL")
or os.environ.get("DATABASE_URL")
or ""
).strip()
return candidate if candidate.startswith("postgresql") else ""
def _proof_is_required() -> bool:
return os.environ.get("GOVOPLAN_REQUIRE_POSTGRES_RETIREMENT_PROOF", "").strip().casefold() in {
"1",
"true",
"yes",
"on",
}
class _AuditCapabilityRegistry:
def __init__(self, recorder: object) -> None:
self._recorder = recorder
def has_capability(self, name: str) -> bool:
return name == CAPABILITY_AUDIT_RECORDER
def require_capability(self, name: str) -> object:
if not self.has_capability(name):
raise LookupError(name)
return self._recorder
class PostgreSQLRetirementAtomicityTests(unittest.TestCase):
"""Production-reference proof for destructive module retirement.
Every test owns a random PostgreSQL schema and removes it afterward. The
fixture stores only synthetic credential material. Assertions and audit
inspection expose booleans and allow-listed metadata, never stored values
or the database URL.
"""
def setUp(self) -> None:
database_url = _postgres_database_url()
if not database_url:
if _proof_is_required():
self.fail("the required PostgreSQL retirement proof has no PostgreSQL database")
self.skipTest("a PostgreSQL integration database is not configured")
try:
from govoplan_access.backend.db import models as access_models # noqa: F401
from govoplan_audit.backend.db.models import AuditLog
from govoplan_audit.backend.recording import SqlAuditRecorder
from govoplan_files.backend.db.models import FileConnectorCredential, FileConnectorProfile
from govoplan_files.backend.manifest import manifest as files_manifest
except ModuleNotFoundError as exc:
if _proof_is_required():
self.fail(f"the required PostgreSQL retirement proof is missing package: {exc.name}")
self.skipTest(f"the full-stack integration packages are not installed: {exc.name}")
self.audit_log_model = AuditLog
self.credential_model = FileConnectorCredential
self.profile_model = FileConnectorProfile
self.files_manifest = files_manifest
self.previous_runtime = get_runtime_context()
self.admin_engine: Engine | None = None
self.engine: Engine | None = None
self.schema_name = f"govoplan_retirement_{uuid4().hex}"
self.capture_retirement_ddl = False
self.retirement_ddl_backend_pids: list[int] = []
self.ddl_listener_registered = False
try:
self.admin_engine = create_engine(database_url, pool_pre_ping=True)
with self.admin_engine.begin() as connection:
if connection.dialect.name != "postgresql":
self.skipTest("the configured integration database is not PostgreSQL")
connection.execute(CreateSchema(self.schema_name))
self.engine = create_engine(
database_url,
pool_pre_ping=True,
pool_size=5,
max_overflow=0,
connect_args={
"options": (
f"-csearch_path={self.schema_name},pg_catalog "
"-clock_timeout=500ms -cstatement_timeout=10s "
"-capplication_name=govoplan-retirement-atomicity"
),
},
)
event.listen(self.engine, "before_cursor_execute", self._capture_ddl_connection)
self.ddl_listener_registered = True
self._create_fixture_schema()
self._seed_credential()
recorder = SqlAuditRecorder()
configure_runtime(ModuleContext(registry=_AuditCapabilityRegistry(recorder), settings=object()))
except Exception:
self._cleanup()
raise
def tearDown(self) -> None:
self._cleanup()
def test_scrub_audit_and_table_retirement_commit_together(self) -> None:
assert self.engine is not None
with Session(self.engine) as session:
installer_backend_pid = self._session_backend_pid(session)
plan = self._retirement_plan(session)
self._execute_retirement(session, plan)
session.commit()
self.assertTrue(self.retirement_ddl_backend_pids, "the retirement provider emitted no DROP TABLE statement")
self.assertEqual(
{installer_backend_pid},
set(self.retirement_ddl_backend_pids),
"destructive DDL did not use the installer Session connection",
)
with self.engine.connect() as connection:
self.assertFalse(inspect(connection).has_table(self.credential_model.__tablename__))
self.assertFalse(inspect(connection).has_table(self.profile_model.__tablename__))
probe = connection.execute(text(
"SELECT password_removed, token_removed, username_removed, "
"secret_reference_removed, metadata_removed "
"FROM retirement_scrub_probe"
)).mappings().one()
self.assertTrue(all(probe.values()), "the credential scrub probe did not observe every removal")
audit_rows = connection.execute(text(
"SELECT action, object_type, object_id, details "
"FROM audit_log ORDER BY created_at, id"
)).mappings().all()
self.assertEqual(1, len(audit_rows))
audit_row = audit_rows[0]
self.assertEqual("files.connector_credential_deleted", audit_row["action"])
self.assertEqual("file_connector_credential", audit_row["object_type"])
self.assertEqual(_CREDENTIAL_ID, audit_row["object_id"])
self._assert_non_secret_audit_details(dict(audit_row["details"] or {}))
self.assertEqual(1, self._row_count(connection, "core_change_sequence"))
def test_database_audit_failure_rolls_back_scrub_and_audit_state(self) -> None:
assert self.engine is not None
with self.engine.begin() as connection:
connection.exec_driver_sql(
"""
CREATE FUNCTION reject_retirement_audit() RETURNS trigger
LANGUAGE plpgsql AS $$
BEGIN
IF NEW.action = 'files.connector_credential_deleted' THEN
RAISE EXCEPTION 'injected retirement audit failure';
END IF;
RETURN NEW;
END;
$$
"""
)
connection.exec_driver_sql(
"""
CREATE TRIGGER reject_retirement_audit_insert
BEFORE INSERT ON audit_log
FOR EACH ROW EXECUTE FUNCTION reject_retirement_audit()
"""
)
with Session(self.engine) as session:
self._session_backend_pid(session)
plan = self._retirement_plan(session)
with self.assertRaises(DBAPIError) as caught:
self._execute_retirement(session, plan)
self.assertEqual("P0001", getattr(caught.exception.orig, "sqlstate", None))
session.rollback()
self.assertEqual([], self.retirement_ddl_backend_pids, "DDL ran after the injected audit failure")
self._assert_failed_retirement_rolled_back()
def test_database_ddl_failure_rolls_back_scrub_audit_and_prior_drop(self) -> None:
assert self.engine is not None
with self.engine.begin() as connection:
connection.exec_driver_sql(
"""
CREATE TABLE retirement_drop_guard (
id bigint PRIMARY KEY,
credential_id varchar(255) NOT NULL
REFERENCES file_connector_credentials(id)
)
"""
)
connection.execute(
text("INSERT INTO retirement_drop_guard (id, credential_id) VALUES (1, :credential_id)"),
{"credential_id": _CREDENTIAL_ID},
)
with Session(self.engine) as session:
installer_backend_pid = self._session_backend_pid(session)
plan = self._retirement_plan(session)
with self.assertRaises(DBAPIError) as caught:
self._execute_retirement(session, plan)
self.assertEqual("2BP01", getattr(caught.exception.orig, "sqlstate", None))
session.rollback()
self.assertTrue(self.retirement_ddl_backend_pids, "the injected DDL failure was not reached")
self.assertEqual(
{installer_backend_pid},
set(self.retirement_ddl_backend_pids),
"destructive DDL did not use the installer Session connection",
)
self._assert_failed_retirement_rolled_back()
def _create_fixture_schema(self) -> None:
assert self.engine is not None
from govoplan_core.core.change_sequence import ChangeSequenceEntry
with self.engine.begin() as connection:
# Audit and Files reference Access actors. The proof does not need
# actor rows, so small FK targets keep its schema bounded.
connection.exec_driver_sql("CREATE TABLE access_users (id varchar(36) PRIMARY KEY)")
connection.exec_driver_sql("CREATE TABLE access_api_keys (id varchar(36) PRIMARY KEY)")
Base.metadata.create_all(
bind=connection,
tables=[
ChangeSequenceEntry.__table__,
self.audit_log_model.__table__,
self.credential_model.__table__,
self.profile_model.__table__,
],
)
connection.exec_driver_sql(
"""
CREATE TABLE retirement_scrub_probe (
id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
password_removed boolean NOT NULL,
token_removed boolean NOT NULL,
username_removed boolean NOT NULL,
secret_reference_removed boolean NOT NULL,
metadata_removed boolean NOT NULL
)
"""
)
connection.exec_driver_sql(
"""
CREATE FUNCTION record_retirement_scrub() RETURNS trigger
LANGUAGE plpgsql AS $$
BEGIN
INSERT INTO retirement_scrub_probe (
password_removed,
token_removed,
username_removed,
secret_reference_removed,
metadata_removed
) VALUES (
OLD.password_encrypted IS NOT NULL AND NEW.password_encrypted IS NULL,
OLD.token_encrypted IS NOT NULL AND NEW.token_encrypted IS NULL,
OLD.username IS NOT NULL AND NEW.username IS NULL,
OLD.secret_ref IS NOT NULL AND NEW.secret_ref IS NULL,
OLD.metadata IS NOT NULL AND NEW.metadata::jsonb = '{}'::jsonb
);
RETURN NEW;
END;
$$
"""
)
connection.exec_driver_sql(
"""
CREATE TRIGGER record_retirement_credential_scrub
BEFORE UPDATE ON file_connector_credentials
FOR EACH ROW EXECUTE FUNCTION record_retirement_scrub()
"""
)
def _seed_credential(self) -> None:
assert self.engine is not None
with Session(self.engine) as session:
session.add(self.credential_model(
id=_CREDENTIAL_ID,
tenant_id=_TENANT_ID,
scope_type="tenant",
scope_id=_TENANT_ID,
label="Retirement integration credential",
provider="webdav",
enabled=True,
credential_mode="basic",
username="integration-user",
password_encrypted=_PRIVATE_FIXTURE_VALUES[0],
token_encrypted=_PRIVATE_FIXTURE_VALUES[1],
password_env="INTEGRATION_PASSWORD",
token_env="INTEGRATION_TOKEN",
secret_ref=_PRIVATE_FIXTURE_VALUES[2],
policy={},
metadata_={"private_hint": _PRIVATE_FIXTURE_VALUES[3]},
))
session.commit()
def _retirement_plan(self, session: Session) -> MigrationRetirementPlan:
migration = self.files_manifest.migration_spec
self.assertIsNotNone(migration)
assert migration is not None
self.assertIsNotNone(migration.retirement_provider)
assert migration.retirement_provider is not None
plan = migration.retirement_provider(session, "files")
self.assertTrue(plan.destroy_data_supported)
self.assertIsNotNone(plan.destroy_data_executor)
return plan
def _execute_retirement(self, session: Session, plan: MigrationRetirementPlan) -> None:
executor = plan.destroy_data_executor
self.assertIsNotNone(executor)
self.capture_retirement_ddl = True
try:
executor(session, "files")
finally:
self.capture_retirement_ddl = False
def _capture_ddl_connection(
self,
connection: Connection,
_cursor: object,
statement: str,
_parameters: object,
_context: object,
_executemany: bool,
) -> None:
if not self.capture_retirement_ddl or not statement.lstrip().upper().startswith("DROP TABLE"):
return
self.retirement_ddl_backend_pids.append(self._connection_backend_pid(connection))
def _session_backend_pid(self, session: Session) -> int:
connection = session.connection()
sql_pid = int(connection.execute(text("SELECT pg_backend_pid()")).scalar_one())
driver_pid = self._connection_backend_pid(connection)
self.assertEqual(sql_pid, driver_pid)
return sql_pid
@staticmethod
def _connection_backend_pid(connection: Connection) -> int:
driver_connection = connection.connection.driver_connection
return int(driver_connection.info.backend_pid)
def _assert_failed_retirement_rolled_back(self) -> None:
assert self.engine is not None
with self.engine.connect() as connection:
inspector = inspect(connection)
self.assertTrue(inspector.has_table(self.credential_model.__tablename__))
self.assertTrue(inspector.has_table(self.profile_model.__tablename__))
state = connection.execute(
text(
"SELECT enabled, password_encrypted IS NOT NULL AS has_password, "
"token_encrypted IS NOT NULL AS has_token, username IS NOT NULL AS has_username, "
"secret_ref IS NOT NULL AS has_secret_reference, metadata::jsonb <> '{}'::jsonb AS has_metadata "
"FROM file_connector_credentials WHERE id = :credential_id"
),
{"credential_id": _CREDENTIAL_ID},
).mappings().one()
self.assertTrue(all(state.values()), "credential state was not fully restored")
self.assertEqual(0, self._row_count(connection, "retirement_scrub_probe"))
self.assertEqual(0, self._row_count(connection, "audit_log"))
self.assertEqual(0, self._row_count(connection, "core_change_sequence"))
def _assert_non_secret_audit_details(self, details: dict[str, object]) -> None:
self.assertEqual("module_data_retired", details.get("deletion_reason"))
self.assertEqual("unowned_external_reference_detached", details.get("storage_backend"))
self.assertEqual("<redacted>", details.get("deleted_secret_kinds"))
self.assertEqual(
["password_env", "token_env", "unowned_external_secret_ref"],
details.get("removed_reference_kinds"),
)
self.assertTrue(details.get("removed_metadata"))
serialized = json.dumps(details, sort_keys=True)
self.assertFalse(
any(private_value in serialized for private_value in _PRIVATE_FIXTURE_VALUES),
"audit diagnostics contain fixture-private material",
)
@staticmethod
def _row_count(connection: Connection, table_name: str) -> int:
query = _ROW_COUNT_QUERIES.get(table_name)
if query is None:
raise ValueError("unsupported retirement proof table")
return int(connection.execute(query).scalar_one())
def _cleanup(self) -> None:
if getattr(self, "previous_runtime", None) is None:
clear_runtime()
else:
configure_runtime(self.previous_runtime)
if self.engine is not None:
if self.ddl_listener_registered:
event.remove(self.engine, "before_cursor_execute", self._capture_ddl_connection)
self.engine.dispose()
self.engine = None
if self.admin_engine is not None:
try:
with self.admin_engine.begin() as connection:
connection.execute(DropSchema(self.schema_name, cascade=True, if_exists=True))
finally:
self.admin_engine.dispose()
self.admin_engine = None
if __name__ == "__main__":
unittest.main()