"""Regression: migration 459 resumes after TiDB generated-column error 3106. The first production run created the principal tables and projection columns, then TiDB rejected ``MODIFY principal_email_group_id`` on ``data_access_bindings`` because ``active_semantic_key`` was generated from that Group column. TiDB DDL is non-transactional, so Alembic remained at 458 with the earlier DDL committed. This test builds that exact pre-459 shape, simulates the already-completed request-column ALTER, and verifies that the revised 459 both converges and is safe to retry before Alembic stamps it. """ from __future__ import annotations import importlib.util import os import sys import uuid from pathlib import Path import pytest import sqlalchemy as sa from alembic.migration import MigrationContext from alembic.operations import Operations SERVER_DIR = Path(__file__).resolve().parents[2] MIGRATION_459 = ( SERVER_DIR / "alembic/versions/459_data_access_service_account_principals.py" ) TEST_SYNC_URL = os.environ.get( "TEST_DATABASE_URL_SYNC", "mysql+pymysql://helix:helix@127.0.0.1:3306/helix", ) def _load_migration(): spec = importlib.util.spec_from_file_location("migration_459_prod", MIGRATION_459) assert spec and spec.loader module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module def _create_pre_459_schema(connection: sa.Connection) -> None: connection.execute(sa.text( "CREATE TABLE identity_persons (" "id CHAR(36) COLLATE utf8mb4_unicode_ci PRIMARY KEY" ") CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci" )) connection.execute(sa.text( "CREATE TABLE email_groups (" "id CHAR(36) COLLATE utf8mb4_unicode_ci PRIMARY KEY, " "google_group_id VARCHAR(128), email VARCHAR(255) NOT NULL, " "status VARCHAR(32) NOT NULL" ") CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci" )) connection.execute(sa.text( "CREATE TABLE data_access_requests (" "id CHAR(36) PRIMARY KEY, requester_person_id CHAR(36) NOT NULL, " "principal_email_group_id CHAR(36) NOT NULL, " "idempotency_key VARCHAR(128), " "CONSTRAINT uq_data_access_requests_idempotency_key " "UNIQUE (idempotency_key)" ") CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci" )) connection.execute(sa.text( "CREATE TABLE data_access_bindings (" "id CHAR(36) PRIMARY KEY, resource_id CHAR(36) NOT NULL, " "principal_email_group_id CHAR(36) NOT NULL, " "entitlement_id CHAR(36) NOT NULL, status VARCHAR(32) NOT NULL, " "active_semantic_key VARCHAR(64) GENERATED ALWAYS AS (" "CASE WHEN status IN ('active', 'revoked_pending') THEN " "SHA2(CONCAT_WS('|', resource_id, principal_email_group_id, " "entitlement_id), 256) ELSE NULL END" ") VIRTUAL, " "CONSTRAINT uq_data_access_bindings_active_semantic " "UNIQUE (active_semantic_key)" ") CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci" )) connection.execute(sa.text( "CREATE TABLE data_access_compiled_effects (" "id CHAR(36) PRIMARY KEY, principal_email_group_id CHAR(36) NOT NULL" ") CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci" )) connection.execute(sa.text( "CREATE TABLE data_access_effect_runs (" "id CHAR(36) PRIMARY KEY, effect_id CHAR(36) NOT NULL" ") CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci" )) connection.execute(sa.text( "INSERT INTO email_groups (id, google_group_id, email, status) VALUES " "('00000000-0000-0000-0000-000000000001', 'group-1', " "'analytics@example.com', 'active')" )) connection.execute(sa.text( "INSERT INTO data_access_requests " "(id, requester_person_id, principal_email_group_id, idempotency_key) " "VALUES ('00000000-0000-0000-0000-000000000010', " "'00000000-0000-0000-0000-000000000011', " "'00000000-0000-0000-0000-000000000001', 'request-key')" )) connection.execute(sa.text( "INSERT INTO data_access_bindings " "(id, resource_id, principal_email_group_id, entitlement_id, status) VALUES " "('00000000-0000-0000-0000-000000000020', " "'00000000-0000-0000-0000-000000000021', " "'00000000-0000-0000-0000-000000000001', " "'00000000-0000-0000-0000-000000000022', 'active')" )) connection.execute(sa.text( "INSERT INTO data_access_compiled_effects (id, principal_email_group_id) " "VALUES ('00000000-0000-0000-0000-000000000030', " "'00000000-0000-0000-0000-000000000001')" )) connection.execute(sa.text( "INSERT INTO data_access_effect_runs (id, effect_id) VALUES " "('00000000-0000-0000-0000-000000000040', " "'00000000-0000-0000-0000-000000000030')" )) def _nullable(connection: sa.Connection, table: str) -> bool: return connection.execute(sa.text( "SELECT is_nullable FROM information_schema.columns " "WHERE table_schema = DATABASE() AND table_name = :table " "AND column_name = 'principal_email_group_id'" ), {"table": table}).scalar() == "YES" def test_migration_459_resumes_prod_partial_generated_column_state() -> None: scratch = f"helix_m459_{uuid.uuid4().hex[:8]}" admin_url = os.environ.get("TEST_DATABASE_ADMIN_URL") base = admin_url.rstrip("/") if admin_url else TEST_SYNC_URL.rpartition("/")[0] admin = sa.create_engine(base + "/", isolation_level="AUTOCOMMIT") try: with admin.connect() as connection: connection.execute(sa.text( f"CREATE DATABASE {scratch} CHARACTER SET utf8mb4 " "COLLATE utf8mb4_unicode_ci" )) except sa.exc.SQLAlchemyError as exc: admin.dispose() pytest.skip(f"cannot create migration scratch schema: {exc}") engine = sa.create_engine(f"{base}/{scratch}", isolation_level="AUTOCOMMIT") try: with engine.connect() as connection: _create_pre_459_schema(connection) migration = _load_migration() context = MigrationContext.configure(connection) with Operations.context(context): # Production had completed the request ALTER, then failed on # bindings before it could make the remaining legacy columns # nullable. The preceding 459 DDL had already committed. migration._create_principal_tables(connection) migration._project_legacy_groups(connection) migration._add_principal_projection_columns(connection) migration._backfill_principal_projection(connection) connection.execute(sa.text( "ALTER TABLE data_access_requests MODIFY COLUMN " "principal_email_group_id CHAR(36) " "COLLATE utf8mb4_unicode_ci NULL" )) assert _nullable(connection, "data_access_requests") assert not _nullable(connection, "data_access_bindings") migration.upgrade() # Alembic has not yet written revision 459 if a later DDL # stage fails, so a second call must converge without dupe FK # or index failures. migration.upgrade() for table in ( "data_access_requests", "data_access_bindings", "data_access_compiled_effects", ): assert _nullable(connection, table), table generated = connection.execute(sa.text( "SELECT generation_expression FROM information_schema.columns " "WHERE table_schema = DATABASE() " "AND table_name = 'data_access_bindings' " "AND column_name = 'active_semantic_key'" )).scalar() normalized = "".join(str(generated).lower().split()).replace("`", "") assert "coalesce(principal_id,principal_email_group_id)" in normalized indexes = { index[0] for index in connection.execute(sa.text( "SELECT DISTINCT index_name FROM information_schema.statistics " "WHERE table_schema = DATABASE() " "AND table_name = 'data_access_bindings'" )) } assert "uq_data_access_bindings_active_semantic" in indexes foreign_keys = { row[0] for row in connection.execute(sa.text( "SELECT constraint_name FROM information_schema.table_constraints " "WHERE table_schema = DATABASE() " "AND constraint_type = 'FOREIGN KEY' " "AND table_name IN ('data_access_requests', " "'data_access_bindings', 'data_access_compiled_effects', " "'data_access_effect_runs')" )) } for table in ( "data_access_requests", "data_access_bindings", "data_access_compiled_effects", "data_access_effect_runs", ): assert f"fk_{table}_principal_v2" in foreign_keys finally: engine.dispose() with admin.connect() as connection: connection.execute(sa.text(f"DROP DATABASE IF EXISTS {scratch}")) admin.dispose()