65 lines
1.9 KiB
Python
65 lines
1.9 KiB
Python
from pathlib import Path
|
|
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
from sqlalchemy import inspect, text
|
|
|
|
import app.models # noqa: F401
|
|
from app.db.session import Base, build_engine
|
|
|
|
|
|
BACKEND_DIR = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def build_alembic_config(database_url: str) -> Config:
|
|
config = Config(str(BACKEND_DIR / "alembic.ini"))
|
|
config.set_main_option("script_location", str(BACKEND_DIR / "alembic"))
|
|
config.set_main_option("sqlalchemy.url", database_url)
|
|
return config
|
|
|
|
|
|
def test_alembic_upgrade_creates_expected_schema(tmp_path):
|
|
db_path = tmp_path / "blank.sqlite3"
|
|
database_url = f"sqlite:///{db_path.as_posix()}"
|
|
|
|
command.upgrade(build_alembic_config(database_url), "head")
|
|
|
|
inspector = inspect(build_engine(database_url))
|
|
assert set(inspector.get_table_names()) == {
|
|
"actions",
|
|
"alembic_version",
|
|
"audit_logs",
|
|
"managed_services",
|
|
"servers",
|
|
"telegram_links",
|
|
"users",
|
|
}
|
|
assert "token_version" in {
|
|
column["name"] for column in inspector.get_columns("users")
|
|
}
|
|
|
|
|
|
def test_alembic_upgrade_stamps_existing_runtime_schema(tmp_path):
|
|
db_path = tmp_path / "runtime.sqlite3"
|
|
database_url = f"sqlite:///{db_path.as_posix()}"
|
|
engine = build_engine(database_url)
|
|
|
|
Base.metadata.create_all(bind=engine)
|
|
with engine.begin() as connection:
|
|
version_table_exists = connection.execute(
|
|
text(
|
|
"SELECT name FROM sqlite_master "
|
|
"WHERE type='table' AND name='alembic_version'"
|
|
)
|
|
).scalar_one_or_none()
|
|
assert version_table_exists is None
|
|
|
|
command.upgrade(build_alembic_config(database_url), "head")
|
|
|
|
with engine.begin() as connection:
|
|
revision = connection.execute(
|
|
text("SELECT version_num FROM alembic_version")
|
|
).scalar_one()
|
|
|
|
assert revision == "20260607_0001"
|