# 테스트는 APP_ENV=test 로 실행한다 (DB 이름 기본값이 web4ai_test_db 로 갈린다, dev DB 와 분리). # 이 픽스처들은 TRUNCATE 를 하므로 dev DB(web4ai_db)와 절대 공유하면 안 된다(아래 db_engine 안전가드 참고). # config.server_configs 가 import 되는 순간 config..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 UserRole, UserStatus 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, 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 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