generated from john/python-template
132 lines
4.6 KiB
Python
132 lines
4.6 KiB
Python
"""Tests for transcription.app."""
|
|
|
|
from contextlib import asynccontextmanager
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from transcription.app import create_app
|
|
from transcription.config import Settings
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestAppFactory:
|
|
"""Verify FastAPI app factory wiring."""
|
|
|
|
def test_create_app_returns_fastapi_instance(self, monkeypatch):
|
|
"""create_app returns a FastAPI application instance."""
|
|
monkeypatch.setattr("transcription.app.register_pages", lambda _app: None)
|
|
app = create_app()
|
|
assert isinstance(app, FastAPI)
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestAppLifespan:
|
|
"""Verify startup and shutdown lifecycle behavior."""
|
|
|
|
def test_startup_initializes_runtime_dependencies(self, monkeypatch, tmp_path):
|
|
"""Startup initializes logging, schema, directories, and worker resources."""
|
|
calls = []
|
|
|
|
monkeypatch.setattr("transcription.app.configure_logging", lambda _settings: calls.append("logging"))
|
|
monkeypatch.setattr("transcription.app.register_pages", lambda _app: None)
|
|
|
|
async def _create_all(**_kwargs):
|
|
calls.append("schema")
|
|
|
|
monkeypatch.setattr("transcription.app.create_all", _create_all)
|
|
monkeypatch.setattr(
|
|
"transcription.app.initialize_database_runtime",
|
|
lambda **_kwargs: type("_Runtime", (), {"engine": object(), "session_factory": object()})(),
|
|
)
|
|
|
|
async def _dispose_runtime():
|
|
calls.append("dispose_db")
|
|
|
|
monkeypatch.setattr("transcription.app.dispose_database_runtime", _dispose_runtime)
|
|
|
|
async def _recover_stale(_app):
|
|
calls.append("recover")
|
|
|
|
monkeypatch.setattr("transcription.app._recover_stale_processing_jobs", _recover_stale)
|
|
|
|
@asynccontextmanager
|
|
async def _worker_lifespan(**_kwargs):
|
|
calls.append("worker_start")
|
|
yield object(), object(), object()
|
|
calls.append("worker_stop")
|
|
|
|
monkeypatch.setattr("transcription.app.worker_consumer_lifespan", _worker_lifespan)
|
|
|
|
settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
environment="test",
|
|
bootstrap_schema_on_startup=True,
|
|
upload_dir=tmp_path / "uploads",
|
|
prompt_dir=tmp_path / "prompts",
|
|
)
|
|
monkeypatch.setattr("transcription.app.get_settings", lambda: settings)
|
|
|
|
app = create_app()
|
|
with TestClient(app):
|
|
pass
|
|
|
|
assert "logging" in calls
|
|
assert "schema" in calls
|
|
assert "recover" in calls
|
|
assert "worker_start" in calls
|
|
assert "worker_stop" in calls
|
|
assert "dispose_db" in calls
|
|
assert settings.upload_dir.exists()
|
|
assert settings.prompt_dir.exists()
|
|
|
|
def test_shutdown_stops_worker_resources(self, monkeypatch, tmp_path):
|
|
"""Shutdown signals and stops worker resources cleanly."""
|
|
calls = []
|
|
|
|
monkeypatch.setattr("transcription.app.configure_logging", lambda _settings: calls.append("logging"))
|
|
monkeypatch.setattr("transcription.app.register_pages", lambda _app: None)
|
|
|
|
async def _create_all(**_kwargs):
|
|
calls.append("schema")
|
|
|
|
monkeypatch.setattr("transcription.app.create_all", _create_all)
|
|
monkeypatch.setattr(
|
|
"transcription.app.initialize_database_runtime",
|
|
lambda **_kwargs: type("_Runtime", (), {"engine": object(), "session_factory": object()})(),
|
|
)
|
|
|
|
async def _dispose_runtime():
|
|
calls.append("dispose_db")
|
|
|
|
monkeypatch.setattr("transcription.app.dispose_database_runtime", _dispose_runtime)
|
|
|
|
async def _recover_stale(_app):
|
|
calls.append("recover")
|
|
|
|
monkeypatch.setattr("transcription.app._recover_stale_processing_jobs", _recover_stale)
|
|
|
|
@asynccontextmanager
|
|
async def _worker_lifespan(**_kwargs):
|
|
calls.append("worker_start")
|
|
yield object(), object(), object()
|
|
calls.append("worker_stop")
|
|
|
|
monkeypatch.setattr("transcription.app.worker_consumer_lifespan", _worker_lifespan)
|
|
|
|
settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
environment="test",
|
|
bootstrap_schema_on_startup=True,
|
|
upload_dir=tmp_path / "uploads",
|
|
prompt_dir=tmp_path / "prompts",
|
|
)
|
|
monkeypatch.setattr("transcription.app.get_settings", lambda: settings)
|
|
|
|
app = create_app()
|
|
with TestClient(app):
|
|
pass
|
|
|
|
assert calls == ["logging", "schema", "recover", "worker_start", "worker_stop", "dispose_db"]
|