test(12-05): add migration 0005→0006 integration regression and RUNBOOK migration ops
- test_migration_0006.py: proves display_name_override, cloud_items, cloud_item_topics, cloud_folder_states, unique constraints, indexes, and DDL privilege boundary - Tests skip cleanly without INTEGRATION=1/INTEGRATION_DATABASE_URL (no dev DB mutation) - RUNBOOK: add Database Migrations section with revision check, upgrade, recovery, and emergency schema-drift remediation commands
This commit is contained in:
curo1305
@@ -0,0 +1,288 @@
|
||||
"""
|
||||
Integration test: PostgreSQL upgrade from revision 0005 to Phase 12 head (0006).
|
||||
|
||||
Prerequisite: a disposable PostgreSQL database accessible via
|
||||
INTEGRATION_DATABASE_URL (psycopg sync DSN, e.g. postgresql://user:pw@host/db)
|
||||
or
|
||||
INTEGRATION=1 with DATABASE_MIGRATE_URL pointing at the test database.
|
||||
|
||||
If neither is set the tests are skipped with an explicit message.
|
||||
|
||||
The test uses an isolated Alembic migration run against this disposable DB — it
|
||||
NEVER touches the normal development database and NEVER runs downgrade on it.
|
||||
|
||||
Requirements covered:
|
||||
CONN-04 — Connections endpoint requires display_name_override column
|
||||
CLOUD-01 — cloud_items/cloud_folder_states tables must exist
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
REQUIRED_ENV = (
|
||||
"INTEGRATION_DATABASE_URL",
|
||||
"DATABASE_MIGRATE_URL",
|
||||
)
|
||||
|
||||
PHASE_12_TABLES = {
|
||||
"cloud_items",
|
||||
"cloud_item_topics",
|
||||
"cloud_folder_states",
|
||||
}
|
||||
|
||||
PHASE_12_UNIQUE_CONSTRAINTS = {
|
||||
"uq_cloud_items_connection_provider_item",
|
||||
"uq_cloud_folder_states_connection_parent",
|
||||
}
|
||||
|
||||
PHASE_12_INDEXES = {
|
||||
"ix_cloud_items_user_id",
|
||||
"ix_cloud_items_connection_id",
|
||||
"ix_cloud_items_connection_parent",
|
||||
"ix_cloud_folder_states_connection",
|
||||
}
|
||||
|
||||
|
||||
def _get_migrate_url() -> str | None:
|
||||
"""Return a disposable database URL for migration tests, or None to skip."""
|
||||
# Highest priority: dedicated integration URL
|
||||
url = os.environ.get("INTEGRATION_DATABASE_URL")
|
||||
if url:
|
||||
return url
|
||||
# Second: INTEGRATION=1 activates the normal DATABASE_MIGRATE_URL
|
||||
if os.environ.get("INTEGRATION") == "1":
|
||||
return os.environ.get("DATABASE_MIGRATE_URL")
|
||||
return None
|
||||
|
||||
|
||||
def _psycopg_conn(dsn: str):
|
||||
"""Return a psycopg3 (sync) connection or raise ImportError / OperationalError."""
|
||||
try:
|
||||
import psycopg # psycopg v3 sync
|
||||
except ImportError as exc:
|
||||
pytest.skip(f"psycopg (v3) not installed — cannot run migration integration test: {exc}")
|
||||
return psycopg.connect(dsn, autocommit=True)
|
||||
|
||||
|
||||
def _run_alembic(target: str, *, dsn: str) -> None:
|
||||
"""Run alembic upgrade/downgrade to target using the given DSN."""
|
||||
env = {**os.environ, "DATABASE_MIGRATE_URL": dsn}
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-m", "alembic", "upgrade", target],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
raise RuntimeError(
|
||||
f"alembic upgrade {target} failed (rc={result.returncode}):\n"
|
||||
f"STDOUT: {result.stdout}\nSTDERR: {result.stderr}"
|
||||
)
|
||||
|
||||
|
||||
def _run_alembic_downgrade(target: str, *, dsn: str) -> None:
|
||||
"""Run alembic downgrade to target using the given DSN."""
|
||||
env = {**os.environ, "DATABASE_MIGRATE_URL": dsn}
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-m", "alembic", "downgrade", target],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
raise RuntimeError(
|
||||
f"alembic downgrade {target} failed (rc={result.returncode}):\n"
|
||||
f"STDOUT: {result.stdout}\nSTDERR: {result.stderr}"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def migrate_dsn() -> str:
|
||||
"""Provide the integration database DSN, or skip the whole module."""
|
||||
dsn = _get_migrate_url()
|
||||
if not dsn:
|
||||
pytest.skip(
|
||||
"Migration integration tests skipped — set INTEGRATION_DATABASE_URL or "
|
||||
"INTEGRATION=1 (with DATABASE_MIGRATE_URL) to activate. "
|
||||
"These tests require a disposable PostgreSQL database and must never "
|
||||
"run against the normal development database."
|
||||
)
|
||||
return dsn
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def pg_conn(migrate_dsn):
|
||||
"""Open a psycopg3 connection to the integration database."""
|
||||
conn = _psycopg_conn(migrate_dsn)
|
||||
yield conn
|
||||
conn.close()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test: upgrade 0005 → 0006/head
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestMigration0005To0006:
|
||||
"""Upgrade integration tests that prove the 0005->0006 transition."""
|
||||
|
||||
def test_upgrade_to_0005(self, migrate_dsn):
|
||||
"""Apply all migrations through 0005 (the pre-phase-12 baseline)."""
|
||||
_run_alembic("0005", dsn=migrate_dsn)
|
||||
|
||||
def test_0005_lacks_display_name_override(self, pg_conn):
|
||||
"""After 0005 only, display_name_override must NOT exist yet."""
|
||||
cur = pg_conn.execute(
|
||||
"""
|
||||
SELECT column_name
|
||||
FROM information_schema.columns
|
||||
WHERE table_name = 'cloud_connections'
|
||||
AND column_name = 'display_name_override'
|
||||
"""
|
||||
)
|
||||
row = cur.fetchone()
|
||||
assert row is None, (
|
||||
"display_name_override should not exist after revision 0005; "
|
||||
"migration 0006 has not run yet"
|
||||
)
|
||||
|
||||
def test_0005_lacks_phase12_tables(self, pg_conn):
|
||||
"""After 0005 only, cloud_items/cloud_item_topics/cloud_folder_states must be absent."""
|
||||
for table in PHASE_12_TABLES:
|
||||
cur = pg_conn.execute(
|
||||
"SELECT table_name FROM information_schema.tables "
|
||||
"WHERE table_schema='public' AND table_name=%s",
|
||||
(table,),
|
||||
)
|
||||
assert cur.fetchone() is None, (
|
||||
f"Table '{table}' must not exist after revision 0005"
|
||||
)
|
||||
|
||||
def test_upgrade_to_head(self, migrate_dsn):
|
||||
"""Upgrade from 0005 to head (0006)."""
|
||||
_run_alembic("head", dsn=migrate_dsn)
|
||||
|
||||
def test_alembic_version_is_0006(self, pg_conn):
|
||||
"""alembic_version must report '0006' after upgrade to head."""
|
||||
cur = pg_conn.execute("SELECT version_num FROM alembic_version")
|
||||
row = cur.fetchone()
|
||||
assert row is not None, "alembic_version table has no rows after migration"
|
||||
assert row[0] == "0006", (
|
||||
f"Expected alembic_version='0006', got {row[0]!r}"
|
||||
)
|
||||
|
||||
def test_display_name_override_column_exists(self, pg_conn):
|
||||
"""cloud_connections.display_name_override must exist after 0006."""
|
||||
cur = pg_conn.execute(
|
||||
"""
|
||||
SELECT column_name, is_nullable, data_type
|
||||
FROM information_schema.columns
|
||||
WHERE table_name = 'cloud_connections'
|
||||
AND column_name = 'display_name_override'
|
||||
"""
|
||||
)
|
||||
row = cur.fetchone()
|
||||
assert row is not None, (
|
||||
"cloud_connections.display_name_override column is missing after migration 0006"
|
||||
)
|
||||
col_name, is_nullable, data_type = row
|
||||
assert is_nullable == "YES", (
|
||||
"display_name_override must be nullable (optional user-defined override)"
|
||||
)
|
||||
|
||||
def test_cloud_items_table_exists_with_key_columns(self, pg_conn):
|
||||
"""cloud_items table must exist with essential columns."""
|
||||
expected_columns = {
|
||||
"id", "user_id", "connection_id", "provider_item_id", "name", "kind",
|
||||
"analysis_status", "semantic_index_status", "created_at", "updated_at",
|
||||
}
|
||||
cur = pg_conn.execute(
|
||||
"SELECT column_name FROM information_schema.columns "
|
||||
"WHERE table_name='cloud_items'"
|
||||
)
|
||||
actual = {row[0] for row in cur.fetchall()}
|
||||
missing = expected_columns - actual
|
||||
assert not missing, f"cloud_items is missing columns: {sorted(missing)}"
|
||||
|
||||
def test_cloud_item_topics_table_exists(self, pg_conn):
|
||||
"""cloud_item_topics association table must exist."""
|
||||
cur = pg_conn.execute(
|
||||
"SELECT table_name FROM information_schema.tables "
|
||||
"WHERE table_schema='public' AND table_name='cloud_item_topics'"
|
||||
)
|
||||
assert cur.fetchone() is not None, "cloud_item_topics table missing after migration 0006"
|
||||
|
||||
def test_cloud_folder_states_table_exists_with_key_columns(self, pg_conn):
|
||||
"""cloud_folder_states table must exist with essential columns."""
|
||||
expected_columns = {
|
||||
"id", "user_id", "connection_id", "parent_ref", "refresh_state",
|
||||
"last_refreshed_at", "error_code", "created_at", "updated_at",
|
||||
}
|
||||
cur = pg_conn.execute(
|
||||
"SELECT column_name FROM information_schema.columns "
|
||||
"WHERE table_name='cloud_folder_states'"
|
||||
)
|
||||
actual = {row[0] for row in cur.fetchall()}
|
||||
missing = expected_columns - actual
|
||||
assert not missing, f"cloud_folder_states is missing columns: {sorted(missing)}"
|
||||
|
||||
def test_unique_constraints_exist(self, pg_conn):
|
||||
"""Named unique constraints from 0006 must exist in the catalog."""
|
||||
for constraint in PHASE_12_UNIQUE_CONSTRAINTS:
|
||||
cur = pg_conn.execute(
|
||||
"""
|
||||
SELECT constraint_name
|
||||
FROM information_schema.table_constraints
|
||||
WHERE constraint_type = 'UNIQUE'
|
||||
AND constraint_name = %s
|
||||
""",
|
||||
(constraint,),
|
||||
)
|
||||
row = cur.fetchone()
|
||||
assert row is not None, (
|
||||
f"Unique constraint '{constraint}' is missing after migration 0006"
|
||||
)
|
||||
|
||||
def test_indexes_exist(self, pg_conn):
|
||||
"""Named indexes from 0006 must exist in pg_indexes."""
|
||||
for idx in PHASE_12_INDEXES:
|
||||
cur = pg_conn.execute(
|
||||
"SELECT indexname FROM pg_indexes WHERE indexname=%s",
|
||||
(idx,),
|
||||
)
|
||||
row = cur.fetchone()
|
||||
assert row is not None, (
|
||||
f"Index '{idx}' is missing after migration 0006"
|
||||
)
|
||||
|
||||
def test_docuvault_app_cannot_create_table(self, migrate_dsn, pg_conn):
|
||||
"""docuvault_app (DML-only role) must not be able to create tables (DDL forbidden)."""
|
||||
# Derive app DSN from migrate DSN by substituting the user
|
||||
# The test DB is shared; we test privilege by executing as the app role.
|
||||
# We check the privilege via information_schema — if the role has CREATE
|
||||
# privilege on the public schema, that is a misconfiguration.
|
||||
#
|
||||
# NOTE: In the Docker initdb.d setup, docuvault_app is granted USAGE but
|
||||
# not CREATE on the public schema. We assert this via has_schema_privilege.
|
||||
cur = pg_conn.execute(
|
||||
"SELECT has_schema_privilege('docuvault_app', 'public', 'CREATE')"
|
||||
)
|
||||
row = cur.fetchone()
|
||||
# If the user doesn't exist (integration DB doesn't have it), skip gracefully
|
||||
if row is None:
|
||||
pytest.skip("docuvault_app role does not exist in the integration database")
|
||||
assert row[0] is False, (
|
||||
"docuvault_app must NOT have CREATE privilege on the public schema — "
|
||||
"only docuvault_migrate should be able to run DDL"
|
||||
)
|
||||
Reference in New Issue
Block a user