64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
"""Runtime migration upgrade/downgrade verification."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
|
|
from noteflow.domain.entities import Meeting
|
|
from noteflow.infrastructure.persistence.unit_of_work import SqlAlchemyUnitOfWork
|
|
from support.db_utils import (
|
|
cleanup_test_schema,
|
|
create_test_engine,
|
|
create_test_session_factory,
|
|
get_or_create_container,
|
|
)
|
|
|
|
|
|
def _alembic_config() -> Config:
|
|
config = Config("alembic.ini")
|
|
config.set_main_option(
|
|
"script_location",
|
|
"src/noteflow/infrastructure/persistence/migrations",
|
|
)
|
|
return config
|
|
|
|
|
|
async def _reset_schema(database_url: str) -> None:
|
|
engine = create_test_engine(database_url)
|
|
async with engine.begin() as conn:
|
|
await cleanup_test_schema(conn)
|
|
await engine.dispose()
|
|
|
|
|
|
async def _crud_roundtrip(database_url: str, meetings_dir: Path) -> None:
|
|
engine = create_test_engine(database_url)
|
|
session_factory = create_test_session_factory(engine)
|
|
|
|
async with SqlAlchemyUnitOfWork(session_factory, meetings_dir) as uow:
|
|
meeting = Meeting.create(title="Migration roundtrip")
|
|
await uow.meetings.create(meeting)
|
|
await uow.commit()
|
|
|
|
async with SqlAlchemyUnitOfWork(session_factory, meetings_dir) as uow:
|
|
loaded = await uow.meetings.get(meeting.id)
|
|
assert loaded is not None, "Expected meeting to persist after migration upgrade"
|
|
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_migrations_upgrade_and_downgrade(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Upgrade to head, run CRUD, then downgrade to base without errors."""
|
|
_, database_url = get_or_create_container()
|
|
monkeypatch.setenv("NOTEFLOW_DATABASE_URL", database_url)
|
|
|
|
asyncio.run(_reset_schema(database_url))
|
|
command.upgrade(_alembic_config(), "head")
|
|
asyncio.run(_crud_roundtrip(database_url, tmp_path))
|
|
command.downgrade(_alembic_config(), "base")
|