generated from john/python-template
Step 1 complete!
This commit is contained in:
+61
-2
@@ -1,18 +1,77 @@
|
||||
"""Tests for transcription.db — schema bootstrap and session factory."""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
from sqlalchemy import inspect, text
|
||||
from sqlmodel import Session, SQLModel, create_engine
|
||||
from sqlmodel.pool import StaticPool
|
||||
|
||||
|
||||
def _in_memory_engine():
|
||||
"""Create a fresh in-memory SQLite engine for isolated db tests."""
|
||||
return create_engine(
|
||||
"sqlite://",
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
|
||||
|
||||
class TestSchemaBootstrap:
|
||||
"""Verify create_all produces the expected table set."""
|
||||
|
||||
def test_create_all_creates_expected_tables(self):
|
||||
"""After create_all(), document, job, and transcript tables exist."""
|
||||
engine = _in_memory_engine()
|
||||
# Ensure models are imported so metadata is populated
|
||||
from transcription.models import Document, Job, Transcript # noqa: F401
|
||||
|
||||
SQLModel.metadata.create_all(engine)
|
||||
|
||||
inspector = inspect(engine)
|
||||
table_names = set(inspector.get_table_names())
|
||||
assert "document" in table_names
|
||||
assert "job" in table_names
|
||||
assert "transcript" in table_names
|
||||
|
||||
|
||||
class TestSessionFactory:
|
||||
"""Verify get_session yields and cleans up sessions."""
|
||||
|
||||
def test_get_session_yields_session(self):
|
||||
def test_get_session_yields_session(self, monkeypatch):
|
||||
"""get_session() yields a usable Session object."""
|
||||
monkeypatch.setenv("OPENROUTER_API_KEY", "test-key-for-db")
|
||||
# Clear the lru_cache so Settings is re-created with our env var
|
||||
from transcription.config import get_settings
|
||||
get_settings.cache_clear()
|
||||
|
||||
def test_session_is_closed_after_generator_exit(self):
|
||||
engine = _in_memory_engine()
|
||||
SQLModel.metadata.create_all(engine)
|
||||
|
||||
import transcription.db as db_module
|
||||
with patch.object(db_module, "engine", engine):
|
||||
with db_module.get_session() as session:
|
||||
assert isinstance(session, Session)
|
||||
|
||||
get_settings.cache_clear()
|
||||
|
||||
def test_session_is_closed_after_generator_exit(self, monkeypatch):
|
||||
"""After the context manager exits, the session is closed."""
|
||||
monkeypatch.setenv("OPENROUTER_API_KEY", "test-key-for-db")
|
||||
from transcription.config import get_settings
|
||||
get_settings.cache_clear()
|
||||
|
||||
engine = _in_memory_engine()
|
||||
SQLModel.metadata.create_all(engine)
|
||||
|
||||
import transcription.db as db_module
|
||||
with patch.object(db_module, "engine", engine):
|
||||
with db_module.get_session() as session:
|
||||
# Session is usable inside the context
|
||||
session.execute(text("SELECT 1"))
|
||||
captured = session
|
||||
|
||||
# After exiting, the session's internal connection is released
|
||||
# (no active transaction bound to the session)
|
||||
assert captured._transaction is None
|
||||
|
||||
get_settings.cache_clear()
|
||||
|
||||
Reference in New Issue
Block a user