54 lines
1.6 KiB
Python
54 lines
1.6 KiB
Python
|
import pytest
|
||
|
|
from sqlalchemy import text
|
||
|
|
|
||
|
|
from alembic import command
|
||
|
|
from alembic.config import Config
|
||
|
|
|
||
|
|
from app.domain.validation import parse_scoring_model
|
||
|
|
from app.persistence.db import SessionLocal, engine
|
||
|
|
from app.persistence.models import ScoringModelRecord
|
||
|
|
from scripts.bootstrap import run_bootstrap
|
||
|
|
|
||
|
|
pytestmark = pytest.mark.integration
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture(scope="module", autouse=True)
|
||
|
|
def require_postgres() -> None:
|
||
|
|
try:
|
||
|
|
with engine.connect() as connection:
|
||
|
|
connection.execute(text("SELECT 1"))
|
||
|
|
except Exception as exc:
|
||
|
|
pytest.skip(f"PostgreSQL not available: {exc}")
|
||
|
|
|
||
|
|
|
||
|
|
def test_upgrade_is_idempotent() -> None:
|
||
|
|
config = Config("alembic.ini")
|
||
|
|
command.upgrade(config, "head")
|
||
|
|
command.upgrade(config, "head")
|
||
|
|
|
||
|
|
|
||
|
|
def test_bootstrap_is_idempotent_and_rejects_hash_change(tmp_path, monkeypatch) -> None:
|
||
|
|
session = SessionLocal()
|
||
|
|
try:
|
||
|
|
run_bootstrap(session)
|
||
|
|
run_bootstrap(session)
|
||
|
|
row = session.get(ScoringModelRecord, "generic-v1")
|
||
|
|
assert row is not None
|
||
|
|
model = parse_scoring_model(row.definition)
|
||
|
|
assert row.definition_hash == model.definition_hash
|
||
|
|
row.definition_hash = "0" * 64
|
||
|
|
session.commit()
|
||
|
|
with pytest.raises(SystemExit):
|
||
|
|
run_bootstrap(session)
|
||
|
|
row.definition_hash = model.definition_hash
|
||
|
|
session.commit()
|
||
|
|
finally:
|
||
|
|
session.close()
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.slow
|
||
|
|
def test_downgrade_and_upgrade_on_disposable_db() -> None:
|
||
|
|
config = Config("alembic.ini")
|
||
|
|
command.downgrade(config, "base")
|
||
|
|
command.upgrade(config, "head")
|