50 lines
2.0 KiB
Python
50 lines
2.0 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine, text
|
|
|
|
from services.shared.schema_migrations import (
|
|
ensure_schema_migrations_table,
|
|
resolve_schema_management_mode,
|
|
validate_schema_migrations_applied,
|
|
)
|
|
|
|
|
|
def test_resolve_schema_management_mode_defaults_by_backend():
|
|
assert resolve_schema_management_mode(None, backend_name="sqlite") == "legacy"
|
|
assert resolve_schema_management_mode(None, backend_name="postgresql") == "migrations"
|
|
|
|
|
|
def test_resolve_schema_management_mode_rejects_postgres_legacy():
|
|
with pytest.raises(RuntimeError, match="SCHEMA_MANAGEMENT_MODE=legacy"):
|
|
resolve_schema_management_mode("legacy", backend_name="postgresql")
|
|
|
|
|
|
def test_validate_schema_migrations_applied_reports_missing_migration_table(tmp_path: Path):
|
|
migrations_dir = tmp_path / "migrations"
|
|
migrations_dir.mkdir()
|
|
(migrations_dir / "0001_dummy_sqlite.sql").write_text("SELECT 1;", encoding="utf-8")
|
|
db_path = tmp_path / "schema_missing.sqlite"
|
|
db_engine = create_engine(f"sqlite:///{db_path.as_posix()}")
|
|
|
|
with pytest.raises(RuntimeError, match="python scripts/migrate_core_db.py"):
|
|
validate_schema_migrations_applied(db_engine=db_engine, migrations_dir=migrations_dir)
|
|
|
|
|
|
def test_validate_schema_migrations_applied_accepts_up_to_date_schema(tmp_path: Path):
|
|
migrations_dir = tmp_path / "migrations"
|
|
migrations_dir.mkdir()
|
|
migration_file = migrations_dir / "0001_dummy_sqlite.sql"
|
|
migration_file.write_text("SELECT 1;", encoding="utf-8")
|
|
db_path = tmp_path / "schema_ready.sqlite"
|
|
db_engine = create_engine(f"sqlite:///{db_path.as_posix()}")
|
|
|
|
ensure_schema_migrations_table(db_engine=db_engine)
|
|
with db_engine.begin() as conn:
|
|
conn.execute(
|
|
text("INSERT INTO schema_migrations(version, applied_at) VALUES (:v, CURRENT_TIMESTAMP)"),
|
|
{"v": migration_file.name},
|
|
)
|
|
|
|
validate_schema_migrations_applied(db_engine=db_engine, migrations_dir=migrations_dir)
|