236 lines
9.1 KiB
Python
236 lines
9.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")
|
|
}
|
|
|
|
|
|
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()
|
|
|
|
|
|
class TestVisionRecipeDirMigration:
|
|
"""016_vision_recipe_dir: recipe_tasks.vision_recipe_dir, su e giu (#46)."""
|
|
|
|
def test_016_upgrade_and_downgrade(self, sqlite_database_url):
|
|
db_path, _ = sqlite_database_url
|
|
cfg = _alembic_config()
|
|
|
|
def cols():
|
|
return {c["name"] for c in _inspect(db_path).get_columns("recipe_tasks")}
|
|
|
|
command.upgrade(cfg, "head")
|
|
assert "vision_recipe_dir" in cols()
|
|
command.downgrade(cfg, "015_api_key_hash")
|
|
assert "vision_recipe_dir" not in cols()
|
|
command.upgrade(cfg, "head") # e di nuovo su
|
|
assert "vision_recipe_dir" in cols()
|
|
|
|
|
|
class TestUserPinMigration:
|
|
"""017_user_pin: users.pin_hash / pin_failed_count / pin_locked_until, su e giu (#59)."""
|
|
|
|
def test_017_upgrade_and_downgrade(self, sqlite_database_url):
|
|
db_path, _ = sqlite_database_url
|
|
cfg = _alembic_config()
|
|
new = {"pin_hash", "pin_failed_count", "pin_locked_until"}
|
|
|
|
def cols():
|
|
return {c["name"] for c in _inspect(db_path).get_columns("users")}
|
|
|
|
command.upgrade(cfg, "head")
|
|
assert new <= cols()
|
|
command.downgrade(cfg, "016_vision_recipe_dir")
|
|
assert not (new & cols())
|
|
command.upgrade(cfg, "head") # e di nuovo su
|
|
assert new <= cols()
|
|
|
|
|
|
_SEED_USER_RECIPE_VERSION = [
|
|
"INSERT INTO users (username, password_hash, display_name, roles, is_admin, active,"
|
|
" language_pref, theme_pref) VALUES ('u', 'x', 'U', '[]', 1, 1, 'en', 'light')",
|
|
"INSERT INTO recipes (code, name, active, created_by) VALUES ('R', 'R', 1, 1)",
|
|
"INSERT INTO recipe_versions (recipe_id, version_number, is_current, created_by)"
|
|
" VALUES (1, 1, 1, 1)",
|
|
]
|
|
|
|
|
|
class TestTaskLoopResultsMigration:
|
|
"""018_task_loop_results: in_loop/at_close/fields, tipo check, task_results (#53 #54 #55)."""
|
|
|
|
def test_018_upgrade_and_downgrade(self, sqlite_database_url):
|
|
db_path, _ = sqlite_database_url
|
|
cfg = _alembic_config()
|
|
|
|
def cols():
|
|
return {c["name"] for c in _inspect(db_path).get_columns("recipe_tasks")}
|
|
|
|
command.upgrade(cfg, "head")
|
|
assert {"in_loop", "at_close", "fields"} <= cols()
|
|
assert "task_results" in _inspect(db_path).get_table_names()
|
|
command.downgrade(cfg, "017_user_pin")
|
|
assert not ({"in_loop", "at_close", "fields"} & cols())
|
|
assert "task_results" not in _inspect(db_path).get_table_names()
|
|
command.upgrade(cfg, "head") # e di nuovo su
|
|
|
|
def test_018_marks_existing_measure_tasks_in_loop(self, sqlite_database_url):
|
|
db_path, _ = sqlite_database_url
|
|
cfg = _alembic_config()
|
|
command.upgrade(cfg, "017_user_pin")
|
|
engine = create_engine(f"sqlite:///{db_path}")
|
|
with engine.begin() as conn:
|
|
for stmt in _SEED_USER_RECIPE_VERSION:
|
|
conn.execute(text(stmt))
|
|
for i, t in enumerate(("measure", "camera_measure", "dxf_compare", "note", "drawing")):
|
|
conn.execute(text(
|
|
"INSERT INTO recipe_tasks (version_id, order_index, task_type, title)"
|
|
f" VALUES (1, {i}, '{t}', '{t}')"))
|
|
engine.dispose()
|
|
|
|
command.upgrade(cfg, "head")
|
|
engine = create_engine(f"sqlite:///{db_path}")
|
|
with engine.connect() as conn:
|
|
rows = dict(conn.execute(text("SELECT task_type, in_loop FROM recipe_tasks")).all())
|
|
engine.dispose()
|
|
assert rows == {"measure": 1, "camera_measure": 1, "dxf_compare": 1, "note": 0, "drawing": 0}
|
|
|
|
def test_018_downgrade_refused_when_a_check_task_exists(self, sqlite_database_url):
|
|
db_path, _ = sqlite_database_url
|
|
cfg = _alembic_config()
|
|
command.upgrade(cfg, "head")
|
|
engine = create_engine(f"sqlite:///{db_path}")
|
|
with engine.begin() as conn:
|
|
for stmt in _SEED_USER_RECIPE_VERSION:
|
|
conn.execute(text(stmt))
|
|
conn.execute(text(
|
|
"INSERT INTO recipe_tasks (version_id, order_index, task_type, title)"
|
|
" VALUES (1, 0, 'check', 'c')"))
|
|
engine.dispose()
|
|
|
|
with pytest.raises(RuntimeError, match="check"):
|
|
command.downgrade(cfg, "017_user_pin")
|