Spaces:
Configuration error
Configuration error
File size: 2,999 Bytes
785dbd5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | """Shared fixtures for the backend test suite.
These tests deliberately avoid loading TensorFlow or any real model.
The route layer depends on ``PredictorService`` only through duck-typed
attributes (``model_version``, ``decode_strategy``, ``max_upload_bytes``,
``caption_image_bytes``), so a small fake stands in cleanly and keeps the
whole suite under one second.
We also bypass the FastAPI lifespan entirely. The lifespan builds a real
``CaptionPredictor`` from disk, which requires weights, a tokenizer, and a
TF graph build. Tests build a fresh ``FastAPI`` instance, wire the same
router and middleware, and stash the fake service directly on
``app.state.predictor_service`` — the exact slot the lifespan would have
populated in production.
"""
from __future__ import annotations
from collections.abc import Callable, Iterator
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.api.routes import router
from app.core.config import BackendSettings
from app.core.logging import RequestContextMiddleware, configure_app_logging
from app.utils.image import ImageDecodeError
configure_app_logging()
class FakePredictorService:
"""Duck-typed stand-in for ``PredictorService``."""
def __init__(
self,
*,
caption: str = "a test caption",
latency_ms: float = 1.23,
decode_strategy: str = "greedy",
model_version: str = "test-v0",
max_upload_bytes: int = 1024,
raise_decode_error: bool = False,
) -> None:
self.model_version = model_version
self.decode_strategy = decode_strategy
self.max_upload_bytes = max_upload_bytes
self._caption = caption
self._latency_ms = latency_ms
self._raise = raise_decode_error
self.calls: list[bytes] = []
async def caption_image_bytes(self, image_bytes: bytes) -> tuple[str, float]:
self.calls.append(image_bytes)
if self._raise:
raise ImageDecodeError("synthetic decode failure")
return self._caption, self._latency_ms
def _build_app(service: FakePredictorService | None) -> FastAPI:
app = FastAPI()
app.state.backend_settings = BackendSettings()
app.state.predictor_service = service
app.add_middleware(RequestContextMiddleware)
app.include_router(router)
return app
@pytest.fixture
def fake_service() -> FakePredictorService:
return FakePredictorService()
@pytest.fixture
def client(fake_service: FakePredictorService) -> Iterator[TestClient]:
with TestClient(_build_app(fake_service)) as test_client:
yield test_client
@pytest.fixture
def client_without_service() -> Iterator[TestClient]:
with TestClient(_build_app(None)) as test_client:
yield test_client
@pytest.fixture
def build_client() -> Callable[[FakePredictorService | None], TestClient]:
def _make(service: FakePredictorService | None) -> TestClient:
return TestClient(_build_app(service))
return _make
|