Adriano
e20299548f
upgrade head partiva dalla 001, che sa solo aggiungere colonne a tabelle
che assume gia' esistenti: su un DB davvero vuoto falliva subito. Nuova
000_base_schema.py (revision "000_base", down_revision None) crea lo
schema pre-001 esplicitamente (create_table, non Base.metadata.create_all,
che porterebbe gia' le colonne di 001-013 e farebbe fallire 001 su
"Duplicate column"). No-op se la tabella recipes esiste gia'
(sa.inspect(bind).has_table("recipes")). 001 ora ha down_revision =
"000_base".
setup.py: init-db e reset-db, dopo create_all/drop_all+create_all,
eseguono l'equivalente di "alembic stamp head" (nuovo _stamp_head(),
command.stamp con la config di src/backend/migrations/alembic.ini) cosi'
un DB nato da create_all non ritenta le migrazioni 001-013 al prossimo
deploy.
Test: catena upgrade head su SQLite vuoto, tabelle/colonne confrontate
con Base.metadata; DB con recipes gia' nello stato pre-001 -> upgrade
head non fallisce (000 no-op). _stamp_head si appoggia a
migrations/env.py, che costruisce una connessione propria da
settings.database_url (non modificabile dai test tramite l'engine
patchato di setup.py): nei test di setup.py e' quindi no-op via
monkeypatch, cosi' init-db/reset-db continuano a girare su SQLite senza
toccare un MySQL vero.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
110 lines
4.1 KiB
Python
110 lines
4.1 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")
|
|
}
|