govoplan-scheduling/tests/test_migrations.py

from __future__ import annotations

import tempfile
import unittest
from datetime import datetime
from pathlib import Path

from alembic.runtime.migration import MigrationContext
from sqlalchemy import create_engine, func, inspect, select, text
from sqlalchemy.orm import Session

from govoplan_core.db.migrations import migrate_database
from govoplan_poll.backend.manifest import get_manifest as get_poll_manifest
from govoplan_scheduling.backend.db.models import (
    SchedulingCandidateSlot,
    SchedulingNotification,
    SchedulingParticipant,
    SchedulingRequest,
)
from govoplan_scheduling.backend.manifest import get_manifest as get_scheduling_manifest


_SCHEDULING_HEAD = "9c2f4a7d1e6b"
_ENABLED_MODULES = ("poll", "scheduling")
_MANIFEST_FACTORIES = (get_poll_manifest, get_scheduling_manifest)


class SchedulingMigrationTests(unittest.TestCase):
    def _migrate(self, url: str) -> None:
        migrate_database(
            database_url=url,
            enabled_modules=_ENABLED_MODULES,
            manifest_factories=_MANIFEST_FACTORIES,
        )

    @staticmethod
    def _remove_scheduling_revision(
        connection, *, remove_privacy_column: bool = True
    ) -> None:
        connection.execute(
            text("DELETE FROM alembic_version WHERE version_num = :revision"),
            {"revision": _SCHEDULING_HEAD},
        )
        if remove_privacy_column:
            connection.execute(
                text(
                    "ALTER TABLE scheduling_requests DROP COLUMN participant_visibility"
                )
            )

    @staticmethod
    def _seed_scheduling_rows(engine) -> None:
        with Session(engine) as session:
            request = SchedulingRequest(
                id="request-1",
                tenant_id="tenant-1",
                title="Migration adoption probe",
            )
            request.slots.append(
                SchedulingCandidateSlot(
                    id="slot-1",
                    tenant_id="tenant-1",
                    label="First slot",
                    start_at=datetime(2026, 7, 21, 8, 0),
                    end_at=datetime(2026, 7, 21, 9, 0),
                )
            )
            request.participants.append(
                SchedulingParticipant(
                    id="participant-1",
                    tenant_id="tenant-1",
                    display_name="Ada",
                    email="ada@example.test",
                )
            )
            session.add(request)
            session.flush()
            session.add(
                SchedulingNotification(
                    id="notification-1",
                    tenant_id="tenant-1",
                    request_id=request.id,
                    event_kind="invitation",
                    recipient="ada@example.test",
                )
            )
            session.commit()

    def test_adopts_complete_unversioned_v018_schema_and_preserves_rows(self) -> None:
        with tempfile.TemporaryDirectory(
            prefix="govoplan-scheduling-migration-"
        ) as directory:
            url = f"sqlite:///{Path(directory) / 'scheduling.db'}"
            self._migrate(url)
            engine = create_engine(url)
            try:
                self._seed_scheduling_rows(engine)
                with engine.begin() as connection:
                    self._remove_scheduling_revision(connection)

                self._migrate(url)

                with engine.connect() as connection:
                    heads = set(
                        MigrationContext.configure(connection).get_current_heads()
                    )
                    columns = {
                        item["name"]
                        for item in inspect(connection).get_columns(
                            "scheduling_requests"
                        )
                    }
                    visibility = connection.execute(
                        text(
                            "SELECT participant_visibility FROM scheduling_requests "
                            "WHERE id = 'request-1'"
                        )
                    ).scalar_one()
                    counts = {
                        model.__tablename__: connection.execute(
                            select(func.count()).select_from(model)
                        ).scalar_one()
                        for model in (
                            SchedulingRequest,
                            SchedulingCandidateSlot,
                            SchedulingParticipant,
                            SchedulingNotification,
                        )
                    }

                self.assertIn(_SCHEDULING_HEAD, heads)
                self.assertIn("participant_visibility", columns)
                self.assertEqual(visibility, "aggregates_only")
                self.assertEqual(set(counts.values()), {1})
            finally:
                engine.dispose()

    def test_adopts_already_present_compatible_privacy_column(self) -> None:
        with tempfile.TemporaryDirectory(
            prefix="govoplan-scheduling-migration-"
        ) as directory:
            url = f"sqlite:///{Path(directory) / 'scheduling.db'}"
            self._migrate(url)
            engine = create_engine(url)
            try:
                with engine.begin() as connection:
                    self._remove_scheduling_revision(
                        connection, remove_privacy_column=False
                    )

                self._migrate(url)

                with engine.connect() as connection:
                    heads = set(
                        MigrationContext.configure(connection).get_current_heads()
                    )
                    matching_columns = [
                        item
                        for item in inspect(connection).get_columns(
                            "scheduling_requests"
                        )
                        if item["name"] == "participant_visibility"
                    ]

                self.assertIn(_SCHEDULING_HEAD, heads)
                self.assertEqual(len(matching_columns), 1)
            finally:
                engine.dispose()

    def test_rejects_incomplete_existing_baseline(self) -> None:
        with tempfile.TemporaryDirectory(
            prefix="govoplan-scheduling-migration-"
        ) as directory:
            url = f"sqlite:///{Path(directory) / 'scheduling.db'}"
            self._migrate(url)
            engine = create_engine(url)
            try:
                with engine.begin() as connection:
                    self._remove_scheduling_revision(connection)
                    connection.execute(text("DROP INDEX ix_scheduling_requests_status"))

                with self.assertRaisesRegex(
                    RuntimeError,
                    "scheduling_requests missing indexes: ix_scheduling_requests_status",
                ):
                    self._migrate(url)

                with engine.connect() as connection:
                    heads = set(
                        MigrationContext.configure(connection).get_current_heads()
                    )
                self.assertNotIn(_SCHEDULING_HEAD, heads)
            finally:
                engine.dispose()

    def test_rejects_partial_existing_calendar_extension(self) -> None:
        with tempfile.TemporaryDirectory(
            prefix="govoplan-scheduling-migration-"
        ) as directory:
            url = f"sqlite:///{Path(directory) / 'scheduling.db'}"
            self._migrate(url)
            engine = create_engine(url)
            try:
                with engine.begin() as connection:
                    self._remove_scheduling_revision(connection)
                    connection.execute(text("DROP TABLE scheduling_notifications"))

                with self.assertRaisesRegex(
                    RuntimeError,
                    "missing table: scheduling_notifications",
                ):
                    self._migrate(url)

                with engine.connect() as connection:
                    heads = set(
                        MigrationContext.configure(connection).get_current_heads()
                    )
                self.assertNotIn(_SCHEDULING_HEAD, heads)
            finally:
                engine.dispose()


if __name__ == "__main__":
    unittest.main()