fix(migrations): adopt existing scheduling schema
This commit is contained in:
224
tests/test_migrations.py
Normal file
224
tests/test_migrations.py
Normal file
@@ -0,0 +1,224 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user