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

# 테스트는 APP_ENV=test 로 실행한다 (DB 이름 기본값이 web4ai_test_db 로 갈린다, dev DB 와 분리).
# 이 픽스처들은 TRUNCATE 를 하므로 dev DB(web4ai_db)와 절대 공유하면 안 된다(아래 db_engine 안전가드 참고).
# config.server_configs 가 import 되는 순간 config.<APP_ENV>.toml 을 읽으므로 가장 먼저 설정.
import os

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

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 CompanyStatus
from config.server_configs import main_db_config


# 모델이 쓰는 스키마. test DB 는 비어 있을 수 있어 create_all 전에 직접 만든다.
# 또 TRUNCATE 가 unqualified 테이블명을 쓰므로 이 스키마들을 search_path 에 얹어 해석시킨다.
# 도메인 모듈(places/facts/sites/local/reports)이 붙을 때마다 스키마를 여기에 추가한다.
_SCHEMAS = ("company", "place", "fact", "local", "site", "job")


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 생성용 관리 접속. CREATE DATABASE 는 대상 DB 안에서 못 하므로 기본 'postgres' 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 를 지운다(있으면). recreate=True 면 지운 뒤 새로 만든다.
    WITH (FORCE): 남아있는 커넥션을 끊고 drop (PG13+). 관리 접속은 기본 'postgres' 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 를 만들고, 끝나면 내린다.

    매 세션 '깨끗한 새 DB'로 시작하므로 스키마 낡음(드리프트)이 원천 차단되고, 끝나면 남는 DB 도 없다.
    (테이블 구조는 db_engine 의 create_all 이 현재 모델 기준으로 채운다.)
    안전가드: 이름에 'test' 있는 DB 만 만들고/지운다(dev 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):
    """테스트용 스키마를 보장하고, 매 테스트 시작 시 테이블을 비워 격리한다.

    ⚠ 이 픽스처는 TRUNCATE 한다 → dev DB(web4ai_db)를 가리키면 실데이터가 날아간다.
    그래서 test 전용 DB(이름에 'test')가 아니면 즉시 중단한다(APP_ENV=test).
    앱(DB_SESSION_MNG)도 APP_ENV=test 면 같은 test DB 에 접속하므로 여기서 만든 스키마를 공유한다.
    """
    # 안전가드: dev DB 오염 방지. web4ai_test_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),
        connect_args={"server_settings": {"search_path": ",".join(_SCHEMAS) + ",public"}},
    )
    async with engine.begin() as conn:
        for sch in _SCHEMAS:
            await conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {sch}"))
        await conn.run_sync(MAIN_BASE.metadata.create_all)  # 이미 있으면 skip
        # 도메인 테이블 전부 비워 격리 (CASCADE: FK 미설정이라 안전망)
        await conn.execute(
            text(
                "TRUNCATE TABLE users, companies, places, place_aliases, place_links, units, media, "
                "facts, faqs, local_contents, routes, nearby_links, "
                "sites, site_versions, publish_logs, ai_check_results, jobs RESTART IDENTITY CASCADE"
            )
        )
    yield engine
    await engine.dispose()


@pytest_asyncio.fixture
async def company_id(db_engine) -> str:
    """테스트용 소속사 1개를 시드하고 company_id(uuid str)를 돌려준다.
    users 는 company_id 를 요구하므로 계정 생성 테스트의 선행 조건이다.
    """
    cid = uuid.uuid4()
    async with db_engine.begin() as conn:
        # status 는 NOT NULL(모델 default 는 ORM 전용이라 raw INSERT 엔 안 먹음) → 명시.
        await conn.execute(
            text("INSERT INTO companies (company_id, name, status) VALUES (:cid, :name, :status)"),
            {"cid": cid, "name": "테스트사", "status": CompanyStatus.ACTIVE.value},
        )
    return str(cid)


@pytest_asyncio.fixture
async def other_company_id(db_engine) -> str:
    """company_id 와 다른 소속사 1개(회사 스코프/IDOR 격리 테스트용)."""
    cid = uuid.uuid4()
    async with db_engine.begin() as conn:
        await conn.execute(
            text("INSERT INTO companies (company_id, name, status) VALUES (:cid, :name, :status)"),
            {"cid": cid, "name": "다른회사", "status": CompanyStatus.ACTIVE.value},
        )
    return str(cid)


@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, company_id):
    """테스트 유저를 시드하고 로그인 헤더(Bearer)를 돌려주는 팩토리.

    계정 생성 API 가 없으므로 users 행을 직접 INSERT(비번 bcrypt 해시)한 뒤 /v1/auth/login 으로 토큰을 받는다.
    company 미지정 시 기본 소속사(company_id 픽스처). role 로 OWNER 계정도 만들 수 있다.
    호출: `h = await auth_headers("user1")` / `await auth_headers("userB", other_company_id)`.
    """
    from common.enums import UserRole, UserStatus
    from router.v1.validator.dependencies import GetHashedPW

    async def _make(login_id, company=None, *, password="pw1234", role=UserRole.USER.value, name="n"):
        cid = company or company_id
        hashed = await GetHashedPW(password)
        async with db_engine.begin() as conn:
            # status·role 은 NOT NULL — ORM default 는 raw INSERT 에 안 먹으므로 명시(companies.status 와 동일).
            await conn.execute(
                text(
                    "INSERT INTO users (user_id, company_id, id, password, name, status, role, last_accessed_at) "
                    "VALUES (:uid, :cid, :id, :pw, :name, :status, :role, now())"
                ),
                {
                    "uid": uuid.uuid4(), "cid": uuid.UUID(cid), "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) 대역.

    ★ 왜 필요한가
      발행 게이트는 이제 **실제로 나갈 HTML** 을 보고 판정한다. 그 HTML 은 Node 렌더러가
      굽고, BUILD 잡은 그 결과 보고서를 기다린다(services/render_report.wait_for).
      파이썬 테스트 환경에는 Node 렌더러가 없으므로, payload 를 읽어 보고서를 만들어 주는
      대역을 끼운다 — 여기서 검사하려는 건 **백엔드가 보고서를 어떻게 처리하는가** 다.

    ★ 구조화 데이터 ↔ 화면 값 대조 자체는 렌더러 쪽 테스트가 본다
      (solution/site/src/seo/verify.test.ts). 그 규칙을 여기서 다시 구현하지 않는다 —
      두 벌로 두면 어긋나고, 어긋난 걸 아무도 모르는 게 원래 문제였다.
    """
    import json

    from services import render_report, 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 _count(payload: dict) -> int:
        def long(value) -> bool:
            return len(str(value or "").strip()) >= MIN_UNIQUE_TEXT

        count = 0
        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")))
        count += sum(
            1 for q in payload.get("faqs") or []
            if 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 _wait_for(slug, site_version, timeout_sec, poll_sec=1.0):
        path = payload_dir / f"{slug}.json"
        if not path.is_file():
            return None
        payload = json.loads(path.read_text(encoding="utf-8"))
        if payload["site"]["version"] != site_version:
            return None
        place = payload.get("place") or {}
        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(payload),
            "jsonld": [{"@type": "LodgingBusiness", "name": place.get("name"),
                        "telephone": place.get("phone")}],
            "mismatches": [],
            "error": None,
        }

    monkeypatch.setattr(render_report, "wait_for", _wait_for)
    return payload_dir