o2o-site-AEO/solution/backend/conftest.py

# 테스트는 APP_ENV=test 로 실행한다 (DB 이름 기본값이 web4ai_test_db 로 갈린다, dev DB 와 분리).
import os

os.environ.setdefault("APP_ENV", "test")
os.environ.setdefault("THUMBNAIL_CAPTURE", "0")

import uuid

import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine

from common.database.model.models import MAIN_BASE
from common.enums import UserRole, UserStatus
from config.server_configs import main_db_config


# 비울 표는 **ORM 이 아는 것**에서 뽑는다.
def _truncate_sql() -> str:
    names = ", ".join(t.name for t in MAIN_BASE.metadata.sorted_tables)
    return f"TRUNCATE TABLE {names} RESTART IDENTITY CASCADE"


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}"


def _admin_url(cfg) -> str:
    """DB 생성용 관리 접속."""
    pw = f":{cfg.write_pw}" if cfg.write_pw else ""
    return f"postgresql+asyncpg://{cfg.write_id}{pw}@{cfg.write_host}:{cfg.write_port}/postgres"


async def _drop_test_db(*, recreate: bool):
    """test DB 를 지운다(있으면)."""
    engine = create_async_engine(_admin_url(main_db_config), isolation_level="AUTOCOMMIT")
    try:
        async with engine.connect() as conn:
            await conn.execute(text(f'DROP DATABASE IF EXISTS "{main_db_config.name}" WITH (FORCE)'))
            if recreate:
                await conn.execute(text(f'CREATE DATABASE "{main_db_config.name}"'))
    finally:
        await engine.dispose()


@pytest_asyncio.fixture(scope="session", autouse=True)
async def _test_db_lifecycle():
    """테스트 세션 동안만 test DB 를 만들고, 끝나면 내린다."""
    assert "test" in main_db_config.name, (
        f"비-test DB('{main_db_config.name}') 는 만들거나 지우지 않는다. APP_ENV=test 로 실행하세요."
    )
    await _drop_test_db(recreate=True)  # 세션 시작: 깨끗한 새 DB
    yield
    # 세션 종료: 앱 싱글톤 커넥션부터 정리(활성 커넥션 있으면 FORCE 로 끊김) 후 DB 를 내린다.
    from common.database.db_session_manager import DB_SESSION_MNG

    await DB_SESSION_MNG.dispose_all()
    await _drop_test_db(recreate=False)


@pytest_asyncio.fixture
async def db_engine(_test_db_lifecycle):
    """테스트용 스키마를 보장하고, 매 테스트 시작 시 테이블을 비워 격리한다."""
    # 안전가드: dev DB 오염 방지.
    assert "test" in main_db_config.name, (
        f"테스트가 비-test DB('{main_db_config.name}')를 가리킵니다. "
        "APP_ENV=test 로 실행하세요. dev 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
        # 도메인 테이블 전부 비워 격리 (CASCADE: FK 미설정이라 안전망)
        await conn.execute(text(_truncate_sql()))
    yield engine
    await engine.dispose()


@pytest_asyncio.fixture
async def owner_id(db_engine) -> str:
    """사장님 계정 1개를 시드하고 user_id(uuid str)를 돌려준다."""
    uid = uuid.uuid4()
    async with db_engine.begin() as conn:
        # status·role 은 NOT NULL(모델 default 는 ORM 전용이라 raw INSERT 엔 안 먹음) → 명시.
        await conn.execute(
            text(
                "INSERT INTO users (user_id, id, password, name, status, role, last_accessed_at) "
                "VALUES (:uid, :id, NULL, :name, :status, :role, now())"
            ),
            {"uid": uid, "id": f"seed{uid.hex[:8]}", "name": "시드사장",
             "status": UserStatus.ACTIVE.value, "role": UserRole.USER.value},
        )
    return str(uid)


@pytest_asyncio.fixture
async def client(db_engine):
    """앱을 실제 네트워크 없이 호출하는 httpx 클라이언트 (ASGITransport)."""
    from router.router import app

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac


@pytest_asyncio.fixture
async def auth_headers(db_engine, client):
    """테스트 유저를 시드하고 로그인 헤더(Bearer)를 돌려주는 팩토리."""
    from router.v1.validator.dependencies import GetHashedPW

    async def _make(login_id, *, password="pw1234", role=UserRole.USER.value, name="n"):
        hashed = await GetHashedPW(password)
        async with db_engine.begin() as conn:
            # status·role 은 NOT NULL — ORM default 는 raw INSERT 에 안 먹으므로 명시.
            await conn.execute(
                text(
                    "INSERT INTO users (user_id, id, password, name, status, role, last_accessed_at) "
                    "VALUES (:uid, :id, :pw, :name, :status, :role, now())"
                ),
                {
                    "uid": uuid.uuid4(), "id": login_id, "pw": hashed,
                    "name": name, "status": UserStatus.ACTIVE.value, "role": role,
                },
            )
        r = await client.post("/v1/auth/login", json={"id": login_id, "password": password})
        return {"Authorization": f"Bearer {r.json()['access_token']}"}

    return _make


# ── 렌더러 스텁 ──────────────────────────────────────────────────────────────
@pytest.fixture(autouse=True)
def fake_renderer(monkeypatch, tmp_path_factory):
    """정적 렌더러(solution/site) 대역."""
    import json
    from pathlib import Path

    from services import render_report, render_service, site_payload

    payload_dir = tmp_path_factory.mktemp("payloads")
    monkeypatch.setattr(site_payload, "payload_dir", lambda: payload_dir)
    monkeypatch.setattr(render_report, "payload_dir", lambda: payload_dir)

    # 고유 콘텐츠 계수 규칙은 렌더러(prerender.ts countUniqueContent)와 같아야 한다.
    MIN_UNIQUE_TEXT = 8

    def _has_long_text(value, long) -> bool:
        if isinstance(value, str):
            return long(value)
        if isinstance(value, list):
            return any(_has_long_text(v, long) for v in value)
        if isinstance(value, dict):
            return any(_has_long_text(v, long) for v in value.values())
        return False

    def _count(payload: dict) -> int:
        def long(value) -> bool:
            return len(str(value or "").strip()) >= MIN_UNIQUE_TEXT

        count = 0
        intro = next(
            (s for s in (payload.get("theme") or {}).get("sections") or [] if s.get("id") == "intro"),
            None,
        )
        if intro and intro.get("enabled") and long(intro.get("body")):
            count += 1
        # 붙여넣기 아이템(theme.sections[].data)의 항목도 고유 콘텐츠다 — 렌더러와 같은 규칙.
        for section in (payload.get("theme") or {}).get("sections") or []:
            if not section.get("enabled") or not section.get("data"):
                continue
            try:
                envelope = json.loads(section["data"])
            except (ValueError, TypeError):
                continue
            for item in (envelope or {}).get("items") or []:
                if _has_long_text(item, long):
                    count += 1
        facts = list(payload.get("facts") or [])
        for unit in payload.get("units") or []:
            facts.extend(unit.get("facts") or [])
        count += sum(1 for f in facts if f.get("type") == "text" and long(f.get("value")))
        # 문의 안내(sourceType=TEMPLATE)는 모든 가게에 같은 문구라 고유 콘텐츠가 아니다.
        count += sum(
            1 for q in payload.get("faqs") or []
            if q.get("sourceType") != 5
            and str(q.get("question") or "").strip() and long(q.get("answer"))
        )
        count += sum(1 for m in payload.get("media") or [] if long(m.get("alt")))
        return count

    async def _render_site(payload_path, site_version, timeout_sec):
        path = Path(payload_path)
        slug = path.stem
        if not path.is_file():
            raise render_service.RenderFailed(f"payload 파일이 없다: {payload_path}")
        payload = json.loads(path.read_text(encoding="utf-8"))
        if payload["site"]["version"] != site_version:
            raise render_service.RenderFailed(
                f"버전이 다르다(payload={payload['site']['version']} 요청={site_version})"
            )
        place = payload.get("place") or {}
        count = _count(payload)
        # 고유 콘텐츠 0건이면 렌더러는 **페이지를 쓰지 않는다** (prerender.ts NoUniqueContentError — 백엔드가 나중에 거부해도 그 전에 디스크에 나가 있으면 크롤러가 읽는다).
        if count <= 0:
            return {
                "schemaVersion": 1,
                "slug": slug,
                "siteId": payload["site"]["siteId"],
                "placeId": payload["site"]["placeId"],
                "siteVersion": site_version,
                "ok": False,
                "renderedAt": "2026-01-01T00:00:00.000Z",
                "routes": 0,
                "bundle": "test.js",
                "uniqueContentCount": count,
                "jsonld": None,
                "mismatches": [],
                "error": "고유 콘텐츠가 0건이다 — 이 가게에만 있는 내용이 없으면 발행하지 않는다",
            }
        return {
            "schemaVersion": 1,
            "slug": slug,
            "siteId": payload["site"]["siteId"],
            "placeId": payload["site"]["placeId"],
            "siteVersion": site_version,
            "ok": True,
            "renderedAt": "2026-01-01T00:00:00.000Z",
            "routes": 3,
            "bundle": "test.js",
            "uniqueContentCount": count,
            "jsonld": [{"@type": "LodgingBusiness", "name": place.get("name"),
                        "telephone": place.get("phone")}],
            "mismatches": [],
            "error": None,
        }

    monkeypatch.setattr(render_service, "render_site", _render_site)
    async def _activate_site(slug, site_version):
        return None
    monkeypatch.setattr(render_service, "activate_site", _activate_site)
    return payload_dir