core-api/tests/test_book_migration.py

"""
Tests for Book migration 005.

Runs against `core_api_test` only (see conftest.py — skip when
TEST_DATABASE_URL is unset, hard failure if it resolves to `core_api`).
Self-contained: starts and ends at head regardless of what ran before it,
so test order never matters for the schema the rest of the suite needs.
"""
import pytest
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.pool import NullPool

from tests.conftest import run_alembic


async def _table_exists(url: str, table: str) -> bool:
    asyncpg_url = url.replace("postgresql://", "postgresql+asyncpg://", 1)
    engine = create_async_engine(asyncpg_url, poolclass=NullPool)
    try:
        async with engine.connect() as conn:
            result = await conn.execute(
                text("SELECT to_regclass(:name) IS NOT NULL AS present"),
                {"name": table},
            )
            return bool(result.scalar())
    finally:
        await engine.dispose()


async def _scalar(url: str, query: str, **params) -> object:
    asyncpg_url = url.replace("postgresql://", "postgresql+asyncpg://", 1)
    engine = create_async_engine(asyncpg_url, poolclass=NullPool)
    try:
        async with engine.connect() as conn:
            result = await conn.execute(text(query), params)
            return result.scalar()
    finally:
        await engine.dispose()


class TestMigration005RoundTrip:
    """Migration 005 must upgrade and downgrade cleanly (dossier verification #2)."""

    @pytest.mark.asyncio
    async def test_upgrade_creates_book_tables(self, test_database_url):
        result = run_alembic(test_database_url, "upgrade", "head")
        assert result.returncode == 0, result.stderr

        for table in (
            "book_records", "book_tags", "book_record_links", "book_attachments",
            "book_events", "book_tenant_counters",
        ):
            assert await _table_exists(test_database_url, table), f"{table} missing after upgrade"

    @pytest.mark.asyncio
    async def test_downgrade_then_upgrade_round_trips_cleanly(self, test_database_url):
        # Baseline: make sure we start at head.
        baseline = run_alembic(test_database_url, "upgrade", "head")
        assert baseline.returncode == 0, baseline.stderr

        down = run_alembic(test_database_url, "downgrade", "004")
        assert down.returncode == 0, down.stderr
        assert not await _table_exists(test_database_url, "book_records")
        assert not await _table_exists(test_database_url, "book_events")
        assert not await _table_exists(test_database_url, "book_tenant_counters")

        # Leave the schema at head for the rest of the suite.
        up = run_alembic(test_database_url, "upgrade", "005")
        assert up.returncode == 0, up.stderr
        assert await _table_exists(test_database_url, "book_records")
        assert await _table_exists(test_database_url, "book_events")
        assert await _table_exists(test_database_url, "book_tenant_counters")


class TestMigration005TenantAndNumber:
    """core-api T-7: owner_tenant/number amended onto 005 before its first release."""

    @pytest.mark.asyncio
    async def test_book_records_owner_tenant_and_number_are_not_null(self, test_database_url):
        run_alembic(test_database_url, "upgrade", "head")
        for column in ("owner_tenant", "number"):
            nullable = await _scalar(
                test_database_url,
                "SELECT is_nullable FROM information_schema.columns "
                "WHERE table_name = 'book_records' AND column_name = :column",
                column=column,
            )
            assert nullable == "NO", f"book_records.{column} must be NOT NULL, got {nullable!r}"

    @pytest.mark.asyncio
    async def test_book_records_has_tenant_number_unique_constraint(self, test_database_url):
        run_alembic(test_database_url, "upgrade", "head")
        found = await _scalar(
            test_database_url,
            "SELECT conname FROM pg_constraint WHERE conname = 'uq_book_records_tenant_number'",
        )
        assert found == "uq_book_records_tenant_number"

    @pytest.mark.asyncio
    async def test_book_records_has_tenant_kind_status_index(self, test_database_url):
        run_alembic(test_database_url, "upgrade", "head")
        found = await _scalar(
            test_database_url,
            "SELECT indexname FROM pg_indexes WHERE indexname = 'ix_book_records_tenant_kind_status'",
        )
        assert found == "ix_book_records_tenant_kind_status"


class TestBookEventsAppendOnly:
    """The trigger must block UPDATE and DELETE on book_events, no exceptions."""

    @pytest.mark.asyncio
    async def test_update_on_book_events_raises(self, clean_book_tables):
        from src.domains.book.models import BookEvent, BookRecord

        session = clean_book_tables
        record = BookRecord(
            owner_tenant="jpmschweitzer", number=1,
            kind="note", body="trigger test", status="active", created_by="test-harness",
        )
        session.add(record)
        await session.flush()
        event = BookEvent(record_id=record.id, identity="test-harness", client=None, action="add", diff={})
        session.add(event)
        await session.commit()

        with pytest.raises(Exception) as exc_info:
            await session.execute(
                text("UPDATE book_events SET action = 'tampered' WHERE id = :id").bindparams(id=event.id)
            )
            await session.commit()
        assert "append-only" in str(exc_info.value).lower() or "book_events" in str(exc_info.value)
        await session.rollback()

    @pytest.mark.asyncio
    async def test_delete_on_book_events_raises(self, clean_book_tables):
        from src.domains.book.models import BookEvent, BookRecord

        session = clean_book_tables
        record = BookRecord(
            owner_tenant="jpmschweitzer", number=1,
            kind="note", body="trigger test", status="active", created_by="test-harness",
        )
        session.add(record)
        await session.flush()
        event = BookEvent(record_id=record.id, identity="test-harness", client=None, action="add", diff={})
        session.add(event)
        await session.commit()

        with pytest.raises(Exception) as exc_info:
            await session.execute(
                text("DELETE FROM book_events WHERE id = :id").bindparams(id=event.id)
            )
            await session.commit()
        assert "append-only" in str(exc_info.value).lower() or "book_events" in str(exc_info.value)
        await session.rollback()