tegwick
82ea38b180
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a049a4-ee9f-78e1-9d66-2cb0f9bea3e3
664 lines
26 KiB
Python
664 lines
26 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from alembic.migration import MigrationContext
|
|
from alembic.operations import Operations
|
|
from sqlalchemy import inspect, select, text
|
|
from sqlalchemy.exc import DBAPIError, IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
|
from api.models.decision import Decision
|
|
from api.models.domain import Domain
|
|
from api.models.managed_repo import ManagedRepo
|
|
from api.models.progress_event import ProgressEvent
|
|
from api.models.repository_rename import (
|
|
RepositoryForgeIdentity,
|
|
RepositoryRenameOperation,
|
|
RepositorySlug,
|
|
)
|
|
from api.models.sbom_snapshot import SBOMSnapshot
|
|
from api.models.task import Task
|
|
from api.models.token_event import TokenEvent
|
|
from api.models.workplan import Workplan
|
|
from tests.conftest import create_test_domain, create_test_repo
|
|
|
|
|
|
def _operation(
|
|
*,
|
|
repo_id: uuid.UUID,
|
|
identity_id: uuid.UUID,
|
|
old_slug: str,
|
|
new_slug: str,
|
|
forge_repository_id: int,
|
|
phase: str = "draft",
|
|
) -> RepositoryRenameOperation:
|
|
now = datetime.now(timezone.utc)
|
|
return RepositoryRenameOperation(
|
|
repo_id=repo_id,
|
|
forge_identity_id=identity_id,
|
|
forge_identity_state="verified",
|
|
expected_provider="forgejo",
|
|
expected_forge_instance="https://forgejo.coulomb.social",
|
|
expected_forge_owner="coulomb",
|
|
expected_forge_repository_id=forge_repository_id,
|
|
expected_source_commit="a" * 40,
|
|
expected_default_branch="main",
|
|
old_slug=old_slug,
|
|
new_slug=new_slug,
|
|
old_coordinates={"slug": old_slug},
|
|
new_coordinates={"slug": new_slug},
|
|
phase=phase,
|
|
actor="test",
|
|
phase_changed_at=now,
|
|
preflighted_at=now if phase != "draft" else None,
|
|
preflight_expires_at=(now + timedelta(minutes=30)) if phase != "draft" else None,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_repo_registration_creates_unverified_identity_and_canonical_slug(
|
|
client, test_engine
|
|
):
|
|
await create_test_domain(client)
|
|
repo = await create_test_repo(client, slug="identity-test")
|
|
factory = async_sessionmaker(
|
|
test_engine, class_=AsyncSession, expire_on_commit=False
|
|
)
|
|
|
|
async with factory() as session:
|
|
identity = await session.scalar(
|
|
select(RepositoryForgeIdentity).where(
|
|
RepositoryForgeIdentity.repo_id == uuid.UUID(repo["id"])
|
|
)
|
|
)
|
|
slug = await session.scalar(
|
|
select(RepositorySlug).where(
|
|
RepositorySlug.repo_id == uuid.UUID(repo["id"])
|
|
)
|
|
)
|
|
|
|
assert identity is not None
|
|
assert identity.verification_state == "unverified"
|
|
assert identity.provider is None
|
|
assert identity.forge_instance is None
|
|
assert identity.forge_owner is None
|
|
assert identity.forge_repository_id is None
|
|
assert slug is not None
|
|
assert (slug.slug, slug.kind, slug.protected) == (
|
|
"identity-test",
|
|
"canonical",
|
|
True,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_protected_alias_cannot_be_claimed_by_another_repository(
|
|
client, test_engine
|
|
):
|
|
await create_test_domain(client)
|
|
source = await create_test_repo(client, slug="access-engine")
|
|
other = await create_test_repo(client, slug="another-repo")
|
|
factory = async_sessionmaker(
|
|
test_engine, class_=AsyncSession, expire_on_commit=False
|
|
)
|
|
|
|
async with factory() as session:
|
|
session.add(
|
|
RepositorySlug(
|
|
repo_id=uuid.UUID(source["id"]),
|
|
slug="flex-auth",
|
|
kind="alias",
|
|
protected=True,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
response = await client.post(
|
|
"/repos/",
|
|
json={
|
|
"domain_slug": "infotech",
|
|
"slug": "flex-auth",
|
|
"name": "Conflicting repository",
|
|
},
|
|
)
|
|
assert response.status_code == 409
|
|
|
|
async with factory() as session:
|
|
session.add(
|
|
RepositorySlug(
|
|
repo_id=uuid.UUID(other["id"]),
|
|
slug="flex-auth",
|
|
kind="alias",
|
|
protected=True,
|
|
)
|
|
)
|
|
with pytest.raises(IntegrityError):
|
|
await session.commit()
|
|
|
|
async with factory() as session:
|
|
session.add(
|
|
RepositorySlug(
|
|
repo_id=uuid.UUID(source["id"]),
|
|
slug="second-canonical",
|
|
kind="canonical",
|
|
protected=True,
|
|
)
|
|
)
|
|
with pytest.raises(IntegrityError):
|
|
await session.commit()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_forge_identity_is_unique_and_unverified_repo_cannot_open_operation(
|
|
client, test_engine
|
|
):
|
|
await create_test_domain(client)
|
|
first = await create_test_repo(client, slug="first-repo")
|
|
second = await create_test_repo(client, slug="second-repo")
|
|
factory = async_sessionmaker(
|
|
test_engine, class_=AsyncSession, expire_on_commit=False
|
|
)
|
|
now = datetime.now(timezone.utc)
|
|
|
|
async with factory() as session:
|
|
first_identity = await session.scalar(
|
|
select(RepositoryForgeIdentity).where(
|
|
RepositoryForgeIdentity.repo_id == uuid.UUID(first["id"])
|
|
)
|
|
)
|
|
assert first_identity is not None
|
|
first_identity.provider = "forgejo"
|
|
first_identity.forge_instance = "https://forgejo.coulomb.social"
|
|
first_identity.forge_owner = "coulomb"
|
|
first_identity.forge_repository_id = 42
|
|
first_identity.verification_state = "verified"
|
|
first_identity.verified_at = now
|
|
first_identity.verified_by = "test"
|
|
first_identity.verification_evidence = {"source": "forge-api"}
|
|
await session.commit()
|
|
|
|
async with factory() as session:
|
|
second_identity = await session.scalar(
|
|
select(RepositoryForgeIdentity).where(
|
|
RepositoryForgeIdentity.repo_id == uuid.UUID(second["id"])
|
|
)
|
|
)
|
|
assert second_identity is not None
|
|
session.add(
|
|
_operation(
|
|
repo_id=uuid.UUID(second["id"]),
|
|
identity_id=second_identity.id,
|
|
old_slug="second-repo",
|
|
new_slug="second-repo-renamed",
|
|
forge_repository_id=43,
|
|
)
|
|
)
|
|
with pytest.raises(IntegrityError):
|
|
await session.commit()
|
|
|
|
async with factory() as session:
|
|
second_identity = await session.scalar(
|
|
select(RepositoryForgeIdentity).where(
|
|
RepositoryForgeIdentity.repo_id == uuid.UUID(second["id"])
|
|
)
|
|
)
|
|
assert second_identity is not None
|
|
second_identity.provider = "forgejo"
|
|
second_identity.forge_instance = "https://forgejo.coulomb.social"
|
|
second_identity.forge_owner = "coulomb"
|
|
second_identity.forge_repository_id = 42
|
|
second_identity.verification_state = "verified"
|
|
second_identity.verified_at = now
|
|
second_identity.verified_by = "test"
|
|
with pytest.raises(IntegrityError):
|
|
await session.commit()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_only_one_active_operation_can_hold_a_repo_or_target_slug(
|
|
client, test_engine
|
|
):
|
|
await create_test_domain(client)
|
|
first = await create_test_repo(client, slug="first-active")
|
|
second = await create_test_repo(client, slug="second-active")
|
|
factory = async_sessionmaker(
|
|
test_engine, class_=AsyncSession, expire_on_commit=False
|
|
)
|
|
now = datetime.now(timezone.utc)
|
|
identities: dict[str, RepositoryForgeIdentity] = {}
|
|
|
|
async with factory() as session:
|
|
for repo, forge_repository_id in ((first, 101), (second, 102)):
|
|
identity = await session.scalar(
|
|
select(RepositoryForgeIdentity).where(
|
|
RepositoryForgeIdentity.repo_id == uuid.UUID(repo["id"])
|
|
)
|
|
)
|
|
assert identity is not None
|
|
identity.provider = "forgejo"
|
|
identity.forge_instance = "https://forgejo.coulomb.social"
|
|
identity.forge_owner = "coulomb"
|
|
identity.forge_repository_id = forge_repository_id
|
|
identity.verification_state = "verified"
|
|
identity.verified_at = now
|
|
identity.verified_by = "test"
|
|
identities[repo["slug"]] = identity
|
|
await session.commit()
|
|
|
|
async with factory() as session:
|
|
session.add(
|
|
_operation(
|
|
repo_id=uuid.UUID(first["id"]),
|
|
identity_id=identities["first-active"].id,
|
|
old_slug="first-active",
|
|
new_slug="shared-target",
|
|
forge_repository_id=101,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
async with factory() as session:
|
|
session.add(
|
|
_operation(
|
|
repo_id=uuid.UUID(first["id"]),
|
|
identity_id=identities["first-active"].id,
|
|
old_slug="first-active",
|
|
new_slug="another-target",
|
|
forge_repository_id=101,
|
|
)
|
|
)
|
|
with pytest.raises(IntegrityError):
|
|
await session.commit()
|
|
|
|
async with factory() as session:
|
|
session.add(
|
|
_operation(
|
|
repo_id=uuid.UUID(second["id"]),
|
|
identity_id=identities["second-active"].id,
|
|
old_slug="second-active",
|
|
new_slug="shared-target",
|
|
forge_repository_id=102,
|
|
)
|
|
)
|
|
with pytest.raises(IntegrityError):
|
|
await session.commit()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_flex_auth_preflight_representation_preserves_all_existing_ids(test_engine):
|
|
factory = async_sessionmaker(
|
|
test_engine, class_=AsyncSession, expire_on_commit=False
|
|
)
|
|
repo_id = uuid.UUID("fda8ad85-a7d7-4055-8f21-902a533e59df")
|
|
domain_id = uuid.uuid4()
|
|
workplan_id = uuid.uuid4()
|
|
task_id = uuid.uuid4()
|
|
progress_id = uuid.uuid4()
|
|
decision_id = uuid.uuid4()
|
|
token_id = uuid.uuid4()
|
|
sbom_id = uuid.uuid4()
|
|
identity_id = uuid.uuid4()
|
|
now = datetime.now(timezone.utc)
|
|
|
|
async with factory() as session:
|
|
session.add_all(
|
|
[
|
|
Domain(id=domain_id, slug="net-kingdom", name="Net Kingdom"),
|
|
ManagedRepo(
|
|
id=repo_id,
|
|
domain_id=domain_id,
|
|
slug="flex-auth",
|
|
name="Flex Auth",
|
|
),
|
|
]
|
|
)
|
|
await session.flush()
|
|
session.add_all(
|
|
[
|
|
RepositoryForgeIdentity(
|
|
id=identity_id,
|
|
repo_id=repo_id,
|
|
provider="forgejo",
|
|
forge_instance="https://forgejo.coulomb.social",
|
|
forge_owner="coulomb",
|
|
forge_repository_id=42,
|
|
verification_state="verified",
|
|
verified_at=now,
|
|
verified_by="test",
|
|
verification_evidence={"source": "forge-api", "repository_id": 42},
|
|
),
|
|
RepositorySlug(
|
|
repo_id=repo_id,
|
|
slug="flex-auth",
|
|
kind="canonical",
|
|
protected=True,
|
|
),
|
|
Workplan(
|
|
id=workplan_id,
|
|
repo_id=repo_id,
|
|
slug="flex-wp-0001",
|
|
title="Existing work",
|
|
),
|
|
TokenEvent(
|
|
id=token_id,
|
|
repo_id=repo_id,
|
|
tokens_in=10,
|
|
tokens_out=5,
|
|
),
|
|
SBOMSnapshot(
|
|
id=sbom_id,
|
|
repo_id=repo_id,
|
|
snapshot_at=now,
|
|
entry_count=0,
|
|
created_at=now,
|
|
),
|
|
]
|
|
)
|
|
await session.flush()
|
|
session.add_all(
|
|
[
|
|
Task(
|
|
id=task_id,
|
|
workplan_id=workplan_id,
|
|
record_id="FLEX-WP-0001-T01",
|
|
title="Existing task",
|
|
),
|
|
Decision(
|
|
id=decision_id,
|
|
workplan_id=workplan_id,
|
|
title="Existing decision",
|
|
),
|
|
]
|
|
)
|
|
await session.flush()
|
|
session.add(
|
|
ProgressEvent(
|
|
id=progress_id,
|
|
workplan_id=workplan_id,
|
|
task_id=task_id,
|
|
decision_id=decision_id,
|
|
event_type="note",
|
|
summary="Existing evidence",
|
|
)
|
|
)
|
|
session.add(
|
|
_operation(
|
|
repo_id=repo_id,
|
|
identity_id=identity_id,
|
|
old_slug="flex-auth",
|
|
new_slug="access-engine",
|
|
forge_repository_id=42,
|
|
phase="preflighted",
|
|
)
|
|
)
|
|
await session.flush()
|
|
|
|
assert await session.scalar(
|
|
select(ManagedRepo.slug).where(ManagedRepo.id == repo_id)
|
|
) == "flex-auth"
|
|
assert await session.scalar(
|
|
select(RepositoryRenameOperation.expected_forge_repository_id).where(
|
|
RepositoryRenameOperation.repo_id == repo_id
|
|
)
|
|
) == 42
|
|
assert await session.scalar(select(Workplan.id)) == workplan_id
|
|
assert await session.scalar(select(Task.id)) == task_id
|
|
assert await session.scalar(select(ProgressEvent.id)) == progress_id
|
|
assert await session.scalar(select(Decision.id)) == decision_id
|
|
assert await session.scalar(select(TokenEvent.id)) == token_id
|
|
assert await session.scalar(select(SBOMSnapshot.id)) == sbom_id
|
|
|
|
# No commit: this proves representability without touching either live
|
|
# State Hub or Forgejo and leaves the test database unchanged.
|
|
await session.rollback()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_migration_upgrade_and_downgrade_are_additive(test_engine):
|
|
migration = importlib.import_module(
|
|
"migrations.versions.f3c4d5e6a7b8_repository_rename_identity"
|
|
)
|
|
rollback_migration = importlib.import_module(
|
|
"migrations.versions.a4d5e6f7b8c9_allow_guarded_slug_rollback"
|
|
)
|
|
schema = f"rename_migration_{uuid.uuid4().hex}"
|
|
repo_ids = [uuid.uuid4(), uuid.uuid4()]
|
|
sentinel_ids = {
|
|
"workplans": uuid.uuid4(),
|
|
"tasks": uuid.uuid4(),
|
|
"progress_events": uuid.uuid4(),
|
|
"decisions": uuid.uuid4(),
|
|
"token_events": uuid.uuid4(),
|
|
"sbom_snapshots": uuid.uuid4(),
|
|
}
|
|
|
|
async with test_engine.begin() as connection:
|
|
def run(sync_connection):
|
|
quoted_schema = sync_connection.dialect.identifier_preparer.quote(schema)
|
|
sync_connection.exec_driver_sql(f"CREATE SCHEMA {quoted_schema}")
|
|
sync_connection.exec_driver_sql(
|
|
f"SET LOCAL search_path TO {quoted_schema}, public"
|
|
)
|
|
sync_connection.exec_driver_sql(
|
|
"CREATE TABLE managed_repos "
|
|
"(id uuid PRIMARY KEY, slug varchar(100) NOT NULL UNIQUE, name text)"
|
|
)
|
|
for table_name in sentinel_ids:
|
|
sync_connection.exec_driver_sql(
|
|
f"CREATE TABLE {table_name} (id uuid PRIMARY KEY, marker text)"
|
|
)
|
|
sync_connection.execute(
|
|
text(
|
|
"INSERT INTO managed_repos (id, slug, name) "
|
|
"VALUES (:first, 'flex-auth', 'Flex Auth'), "
|
|
"(:second, 'state-hub', 'State Hub')"
|
|
),
|
|
{"first": repo_ids[0], "second": repo_ids[1]},
|
|
)
|
|
for table_name, sentinel_id in sentinel_ids.items():
|
|
sync_connection.execute(
|
|
text(f"INSERT INTO {table_name} (id, marker) VALUES (:id, 'keep')"),
|
|
{"id": sentinel_id},
|
|
)
|
|
|
|
original_op = migration.op
|
|
original_rollback_op = rollback_migration.op
|
|
migration.op = Operations(MigrationContext.configure(sync_connection))
|
|
rollback_migration.op = migration.op
|
|
try:
|
|
migration.upgrade()
|
|
rollback_migration.upgrade()
|
|
tables = set(inspect(sync_connection).get_table_names(schema=schema))
|
|
assert {
|
|
"repository_forge_identities",
|
|
"repository_rename_operations",
|
|
"repository_slugs",
|
|
}.issubset(tables)
|
|
identity_rows = sync_connection.execute(
|
|
text(
|
|
"SELECT repo_id, verification_state, provider, forge_repository_id "
|
|
"FROM repository_forge_identities ORDER BY repo_id"
|
|
)
|
|
).all()
|
|
assert len(identity_rows) == 2
|
|
assert all(
|
|
row.verification_state == "unverified"
|
|
and row.provider is None
|
|
and row.forge_repository_id is None
|
|
for row in identity_rows
|
|
)
|
|
slug_rows = sync_connection.execute(
|
|
text(
|
|
"SELECT repo_id, slug, kind, protected "
|
|
"FROM repository_slugs ORDER BY slug"
|
|
)
|
|
).all()
|
|
assert {(row.slug, row.kind, row.protected) for row in slug_rows} == {
|
|
("flex-auth", "canonical", True),
|
|
("state-hub", "canonical", True),
|
|
}
|
|
identity_id = sync_connection.execute(
|
|
text(
|
|
"UPDATE repository_forge_identities SET "
|
|
"provider = 'forgejo', "
|
|
"forge_instance = 'https://forgejo.coulomb.social', "
|
|
"forge_owner = 'coulomb', forge_repository_id = 42, "
|
|
"verification_state = 'verified', verified_at = now(), "
|
|
"verified_by = 'migration-test' "
|
|
"WHERE repo_id = :repo_id RETURNING id"
|
|
),
|
|
{"repo_id": repo_ids[0]},
|
|
).scalar_one()
|
|
with pytest.raises(DBAPIError):
|
|
with sync_connection.begin_nested():
|
|
sync_connection.execute(
|
|
text(
|
|
"UPDATE repository_forge_identities "
|
|
"SET forge_repository_id = 43 WHERE id = :id"
|
|
),
|
|
{"id": identity_id},
|
|
)
|
|
with pytest.raises(DBAPIError):
|
|
with sync_connection.begin_nested():
|
|
sync_connection.execute(
|
|
text(
|
|
"DELETE FROM repository_slugs "
|
|
"WHERE repo_id = :repo_id"
|
|
),
|
|
{"repo_id": repo_ids[0]},
|
|
)
|
|
|
|
operation_id = uuid.uuid4()
|
|
sync_connection.execute(
|
|
text(
|
|
"INSERT INTO repository_rename_operations ("
|
|
"id, repo_id, forge_identity_id, forge_identity_state, "
|
|
"expected_provider, expected_forge_instance, "
|
|
"expected_forge_owner, expected_forge_repository_id, "
|
|
"expected_source_commit, expected_default_branch, "
|
|
"old_slug, new_slug, old_coordinates, new_coordinates, "
|
|
"phase, actor, phase_changed_at, evidence, created_at, updated_at"
|
|
") VALUES ("
|
|
":id, :repo_id, :identity_id, 'verified', 'forgejo', "
|
|
"'https://forgejo.coulomb.social', 'coulomb', 42, "
|
|
":commit, 'main', 'flex-auth', 'access-engine', "
|
|
"'{\"slug\": \"flex-auth\"}'::jsonb, "
|
|
"'{\"slug\": \"access-engine\"}'::jsonb, "
|
|
"'draft', 'migration-test', now(), '{}'::jsonb, now(), now()"
|
|
")"
|
|
),
|
|
{
|
|
"id": operation_id,
|
|
"repo_id": repo_ids[0],
|
|
"identity_id": identity_id,
|
|
"commit": "a" * 40,
|
|
},
|
|
)
|
|
sync_connection.execute(
|
|
text(
|
|
"UPDATE repository_rename_operations SET "
|
|
"phase = 'preflighted', preflighted_at = now(), "
|
|
"phase_changed_at = now() WHERE id = :id"
|
|
),
|
|
{"id": operation_id},
|
|
)
|
|
with pytest.raises(DBAPIError):
|
|
with sync_connection.begin_nested():
|
|
sync_connection.execute(
|
|
text(
|
|
"UPDATE repository_rename_operations "
|
|
"SET old_slug = 'rewritten' WHERE id = :id"
|
|
),
|
|
{"id": operation_id},
|
|
)
|
|
# A forward cutover may make the old canonical an alias. T03
|
|
# permits restoring it only while its own operation is in the
|
|
# durable rollback-preflight phase.
|
|
sync_connection.execute(
|
|
text(
|
|
"UPDATE repository_slugs SET kind = 'alias', "
|
|
"source_operation_id = :operation_id "
|
|
"WHERE repo_id = :repo_id AND slug = 'flex-auth'"
|
|
),
|
|
{"operation_id": operation_id, "repo_id": repo_ids[0]},
|
|
)
|
|
sync_connection.execute(
|
|
text(
|
|
"INSERT INTO repository_slugs "
|
|
"(id, repo_id, slug, kind, protected, source_operation_id, created_at, updated_at) "
|
|
"VALUES (gen_random_uuid(), :repo_id, 'access-engine', 'canonical', true, "
|
|
":operation_id, now(), now())"
|
|
),
|
|
{"operation_id": operation_id, "repo_id": repo_ids[0]},
|
|
)
|
|
sync_connection.execute(
|
|
text(
|
|
"UPDATE repository_slugs SET kind = 'alias' "
|
|
"WHERE repo_id = :repo_id AND slug = 'access-engine'"
|
|
),
|
|
{"repo_id": repo_ids[0]},
|
|
)
|
|
with pytest.raises(DBAPIError):
|
|
with sync_connection.begin_nested():
|
|
sync_connection.execute(
|
|
text(
|
|
"UPDATE repository_slugs SET kind = 'canonical' "
|
|
"WHERE repo_id = :repo_id AND slug = 'flex-auth'"
|
|
),
|
|
{"repo_id": repo_ids[0]},
|
|
)
|
|
sync_connection.execute(
|
|
text(
|
|
"UPDATE repository_rename_operations SET "
|
|
"phase = 'rollback-preflight', phase_changed_at = now() "
|
|
"WHERE id = :id"
|
|
),
|
|
{"id": operation_id},
|
|
)
|
|
sync_connection.execute(
|
|
text(
|
|
"UPDATE repository_slugs SET kind = 'canonical' "
|
|
"WHERE repo_id = :repo_id AND slug = 'flex-auth'"
|
|
),
|
|
{"repo_id": repo_ids[0]},
|
|
)
|
|
with pytest.raises(DBAPIError):
|
|
with sync_connection.begin_nested():
|
|
sync_connection.execute(
|
|
text(
|
|
"DELETE FROM repository_rename_operations "
|
|
"WHERE id = :id"
|
|
),
|
|
{"id": operation_id},
|
|
)
|
|
assert sync_connection.execute(
|
|
text("SELECT id, slug FROM managed_repos ORDER BY slug")
|
|
).all() == sorted(
|
|
[(repo_ids[0], "flex-auth"), (repo_ids[1], "state-hub")],
|
|
key=lambda item: item[1],
|
|
)
|
|
for table_name, sentinel_id in sentinel_ids.items():
|
|
assert sync_connection.execute(
|
|
text(f"SELECT id, marker FROM {table_name}")
|
|
).one() == (sentinel_id, "keep")
|
|
|
|
rollback_migration.downgrade()
|
|
migration.downgrade()
|
|
tables = set(inspect(sync_connection).get_table_names(schema=schema))
|
|
assert "repository_forge_identities" not in tables
|
|
assert "repository_rename_operations" not in tables
|
|
assert "repository_slugs" not in tables
|
|
for table_name, sentinel_id in sentinel_ids.items():
|
|
assert sync_connection.execute(
|
|
text(f"SELECT id, marker FROM {table_name}")
|
|
).one() == (sentinel_id, "keep")
|
|
finally:
|
|
migration.op = original_op
|
|
rollback_migration.op = original_rollback_op
|
|
sync_connection.exec_driver_sql("SET LOCAL search_path TO public")
|
|
sync_connection.exec_driver_sql(f"DROP SCHEMA {quoted_schema} CASCADE")
|
|
|
|
await connection.run_sync(run)
|