lamhieu/docsifer

1"""Shared pytest fixtures."""2 3from __future__ import annotations4 5from collections.abc import AsyncIterator6 7import pytest8import pytest_asyncio9from httpx import ASGITransport, AsyncClient10 11from docsifer.analytics import AnalyticsService, InMemoryStore12from docsifer.config import Settings, get_settings13from docsifer.main import create_app14 15 16@pytest.fixture17def settings(monkeypatch: pytest.MonkeyPatch, tmp_path) -> Settings:18    monkeypatch.setenv("DOCSIFER_ENVIRONMENT", "development")19    monkeypatch.setenv("DOCSIFER_LOG_JSON", "false")20    monkeypatch.setenv("DOCSIFER_ANALYTICS_ENABLED", "false")21    monkeypatch.setenv("DOCSIFER_MAX_UPLOAD_BYTES", str(2 * 1024 * 1024))22    monkeypatch.setenv("DOCSIFER_MAX_CONCURRENT_CONVERSIONS", "2")23    monkeypatch.setenv("DOCSIFER_MAX_QUEUE_DEPTH", "4")24    monkeypatch.setenv("DOCSIFER_MAX_PER_IP_CONCURRENT", "5")25    monkeypatch.setenv("DOCSIFER_TMP_DIR", str(tmp_path))26    get_settings.cache_clear()27    return get_settings()28 29 30@pytest_asyncio.fixture31async def app(settings: Settings):32    application = create_app(settings)33    async with application.router.lifespan_context(application):34        yield application35 36 37@pytest_asyncio.fixture38async def client(app) -> AsyncIterator[AsyncClient]:39    transport = ASGITransport(app=app)40    async with AsyncClient(transport=transport, base_url="http://test") as ac:41        yield ac42 43 44@pytest_asyncio.fixture45async def in_memory_analytics() -> AsyncIterator[AnalyticsService]:46    svc = AnalyticsService(store=InMemoryStore(), sync_interval_sec=1)47    await svc.start()48    try:49        yield svc50    finally:51        await svc.stop()52