# 테스트는 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