generated from john/python-template
zoltan57andCopilot App
2ccea77520
Pure remediation; no behavior change. Every item traces to a finding in docs/architecture_code_review_2026-08-17.md. Deletions - Delete app_state.py, which had zero importers and whose get_session_factory raised TypeError at runtime [HIGH-01]. - Delete services/transcription.py and point build_prompt_execution imports at services/sources.py; drop the store.py compatibility aliases [MED-05]. - Delete ServiceBase.queue and its unparameterized asyncio.Queue [MED-07]. - Delete db/operations.get_next_queued_job, a divergent duplicate [CRIT-01]. - Drop the discarded load_docs parameter from list_jobs [LOW-03]. Config - Delete worker_retry_backoff_seconds; no backoff behavior existed anywhere, so wiring it would have been a new feature [MED-02]. - Wire sqlite_check_same_thread through get_engine. The engine hardcoded the setting's own default, so this preserves behavior exactly [MED-02]. - Replace DATABASE_URL in docker-compose.yml with the nested DATABASE__DRIVER / DATABASE__PATH names. Settings uses env_nested_delimiter with extra="ignore", so DATABASE_URL was silently discarded [MED-10]. UI - Move the 23KB inline VIBESCRIBE_LOGO_SVG to ui/static/vibescribe_logo.svg and load it through a cached read_svg sibling of read_css [MED-09]. - Route the portrait upload failure through error_presenter.show_error [LOW-07]. - Cancel the job detail auto-refresh timer instead of only deactivating it, and name its interval constant [LOW-06]. Worker - Make WorkerNotifier runtime_checkable and validate the resolved object in resolve_worker_notifier, which previously returned any non-None attribute unchecked [LOW-04]. Docs and lint - Fix two stale paths in services.instructions.md, one of which pointed at the module deleted here [LOW-02]. - ruff check --fix to zero [LOW-01]. Verified: 264 passed, 4 skipped; ruff check clean. Co-authored-by: Copilot App <[email protected]>
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()
|
|
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()
|
|
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"]
|