11a7be7154
backend 와 동일한 프레임워크로 lps/ 폴더를 신설한다. 구체적인 도메인 로직(크롤링/오픈API/최저가 산정)은 미정이라 골격만 구성한다. - FastAPI 부트스트랩(web_main/router) + lifespan·CORS·gzip·로그 미들웨어 + /healthz - TOML 설정 로더(APP_ENV) + pydantic config + DB env override - DB 세션 매니저(논리 DB × R/W, service→람다 위임) — 엔진 lazy 라 DB 없이도 부팅 - 공통 응답 규약(gmodel) + ErrorType/DBType/DBWRType - router→services→crud 계층 컨벤션(services/crud 는 빈 폴더로 자리만) - 포트 9400(backend 9300·agent 9500 회피), Dockerfile/run_local_server.sh/README - tests/test_health.py 스모크(healthz, DB 없이 통과) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
56 lines
2.1 KiB
Python
56 lines
2.1 KiB
Python
# 테스트도 APP_ENV=local 로 실행한다 (config.local.toml 사용).
|
|
# config.server_configs 가 import 되는 순간 config.<APP_ENV>.toml 을 읽으므로 가장 먼저 설정.
|
|
import os
|
|
|
|
os.environ.setdefault("APP_ENV", "local")
|
|
|
|
import pytest_asyncio
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
|
|
from common.database.model.models import MAIN_BASE
|
|
from config.server_configs import main_db_config
|
|
|
|
|
|
def _write_url(cfg) -> str:
|
|
pw = f":{cfg.write_pw}" if cfg.write_pw else ""
|
|
return f"postgresql+asyncpg://{cfg.write_id}{pw}@{cfg.write_host}:{cfg.write_port}/{cfg.name}"
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def db_engine():
|
|
"""테스트용 스키마를 보장한다(실제 DB 필요). 도메인 테이블이 생기면 이 fixture 를 쓰는 테스트를 추가한다.
|
|
|
|
앱(DB_SESSION_MNG)은 자체 엔진으로 같은 DB 에 접속하므로 여기서 만든 스키마를 그대로 공유한다.
|
|
"""
|
|
engine = create_async_engine(_write_url(main_db_config))
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(MAIN_BASE.metadata.create_all) # 이미 있으면 skip
|
|
yield engine
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="session", autouse=True)
|
|
async def _dispose_app_engines():
|
|
"""테스트 세션이 끝날 때 앱 싱글톤 엔진을 정리한다.
|
|
(이벤트 루프 종료 후 커넥션이 GC 되며 나오는 'Event loop is closed' 경고 제거)
|
|
"""
|
|
yield
|
|
from common.database.db_session_manager import DB_SESSION_MNG
|
|
|
|
await DB_SESSION_MNG.dispose_all()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def client():
|
|
"""앱을 실제 네트워크 없이 호출하는 httpx 클라이언트 (ASGITransport).
|
|
|
|
아직 도메인 테이블이 없어 DB 없이도 부팅되므로 db_engine 에 의존하지 않는다.
|
|
DB 를 쓰는 도메인 테스트를 추가할 땐 인자에 db_engine 을 받아 스키마를 보장한다.
|
|
"""
|
|
from router.router import app
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
|
yield ac
|