134 lines
5.0 KiB
Python
134 lines
5.0 KiB
Python
"""Tests for the Alembic migration chain, and specifically for
|
|
000_base_schema.py (issue #6): a genuinely empty database has no starting
|
|
point before it, since 001-013 all assume tables 001 was written against
|
|
already exist.
|
|
|
|
These run the real `alembic upgrade head` (via `alembic.command`), not the
|
|
async app engine: `migrations/env.py` builds its own engine from
|
|
`settings.database_url`, so `Settings.database_url` is patched at the class
|
|
level (a plain instance monkeypatch cannot override a `@property`) to point
|
|
at a throwaway SQLite file for the duration of each test.
|
|
"""
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
from sqlalchemy import create_engine, inspect as sa_inspect, text
|
|
|
|
from src.backend.config import Settings
|
|
from src.backend.database import Base
|
|
|
|
_ALEMBIC_INI = (
|
|
Path(__file__).resolve().parents[1] / "migrations" / "alembic.ini"
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sqlite_database_url(tmp_path, monkeypatch):
|
|
"""Point settings.database_url (and therefore migrations/env.py) at a
|
|
throwaway SQLite file, and return its async URL."""
|
|
db_path = tmp_path / "migration_test.db"
|
|
async_url = f"sqlite+aiosqlite:///{db_path}"
|
|
monkeypatch.setattr(Settings, "database_url", property(lambda self: async_url))
|
|
return db_path, async_url
|
|
|
|
|
|
def _alembic_config() -> Config:
|
|
return Config(str(_ALEMBIC_INI))
|
|
|
|
|
|
def _inspect(db_path: Path):
|
|
engine = create_engine(f"sqlite:///{db_path}")
|
|
try:
|
|
return sa_inspect(engine)
|
|
finally:
|
|
engine.dispose()
|
|
|
|
|
|
class TestUpgradeHeadFromEmpty:
|
|
"""Full chain on a genuinely empty SQLite database (#6)."""
|
|
|
|
def test_upgrade_head_succeeds_on_empty_database(self, sqlite_database_url):
|
|
db_path, _ = sqlite_database_url
|
|
command.upgrade(_alembic_config(), "head") # must not raise
|
|
|
|
insp = _inspect(db_path)
|
|
assert set(Base.metadata.tables) <= set(insp.get_table_names())
|
|
|
|
def test_tables_and_columns_match_base_metadata(self, sqlite_database_url):
|
|
"""Every table and column ORM models declare must exist after
|
|
`upgrade head` - the exact assertion the analysis asks for."""
|
|
db_path, _ = sqlite_database_url
|
|
command.upgrade(_alembic_config(), "head")
|
|
|
|
insp = _inspect(db_path)
|
|
db_tables = set(insp.get_table_names())
|
|
|
|
for table_name, table in Base.metadata.tables.items():
|
|
assert table_name in db_tables, f"missing table {table_name}"
|
|
db_columns = {c["name"] for c in insp.get_columns(table_name)}
|
|
model_columns = {c.name for c in table.columns}
|
|
missing = model_columns - db_columns
|
|
assert not missing, f"{table_name} missing columns: {missing}"
|
|
|
|
|
|
class TestBaseMigrationNoopsOnExistingSchema:
|
|
"""000_base_schema.py does nothing when `recipes` already exists (#6)."""
|
|
|
|
def test_upgrade_head_noops_000_when_pre_001_schema_already_there(
|
|
self, sqlite_database_url
|
|
):
|
|
db_path, _ = sqlite_database_url
|
|
cfg = _alembic_config()
|
|
|
|
# Build the pre-001 schema and record it as migrated.
|
|
command.upgrade(cfg, "000_base")
|
|
insp = _inspect(db_path)
|
|
assert "recipes" in insp.get_table_names()
|
|
assert "image_path" not in {
|
|
c["name"] for c in insp.get_columns("recipes")
|
|
}
|
|
|
|
# Erase the tracking: alembic now believes nothing has run, exactly
|
|
# like a database that was born from Base.metadata.create_all() and
|
|
# never stamped. Running 000's upgrade() again on a database that
|
|
# already has `recipes` must be a no-op, not a "table already
|
|
# exists" failure.
|
|
sync_engine = create_engine(f"sqlite:///{db_path}")
|
|
with sync_engine.begin() as conn:
|
|
conn.execute(text("DELETE FROM alembic_version"))
|
|
sync_engine.dispose()
|
|
|
|
command.upgrade(cfg, "head") # must not raise (no duplicate table)
|
|
|
|
insp = _inspect(db_path)
|
|
assert set(Base.metadata.tables) <= set(insp.get_table_names())
|
|
assert "image_path" in {
|
|
c["name"] for c in insp.get_columns("recipes")
|
|
}
|
|
|
|
|
|
class TestApiKeyHashMigration:
|
|
"""015_api_key_hash: users.api_key -> api_keys table, and back (#41)."""
|
|
|
|
def test_015_upgrade_and_downgrade(self, sqlite_database_url):
|
|
db_path, _ = sqlite_database_url
|
|
cfg = _alembic_config()
|
|
command.upgrade(cfg, "head")
|
|
|
|
insp = _inspect(db_path)
|
|
assert "api_keys" in insp.get_table_names()
|
|
assert "api_key" not in {c["name"] for c in insp.get_columns("users")}
|
|
assert {c["name"] for c in insp.get_columns("api_keys")} >= {
|
|
"user_id", "key_hash", "created_at", "expires_at", "last_used_at",
|
|
}
|
|
|
|
command.downgrade(cfg, "014_vision_fk_ondelete")
|
|
insp = _inspect(db_path)
|
|
assert "api_keys" not in insp.get_table_names()
|
|
assert "api_key" in {c["name"] for c in insp.get_columns("users")}
|
|
|
|
command.upgrade(cfg, "head") # and up again
|
|
assert "api_keys" in _inspect(db_path).get_table_names()
|