ahtlon
db87625384
This was anoying to fix (and even debug) so I used the local llm for some code changes Assisted-by: local-qwen3.8 <localllm@ahtlon.de>
61 lines
2.0 KiB
Python
61 lines
2.0 KiB
Python
from alembic.config import Config
|
|
from alembic.script import ScriptDirectory
|
|
from sqlalchemy import create_engine, text
|
|
from sqlmodel import Session
|
|
|
|
from app.model.models import UserDB
|
|
from app.services.database import (
|
|
add_and_refresh,
|
|
configure_engine,
|
|
create_db_and_tables,
|
|
get_engine,
|
|
reset_engine,
|
|
)
|
|
from app.services.settings import settings
|
|
|
|
|
|
def test_configure_engine_and_reset(tmp_path, db_session):
|
|
"""configure_engine() is the single injection point for the whole app.
|
|
reset_engine() drops the engine again and rebuilds from settings
|
|
conftest's autouse _installed_engine fixture puts the shared test engine back.
|
|
"""
|
|
from app.services import database as database_service
|
|
|
|
url = f"sqlite:///{tmp_path / 'injected.db'}"
|
|
injected = create_engine(url, connect_args={"check_same_thread": False})
|
|
configure_engine(injected)
|
|
assert get_engine() is injected
|
|
|
|
# Clean database: create_db_and_tables() must create the file and stamp
|
|
# *the configured engine*, not the URL from alembic.ini/settings.
|
|
create_db_and_tables()
|
|
assert tmp_path.joinpath("injected.db").exists()
|
|
with Session(injected) as session:
|
|
stamped = session.exec(text("SELECT version_num FROM alembic_version")).all()
|
|
head = ScriptDirectory.from_config(
|
|
Config(settings.alembic_config)
|
|
).get_current_head()
|
|
assert [row[0] for row in stamped] == [head]
|
|
|
|
reset_engine()
|
|
rebuilt = get_engine()
|
|
assert rebuilt is not injected
|
|
assert str(rebuilt.url) == str(database_service.settings.sqlalchemy_database_url)
|
|
|
|
|
|
def test_add_and_refresh(db_session):
|
|
"""Test add_and_refresh helper function."""
|
|
user = UserDB(name="Test User", passwordhash="hashed")
|
|
|
|
# Add user
|
|
result = add_and_refresh(db_session, user)
|
|
|
|
# Assert that user is now in database with ID
|
|
assert result.id is not None
|
|
assert result.name == "Test User"
|
|
|
|
# Verify in database
|
|
db_user = db_session.get(UserDB, result.id)
|
|
assert db_user is not None
|
|
assert db_user.name == "Test User"
|