"""Shared fixtures for pytest: async DB, httpx test client, mocks.""" import tempfile import uuid from collections.abc import AsyncGenerator from pathlib import Path from unittest.mock import MagicMock import numpy as np import pytest import sqlalchemy from httpx import ASGITransport, AsyncClient from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.pool import NullPool from app.config import settings from app.dependencies import ( get_db, get_face_service, get_garment_service, get_llm_provider, get_vton_service, ) from app.main import app from app.models.face import FaceEmbedding # noqa: F401 from app.models.garment import Garment # noqa: F401 from app.services.face_service import FaceService from app.services.garment_service import GarmentService from app.services.llm.base import ( BaseLLMProvider, GarmentAnalysis, StyleAnalysis, StyleAnalysisTemp, StyleCategoryScore, ) from app.services.vton_service import VtonService GLOBAL_USER_ID = uuid.UUID("00000000-0000-0000-0000-000000000000") # --------------------------------------------------------------------------- # Database — NullPool avoids asyncpg "another operation in progress" issues # --------------------------------------------------------------------------- TEST_ENGINE = create_async_engine(settings.DATABASE_URL, echo=False, poolclass=NullPool) TestSessionFactory = async_sessionmaker(TEST_ENGINE, expire_on_commit=False) # Transaction-scoped session: each test runs inside a transaction that # gets rolled back at the end, so test data never persists in the DB. _test_session: AsyncSession | None = None async def _override_get_db() -> AsyncGenerator[AsyncSession, None]: assert _test_session is not None yield _test_session # --------------------------------------------------------------------------- # Mock LLM provider # --------------------------------------------------------------------------- # 1x1 transparent PNG (valid bytes) _ONE_PX_PNG = ( b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01" b"\x00\x00\x00\x01\x08\x06\x00\x00\x00\x1f\x15\xc4\x89" b"\x00\x00\x00\nIDATx\x9cc\x00\x01\x00\x00\x05\x00\x01" b"\r\n\xb4\x00\x00\x00\x00IEND\xaeB`\x82" ) class MockLLMProvider(BaseLLMProvider): """LLM provider that returns deterministic results without calling real APIs.""" # Tests can mutate these to steer VTON behaviour per-case. vton_note: str | None = "Used all garments." vton_image: bytes | None = _ONE_PX_PNG async def analyze_garment(self, image: bytes) -> list[GarmentAnalysis]: return [ GarmentAnalysis( category_main="top", category_sub="t-shirt", description="검정 반팔 티셔츠", tags={ "color": ["black"], "pattern": "solid", "material": "cotton", "style": ["casual"], "fit": "regular", "occasion": ["daily"], "brand": "unknown", }, ), ] async def generate_ghost_mannequin( self, image: bytes, item: GarmentAnalysis, mode: str, image_model: str | None = None, ) -> bytes | None: return _ONE_PX_PNG async def generate_vton( self, person_image: bytes, ghost_images: list[bytes], categories: list[str], image_model: str | None = None, ) -> tuple[bytes | None, str | None]: return self.vton_image, self.vton_note async def analyze_style(self, image: bytes) -> StyleAnalysis: return StyleAnalysis( color_palette=StyleCategoryScore(score=4, comment="색상 조합이 좋습니다."), silhouette=StyleCategoryScore(score=3, comment="핏이 적절합니다."), detail=StyleCategoryScore(score=4, comment="디테일이 돋보입니다."), overall_score=3.5, overall_comment="전체적으로 균형 잡힌 코디입니다.", style_tags=["casual", "minimal"], ) async def analyze_style_temp(self, image: bytes) -> StyleAnalysisTemp: return StyleAnalysisTemp( time_context="출근룩으로는 무난한 조합입니다.", occasion_context="오늘 일정은 적절한 복장입니다.", weather_context="오늘 날씨는 적합한 옷차림입니다.", color_harmony="전체적인 배색은 조화롭습니다.", improvement_tips="더 완벽한 코디를 위해 액세서리를 추가해보세요.", ) _mock_llm = MockLLMProvider() def _override_get_llm_provider(): return _mock_llm def _override_get_garment_service(): return GarmentService( llm_provider=_mock_llm, segmentation_service=_mock_segmentation, ) # --------------------------------------------------------------------------- # Mock face analyzer # --------------------------------------------------------------------------- def _make_mock_face_analyzer(): """Create a mock FaceAnalyzer that returns a fixed 512-d embedding.""" analyzer = MagicMock() fixed_embedding = np.random.default_rng(42).standard_normal(512).astype(np.float32) fixed_embedding /= np.linalg.norm(fixed_embedding) analyzer.get_largest_face_embedding.return_value = fixed_embedding analyzer.get_embeddings.return_value = [fixed_embedding] return analyzer _mock_analyzer = _make_mock_face_analyzer() def _override_get_face_service(): return FaceService(analyzer=_mock_analyzer) _mock_segmentation = MagicMock() # Pass-through: the real rembg pipeline adds latency + optional model load, so # tests skip background removal by returning the LLM bytes unchanged. _mock_segmentation.remove_background.side_effect = lambda image_bytes, **_: image_bytes def _override_get_vton_service(): return VtonService( llm_provider=_mock_llm, face_analyzer=_mock_analyzer, segmentation_service=_mock_segmentation, ) # --------------------------------------------------------------------------- # HTTPX async test client # --------------------------------------------------------------------------- @pytest.fixture async def client() -> AsyncGenerator[AsyncClient, None]: """Provide an async test client with all dependencies overridden. Each test runs inside a DB transaction that is rolled back at the end, so test data never persists in the real database. """ global _test_session # Reset mock state so test-specific mutations don't bleed across tests. _mock_llm.vton_note = "Used all garments." _mock_llm.vton_image = _ONE_PX_PNG _mock_analyzer.get_embeddings.return_value = [ _mock_analyzer.get_largest_face_embedding.return_value ] app.dependency_overrides[get_db] = _override_get_db app.dependency_overrides[get_llm_provider] = _override_get_llm_provider app.dependency_overrides[get_garment_service] = _override_get_garment_service app.dependency_overrides[get_face_service] = _override_get_face_service app.dependency_overrides[get_vton_service] = _override_get_vton_service # Use a temporary directory for file storage so test images are # automatically cleaned up when the test finishes. original_upload_dir = settings.UPLOAD_DIR with tempfile.TemporaryDirectory() as tmp_dir: settings.UPLOAD_DIR = tmp_dir # Open a connection and begin a transaction that will be rolled back. # Use begin_nested() so that commit() calls inside services create # savepoints instead of real commits. async with TEST_ENGINE.connect() as conn: txn = await conn.begin() _test_session = AsyncSession(bind=conn, expire_on_commit=False) # Start a SAVEPOINT so that session.commit() creates sub-savepoints await conn.begin_nested() # After each commit(), automatically start a new SAVEPOINT @sqlalchemy.event.listens_for(_test_session.sync_session, "after_transaction_end") def restart_savepoint(session, transaction): if transaction.nested and not transaction._parent.nested: session.begin_nested() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as ac: yield ac # Roll back the outer transaction — all test data is discarded await _test_session.close() await txn.rollback() _test_session = None settings.UPLOAD_DIR = original_upload_dir app.dependency_overrides.clear() # --------------------------------------------------------------------------- # Minimal test image (JPEG) # --------------------------------------------------------------------------- @pytest.fixture def sample_jpeg() -> bytes: """Return a minimal valid JPEG image (100x100 red).""" import io from PIL import Image img = Image.new("RGB", (100, 100), color=(255, 0, 0)) buf = io.BytesIO() img.save(buf, format="JPEG") return buf.getvalue()