conftest.py

"""통합 테스트 픽스처.

- 실제 PostgreSQL + pgvector(Testcontainers). SQLite·H2로 대체하지 않는다 — 검증 대상이
  조건부 UPDATE 영향 행 수·FOR UPDATE·<=> 연산자·ON CONFLICT SET 절이라 전부 방언 의존적.
- 격리는 TRUNCATE. 동시성 테스트가 여러 커넥션을 쓰므로 트랜잭션 롤백 격리를 못 쓴다.
- Profile 문자열 리터럴을 테스트에 두지 않는다 — settings fixture 경유(model-profile.md §2.1).
"""
from __future__ import annotations

import asyncio
import os
from pathlib import Path

import asyncpg
import pytest
import pytest_asyncio
from pgvector.asyncpg import register_vector
from testcontainers.postgres import PostgresContainer

from app.core.config import Settings, get_settings
from app.core.db import Database

# 운영·back compose.yaml과 digest까지 일치 유지(재현성). 롤링 태그 금지 — 태그만 고정하면
# 같은 태그가 언제든 다른 이미지를 가리킨다. 값의 정본은 back `compose.yaml`이다.
PGVECTOR_IMAGE = (
    "pgvector/pgvector:0.8.5-pg16"
    "@sha256:1d533553fefe4f12e5d80c7b80622ba0c382abb5758856f52983d8789179f0fb"
)

_SCHEMA_SQL = (Path(__file__).parent / "schema" / "ai_snapshot.sql").read_text(encoding="utf-8")

_AI_TABLES = [
    "ai.context_keyword_analysis",
    "ai.context_keyword",
    "ai.context_embedding",
    "ai.context_ai_state",
    "ai.keyword_preset",
]

# 테스트 설정값. Profile은 여기 한 곳에서만 정의하고 각 테스트는 settings fixture로 받는다.
_TEST_ENV = {
    "GMS_API_KEY": "test-key",
    "KAKAO_REST_API_KEY": "test-kakao-key",
    "GMS_BASE_URL": "https://gms.example/gmsapi/api.openai.com/v1",
    "PINLOG_EMBEDDING_MODEL": "text-embedding-3-small",
    "PINLOG_EMBEDDING_DIMENSION": "1536",
    "PINLOG_EMBEDDING_DISTANCE": "cosine",
    "PINLOG_EMBEDDING_PROFILE": "openai-text-embedding-3-small-1536-cosine-v1",
    # 판정 벤더 체인은 주입하지 않는다 — 정본이 코드에 있고(P45) 통합 테스트는 Fake
    # 클라이언트를 쓰므로 값이 동작에 들어오지 않는다. 기본값 자체는 test_unit.py가 본다.
    "KEYWORD_CANDIDATE_TOP_K": "10",
    "SIMILARITY_FLOOR": "0.30",
    "PROCESSING_EXPIRY_SEC": "600",
    "INTERNAL_SHARED_SECRET": "test-secret",
}

# app.main 로드 시 create_app()이 get_settings()를 호출하므로, .env가 없는 CI에서도
# import가 성공하도록 placeholder를 미리 넣는다. 실제 DATABASE_URL은 settings fixture가
# 컨테이너 dsn으로 덮는다(cache_clear 후).
for _k, _v in {**_TEST_ENV, "DATABASE_URL": "postgresql://placeholder/db"}.items():
    os.environ.setdefault(_k, _v)


@pytest.fixture(scope="session")
def _pg_container():
    with PostgresContainer(PGVECTOR_IMAGE, driver=None) as pg:
        yield pg


@pytest.fixture(scope="session")
def dsn(_pg_container) -> str:
    # asyncpg용 DSN (testcontainers는 SQLAlchemy 스타일 URL을 주므로 재구성)
    return (
        f"postgresql://{_pg_container.username}:{_pg_container.password}"
        f"@{_pg_container.get_container_host_ip()}"
        f":{_pg_container.get_exposed_port(5432)}/{_pg_container.dbname}"
    )


@pytest.fixture(scope="session")
def settings(dsn) -> Settings:
    for key, value in {**_TEST_ENV, "DATABASE_URL": dsn}.items():
        os.environ[key] = value
    # app.main 모듈 로드 시 create_app()이 .env로 get_settings()를 이미 캐시했을 수 있다.
    # 테스트 env로 재설정해 미들웨어(create_app 내부)와 fixture가 같은 설정을 보게 한다.
    get_settings.cache_clear()
    return get_settings()


async def _apply_schema(dsn: str) -> None:
    conn = await asyncpg.connect(dsn)
    try:
        await conn.execute(_SCHEMA_SQL)
    finally:
        await conn.close()


async def _truncate_ai_tables(dsn: str) -> None:
    conn = await asyncpg.connect(dsn)
    try:
        await conn.execute(f"TRUNCATE {', '.join(_AI_TABLES)} RESTART IDENTITY CASCADE")
    finally:
        await conn.close()


# 동기 fixture다. 스크립트 엔트리포인트(`app.bootstrap.load_presets`, `app.smoke.gms_roundtrip`)는
# 자기가 `asyncio.run()`을 부르므로 그것을 검증하는 테스트도 sync여야 하고, sync 테스트는
# async fixture를 받을 수 없다. 스키마 적용 자체는 세션당 1회라 어느 쪽이든 비용이 같다.
# `_SCHEMA_SQL`은 `CREATE SCHEMA ai`로 시작해 멱등이 아니므로 반드시 세션 스코프여야 한다.
@pytest.fixture(scope="session")
def _schema(dsn) -> None:
    asyncio.run(_apply_schema(dsn))


@pytest.fixture
def clean_dsn(settings, _schema) -> str:
    """동기 테스트용 — 스키마가 적용된 DSN을 주고 ai 테이블을 비운다(`db` fixture의 sync판)."""
    asyncio.run(_truncate_ai_tables(settings.database_url))
    return settings.database_url


@pytest_asyncio.fixture
async def db(settings, _schema) -> Database:
    database = Database(settings.database_url)
    await database.connect()
    # 함수별 격리: 모든 ai 테이블 TRUNCATE
    async with database.acquire() as conn:
        await conn.execute(f"TRUNCATE {', '.join(_AI_TABLES)} RESTART IDENTITY CASCADE")
    try:
        yield database
    finally:
        await database.disconnect()


@pytest_asyncio.fixture
async def conn(db) -> asyncpg.Connection:
    """단일 커넥션(register_vector 적용). 저장소·빌더 테스트용."""
    async with db.acquire() as connection:
        yield connection


async def raw_connect(dsn: str) -> asyncpg.Connection:
    """동시성 테스트에서 별도 커넥션이 필요할 때 사용(pgvector 등록 포함)."""
    connection = await asyncpg.connect(dsn)
    await connection.execute("SET search_path = ai, public")
    await register_vector(connection)
    return connection