generated from john/python-template
101 lines
4.3 KiB
Python
101 lines
4.3 KiB
Python
"""Tests for transcription.app."""
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from transcription.app import create_app
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestAppFactory:
|
|
"""Verify FastAPI app factory wiring."""
|
|
|
|
def test_create_app_returns_fastapi_instance(self):
|
|
"""create_app returns a FastAPI application instance."""
|
|
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):
|
|
"""Startup initializes logging, schema, directories, and worker resources."""
|
|
calls = []
|
|
|
|
monkeypatch.setattr("transcription.app.setup_logging", lambda: calls.append("logging"))
|
|
monkeypatch.setattr("transcription.app.get_settings", lambda: object())
|
|
monkeypatch.setattr("transcription.app.enforce_request_access", lambda **_kwargs: None)
|
|
monkeypatch.setattr("transcription.app.create_all", lambda **_kwargs: calls.append("schema"))
|
|
monkeypatch.setattr(
|
|
"transcription.app.initialize_database_runtime",
|
|
lambda **_kwargs: type("_Runtime", (), {"engine": object()})(),
|
|
)
|
|
monkeypatch.setattr("transcription.app.dispose_database_runtime", lambda: calls.append("dispose_db"))
|
|
monkeypatch.setattr("transcription.app.should_bootstrap_schema", lambda _settings: True)
|
|
monkeypatch.setattr("transcription.app._start_worker", lambda _app: calls.append("start_worker"))
|
|
monkeypatch.setattr("transcription.app._stop_worker", lambda _app: calls.append("stop_worker"))
|
|
monkeypatch.setattr("transcription.app.apply_pending_migrations", lambda **_kwargs: calls.append("migrate"))
|
|
monkeypatch.setattr("transcription.app.validate_schema_compatibility", lambda **_kwargs: [])
|
|
|
|
class _Dir:
|
|
def mkdir(self, parents: bool, exist_ok: bool):
|
|
calls.append("mkdir")
|
|
|
|
class _Settings:
|
|
upload_dir = _Dir()
|
|
prompt_dir = _Dir()
|
|
migration_auto_apply_on_startup = False
|
|
validate_schema_on_startup = True
|
|
|
|
monkeypatch.setattr("transcription.app.get_settings", lambda: _Settings())
|
|
|
|
app = create_app()
|
|
with TestClient(app):
|
|
pass
|
|
|
|
assert "logging" in calls
|
|
assert "schema" in calls
|
|
assert "mkdir" in calls
|
|
assert "start_worker" in calls
|
|
assert "dispose_db" in calls
|
|
|
|
def test_shutdown_stops_worker_resources(self, monkeypatch):
|
|
"""Shutdown signals and stops worker resources cleanly."""
|
|
calls = []
|
|
|
|
monkeypatch.setattr("transcription.app.setup_logging", lambda: None)
|
|
monkeypatch.setattr("transcription.app.get_settings", lambda: object())
|
|
monkeypatch.setattr("transcription.app.enforce_request_access", lambda **_kwargs: None)
|
|
monkeypatch.setattr("transcription.app.create_all", lambda **_kwargs: None)
|
|
monkeypatch.setattr(
|
|
"transcription.app.initialize_database_runtime",
|
|
lambda **_kwargs: type("_Runtime", (), {"engine": object()})(),
|
|
)
|
|
monkeypatch.setattr("transcription.app.dispose_database_runtime", lambda: calls.append("dispose_db"))
|
|
monkeypatch.setattr("transcription.app.should_bootstrap_schema", lambda _settings: True)
|
|
monkeypatch.setattr("transcription.app._start_worker", lambda _app: calls.append("start_worker"))
|
|
monkeypatch.setattr("transcription.app._stop_worker", lambda _app: calls.append("stop_worker"))
|
|
monkeypatch.setattr("transcription.app.apply_pending_migrations", lambda **_kwargs: calls.append("migrate"))
|
|
monkeypatch.setattr("transcription.app.validate_schema_compatibility", lambda **_kwargs: [])
|
|
|
|
class _Dir:
|
|
def mkdir(self, parents: bool, exist_ok: bool):
|
|
return None
|
|
|
|
class _Settings:
|
|
upload_dir = _Dir()
|
|
prompt_dir = _Dir()
|
|
migration_auto_apply_on_startup = False
|
|
validate_schema_on_startup = True
|
|
|
|
monkeypatch.setattr("transcription.app.get_settings", lambda: _Settings())
|
|
|
|
app = create_app()
|
|
with TestClient(app):
|
|
pass
|
|
|
|
assert calls == ["start_worker", "stop_worker", "dispose_db"]
|