292 lines
14 KiB
Python
292 lines
14 KiB
Python
# 테스트는 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")
|
|
|
|
|
|
def pytest_configure(config):
|
|
"""순수 단위 테스트(test_itinerary_prompt.py 등)는 DB 없이 실행될 수 있도록 표시한다."""
|
|
args = " ".join(config.invocation_params.args)
|
|
# test_itinerary_prompt.py 만 실행 중인지 확인
|
|
config.SKIP_DB_SETUP = "test_itinerary_prompt.py" in args
|
|
|
|
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
|
|
|
|
|
|
# ★ 스키마는 public 한 벌이다 — 도메인 스키마(company·place·fact·local·site·job)는
|
|
# 2026-09-09 에 걷어냈다(migrations/0005). 그래서 여기서 스키마를 만들지도, search_path 를
|
|
# 얹지도 않는다.
|
|
#
|
|
# ★ 비울 표는 **ORM 이 아는 것**에서 뽑는다. 예전에는 이름을 손으로 나열했는데,
|
|
# 0005 가 표 이름을 옮겼을 때 이 문자열만 옛 이름으로 남아 테스트 13건이 통째로
|
|
# `relation "place_aliases" does not exist` 로 죽었다 — 문자열이라 import 도 타입검사도
|
|
# pyflakes 도 잡지 못한다. 모델에서 뽑으면 다시 어긋날 수 없다.
|
|
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 생성용 관리 접속. 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(request):
|
|
"""테스트 세션 동안만 test DB 를 만들고, 끝나면 내린다.
|
|
|
|
매 세션 '깨끗한 새 DB'로 시작하므로 스키마 낡음(드리프트)이 원천 차단되고, 끝나면 남는 DB 도 없다.
|
|
(테이블 구조는 db_engine 의 create_all 이 현재 모델 기준으로 채운다.)
|
|
안전가드: 이름에 'test' 있는 DB 만 만들고/지운다(dev DB 보호).
|
|
|
|
순수 단위 테스트(DB 필요 없음)는 SKIP_DB_SETUP 플래그로 이 fixture 를 건너뛴다.
|
|
"""
|
|
if getattr(request.config, "SKIP_DB_SETUP", False):
|
|
yield
|
|
return
|
|
|
|
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))
|
|
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)를 돌려준다.
|
|
|
|
★ 예전엔 `company_id`(소속사)였다. 회사(테넌트)를 걷어내면서 사업장이 `owner_user_id` 로
|
|
계정에 직접 매이게 됐다 — DB 를 직접 시드하는 테스트가 place 에 넣을 주인이 이 값이다.
|
|
"""
|
|
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)를 돌려주는 팩토리.
|
|
|
|
계정 생성 API 가 없으므로 users 행을 직접 INSERT(비번 bcrypt 해시)한 뒤 /v1/auth/login 으로 토큰을 받는다.
|
|
★ 회사 인자가 없다. 스코프가 계정 자체이므로 **다른 login_id 로 한 번 더 부르면 그게 남**이다
|
|
— 격리 테스트는 `await auth_headers("o2")` 하나면 된다.
|
|
호출: `h = await auth_headers("user1")`.
|
|
"""
|
|
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) 대역.
|
|
|
|
★ 왜 필요한가
|
|
발행 게이트는 이제 **실제로 나갈 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 _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")))
|
|
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 {}
|
|
count = _count(payload)
|
|
# ★ 고유 콘텐츠 0건이면 렌더러는 **페이지를 쓰지 않는다**
|
|
# (prerender.ts NoUniqueContentError — 백엔드가 나중에 거부해도 그 전에 디스크에
|
|
# 나가 있으면 크롤러가 읽는다). 대역이 늘 ok=True 를 주면 백엔드가 그 실패를
|
|
# NO_UNIQUE_CONTENT 로 되짚는 경로가 통째로 테스트되지 않는다.
|
|
# mismatches 는 비워 둔다 — 사유가 JSONLD_MISMATCH 로 섞이면 화면 문구가 틀린다.
|
|
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_report, "wait_for", _wait_for)
|
|
return payload_dir
|