state-hub/tests/test_repository_rename_persistence.py

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"
    )
    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
            migration.op = Operations(MigrationContext.configure(sync_connection))
            try:
                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},
                        )
                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")

                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
                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)