test: preserve exact historical migration replay boundaries
This commit is contained in:
262
backend/tests/alembic_migration_schema_assertions.py
Normal file
262
backend/tests/alembic_migration_schema_assertions.py
Normal file
@@ -0,0 +1,262 @@
|
||||
# #region Test.Alembic.SchemaAssertions [C:4] [TYPE Module] [SEMANTICS test,alembic,migration,schema,source-binding,postgres]
|
||||
# @BRIEF DDL and behavior assertions for migration-owned tables, independent of production ORM metadata.
|
||||
# @RELATION BINDS_TO -> [Migrations.0051.DeploymentAppliedAt]
|
||||
# @RELATION BINDS_TO -> [Migrations.0052.ReferenceDraftInspection]
|
||||
# @PRE Callers pass an engine bound to a schema already upgraded to the revision under assertion.
|
||||
# @POST Columns, constraints, indexes and insert refusals are checked against hardcoded expected values.
|
||||
# @INVARIANT Expectations are literal fixtures and never imported from production ORM metadata.
|
||||
|
||||
import re
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
|
||||
# #region Test.AlembicMigrations.AppliedAtSchema [C:2] [TYPE Function]
|
||||
# @POST deployment_records.applied_at/target_digest become nullable after 0051 and keep their declared types.
|
||||
def _assert_0051_schema(engine) -> None:
|
||||
from sqlalchemy import inspect
|
||||
import sqlalchemy as sa
|
||||
|
||||
columns = {column["name"]: column for column in inspect(engine).get_columns("deployment_records")}
|
||||
_assert_deployment_applied_columns(columns)
|
||||
assert isinstance(columns["applied_at"]["type"], sa.DateTime)
|
||||
# #endregion Test.AlembicMigrations.AppliedAtSchema
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.DeploymentAppliedColumns [C:2] [TYPE Function]
|
||||
# @POST The 0051 boundary columns are nullable and target_digest keeps the 64-char width.
|
||||
def _assert_deployment_applied_columns(columns: dict) -> None:
|
||||
assert columns["applied_at"]["nullable"] is True
|
||||
assert columns["target_digest"]["nullable"] is True
|
||||
assert columns["target_digest"]["type"].length == 64
|
||||
# #endregion Test.AlembicMigrations.DeploymentAppliedColumns
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.InspectionColumnAbsent [C:1] [TYPE Function]
|
||||
# @POST The 0052-owned inspection column is absent before 0052 is applied.
|
||||
def _assert_no_inspection_snapshot_column(schema_engine) -> None:
|
||||
from sqlalchemy import inspect
|
||||
|
||||
assert "inspection_snapshot" not in {column["name"] for column in inspect(schema_engine).get_columns("dashboard_reference_draft_authorities")}
|
||||
# #endregion Test.AlembicMigrations.InspectionColumnAbsent
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.InspectionSchema [C:2] [TYPE Function]
|
||||
# @POST The 0052-owned inspection column exists exactly once as nullable JSON without a server default.
|
||||
def _assert_0052_schema(engine) -> None:
|
||||
from sqlalchemy import inspect
|
||||
from sqlalchemy.dialects.postgresql import JSONB
|
||||
import sqlalchemy as sa
|
||||
|
||||
columns = [column for column in inspect(engine).get_columns("dashboard_reference_draft_authorities")
|
||||
if column["name"] == "inspection_snapshot"]
|
||||
assert len(columns) == 1
|
||||
assert columns[0]["nullable"] is True
|
||||
assert isinstance(columns[0]["type"], sa.JSON)
|
||||
assert not isinstance(columns[0]["type"], JSONB)
|
||||
assert columns[0].get("default") is None
|
||||
# #endregion Test.AlembicMigrations.InspectionSchema
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.SourceBindingSchemaAssert [C:4] [TYPE Function]
|
||||
# @BRIEF Assert the source-binding DDL independently of production ORM metadata.
|
||||
# @POST Verifies exact columns, FK columns/ondelete, check expressions, index columns/partial predicates.
|
||||
def _assert_source_binding_schema(schema_engine, schema: str) -> None:
|
||||
from sqlalchemy import inspect
|
||||
|
||||
table = "git_repository_source_bindings"
|
||||
inspector = inspect(schema_engine)
|
||||
assert table in set(inspector.get_table_names())
|
||||
with schema_engine.connect() as connection:
|
||||
_assert_source_binding_columns(connection, schema, table)
|
||||
_assert_source_binding_constraints(connection, schema, table)
|
||||
_assert_source_binding_indexes(connection, schema, table)
|
||||
# #endregion Test.AlembicMigrations.SourceBindingSchemaAssert
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.SourceBindingColumnsAssert [C:2] [TYPE Function]
|
||||
# @POST The source-binding table has exactly the expected 16 columns and nullability.
|
||||
def _assert_source_binding_columns(connection, schema: str, table: str) -> None:
|
||||
from sqlalchemy import text as sa_text
|
||||
|
||||
column_rows = connection.execute(sa_text(
|
||||
"SELECT column_name, is_nullable FROM information_schema.columns "
|
||||
"WHERE table_schema = :schema AND table_name = :table"
|
||||
), {"schema": schema, "table": table}).all()
|
||||
expected_columns = {
|
||||
"id": "NO", "repository_id": "NO", "dashboard_uuid": "NO",
|
||||
"source_environment_id": "NO", "source_dashboard_id": "NO",
|
||||
"release_id": "NO", "release_commit_hash": "NO", "status": "NO",
|
||||
"actor_id": "NO", "actor_username": "NO", "created_at": "NO",
|
||||
"retired_at": "YES", "retired_by": "YES", "cas_version": "NO",
|
||||
"request_sha256": "NO", "receipt_sha256": "NO",
|
||||
}
|
||||
assert {row.column_name: row.is_nullable for row in column_rows} == expected_columns
|
||||
# #endregion Test.AlembicMigrations.SourceBindingColumnsAssert
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.SourceBindingConstraintsAssert [C:3] [TYPE Function]
|
||||
# @POST The two FK/RESTRICT constraints and six checks exist with their exact definitions.
|
||||
def _assert_source_binding_constraints(connection, schema: str, table: str) -> None:
|
||||
from sqlalchemy import text as sa_text
|
||||
|
||||
constraint_rows = connection.execute(sa_text(
|
||||
"SELECT conname, contype, pg_get_constraintdef(oid) AS definition "
|
||||
"FROM pg_constraint WHERE conrelid = CAST(:qualified AS regclass)"
|
||||
), {"qualified": f"{schema}.{table}"}).all()
|
||||
definitions = {row.contype: [] for row in constraint_rows}
|
||||
for row in constraint_rows:
|
||||
definitions[row.contype].append(" ".join(row.definition.replace('"', "").lower().split()))
|
||||
assert set(definitions.get("f", [])) == {
|
||||
"foreign key (repository_id) references git_repositories(id) on delete restrict",
|
||||
"foreign key (release_id) references dashboard_releases(id) on delete restrict",
|
||||
}
|
||||
assert len(definitions.get("c", [])) == 6
|
||||
# #endregion Test.AlembicMigrations.SourceBindingConstraintsAssert
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.IndexPredicateNormalize [C:2] [TYPE Function]
|
||||
# @POST Normalizes a PG index predicate to the hardcoded expected form; None means no predicate.
|
||||
def _normalize_index_predicate(row):
|
||||
if row.predicate is None:
|
||||
return None
|
||||
normalized_predicate = "".join(
|
||||
row.predicate.replace('"', "").lower().replace("::text", "").split())
|
||||
if normalized_predicate.startswith("(") and normalized_predicate.endswith(")"):
|
||||
normalized_predicate = normalized_predicate[1:-1]
|
||||
return re.sub(r"\(([a-z_][a-z0-9_]*)\)", r"\1", normalized_predicate)
|
||||
# #endregion Test.AlembicMigrations.IndexPredicateNormalize
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.SourceBindingIndexesAssert [C:3] [TYPE Function]
|
||||
# @POST The primary/plain/partial-unique indexes exist with exact uniqueness, predicate and columns.
|
||||
def _assert_source_binding_indexes(connection, schema: str, table: str) -> None:
|
||||
from sqlalchemy import text as sa_text
|
||||
|
||||
index_rows = connection.execute(sa_text(
|
||||
"SELECT i.relname AS indexname, x.indisunique AS is_unique, "
|
||||
"pg_get_indexdef(x.indexrelid) AS indexdef, "
|
||||
"pg_get_expr(x.indpred, x.indrelid) AS predicate "
|
||||
"FROM pg_index x JOIN pg_class i ON i.oid = x.indexrelid "
|
||||
"WHERE x.indrelid = CAST(:qualified AS regclass)"
|
||||
), {"qualified": f"{schema}.{table}"}).all()
|
||||
expected_indexes = {
|
||||
"git_repository_source_bindings_pkey": (True, None, "(id)"),
|
||||
"ix_git_source_binding_repository": (False, None, "(repository_id)"),
|
||||
"uq_git_source_binding_active_repository": (True, "status='active'", "(repository_id)"),
|
||||
"uq_git_source_binding_active_uuid": (True, "status='active'", "(dashboard_uuid)"),
|
||||
"uq_git_source_binding_active_source": (True, "status='active'",
|
||||
"(source_environment_id, source_dashboard_id)"),
|
||||
}
|
||||
actual_indexes = {row.indexname: row for row in index_rows}
|
||||
assert set(actual_indexes) == set(expected_indexes)
|
||||
for name, (unique, predicate, columns) in expected_indexes.items():
|
||||
row = actual_indexes[name]
|
||||
assert bool(row.is_unique) is unique, name
|
||||
normalized_predicate = _normalize_index_predicate(row)
|
||||
assert normalized_predicate == predicate, name
|
||||
assert columns in row.indexdef.replace('"', "").lower(), name
|
||||
# #endregion Test.AlembicMigrations.SourceBindingIndexesAssert
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.SourceBindingBehavior [C:4] [TYPE Function]
|
||||
# @BRIEF Independent insert-refusal/retired-allowed fixtures for checks and active partial indexes.
|
||||
# @POST All six checks reject invalid rows; retired duplicates are allowed; active duplicates are rejected by DB.
|
||||
# @INVARIANT Expected values are hardcoded here, not imported from production ORM metadata.
|
||||
def _assert_source_binding_behavior(schema_engine) -> None:
|
||||
from datetime import UTC, datetime
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy import text as sa_text
|
||||
|
||||
table = "git_repository_source_bindings"
|
||||
env_id = "00000000-0000-4000-8000-000000000001"
|
||||
repo_a = "00000000-0000-4000-8000-000000000011"
|
||||
repo_b = "00000000-0000-4000-8000-000000000012"
|
||||
repo_c = "00000000-0000-4000-8000-000000000013"
|
||||
release_id = "00000000-0000-4000-8000-000000000021"
|
||||
uuid_a = "a2cef831-4ffd-46ec-aaa1-60f71782a91c"
|
||||
uuid_b = "b2cef831-4ffd-46ec-aaa1-60f71782a91c"
|
||||
uuid_c = "c2cef831-4ffd-46ec-aaa1-60f71782a91c"
|
||||
commit = "a" * 40
|
||||
request_hash = "0" * 64
|
||||
receipt_hash = "1" * 64
|
||||
now = datetime.now(UTC)
|
||||
|
||||
with schema_engine.begin() as connection:
|
||||
connection.execute(sa_text(
|
||||
"INSERT INTO deployment_environments (id, name, superset_url, superset_token, is_active) "
|
||||
"VALUES (:id, :name, :url, :token, true)"
|
||||
), {"id": env_id, "name": "fixture", "url": "http://fixture", "token": ""})
|
||||
for dashboard_id, repo_id in ((810001, repo_a), (810002, repo_b), (810003, repo_c)):
|
||||
connection.execute(sa_text(
|
||||
"INSERT INTO git_repositories "
|
||||
"(id, dashboard_id, local_path, current_branch, sync_status) "
|
||||
"VALUES (:id, :dashboard_id, :path, 'dev', 'CLEAN')"
|
||||
), {"id": repo_id, "dashboard_id": dashboard_id, "path": f"/tmp/{repo_id}"})
|
||||
deployment_id = connection.execute(sa_text(
|
||||
"INSERT INTO deployment_records "
|
||||
"(repository_id, environment_id, commit_hash, content_hash, deployed_at, status, validation_status) "
|
||||
"VALUES (:repo, :env, :commit, :content, :deployed, 'success', 'validated') RETURNING id"
|
||||
), {"repo": repo_a, "env": env_id, "commit": commit, "content": "c" * 64,
|
||||
"deployed": now}).scalar_one()
|
||||
connection.execute(sa_text(
|
||||
"INSERT INTO dashboard_releases "
|
||||
"(id, repository_id, deployment_id, name, version, notes, commit_hash, content_hash, status, created_at, created_by) "
|
||||
"VALUES (:id, :repo, :deployment, 'fixture', '1.0.0', 'fixture', :commit, :content, 'ready_to_publish', :created, 'fixture')"
|
||||
), {"id": release_id, "repo": repo_a, "deployment": deployment_id, "commit": commit,
|
||||
"content": "c" * 64, "created": now})
|
||||
|
||||
# #region Test.AlembicMigrations.SourceBindingInsert [C:1] [TYPE Function]
|
||||
# @BRIEF One valid source-binding row fixture; invalid variants are supplied by the caller.
|
||||
def insert_binding(*, repository_id: str, dashboard_uuid: str, source_environment_id: str,
|
||||
source_dashboard_id: int = 12, status: str = "active", commit_hash: str = commit,
|
||||
cas_version: int = 1) -> None:
|
||||
with schema_engine.begin() as connection:
|
||||
connection.execute(sa_text(
|
||||
f"INSERT INTO {table} "
|
||||
"(id, repository_id, dashboard_uuid, source_environment_id, source_dashboard_id, "
|
||||
"release_id, release_commit_hash, status, actor_id, actor_username, created_at, "
|
||||
"cas_version, request_sha256, receipt_sha256) "
|
||||
"VALUES (:id, :repo, :uuid, :source_env, :source_dashboard, :release, :commit, :status, "
|
||||
"'actor', 'operator', :created, :cas, :request, :receipt)"
|
||||
), {"id": str(uuid.uuid4()), "repo": repository_id, "uuid": dashboard_uuid,
|
||||
"source_env": source_environment_id, "source_dashboard": source_dashboard_id,
|
||||
"release": release_id, "commit": commit_hash, "status": status,
|
||||
"created": now, "cas": cas_version, "request": request_hash, "receipt": receipt_hash})
|
||||
# #endregion Test.AlembicMigrations.SourceBindingInsert
|
||||
|
||||
insert_binding(repository_id=repo_a, dashboard_uuid=uuid_a, source_environment_id="source-a")
|
||||
insert_binding(repository_id=repo_a, dashboard_uuid=uuid_a, source_environment_id="source-a",
|
||||
status="retired")
|
||||
with schema_engine.connect() as connection:
|
||||
assert connection.execute(sa_text(f"SELECT count(*) FROM {table}")).scalar_one() == 2
|
||||
|
||||
for invalid in (
|
||||
{"repository_id": repo_a, "uuid": uuid_b, "source_env": "source-b"},
|
||||
{"repository_id": repo_b, "uuid": uuid_a, "source_env": "source-c"},
|
||||
{"repository_id": repo_c, "uuid": uuid_b, "source_env": "source-a"},
|
||||
):
|
||||
with pytest.raises(IntegrityError):
|
||||
insert_binding(repository_id=invalid["repository_id"], dashboard_uuid=invalid["uuid"],
|
||||
source_environment_id=invalid["source_env"])
|
||||
|
||||
invalid_rows = (
|
||||
{"status": "paused"},
|
||||
{"dashboard_uuid": "short"},
|
||||
{"commit_hash": "short"},
|
||||
{"source_environment_id": ""},
|
||||
{"source_dashboard_id": 0},
|
||||
{"cas_version": 0},
|
||||
)
|
||||
for invalid in invalid_rows:
|
||||
values = {
|
||||
"repository_id": repo_a, "dashboard_uuid": uuid_c,
|
||||
"source_environment_id": "source-check", "source_dashboard_id": 12,
|
||||
"status": "retired", "commit_hash": commit, "cas_version": 1,
|
||||
}
|
||||
values.update(invalid)
|
||||
with pytest.raises(IntegrityError):
|
||||
insert_binding(**values)
|
||||
# #endregion Test.AlembicMigrations.SourceBindingBehavior
|
||||
|
||||
# #endregion Test.Alembic.SchemaAssertions
|
||||
66
backend/tests/alembic_migration_support.py
Normal file
66
backend/tests/alembic_migration_support.py
Normal file
@@ -0,0 +1,66 @@
|
||||
# #region Test.Alembic.MigrationSupport [C:3] [TYPE Module] [SEMANTICS test,alembic,migration,postgres,isolation]
|
||||
# @BRIEF Shared PostgreSQL harness for Alembic migration tests: backend gate and per-test schema isolation.
|
||||
# @RELATION BINDS_TO -> [Alembic.Env.AlembicEnvModule]
|
||||
# @PRE DATABASE_URL is exported by the run-owned test database.
|
||||
# @POST PostgreSQL-only test modules import the same backend gate; every isolation caller owns one private schema.
|
||||
# @INVARIANT The isolation helper drops only its own schema and never downgrades or drops shared worker tables.
|
||||
# @REJECTED Migrating the already-migrated shared worker schema was rejected: both upgrades would be no-ops.
|
||||
|
||||
from contextlib import contextmanager
|
||||
import os
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.engine import make_url
|
||||
|
||||
# Alembic migrations use PostgreSQL-specific DDL (ALTER COLUMN, partial indexes, JSON).
|
||||
_DB_BACKEND = make_url(os.environ.get("DATABASE_URL", "sqlite://")).get_backend_name()
|
||||
_REQUIRES_PG = pytest.mark.skipif(
|
||||
_DB_BACKEND != "postgresql",
|
||||
reason="Alembic migrations are PostgreSQL-specific; use a PostgreSQL DATABASE_URL to test",
|
||||
)
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.SourceBindingIsolation [C:4] [TYPE Function]
|
||||
# @BRIEF Give each migration test its own PostgreSQL schema in the run-owned worker DB.
|
||||
# @POST Alembic sees exactly one search_path (no public fallback); cleanup drops only this test's schema.
|
||||
# @INVARIANT No downgrade/drop of shared worker tables; no credential files or URL logging.
|
||||
@contextmanager
|
||||
def _isolated_pg_schema(monkeypatch: pytest.MonkeyPatch):
|
||||
from sqlalchemy import create_engine, text as sa_text
|
||||
|
||||
schema = f"alembic_sb_{uuid.uuid4().hex}"
|
||||
previous_schema = os.environ.get("DB_SCHEMA")
|
||||
engine = None
|
||||
schema_engine = None
|
||||
created = False
|
||||
try:
|
||||
engine = create_engine(os.environ["DATABASE_URL"])
|
||||
with engine.begin() as connection:
|
||||
connection.execute(sa_text(f'CREATE SCHEMA "{schema}"'))
|
||||
created = True
|
||||
schema_engine = create_engine(
|
||||
os.environ["DATABASE_URL"],
|
||||
connect_args={"options": f"-csearch_path={schema}"},
|
||||
)
|
||||
monkeypatch.setenv("DB_SCHEMA", schema)
|
||||
yield schema, schema_engine
|
||||
finally:
|
||||
try:
|
||||
if schema_engine is not None:
|
||||
schema_engine.dispose()
|
||||
finally:
|
||||
try:
|
||||
if created and engine is not None:
|
||||
with engine.begin() as connection:
|
||||
connection.execute(sa_text(f'DROP SCHEMA "{schema}" CASCADE'))
|
||||
finally:
|
||||
if engine is not None:
|
||||
engine.dispose()
|
||||
if previous_schema is None:
|
||||
monkeypatch.delenv("DB_SCHEMA", raising=False)
|
||||
else:
|
||||
monkeypatch.setenv("DB_SCHEMA", previous_schema)
|
||||
# #endregion Test.AlembicMigrations.SourceBindingIsolation
|
||||
|
||||
# #endregion Test.Alembic.MigrationSupport
|
||||
@@ -6,7 +6,8 @@
|
||||
# @PRE Catalog/publication tables are byte-identical at 0046 and at head: no migration after 0046 touches them
|
||||
# (0047-0049 change draft authorities/receipts/finishes, 0050 adds git_repository_source_bindings,
|
||||
# 0051 adds nullable applied_at/target_digest only to deployment_records;
|
||||
# 0052 adds nullable inspection_snapshot only to draft authorities).
|
||||
# 0052 adds nullable inspection_snapshot only to draft authorities;
|
||||
# 0053-0055 add finish intent/settings and run-notification tables, not catalog/publication tables).
|
||||
# @POST The same rows, payload bytes and digests exist after `alembic upgrade head`, and the current read
|
||||
# path returns them by exact revision, including a receipt-bound publication.
|
||||
# @INVARIANT The upgrade never rewrites an approved catalog row, an approval receipt or a publication operation.
|
||||
@@ -28,7 +29,7 @@ from sqlalchemy import text
|
||||
from tests.test_alembic_migrations import _REQUIRES_PG, _isolated_pg_schema, _run_alembic_upgrade
|
||||
|
||||
PREVIOUS_RELEASE_HEAD = "0046_mapper_llm_plans"
|
||||
CURRENT_HEAD = "0052_draft_inspection"
|
||||
CURRENT_HEAD = "0055_scenario_run_notifications"
|
||||
EXPECTED_FORWARD_REVISIONS = (
|
||||
"0047_reference_draft_projection",
|
||||
"0048_reference_draft_finish",
|
||||
@@ -36,6 +37,9 @@ EXPECTED_FORWARD_REVISIONS = (
|
||||
"0050_git_source_bind",
|
||||
"0051_deployment_applied_at",
|
||||
"0052_draft_inspection",
|
||||
"0053_reference_finish_intent",
|
||||
"0054_reference_finish_settings",
|
||||
"0055_scenario_run_notifications",
|
||||
)
|
||||
_PRE_BUILD_IDENTITY_AUTHORITY_KEYS = (
|
||||
"superset_build",
|
||||
@@ -47,8 +51,9 @@ _PRE_BUILD_IDENTITY_AUTHORITY_KEYS = (
|
||||
|
||||
|
||||
# #region Test.Alembic.InPlaceUpgrade.ExactForwardChain [C:2] [TYPE Function]
|
||||
# @BRIEF The integrated previous-release path includes every exact forward revision through 0052.
|
||||
# @RATIONALE Integration adds the 0052 draft inspection column while preserving every earlier revision and approved-reference assertion.
|
||||
# @BRIEF The integrated previous-release path includes every exact forward revision through CURRENT_HEAD.
|
||||
# @RATIONALE Integration adds the 0052 inspection column and the 0053-0055 finish/settings/notification revisions
|
||||
# while preserving every earlier revision and approved-reference assertion.
|
||||
# @REJECTED Deriving the expected head from the chain under test would silently accept a removed or foreign revision.
|
||||
def test_previous_release_forward_chain_is_exact() -> None:
|
||||
from alembic.script import ScriptDirectory
|
||||
@@ -65,6 +70,11 @@ def test_previous_release_forward_chain_is_exact() -> None:
|
||||
# @PRE The schema is already at PREVIOUS_RELEASE_HEAD inside an isolated PostgreSQL schema.
|
||||
# @POST One published catalog revision with an entry + authority, one draft approval receipt and one
|
||||
# publication operation exist, shaped like the previous release wrote them.
|
||||
# @PRE The previous release wrote the authority envelope before the build-identity members existed,
|
||||
# so the fixture persists that exact shape instead of the current builder output.
|
||||
# @REJECTED Updating the appended envelope afterwards: the PG append-only trigger refuses it by design.
|
||||
# @RATIONALE Draft authorities and approval receipts changed after 0046 (0047-0049), so the
|
||||
# previous-release rows are written with the 0046 column set, not the current ORM.
|
||||
def _seed_approved_reference(engine) -> dict:
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
@@ -89,9 +99,6 @@ def _seed_approved_reference(engine) -> dict:
|
||||
seed = seed_catalog(db, repository=CATALOG_REPOSITORY,
|
||||
dashboard_key=f"dashboard-{wrapper.entry.dashboard_id}", actor_id="dr-audit",
|
||||
reason="previous release catalog seed", idempotency_key=f"dr-audit:seed:{uuid4()}")
|
||||
# @PRE The previous release wrote the authority envelope before the build-identity members existed,
|
||||
# so the fixture persists that exact shape instead of the current builder output.
|
||||
# @REJECTED Updating the appended envelope afterwards: the PG append-only trigger refuses it by design.
|
||||
built = build_published_entry_authority(wrapper=wrapper,
|
||||
catalog_id=seed.revision.catalog_id, catalog_revision_id=invocation_id,
|
||||
release_id=RELEASE, repository_id=CATALOG_REPOSITORY)
|
||||
@@ -130,8 +137,6 @@ def _seed_approved_reference(engine) -> dict:
|
||||
catalog_revision_id=selector.catalog_revision_id).scalar()),
|
||||
authority=dict(db.query(BaselineCatalogEntryAuthority.authority).filter_by(
|
||||
catalog_revision_id=selector.catalog_revision_id).scalar()))
|
||||
# @RATIONALE Draft authorities and approval receipts changed after 0046 (0047-0049), so the
|
||||
# previous-release rows are written with the 0046 column set, not the current ORM.
|
||||
now = datetime.now(UTC)
|
||||
with engine.begin() as connection:
|
||||
connection.execute(text(
|
||||
@@ -183,8 +188,11 @@ def _whole_rows(connection, seeded: dict) -> dict:
|
||||
}
|
||||
return {name: tuple(connection.execute(text(sql), params).scalars().all())
|
||||
for name, (sql, params) in queries.items()}
|
||||
# #endregion Test.Alembic.InPlaceUpgrade.RowSnapshot
|
||||
|
||||
|
||||
# #region Test.Alembic.InPlaceUpgrade.DraftAuthorityRow [C:1] [TYPE Function]
|
||||
# @POST The legacy draft-authority row is compared column-by-column as exact text.
|
||||
def _draft_authority_row(connection, draft_id: str) -> dict:
|
||||
return dict(connection.execute(text(
|
||||
"SELECT draft_id, owner_id, owner_username, environment_id, dashboard_id, "
|
||||
@@ -193,12 +201,21 @@ def _draft_authority_row(connection, draft_id: str) -> dict:
|
||||
"state, cas_version, invalidation_code, created_at, updated_at "
|
||||
"FROM dashboard_reference_draft_authorities WHERE draft_id = :draft_id"),
|
||||
{"draft_id": draft_id}).mappings().one())
|
||||
# #endregion Test.Alembic.InPlaceUpgrade.RowSnapshot
|
||||
# #endregion Test.Alembic.InPlaceUpgrade.DraftAuthorityRow
|
||||
|
||||
|
||||
# #region Test.Alembic.InPlaceUpgrade.SeededReferencePresent [C:2] [TYPE Function]
|
||||
# @POST The seeded catalog, authority and receipt row groups are all non-empty before the upgrade.
|
||||
def _assert_reference_rows_seeded(before: dict) -> None:
|
||||
assert before["catalog_revisions"] and before["catalog_memberships"]
|
||||
assert before["entry_authorities"] and before["publication_operations"]
|
||||
assert before["approval_receipts"]
|
||||
# #endregion Test.Alembic.InPlaceUpgrade.SeededReferencePresent
|
||||
|
||||
|
||||
# #region Test.Alembic.InPlaceUpgrade.PreviousRelease [C:4] [TYPE Function]
|
||||
# @BRIEF A database at the previous release head upgrades to head and keeps every approved reference row.
|
||||
# @INVARIANT The alembic_version row moves 0046_mapper_llm_plans -> 0052_draft_inspection exactly once.
|
||||
# @INVARIANT The alembic_version row moves 0046_mapper_llm_plans -> 0055_scenario_run_notifications exactly once.
|
||||
@_REQUIRES_PG
|
||||
def test_previous_release_upgrades_in_place_preserving_approved_reference(
|
||||
monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
@@ -211,9 +228,7 @@ def test_previous_release_upgrades_in_place_preserving_approved_reference(
|
||||
with schema_engine.connect() as connection:
|
||||
before = _whole_rows(connection, seeded)
|
||||
before_authority = _draft_authority_row(connection, seeded["draft_id"])
|
||||
assert before["catalog_revisions"] and before["catalog_memberships"]
|
||||
assert before["entry_authorities"] and before["publication_operations"]
|
||||
assert before["approval_receipts"]
|
||||
_assert_reference_rows_seeded(before)
|
||||
|
||||
_run_alembic_upgrade()
|
||||
|
||||
@@ -237,7 +252,7 @@ def test_previous_release_upgrades_in_place_preserving_approved_reference(
|
||||
# #endregion Test.Alembic.InPlaceUpgrade.PreviousRelease
|
||||
|
||||
|
||||
# #region Test.Alembic.InPlaceUpgrade.CurrentRead [C:4] [TYPE Function]
|
||||
# #region Test.Alembic.InPlaceUpgrade.CurrentRead [C:3] [TYPE Function]
|
||||
# @POST Exact revision read keeps the seeded digest, receipt binding and the honest legacy authority shape.
|
||||
def _assert_current_code_reads_exact_reference(engine, seeded: dict) -> None:
|
||||
from sqlalchemy.orm import Session
|
||||
@@ -250,28 +265,56 @@ def _assert_current_code_reads_exact_reference(engine, seeded: dict) -> None:
|
||||
|
||||
with Session(engine) as db:
|
||||
head = read_catalog_head(db, seeded["catalog_id"])
|
||||
assert head.cas_version == 2 and head.status == "active"
|
||||
_assert_catalog_head_current(head)
|
||||
snapshot = load_receipted_published_catalog(db, ExactCatalogRevision(kind="revision",
|
||||
catalog_id=seeded["catalog_id"],
|
||||
catalog_revision_id=seeded["catalog_revision_id"]))
|
||||
assert snapshot.integrity_sha256 == seeded["integrity_sha256"]
|
||||
assert snapshot.revision.catalog_digest == seeded["catalog_digest"]
|
||||
assert snapshot.revision.catalog_revision_id == seeded["catalog_revision_id"]
|
||||
assert snapshot.publication.operation_id == seeded["operation_id"]
|
||||
assert snapshot.publication.commit_sha == seeded["commit_sha"]
|
||||
assert snapshot.publication.render_sha256 == seeded["render_sha256"]
|
||||
authority = snapshot.entry_authorities[0].authority
|
||||
# @INVARIANT A pre-build-identity envelope stays readable and is not rewritten into a different shape.
|
||||
assert authority.model_dump(mode="json", exclude_unset=True) == seeded["authority"]
|
||||
_assert_published_snapshot_exact(snapshot, seeded)
|
||||
_assert_legacy_authority_shape(snapshot, seeded)
|
||||
receipt = db.get(ReferenceDraftApprovalReceipt, seeded["receipt_id"])
|
||||
assert receipt is not None
|
||||
assert (receipt.draft_id, receipt.candidate_id, receipt.decision) == (
|
||||
seeded["draft_id"], seeded["candidate_id"], "approved")
|
||||
assert receipt.reason == seeded["receipt_reason"]
|
||||
assert receipt.actor_id == "dr-audit-owner" and receipt.cas_version == 2
|
||||
_assert_receipt_unchanged(receipt, seeded)
|
||||
# #endregion Test.Alembic.InPlaceUpgrade.CurrentRead
|
||||
|
||||
|
||||
# #region Test.Alembic.InPlaceUpgrade.CatalogHeadCurrent [C:1] [TYPE Function]
|
||||
# @POST The readable head is the second active revision from the seeded fixture.
|
||||
def _assert_catalog_head_current(head) -> None:
|
||||
assert head.cas_version == 2 and head.status == "active"
|
||||
# #endregion Test.Alembic.InPlaceUpgrade.CatalogHeadCurrent
|
||||
|
||||
|
||||
# #region Test.Alembic.InPlaceUpgrade.PublishedSnapshotExact [C:2] [TYPE Function]
|
||||
# @POST The receipted snapshot preserves integrity, digest, revision, publication identity and render hash.
|
||||
def _assert_published_snapshot_exact(snapshot, seeded: dict) -> None:
|
||||
assert snapshot.integrity_sha256 == seeded["integrity_sha256"]
|
||||
assert snapshot.revision.catalog_digest == seeded["catalog_digest"]
|
||||
assert snapshot.revision.catalog_revision_id == seeded["catalog_revision_id"]
|
||||
assert snapshot.publication.operation_id == seeded["operation_id"]
|
||||
assert snapshot.publication.commit_sha == seeded["commit_sha"]
|
||||
assert snapshot.publication.render_sha256 == seeded["render_sha256"]
|
||||
# #endregion Test.Alembic.InPlaceUpgrade.PublishedSnapshotExact
|
||||
|
||||
|
||||
# #region Test.Alembic.InPlaceUpgrade.LegacyAuthorityShape [C:2] [TYPE Function]
|
||||
# @BRIEF The pre-build-identity envelope is compared exactly against the seeded authority JSON.
|
||||
# @INVARIANT A pre-build-identity envelope stays readable and is not rewritten into a different shape.
|
||||
def _assert_legacy_authority_shape(snapshot, seeded: dict) -> None:
|
||||
authority = snapshot.entry_authorities[0].authority
|
||||
assert authority.model_dump(mode="json", exclude_unset=True) == seeded["authority"]
|
||||
# #endregion Test.Alembic.InPlaceUpgrade.LegacyAuthorityShape
|
||||
|
||||
|
||||
# #region Test.Alembic.InPlaceUpgrade.ReceiptUnchanged [C:2] [TYPE Function]
|
||||
# @POST The approval receipt keeps its decision, reason and actor/cas version.
|
||||
def _assert_receipt_unchanged(receipt, seeded: dict) -> None:
|
||||
assert receipt is not None
|
||||
assert (receipt.draft_id, receipt.candidate_id, receipt.decision) == (
|
||||
seeded["draft_id"], seeded["candidate_id"], "approved")
|
||||
assert receipt.reason == seeded["receipt_reason"]
|
||||
assert receipt.actor_id == "dr-audit-owner" and receipt.cas_version == 2
|
||||
# #endregion Test.Alembic.InPlaceUpgrade.ReceiptUnchanged
|
||||
|
||||
|
||||
# #region Test.Alembic.InPlaceUpgrade.LegacyRefusal [C:3] [TYPE Function]
|
||||
# @BRIEF An rc2-shaped schema (non-empty, no alembic_version) is refused without losing or stamping data.
|
||||
# @RATIONALE The 2026-10-20 finding names this exact refusal; PG proves transactional DDL leaves no partial state.
|
||||
|
||||
@@ -1,71 +1,50 @@
|
||||
# #region Test.Alembic.Migrations [C:3] [TYPE Module] [SEMANTICS test,alembic,migration,schema,postgres]
|
||||
# @BRIEF Verifies the single baseline Alembic upgrade.
|
||||
# @RELATION BINDS_TO -> [Alembic.AddDeploymentRecords]
|
||||
# @INVARIANT Running `alembic upgrade head` twice is idempotent (no-op on second run) on PostgreSQL.
|
||||
# @PRE Alembic is installed, migration files exist in alembic/versions/.
|
||||
# @POST Schema matches expected state after upgrade.
|
||||
# #region Test.Alembic.Migrations [C:3] [TYPE Module] [SEMANTICS test,alembic,migration,schema,postgres,current-head]
|
||||
# @BRIEF Verify the fresh Alembic chain, the pinned current head, and the fixed 0049→0050 source-binding boundary.
|
||||
# @RELATION BINDS_TO -> [Migrations.0051.DeploymentAppliedAt]
|
||||
# @RELATION BINDS_TO -> [Migrations.0052.ReferenceDraftInspection]
|
||||
# @INVARIANT Running `alembic upgrade head` twice is idempotent (no-op on second run) on PostgreSQL.
|
||||
# @PRE Alembic is installed, migration files exist in alembic/versions/.
|
||||
# @POST Schema matches expected state after upgrade; named current-chain tests assert CURRENT_ALEMBIC_HEAD.
|
||||
# @TEST_INVARIANT: alembic_fresh_upgrade -> VERIFIED_BY: [Test.AlembicMigrations.TestFreshUpgradeCreatesTables]
|
||||
# @TEST_INVARIANT: alembic_current_head_pin -> VERIFIED_BY: [Test.AlembicMigrations.CurrentHeadContract]
|
||||
# @REJECTED Full migration chain idempotency on SQLite — initial migration uses
|
||||
# op.create_index() which conflicts with SQLite auto-created indexes
|
||||
# for UNIQUE constraints. Migrations are designed for PostgreSQL.
|
||||
# @REJECTED Moving `_run_alembic_upgrade` into a helper module was rejected: test_alembic_url_interpolation
|
||||
# execs that top-level function from this module's AST to prove exact URL interpolation.
|
||||
|
||||
from contextlib import contextmanager
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import uuid
|
||||
import pytest
|
||||
import os
|
||||
import sys
|
||||
|
||||
from sqlalchemy.engine import make_url
|
||||
import pytest
|
||||
|
||||
from tests.alembic_migration_schema_assertions import (
|
||||
_assert_0051_schema,
|
||||
_assert_0052_schema,
|
||||
_assert_no_inspection_snapshot_column,
|
||||
_assert_source_binding_behavior,
|
||||
_assert_source_binding_schema,
|
||||
)
|
||||
from tests.alembic_migration_support import ( # noqa: F401 re-exported for sibling test modules
|
||||
_DB_BACKEND,
|
||||
_REQUIRES_PG,
|
||||
_isolated_pg_schema,
|
||||
)
|
||||
|
||||
# One hardcoded pin for the current chain head; bump it when a new migration becomes
|
||||
# head (root integrates 0056_reference_finish_completion after this preparation).
|
||||
CURRENT_ALEMBIC_HEAD = "0055_scenario_run_notifications"
|
||||
|
||||
# Ensure backend/src is importable for model metadata
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||||
|
||||
# Alembic migrations use PostgreSQL-specific features (CREATE INDEX may conflict
|
||||
# with SQLite auto-indexes for UNIQUE constraints). Skip full-chain tests on SQLite.
|
||||
_DB_BACKEND = make_url(os.environ.get("DATABASE_URL", "sqlite://")).get_backend_name()
|
||||
_REQUIRES_PG = pytest.mark.skipif(
|
||||
_DB_BACKEND != "postgresql",
|
||||
reason="Alembic migrations are PostgreSQL-specific; use a PostgreSQL DATABASE_URL to test",
|
||||
)
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.TestFreshUpgradeCreatesTables [C:3] [TYPE Function]
|
||||
# @BRIEF Run alembic upgrade head on a fresh PostgreSQL DB and verify key tables exist.
|
||||
# @POST All expected tables exist after upgrade.
|
||||
# @POST All expected tables exist after upgrade and alembic_version names CURRENT_ALEMBIC_HEAD.
|
||||
# @TEST_INVARIANT: alembic_fresh_upgrade -> VERIFIED_BY: [Test.AlembicMigrations.TestFreshUpgradeCreatesTables]
|
||||
@_REQUIRES_PG
|
||||
# #region Test.AlembicMigrations.AppliedAtSchema
|
||||
def _assert_0051_schema(engine) -> None:
|
||||
from sqlalchemy import inspect
|
||||
import sqlalchemy as sa
|
||||
|
||||
columns = {column["name"]: column for column in inspect(engine).get_columns("deployment_records")}
|
||||
assert columns["applied_at"]["nullable"] is True
|
||||
assert columns["target_digest"]["nullable"] is True
|
||||
assert isinstance(columns["applied_at"]["type"], sa.DateTime)
|
||||
assert columns["target_digest"]["type"].length == 64
|
||||
# #endregion Test.AlembicMigrations.AppliedAtSchema
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.InspectionSchema [C:2] [TYPE Function]
|
||||
# @POST The 0052-owned inspection column exists exactly once as nullable JSON without a server default.
|
||||
def _assert_0052_schema(engine) -> None:
|
||||
from sqlalchemy import inspect
|
||||
from sqlalchemy.dialects.postgresql import JSONB
|
||||
import sqlalchemy as sa
|
||||
|
||||
columns = [column for column in inspect(engine).get_columns("dashboard_reference_draft_authorities")
|
||||
if column["name"] == "inspection_snapshot"]
|
||||
assert len(columns) == 1
|
||||
assert columns[0]["nullable"] is True
|
||||
assert isinstance(columns[0]["type"], sa.JSON)
|
||||
assert not isinstance(columns[0]["type"], JSONB)
|
||||
assert columns[0].get("default") is None
|
||||
# #endregion Test.AlembicMigrations.InspectionSchema
|
||||
|
||||
|
||||
def test_fresh_upgrade_creates_tables() -> None:
|
||||
"""Fresh Alembic upgrade creates all expected tables (PostgreSQL only)."""
|
||||
from sqlalchemy import create_engine, inspect, text as sa_text
|
||||
@@ -82,7 +61,7 @@ def test_fresh_upgrade_creates_tables() -> None:
|
||||
result = conn.execute(sa_text("SELECT version_num FROM alembic_version"))
|
||||
version = result.scalar_one()
|
||||
assert version is not None and len(version) > 0
|
||||
assert version == "0052_draft_inspection"
|
||||
assert version == CURRENT_ALEMBIC_HEAD
|
||||
_assert_0051_schema(engine)
|
||||
_assert_0052_schema(engine)
|
||||
roles_cols = {c["name"] for c in inspector.get_columns("roles")}
|
||||
@@ -94,219 +73,6 @@ def test_fresh_upgrade_creates_tables() -> None:
|
||||
# #endregion Test.AlembicMigrations.TestFreshUpgradeCreatesTables
|
||||
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.SourceBindingIsolation [C:4] [TYPE Function]
|
||||
# @BRIEF Give each migration test its own PostgreSQL schema in the run-owned worker DB.
|
||||
# @POST Alembic sees exactly one search_path (no public fallback); cleanup drops only this test's schema.
|
||||
# @INVARIANT No downgrade/drop of shared worker tables; no credential files or URL logging.
|
||||
@contextmanager
|
||||
def _isolated_pg_schema(monkeypatch: pytest.MonkeyPatch):
|
||||
from sqlalchemy import create_engine, text as sa_text
|
||||
|
||||
schema = f"alembic_sb_{uuid.uuid4().hex}"
|
||||
previous_schema = os.environ.get("DB_SCHEMA")
|
||||
engine = None
|
||||
schema_engine = None
|
||||
created = False
|
||||
try:
|
||||
engine = create_engine(os.environ["DATABASE_URL"])
|
||||
with engine.begin() as connection:
|
||||
connection.execute(sa_text(f'CREATE SCHEMA "{schema}"'))
|
||||
created = True
|
||||
schema_engine = create_engine(
|
||||
os.environ["DATABASE_URL"],
|
||||
connect_args={"options": f"-csearch_path={schema}"},
|
||||
)
|
||||
monkeypatch.setenv("DB_SCHEMA", schema)
|
||||
yield schema, schema_engine
|
||||
finally:
|
||||
try:
|
||||
if schema_engine is not None:
|
||||
schema_engine.dispose()
|
||||
finally:
|
||||
try:
|
||||
if created and engine is not None:
|
||||
with engine.begin() as connection:
|
||||
connection.execute(sa_text(f'DROP SCHEMA "{schema}" CASCADE'))
|
||||
finally:
|
||||
if engine is not None:
|
||||
engine.dispose()
|
||||
if previous_schema is None:
|
||||
monkeypatch.delenv("DB_SCHEMA", raising=False)
|
||||
else:
|
||||
monkeypatch.setenv("DB_SCHEMA", previous_schema)
|
||||
# #endregion Test.AlembicMigrations.SourceBindingIsolation
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.SourceBindingSchemaAssert [C:4] [TYPE Function]
|
||||
# @BRIEF Assert the source-binding DDL independently of production ORM metadata.
|
||||
# @POST Verifies exact columns, FK columns/ondelete, check expressions, index columns/partial predicates.
|
||||
def _assert_source_binding_schema(schema_engine, schema: str) -> None:
|
||||
from sqlalchemy import inspect, text as sa_text
|
||||
|
||||
table = "git_repository_source_bindings"
|
||||
inspector = inspect(schema_engine)
|
||||
assert table in set(inspector.get_table_names())
|
||||
with schema_engine.connect() as connection:
|
||||
column_rows = connection.execute(sa_text(
|
||||
"SELECT column_name, is_nullable FROM information_schema.columns "
|
||||
"WHERE table_schema = :schema AND table_name = :table"
|
||||
), {"schema": schema, "table": table}).all()
|
||||
expected_columns = {
|
||||
"id": "NO", "repository_id": "NO", "dashboard_uuid": "NO",
|
||||
"source_environment_id": "NO", "source_dashboard_id": "NO",
|
||||
"release_id": "NO", "release_commit_hash": "NO", "status": "NO",
|
||||
"actor_id": "NO", "actor_username": "NO", "created_at": "NO",
|
||||
"retired_at": "YES", "retired_by": "YES", "cas_version": "NO",
|
||||
"request_sha256": "NO", "receipt_sha256": "NO",
|
||||
}
|
||||
assert {row.column_name: row.is_nullable for row in column_rows} == expected_columns
|
||||
|
||||
constraint_rows = connection.execute(sa_text(
|
||||
"SELECT conname, contype, pg_get_constraintdef(oid) AS definition "
|
||||
"FROM pg_constraint WHERE conrelid = CAST(:qualified AS regclass)"
|
||||
), {"qualified": f"{schema}.{table}"}).all()
|
||||
definitions = {row.contype: [] for row in constraint_rows}
|
||||
for row in constraint_rows:
|
||||
definitions[row.contype].append(" ".join(row.definition.replace('"', "").lower().split()))
|
||||
assert set(definitions.get("f", [])) == {
|
||||
"foreign key (repository_id) references git_repositories(id) on delete restrict",
|
||||
"foreign key (release_id) references dashboard_releases(id) on delete restrict",
|
||||
}
|
||||
assert len(definitions.get("c", [])) == 6
|
||||
|
||||
index_rows = connection.execute(sa_text(
|
||||
"SELECT i.relname AS indexname, x.indisunique AS is_unique, "
|
||||
"pg_get_indexdef(x.indexrelid) AS indexdef, "
|
||||
"pg_get_expr(x.indpred, x.indrelid) AS predicate "
|
||||
"FROM pg_index x JOIN pg_class i ON i.oid = x.indexrelid "
|
||||
"WHERE x.indrelid = CAST(:qualified AS regclass)"
|
||||
), {"qualified": f"{schema}.{table}"}).all()
|
||||
expected_indexes = {
|
||||
"git_repository_source_bindings_pkey": (True, None, "(id)"),
|
||||
"ix_git_source_binding_repository": (False, None, "(repository_id)"),
|
||||
"uq_git_source_binding_active_repository": (True, "status='active'", "(repository_id)"),
|
||||
"uq_git_source_binding_active_uuid": (True, "status='active'", "(dashboard_uuid)"),
|
||||
"uq_git_source_binding_active_source": (True, "status='active'",
|
||||
"(source_environment_id, source_dashboard_id)"),
|
||||
}
|
||||
actual_indexes = {row.indexname: row for row in index_rows}
|
||||
assert set(actual_indexes) == set(expected_indexes)
|
||||
for name, (unique, predicate, columns) in expected_indexes.items():
|
||||
row = actual_indexes[name]
|
||||
assert bool(row.is_unique) is unique, name
|
||||
if row.predicate is None:
|
||||
normalized_predicate = None
|
||||
else:
|
||||
normalized_predicate = "".join(
|
||||
row.predicate.replace('"', "").lower().replace("::text", "").split())
|
||||
if normalized_predicate.startswith("(") and normalized_predicate.endswith(")"):
|
||||
normalized_predicate = normalized_predicate[1:-1]
|
||||
normalized_predicate = re.sub(
|
||||
r"\(([a-z_][a-z0-9_]*)\)", r"\1", normalized_predicate)
|
||||
assert normalized_predicate == predicate, name
|
||||
assert columns in row.indexdef.replace('"', "").lower(), name
|
||||
# #endregion Test.AlembicMigrations.SourceBindingSchemaAssert
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.SourceBindingBehavior [C:4] [TYPE Function]
|
||||
# @BRIEF Independent insert-refusal/retired-allowed fixtures for checks and active partial indexes.
|
||||
# @POST All six checks reject invalid rows; retired duplicates are allowed; active duplicates are rejected by DB.
|
||||
# @INVARIANT Expected values are hardcoded here, not imported from production ORM metadata.
|
||||
def _assert_source_binding_behavior(schema_engine) -> None:
|
||||
from datetime import UTC, datetime
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy import text as sa_text
|
||||
|
||||
table = "git_repository_source_bindings"
|
||||
env_id = "00000000-0000-4000-8000-000000000001"
|
||||
repo_a = "00000000-0000-4000-8000-000000000011"
|
||||
repo_b = "00000000-0000-4000-8000-000000000012"
|
||||
repo_c = "00000000-0000-4000-8000-000000000013"
|
||||
release_id = "00000000-0000-4000-8000-000000000021"
|
||||
uuid_a = "a2cef831-4ffd-46ec-aaa1-60f71782a91c"
|
||||
uuid_b = "b2cef831-4ffd-46ec-aaa1-60f71782a91c"
|
||||
uuid_c = "c2cef831-4ffd-46ec-aaa1-60f71782a91c"
|
||||
commit = "a" * 40
|
||||
request_hash = "0" * 64
|
||||
receipt_hash = "1" * 64
|
||||
now = datetime.now(UTC)
|
||||
|
||||
with schema_engine.begin() as connection:
|
||||
connection.execute(sa_text(
|
||||
"INSERT INTO deployment_environments (id, name, superset_url, superset_token, is_active) "
|
||||
"VALUES (:id, :name, :url, :token, true)"
|
||||
), {"id": env_id, "name": "fixture", "url": "http://fixture", "token": ""})
|
||||
for dashboard_id, repo_id in ((810001, repo_a), (810002, repo_b), (810003, repo_c)):
|
||||
connection.execute(sa_text(
|
||||
"INSERT INTO git_repositories "
|
||||
"(id, dashboard_id, local_path, current_branch, sync_status) "
|
||||
"VALUES (:id, :dashboard_id, :path, 'dev', 'CLEAN')"
|
||||
), {"id": repo_id, "dashboard_id": dashboard_id, "path": f"/tmp/{repo_id}"})
|
||||
deployment_id = connection.execute(sa_text(
|
||||
"INSERT INTO deployment_records "
|
||||
"(repository_id, environment_id, commit_hash, content_hash, deployed_at, status, validation_status) "
|
||||
"VALUES (:repo, :env, :commit, :content, :deployed, 'success', 'validated') RETURNING id"
|
||||
), {"repo": repo_a, "env": env_id, "commit": commit, "content": "c" * 64,
|
||||
"deployed": now}).scalar_one()
|
||||
connection.execute(sa_text(
|
||||
"INSERT INTO dashboard_releases "
|
||||
"(id, repository_id, deployment_id, name, version, notes, commit_hash, content_hash, status, created_at, created_by) "
|
||||
"VALUES (:id, :repo, :deployment, 'fixture', '1.0.0', 'fixture', :commit, :content, 'ready_to_publish', :created, 'fixture')"
|
||||
), {"id": release_id, "repo": repo_a, "deployment": deployment_id, "commit": commit,
|
||||
"content": "c" * 64, "created": now})
|
||||
|
||||
def insert_binding(*, repository_id: str, dashboard_uuid: str, source_environment_id: str,
|
||||
source_dashboard_id: int = 12, status: str = "active", commit_hash: str = commit,
|
||||
cas_version: int = 1) -> None:
|
||||
with schema_engine.begin() as connection:
|
||||
connection.execute(sa_text(
|
||||
f"INSERT INTO {table} "
|
||||
"(id, repository_id, dashboard_uuid, source_environment_id, source_dashboard_id, "
|
||||
"release_id, release_commit_hash, status, actor_id, actor_username, created_at, "
|
||||
"cas_version, request_sha256, receipt_sha256) "
|
||||
"VALUES (:id, :repo, :uuid, :source_env, :source_dashboard, :release, :commit, :status, "
|
||||
"'actor', 'operator', :created, :cas, :request, :receipt)"
|
||||
), {"id": str(uuid.uuid4()), "repo": repository_id, "uuid": dashboard_uuid,
|
||||
"source_env": source_environment_id, "source_dashboard": source_dashboard_id,
|
||||
"release": release_id, "commit": commit_hash, "status": status,
|
||||
"created": now, "cas": cas_version, "request": request_hash, "receipt": receipt_hash})
|
||||
|
||||
insert_binding(repository_id=repo_a, dashboard_uuid=uuid_a, source_environment_id="source-a")
|
||||
insert_binding(repository_id=repo_a, dashboard_uuid=uuid_a, source_environment_id="source-a",
|
||||
status="retired")
|
||||
with schema_engine.connect() as connection:
|
||||
assert connection.execute(sa_text(f"SELECT count(*) FROM {table}")).scalar_one() == 2
|
||||
|
||||
for invalid in (
|
||||
{"repository_id": repo_a, "uuid": uuid_b, "source_env": "source-b"},
|
||||
{"repository_id": repo_b, "uuid": uuid_a, "source_env": "source-c"},
|
||||
{"repository_id": repo_c, "uuid": uuid_b, "source_env": "source-a"},
|
||||
):
|
||||
with pytest.raises(IntegrityError):
|
||||
insert_binding(repository_id=invalid["repository_id"], dashboard_uuid=invalid["uuid"],
|
||||
source_environment_id=invalid["source_env"])
|
||||
|
||||
invalid_rows = (
|
||||
{"status": "paused"},
|
||||
{"dashboard_uuid": "short"},
|
||||
{"commit_hash": "short"},
|
||||
{"source_environment_id": ""},
|
||||
{"source_dashboard_id": 0},
|
||||
{"cas_version": 0},
|
||||
)
|
||||
for invalid in invalid_rows:
|
||||
values = {
|
||||
"repository_id": repo_a, "dashboard_uuid": uuid_c,
|
||||
"source_environment_id": "source-check", "source_dashboard_id": 12,
|
||||
"status": "retired", "commit_hash": commit, "cas_version": 1,
|
||||
}
|
||||
values.update(invalid)
|
||||
with pytest.raises(IntegrityError):
|
||||
insert_binding(**values)
|
||||
# #endregion Test.AlembicMigrations.SourceBindingBehavior
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.SourceBindingTable [C:4] [TYPE Function]
|
||||
# @BRIEF Fresh PostgreSQL chain must let revision 0050 own the source-binding DDL, not 0001 create_all.
|
||||
# @TEST_INVARIANT: source_binding_owned_by_0050 -> VERIFIED_BY: [Test.AlembicMigrations.SourceBindingTable]
|
||||
@@ -328,12 +94,12 @@ def test_fresh_upgrade_owns_source_binding_table(monkeypatch: pytest.MonkeyPatch
|
||||
_assert_0052_schema(schema_engine)
|
||||
with schema_engine.connect() as connection:
|
||||
version = connection.execute(sa_text("SELECT version_num FROM alembic_version")).scalar_one()
|
||||
assert version == "0052_draft_inspection"
|
||||
assert version == CURRENT_ALEMBIC_HEAD
|
||||
# #endregion Test.AlembicMigrations.SourceBindingTable
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.SourceBindingUpgradePath [C:4] [TYPE Function]
|
||||
# @BRIEF Prove the exact 0049→0050 source-binding boundary, then retain full upgrade-to-0052 coverage.
|
||||
# @BRIEF Prove the exact 0049→0050 source-binding boundary, then replay the full chain to CURRENT_ALEMBIC_HEAD.
|
||||
@_REQUIRES_PG
|
||||
def test_source_binding_upgrade_from_0049_to_0050(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
from sqlalchemy import inspect, text as sa_text
|
||||
@@ -352,7 +118,7 @@ def test_source_binding_upgrade_from_0049_to_0050(monkeypatch: pytest.MonkeyPatc
|
||||
_assert_0052_schema(schema_engine)
|
||||
with schema_engine.connect() as connection:
|
||||
version = connection.execute(sa_text("SELECT version_num FROM alembic_version")).scalar_one()
|
||||
assert version == "0052_draft_inspection"
|
||||
assert version == CURRENT_ALEMBIC_HEAD
|
||||
# #endregion Test.AlembicMigrations.SourceBindingUpgradePath
|
||||
|
||||
|
||||
@@ -367,17 +133,19 @@ def test_deployment_applied_at_upgrade_from_0050_to_0051(monkeypatch: pytest.Mon
|
||||
before = {c["name"] for c in inspect(schema_engine).get_columns("deployment_records")}
|
||||
assert "applied_at" not in before and "target_digest" not in before
|
||||
_run_alembic_upgrade("0051_deployment_applied_at")
|
||||
columns = {c["name"]: c for c in inspect(schema_engine).get_columns("deployment_records")}
|
||||
assert columns["applied_at"]["nullable"] is True
|
||||
assert columns["target_digest"]["nullable"] is True
|
||||
assert columns["target_digest"]["type"].length == 64
|
||||
_assert_0051_schema(schema_engine)
|
||||
with schema_engine.connect() as connection:
|
||||
version = connection.execute(sa_text("SELECT version_num FROM alembic_version")).scalar_one()
|
||||
assert version == "0051_deployment_applied_at"
|
||||
assert "inspection_snapshot" not in {column["name"] for column in inspect(schema_engine).get_columns("dashboard_reference_draft_authorities")}
|
||||
_assert_no_inspection_snapshot_column(schema_engine)
|
||||
# #endregion Test.AlembicMigrations.DeploymentAppliedAt
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.BaselineIncompleteSchema [C:2] [TYPE Function]
|
||||
# @BRIEF A legacy table cannot be silently stamped as the reset baseline.
|
||||
# @POST The baseline upgrade raises its typed refusal before any stamp or DDL runs.
|
||||
# @RELATION VERIFIES -> [Migrations.Baseline.Upgrade]
|
||||
# @TEST_INVARIANT: incomplete_existing_schema_refused -> VERIFIED_BY: [Test.AlembicMigrations.BaselineIncompleteSchema]
|
||||
def test_baseline_refuses_incomplete_existing_schema(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""A legacy table cannot be silently stamped as the reset baseline."""
|
||||
from sqlalchemy import create_engine, text
|
||||
@@ -391,6 +159,7 @@ def test_baseline_refuses_incomplete_existing_schema(tmp_path: Path, monkeypatch
|
||||
monkeypatch.setenv("DATABASE_URL", database_url)
|
||||
with pytest.raises(RuntimeError, match="Refusing to stamp an incomplete existing schema"):
|
||||
_run_alembic_upgrade()
|
||||
# #endregion Test.AlembicMigrations.BaselineIncompleteSchema
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.RevisionIdLength [C:2] [TYPE Function]
|
||||
@@ -413,6 +182,24 @@ def test_revision_ids_fit_varchar32() -> None:
|
||||
# #endregion Test.AlembicMigrations.RevisionIdLength
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.CurrentHeadContract [C:2] [TYPE Function]
|
||||
# @BRIEF Offline structural guard: the real migration scripts resolve to exactly CURRENT_ALEMBIC_HEAD.
|
||||
# @POST ScriptDirectory reports one head and it equals the hardcoded chain pin.
|
||||
# @TEST_INVARIANT: alembic_current_head_pin -> VERIFIED_BY: [Test.AlembicMigrations.CurrentHeadContract]
|
||||
# @REJECTED Reading the head from a database was rejected: this guard must run without PostgreSQL.
|
||||
def test_current_head_revision_is_pinned_offline() -> None:
|
||||
"""Static guard: the declared chain pin matches the single real Alembic head."""
|
||||
from alembic.config import Config as AlembicConfig
|
||||
from alembic.script import ScriptDirectory
|
||||
|
||||
backend_dir = Path(__file__).parent.parent
|
||||
alembic_cfg = AlembicConfig(str(backend_dir / "alembic.ini"))
|
||||
alembic_cfg.set_main_option("script_location", str(backend_dir / "alembic"))
|
||||
heads = tuple(ScriptDirectory.from_config(alembic_cfg).get_heads())
|
||||
assert heads == (CURRENT_ALEMBIC_HEAD,)
|
||||
# #endregion Test.AlembicMigrations.CurrentHeadContract
|
||||
|
||||
|
||||
# #region Test.AlembicMigrations.RunAlembicUpgrade [C:1] [TYPE Function]
|
||||
def _run_alembic_upgrade(revision: str = "head") -> None:
|
||||
"""Run a named Alembic upgrade programmatically against DATABASE_URL."""
|
||||
|
||||
@@ -34,44 +34,107 @@ def install_shape_double(monkeypatch, baseline_owner, *, column=None, drop_mutat
|
||||
columns = [{"name": "draft_id", "type": sa.String(36), "nullable": False, "default": None}]
|
||||
if column is not None:
|
||||
columns.append(column)
|
||||
# #region Test.Migrations.ReferenceDraftInspection.ShapeInspectColumns [C:1] [TYPE Function]
|
||||
# @BRIEF Returns a deep copy of the simulated authority column list.
|
||||
def inspect_columns(table):
|
||||
assert table == _TABLE
|
||||
return deepcopy(columns)
|
||||
# #endregion Test.Migrations.ReferenceDraftInspection.ShapeInspectColumns
|
||||
# #region Test.Migrations.ReferenceDraftInspection.ShapeDropColumn [C:1] [TYPE Function]
|
||||
# @BRIEF Simulates an owned column drop and optionally mutates the simulated column list.
|
||||
def drop_column(table, name):
|
||||
assert (table, name) == (_TABLE, _COLUMN)
|
||||
if drop_mutates:
|
||||
columns[:] = [item for item in columns if item["name"] != name]
|
||||
# #endregion Test.Migrations.ReferenceDraftInspection.ShapeDropColumn
|
||||
inspector = SimpleNamespace(get_columns=inspect_columns)
|
||||
monkeypatch.setattr(baseline_owner.sa, "inspect", lambda _bind: inspector)
|
||||
monkeypatch.setattr(
|
||||
baseline_owner.sa,
|
||||
"inspect",
|
||||
# #region Test.Migrations.ReferenceDraftInspection.ShapeInspectorCallback [C:1] [TYPE Function]
|
||||
# @BRIEF Binds the injected inspector to any inspected bind.
|
||||
lambda _bind: inspector
|
||||
# #endregion Test.Migrations.ReferenceDraftInspection.ShapeInspectorCallback
|
||||
)
|
||||
dropped = Mock(side_effect=drop_column)
|
||||
monkeypatch.setattr(baseline_owner.op, "drop_column", dropped)
|
||||
return columns, dropped
|
||||
# #endregion Test.Migrations.ReferenceDraftInspection.ShapeDouble
|
||||
|
||||
|
||||
# #region Test.Migrations.ReferenceDraftInspection.Ownership [C:3] [TYPE Function]
|
||||
def test_current_metadata_field_is_deferred_before_historical_0041_validation(baseline_owner):
|
||||
# #region Test.Migrations.ReferenceDraftInspection.MetadataDefersInspection [C:1] [TYPE Function]
|
||||
# @POST Current metadata still declares the 0052-owned inspection column.
|
||||
def _assert_metadata_defers_inspection_column(baseline_owner) -> None:
|
||||
assert _COLUMN in baseline_owner.Base.metadata.tables[_TABLE].c
|
||||
# #endregion Test.Migrations.ReferenceDraftInspection.MetadataDefersInspection
|
||||
|
||||
|
||||
# #region Test.Migrations.ReferenceDraftInspection.ExpectedColumns0041 [C:2] [TYPE Function]
|
||||
# @POST The historical 0041 expected-column set excludes the future 0052 inspection column.
|
||||
def _assert_0041_expected_columns_exclude_inspection() -> None:
|
||||
path = Path(__file__).parents[1] / "alembic/versions/0041_reference_draft_authority.py"
|
||||
tree = ast.parse(path.read_text())
|
||||
expected = next(node.value for node in tree.body if isinstance(node, ast.Assign)
|
||||
and any(isinstance(target, ast.Name) and target.id == "_EXPECTED_COLUMNS" for target in node.targets))
|
||||
assert _COLUMN not in [key.value for key in expected.keys]
|
||||
# #endregion Test.Migrations.ReferenceDraftInspection.ExpectedColumns0041
|
||||
|
||||
|
||||
# #region Test.Migrations.ReferenceDraftInspection.BaselineRestoreCalls [C:2] [TYPE Function]
|
||||
# @POST Collects every helper call made by the baseline upgrade body.
|
||||
def _baseline_restore_calls(baseline_owner) -> list:
|
||||
source = ast.parse(Path(baseline_owner.__file__).read_text())
|
||||
upgrade = next(node for node in source.body if isinstance(node, ast.FunctionDef) and node.name == "upgrade")
|
||||
calls = [node.func.id for statement in upgrade.body for node in ast.walk(statement)
|
||||
return [node.func.id for statement in upgrade.body for node in ast.walk(statement)
|
||||
if isinstance(node, ast.Call) and isinstance(node.func, ast.Name)]
|
||||
# #endregion Test.Migrations.ReferenceDraftInspection.BaselineRestoreCalls
|
||||
|
||||
|
||||
# #region Test.Migrations.ReferenceDraftInspection.BaselineRestoreOrder [C:1] [TYPE Function]
|
||||
# @POST The baseline upgrade calls the seven historical-shape restorers in the fixed order.
|
||||
def _assert_baseline_restore_order(baseline_owner) -> None:
|
||||
calls = _baseline_restore_calls(baseline_owner)
|
||||
required = ["_create_frozen_catalog_tables", "_create_metadata_without_visual_capture_tables",
|
||||
"_restore_pre_0047_draft_shape", "_restore_pre_0048_finish_shape", "_restore_pre_0049_release_shape",
|
||||
"_restore_pre_0051_receipt_shape", "_restore_pre_0052_inspection_shape"]
|
||||
assert [call for call in calls if call in required] == required
|
||||
migration = ast.parse((path.parent / "0052_reference_draft_inspection.py").read_text())
|
||||
parent = next(node.value.value for node in migration.body if isinstance(node, ast.Assign)
|
||||
# #endregion Test.Migrations.ReferenceDraftInspection.BaselineRestoreOrder
|
||||
|
||||
|
||||
# #region Test.Migrations.ReferenceDraftInspection.Migration0052Tree [C:1] [TYPE Function]
|
||||
# @POST Parses the 0052 migration source exactly once for the ownership checks.
|
||||
def _0052_migration_tree():
|
||||
path = Path(__file__).parents[1] / "alembic/versions/0052_reference_draft_inspection.py"
|
||||
return ast.parse(path.read_text())
|
||||
# #endregion Test.Migrations.ReferenceDraftInspection.Migration0052Tree
|
||||
|
||||
|
||||
# #region Test.Migrations.ReferenceDraftInspection.DownRevision0052 [C:1] [TYPE Function]
|
||||
# @POST 0052 is chained directly onto the 0051 deployment boundary.
|
||||
def _assert_0052_down_revision(tree) -> None:
|
||||
parent = next(node.value.value for node in tree.body if isinstance(node, ast.Assign)
|
||||
and any(isinstance(target, ast.Name) and target.id == "down_revision" for target in node.targets))
|
||||
assert parent == "0051_deployment_applied_at"
|
||||
add = next(node for node in migration.body if isinstance(node, ast.FunctionDef) and node.name == "upgrade")
|
||||
# #endregion Test.Migrations.ReferenceDraftInspection.DownRevision0052
|
||||
|
||||
|
||||
# #region Test.Migrations.ReferenceDraftInspection.InspectionColumn0052 [C:1] [TYPE Function]
|
||||
# @POST 0052 owns exactly one add_column of the inspection snapshot.
|
||||
def _assert_0052_single_add_column(tree) -> None:
|
||||
add = next(node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == "upgrade")
|
||||
assert len(add.body) == 1
|
||||
assert isinstance(add.body[0], ast.Expr) and add.body[0].value.func.attr == "add_column"
|
||||
# #endregion Test.Migrations.ReferenceDraftInspection.InspectionColumn0052
|
||||
|
||||
|
||||
# #region Test.Migrations.ReferenceDraftInspection.Ownership [C:3] [TYPE Function]
|
||||
def test_current_metadata_field_is_deferred_before_historical_0041_validation(baseline_owner):
|
||||
_assert_metadata_defers_inspection_column(baseline_owner)
|
||||
_assert_0041_expected_columns_exclude_inspection()
|
||||
_assert_baseline_restore_order(baseline_owner)
|
||||
migration = _0052_migration_tree()
|
||||
_assert_0052_down_revision(migration)
|
||||
_assert_0052_single_add_column(migration)
|
||||
# #endregion Test.Migrations.ReferenceDraftInspection.Ownership
|
||||
|
||||
|
||||
@@ -109,43 +172,63 @@ def test_baseline_rechecks_absence_after_drop(monkeypatch, baseline_owner):
|
||||
# #endregion Test.Migrations.ReferenceDraftInspection.DropVerification
|
||||
|
||||
|
||||
# #region Test.Migrations.ReferenceDraftInspection.LegacyRowsBefore0052 [C:2] [TYPE Function]
|
||||
# @POST Captures the pre-0052 column absence and the exact legacy row/authority snapshots.
|
||||
def _assert_legacy_rows_before_0052(engine, seeded: dict):
|
||||
from tests.test_alembic_inplace_upgrade import _whole_rows
|
||||
|
||||
assert _COLUMN not in {column["name"] for column in sa.inspect(engine).get_columns(_TABLE)}
|
||||
with engine.connect() as connection:
|
||||
assert connection.execute(sa.text("SELECT version_num FROM alembic_version")).scalar_one() == "0051_deployment_applied_at"
|
||||
before = _whole_rows(connection, seeded)
|
||||
before_authority = connection.execute(sa.text(
|
||||
"SELECT to_jsonb(t)::text FROM dashboard_reference_draft_authorities t WHERE draft_id=:draft_id"),
|
||||
{"draft_id": seeded["draft_id"]}).scalar_one()
|
||||
assert all(before.values())
|
||||
return before, before_authority
|
||||
# #endregion Test.Migrations.ReferenceDraftInspection.LegacyRowsBefore0052
|
||||
|
||||
|
||||
# #region Test.Migrations.ReferenceDraftInspection.LegacyRows0052 [C:2] [TYPE Function]
|
||||
# @POST Exact 0052 replay adds SQL NULL only; every legacy byte group stays equal and the version stays 0052.
|
||||
def _assert_0052_replay_preserves(engine, seeded: dict, before: dict, before_authority: str) -> None:
|
||||
from tests.test_alembic_inplace_upgrade import _whole_rows
|
||||
from tests.test_alembic_migrations import _assert_0052_schema
|
||||
|
||||
_assert_0052_schema(engine)
|
||||
with engine.connect() as connection:
|
||||
assert connection.execute(sa.text("SELECT version_num FROM alembic_version")).scalar_one() == "0052_draft_inspection"
|
||||
assert _whole_rows(connection, seeded) == before
|
||||
assert connection.execute(sa.text(
|
||||
"SELECT (to_jsonb(t)-'inspection_snapshot')::text FROM dashboard_reference_draft_authorities t "
|
||||
"WHERE draft_id=:draft_id"), {"draft_id": seeded["draft_id"]}).scalar_one() == before_authority
|
||||
added = connection.execute(sa.text(
|
||||
"SELECT inspection_snapshot, inspection_snapshot IS NULL AS inspection_is_sql_null "
|
||||
"FROM dashboard_reference_draft_authorities WHERE draft_id=:draft_id"),
|
||||
{"draft_id": seeded["draft_id"]}).mappings().one()
|
||||
assert dict(added) == {"inspection_snapshot": None, "inspection_is_sql_null": True}
|
||||
# #endregion Test.Migrations.ReferenceDraftInspection.LegacyRows0052
|
||||
|
||||
|
||||
# #region Test.Migrations.ReferenceDraftInspection.LegacyUpgrade [C:4] [TYPE Function]
|
||||
# @PRE Root supplies isolated migrated PostgreSQL; a valid 0046 fixture is advanced explicitly to 0051 before the measured step.
|
||||
# @POST Exact 0051→0052 adds SQL NULL only; every old authority/catalog/membership/publication/receipt byte group stays equal.
|
||||
# @POST Exact 0051→0052 adds SQL NULL only; replaying exact 0052 is a no-op and the version remains 0052.
|
||||
@_REQUIRES_PG
|
||||
def test_0051_to_0052_preserves_all_legacy_rows_and_is_idempotent(monkeypatch):
|
||||
from tests.test_alembic_inplace_upgrade import (
|
||||
PREVIOUS_RELEASE_HEAD, _seed_approved_reference, _whole_rows, _assert_current_code_reads_exact_reference)
|
||||
from tests.test_alembic_migrations import _isolated_pg_schema, _run_alembic_upgrade, _assert_0052_schema
|
||||
PREVIOUS_RELEASE_HEAD, _assert_current_code_reads_exact_reference, _seed_approved_reference)
|
||||
from tests.test_alembic_migrations import _isolated_pg_schema, _run_alembic_upgrade
|
||||
|
||||
with _isolated_pg_schema(monkeypatch) as (_schema, engine):
|
||||
_run_alembic_upgrade(PREVIOUS_RELEASE_HEAD)
|
||||
seeded = _seed_approved_reference(engine)
|
||||
_run_alembic_upgrade("0051_deployment_applied_at")
|
||||
assert _COLUMN not in {column["name"] for column in sa.inspect(engine).get_columns(_TABLE)}
|
||||
with engine.connect() as connection:
|
||||
assert connection.execute(sa.text("SELECT version_num FROM alembic_version")).scalar_one() == "0051_deployment_applied_at"
|
||||
before = _whole_rows(connection, seeded)
|
||||
before_authority = connection.execute(sa.text(
|
||||
"SELECT to_jsonb(t)::text FROM dashboard_reference_draft_authorities t WHERE draft_id=:draft_id"),
|
||||
{"draft_id": seeded["draft_id"]}).scalar_one()
|
||||
assert all(before.values())
|
||||
before, before_authority = _assert_legacy_rows_before_0052(engine, seeded)
|
||||
_run_alembic_upgrade("0052_draft_inspection")
|
||||
for pass_index in (1, 2):
|
||||
_assert_0052_schema(engine)
|
||||
with engine.connect() as connection:
|
||||
assert connection.execute(sa.text("SELECT version_num FROM alembic_version")).scalar_one() == "0052_draft_inspection"
|
||||
assert _whole_rows(connection, seeded) == before
|
||||
assert connection.execute(sa.text(
|
||||
"SELECT (to_jsonb(t)-'inspection_snapshot')::text FROM dashboard_reference_draft_authorities t "
|
||||
"WHERE draft_id=:draft_id"), {"draft_id": seeded["draft_id"]}).scalar_one() == before_authority
|
||||
added = connection.execute(sa.text(
|
||||
"SELECT inspection_snapshot, inspection_snapshot IS NULL AS inspection_is_sql_null "
|
||||
"FROM dashboard_reference_draft_authorities WHERE draft_id=:draft_id"),
|
||||
{"draft_id": seeded["draft_id"]}).mappings().one()
|
||||
assert dict(added) == {"inspection_snapshot": None, "inspection_is_sql_null": True}
|
||||
_assert_0052_replay_preserves(engine, seeded, before, before_authority)
|
||||
if pass_index == 1:
|
||||
_run_alembic_upgrade()
|
||||
_run_alembic_upgrade("0052_draft_inspection")
|
||||
_assert_current_code_reads_exact_reference(engine, seeded)
|
||||
# #endregion Test.Migrations.ReferenceDraftInspection.LegacyUpgrade
|
||||
# #endregion Test.Migrations.ReferenceDraftInspection
|
||||
|
||||
Reference in New Issue
Block a user