370 lines
14 KiB
Python
370 lines
14 KiB
Python
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 = "be8f4d2c1a70"
|
|
_SCHEDULING_RESPONSE_SETTINGS_REVISION = "ad7e3c9b2f10"
|
|
_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"
|
|
)
|
|
}
|
|
participant_columns = {
|
|
item["name"]
|
|
for item in inspect(connection).get_columns(
|
|
"scheduling_participants"
|
|
)
|
|
}
|
|
visibility = connection.execute(
|
|
text(
|
|
"SELECT participant_visibility FROM scheduling_requests "
|
|
"WHERE id = 'request-1'"
|
|
)
|
|
).scalar_one()
|
|
response_defaults = connection.execute(
|
|
text(
|
|
"SELECT notify_on_answers, single_choice, "
|
|
"max_participants_per_option, allow_maybe, allow_comments, "
|
|
"participant_email_required, anonymous_password_protection_enabled, "
|
|
"anonymous_password_hash FROM scheduling_requests "
|
|
"WHERE id = 'request-1'"
|
|
)
|
|
).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.assertIn("max_participants_per_option", columns)
|
|
self.assertIn("response_comment", participant_columns)
|
|
self.assertIn("participation_gateway", participant_columns)
|
|
self.assertEqual(visibility, "aggregates_only")
|
|
self.assertEqual(tuple(response_defaults), (1, 0, None, 1, 0, 0, 0, None))
|
|
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()
|
|
|
|
def test_rejects_partial_existing_response_settings(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,
|
|
)
|
|
connection.execute(
|
|
text(
|
|
"ALTER TABLE scheduling_requests "
|
|
"DROP COLUMN allow_comments"
|
|
)
|
|
)
|
|
|
|
with self.assertRaisesRegex(
|
|
RuntimeError,
|
|
"partial scheduling_requests scheduling response settings",
|
|
):
|
|
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_participant_response_settings(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,
|
|
)
|
|
connection.execute(
|
|
text(
|
|
"ALTER TABLE scheduling_participants "
|
|
"DROP COLUMN participation_gateway"
|
|
)
|
|
)
|
|
|
|
with self.assertRaisesRegex(
|
|
RuntimeError,
|
|
"partial scheduling_participants scheduling response settings",
|
|
):
|
|
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_repairs_missing_participation_gateway_after_previous_head_was_stamped(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:
|
|
connection.execute(
|
|
text(
|
|
"DELETE FROM alembic_version WHERE version_num = :revision"
|
|
),
|
|
{"revision": _SCHEDULING_HEAD},
|
|
)
|
|
connection.execute(
|
|
text(
|
|
"INSERT INTO alembic_version (version_num) VALUES (:revision)"
|
|
),
|
|
{"revision": _SCHEDULING_RESPONSE_SETTINGS_REVISION},
|
|
)
|
|
connection.execute(
|
|
text(
|
|
"ALTER TABLE scheduling_participants "
|
|
"DROP COLUMN participation_gateway"
|
|
)
|
|
)
|
|
|
|
self._migrate(url)
|
|
|
|
with engine.connect() as connection:
|
|
heads = set(
|
|
MigrationContext.configure(connection).get_current_heads()
|
|
)
|
|
participant_columns = {
|
|
item["name"]
|
|
for item in inspect(connection).get_columns(
|
|
"scheduling_participants"
|
|
)
|
|
}
|
|
participant = connection.execute(
|
|
text(
|
|
"SELECT display_name, email, participation_gateway "
|
|
"FROM scheduling_participants WHERE id = 'participant-1'"
|
|
)
|
|
).one()
|
|
|
|
self.assertIn(_SCHEDULING_HEAD, heads)
|
|
self.assertIn("participation_gateway", participant_columns)
|
|
self.assertEqual(
|
|
tuple(participant),
|
|
("Ada", "ada@example.test", None),
|
|
)
|
|
finally:
|
|
engine.dispose()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|