o2o-site-AEO/solution/backend/tests/test_copy_api.py

"""소개문·FAQ 생성 — ★ LLM 은 사실을 만들지 않는다.

이 잡이 절대 하면 안 되는 것:
  - 미검증 fact 를 근거로 문장을 쓰는 것 (그 문장도 미검증이 된다)
  - 근거 없이 생성하는 것 (그게 환각이다)
  - 사람이 정정한 FAQ·소개문(CORRECTED)을 재생성이 덮어쓰는 것

★ 반대로 '생성물을 바로 노출하는 것' 은 이제 금지가 아니다(2026-09-10 결정).
  게이트는 입력 쪽에 있다 — 확인된 fact 로만 쓰고, 근거 없는 FAQ 는 저장되지 않는다.
"""
import uuid

from sqlalchemy import text

from common.enums import ErrorType, FactStatus, JobStatus, SourceType
from crud.job_crud import JobQueue
from services.external import gemini_text
from worker.handlers import build_handler
from worker.runner import Worker


def _copy(intro="조용한 숙소입니다.", faqs=None, rejected=None):
    return gemini_text.GeneratedCopy(
        intro=intro,
        intro_fact_keys=["check_in_time"],
        meta_description="양양 하조대 펜션",
        faqs=faqs if faqs is not None else [
            gemini_text.GeneratedFaq("체크인은 몇 시인가요?", "15시입니다.", ["check_in_time"]),
        ],
        rejected=rejected or [],
    )


def _patch(monkeypatch, copy):
    monkeypatch.setattr(gemini_text, "is_configured", lambda: True)

    async def _gen(place_name, category, facts, **kw):
        return copy

    monkeypatch.setattr(gemini_text, "generate_copy", _gen)


async def _place_with_facts(client, h, n_verified=5):
    pid = (await client.post("/v1/place", headers=h, json={"name": "카피펜션", "category": 1})).json()["place"]["place_id"]
    await client.post(f"/v1/place/{pid}/verify", headers=h, json={"source": 2, "road_address": f"주소{uuid.uuid4().hex[:6]}"})
    keys = ["check_in_time", "check_out_time", "cancel_policy", "cooking_allowed", "pet_allowed"][:n_verified]
    for k in keys:
        await client.post(f"/v1/place/{pid}/fact", headers=h, json={"key": k, "value": "15:00"})
    return pid


async def _faq_rows(db_engine, pid):
    async with db_engine.begin() as c:
        return (await c.execute(
            text("SELECT question, answer, source_fact_ids, status, generated_by FROM place_faqs "
                 "WHERE place_id = :p ORDER BY sort_order"),
            {"p": uuid.UUID(pid)},
        )).all()


async def test_copy_generates_intro_and_faq_as_published(auth_headers, client, db_engine, monkeypatch):
    """검증: 소개문·FAQ 를 생성한다.
    기대결과: ★ 바로 노출값(VERIFIED) — 승인 단계를 두지 않는다(2026-09-10 결정).
    게이트는 앞에 있다: 입력이 확인된 fact 뿐이고 근거 없는 FAQ 는 저장되지 않는다."""
    _patch(monkeypatch, _copy())
    h = await auth_headers("u1")
    pid = await _place_with_facts(client, h)

    body = (await client.post(f"/v1/place/{pid}/copy", headers=h, json={})).json()
    assert body["result"]["success"] is True
    assert body["grounded_facts"] == 5

    await Worker("w", JobQueue(), build_handler(), job_deadline_sec=30).drain()
    job = (await client.get(f"/v1/job/{body['job_id']}", headers=h)).json()["job"]
    assert job["status"] == JobStatus.DONE.value, job.get("last_error")
    assert job["result"]["intro"] is True
    assert job["result"]["faqs"] == 1

    rows = await _faq_rows(db_engine, pid)
    assert rows[0][3] == FactStatus.VERIFIED.value, "★ 생성된 FAQ 는 바로 노출값이어야 한다"
    assert rows[0][2] == ["check_in_time"], "근거 fact 가 기록돼야 한다"

    # 소개문도 fact 로 들어가고, 승인 없이 바로 노출값이다
    facts = (await client.get(f"/v1/place/{pid}/fact/list", headers=h)).json()
    intro = [f for f in facts["facts"] if f["key"] == "intro"]
    assert intro and intro[0]["status"] == FactStatus.VERIFIED.value
    assert intro[0]["source_type"] == SourceType.LLM.value


async def _must_not_call_llm(*args, **kwargs):
    raise AssertionError("★ 근거 없이 LLM 을 불렀다")


async def test_copy_refuses_without_facts_when_no_catalog(auth_headers, client, monkeypatch):
    """검증: 확인된 fact 가 하나도 없는 **호텔**(펜션 카탈로그 제외 대상)에서 생성을 시도한다.
    기대결과: FAQ_UNGROUNDED — ★ 잡을 만들지 않는다. 쓸 근거도, 채울 공통 질문도 없다."""
    monkeypatch.setattr(gemini_text, "is_configured", lambda: True)
    h = await auth_headers("u1")
    pid = (await client.post("/v1/place", headers=h, json={"name": "빈호텔", "category": 1})).json()["place"]["place_id"]
    await client.post(f"/v1/place/{pid}/verify", headers=h,
                      json={"source": 2, "road_address": "빈호텔주소", "category_name": "호텔"})

    r = await client.post(f"/v1/place/{pid}/copy", headers=h, json={})
    assert r.json()["result"]["code"] == ErrorType.FAQ_UNGROUNDED.value


async def test_unverified_facts_are_not_used_as_grounding(auth_headers, client, db_engine, monkeypatch):
    """검증: 크롤링으로 들어온 미검증 fact 만 있는 펜션.
    기대결과: ★ 근거 0건으로 친다 — LLM 을 부르지 않는다(미검증 값으로 쓴 문장도 미검증이다).
      FAQ 는 문의 안내로 20개를 채우고, 미검증인 체크인도 '답이 있는 주제' 로 치지 않는다."""
    monkeypatch.setattr(gemini_text, "is_configured", lambda: True)
    monkeypatch.setattr(gemini_text, "generate_copy", _must_not_call_llm)
    h = await auth_headers("u1")
    pid = (await client.post("/v1/place", headers=h, json={"name": "미검증펜션", "category": 1})).json()["place"]["place_id"]
    await client.post(f"/v1/place/{pid}/verify", headers=h, json={"source": 2, "road_address": "미검증주소"})
    await client.post(f"/v1/place/{pid}/fact", headers=h, json={
        "key": "check_in_time", "value": "15:00",
        "source_type": SourceType.CRAWL.value, "source_url": "https://ota.test/1"})

    body = (await client.post(f"/v1/place/{pid}/copy", headers=h, json={})).json()
    assert body["result"]["success"] is True and body["grounded_facts"] == 0
    await Worker("w", JobQueue(), build_handler(), job_deadline_sec=30).drain()
    job = (await client.get(f"/v1/job/{body['job_id']}", headers=h)).json()["job"]
    assert job["status"] == JobStatus.DONE.value, job.get("last_error")
    assert job["result"]["faqs"] == 0 and job["result"]["faq_fill"] == 20

    live = [r for r in await _faq_rows(db_engine, pid) if r[3] in _PUBLISHABLE]
    assert len(live) == 20 and all(r[4] == SourceType.TEMPLATE.value for r in live)
    assert any("체크인" in r[0] for r in live)


async def test_zero_facts_still_fill_twenty_without_api_key(auth_headers, client, db_engine, monkeypatch):
    """검증: fact 0건 · GEMINI 키 없음인 펜션(분류 없음 — 스테이머뭄처럼).
    기대결과: ★ 잡이 만들어지고 FAQ 가 정확히 20건 — 채우기는 LLM 을 안 쓰므로 키가 필요 없다."""
    monkeypatch.setattr(gemini_text, "is_configured", lambda: False)
    monkeypatch.setattr(gemini_text, "generate_copy", _must_not_call_llm)
    h = await auth_headers("u1")
    pid = (await client.post("/v1/place", headers=h, json={"name": "무근거펜션", "category": 1})).json()["place"]["place_id"]
    await client.post(f"/v1/place/{pid}/verify", headers=h, json={"source": 2, "road_address": "무근거주소"})

    body = (await client.post(f"/v1/place/{pid}/copy", headers=h, json={})).json()
    assert body["result"]["success"] is True, body["result"]
    await Worker("w", JobQueue(), build_handler(), job_deadline_sec=30).drain()
    job = (await client.get(f"/v1/job/{body['job_id']}", headers=h)).json()["job"]
    assert job["status"] == JobStatus.DONE.value, job.get("last_error")
    assert job["result"]["faq_fill"] == 20

    live = [r for r in await _faq_rows(db_engine, pid) if r[3] in _PUBLISHABLE]
    assert len(live) == 20 and len({r[0] for r in live}) == 20


async def test_faq_without_grounding_is_dropped(auth_headers, client, db_engine, monkeypatch):
    """검증: 근거 fact 가 비어 있는 FAQ 가 생성물에 섞여 온다.
    기대결과: 저장하지 않고 반려 목록에 남는다."""
    _patch(monkeypatch, _copy(faqs=[
        gemini_text.GeneratedFaq("체크인은?", "15시입니다.", ["check_in_time"]),
        gemini_text.GeneratedFaq("수영장 있나요?", "네 있습니다.", []),
    ]))
    h = await auth_headers("u1")
    pid = await _place_with_facts(client, h)
    job_id = (await client.post(f"/v1/place/{pid}/copy", headers=h, json={})).json()["job_id"]
    await Worker("w", JobQueue(), build_handler(), job_deadline_sec=30).drain()

    job = (await client.get(f"/v1/job/{job_id}", headers=h)).json()["job"]
    assert job["result"]["faqs"] == 1
    assert any("수영장" in str(r) for r in job["result"]["rejected"])
    rows = await _faq_rows(db_engine, pid)
    assert len([r for r in rows if r[4] == SourceType.LLM.value]) == 1
    # 채우기가 넣은 문의 안내에도 '수영장 있나요?' 의 거짓 답("네 있습니다")은 없다.
    assert not any("있습니다" in r[1] for r in rows if r[4] == SourceType.TEMPLATE.value)


async def test_rejected_sentences_are_reported(auth_headers, client, monkeypatch):
    """검증: 클라이언트가 반려한 문장이 있다.
    기대결과: 잡 result 에 사유와 함께 실린다 — 소개문이 왜 안 나왔는지 알 수 있어야 한다."""
    _patch(monkeypatch, _copy(intro=None, rejected=[("수영장을 갖추고 있습니다", "fact 에 없는 시설 '수영장'")]))
    h = await auth_headers("u1")
    pid = await _place_with_facts(client, h)
    job_id = (await client.post(f"/v1/place/{pid}/copy", headers=h, json={})).json()["job_id"]
    await Worker("w", JobQueue(), build_handler(), job_deadline_sec=30).drain()

    job = (await client.get(f"/v1/job/{job_id}", headers=h)).json()["job"]
    assert job["result"]["intro"] is False
    assert any("수영장" in str(r) for r in job["result"]["rejected"])


async def test_regeneration_keeps_faq_the_owner_corrected(auth_headers, client, db_engine, monkeypatch):
    """검증: 사장님이 고친 FAQ 가 있는 상태에서 재생성한다.
    기대결과: ★ 고친 FAQ 는 남는다 — 재생성이 사람의 판단을 덮어쓰면 안 된다.

    ★ '승인' 이 아니라 '정정' 으로 검증한다(2026-09-10). 생성분이 곧바로 VERIFIED 로 들어가면서
      status 만으로는 사람이 손댔는지 알 수 없게 됐다. 책임 주체를 적어 두는 자리는
      generated_by 이고, 사장님이 정정하면 faq_service 가 그 값을 OWNER 로 바꾼다."""
    _patch(monkeypatch, _copy())
    h = await auth_headers("u1")
    pid = await _place_with_facts(client, h)
    await client.post(f"/v1/place/{pid}/copy", headers=h, json={})
    await Worker("w", JobQueue(), build_handler(), job_deadline_sec=30).drain()

    # 사장님이 문구를 고쳐 승인 — 잠긴 상태(CORRECTED)가 되고 책임 주체가 사람으로 넘어온다.
    async with db_engine.begin() as c:
        await c.execute(
            text("UPDATE place_faqs SET status = :s, generated_by = :g, question = :q "
                 "WHERE place_id = :p AND generated_by = :llm"),
            {"s": FactStatus.CORRECTED.value, "g": SourceType.OWNER.value,
             "q": "체크인 시간이 어떻게 되나요?", "p": uuid.UUID(pid), "llm": SourceType.LLM.value},
        )

    _patch(monkeypatch, _copy(faqs=[gemini_text.GeneratedFaq("새 질문?", "새 답변", ["check_in_time"])]))
    await client.post(f"/v1/place/{pid}/copy", headers=h, json={})
    await Worker("w", JobQueue(), build_handler(), job_deadline_sec=30).drain()

    rows = await _faq_rows(db_engine, pid)
    kept = [r for r in rows if r[3] == FactStatus.CORRECTED.value]
    assert kept and kept[0][0] == "체크인 시간이 어떻게 되나요?", "★ 사장님이 고친 FAQ 가 재생성에 밀려났다"
    # 새로 생성된 것은 바로 노출값으로 함께 선다.
    assert any(r[0] == "새 질문?" and r[3] == FactStatus.VERIFIED.value for r in rows)


_PUBLISHABLE = (FactStatus.VERIFIED.value, FactStatus.CORRECTED.value)


async def test_copy_fills_faqs_to_target_without_duplicates(auth_headers, client, db_engine, monkeypatch):
    """검증: 생성된 FAQ 가 1건뿐이고, 사장님이 직접 쓴 FAQ 가 1건 있다.
    기대결과: 노출 중 FAQ 가 정확히 20건 —
      · 모자란 만큼은 문의 안내(TEMPLATE)로 채운다
      · fact 로 답할 수 있는 주제(체크인·체크아웃·취소·취사·반려동물)에는 문의 안내를 붙이지 않는다
      · 사장님이 이미 쓴 주제(보증금)도 다시 넣지 않는다"""
    _patch(monkeypatch, _copy())
    h = await auth_headers("u1")
    pid = await _place_with_facts(client, h)
    owner = await client.post(f"/v1/place/{pid}/faq", headers=h, json={"question": "보증금이 있나요?", "answer": "없습니다."})
    assert owner.json()["result"]["success"] is True

    job_id = (await client.post(f"/v1/place/{pid}/copy", headers=h, json={})).json()["job_id"]
    await Worker("w", JobQueue(), build_handler(), job_deadline_sec=30).drain()
    job = (await client.get(f"/v1/job/{job_id}", headers=h)).json()["job"]
    assert job["status"] == JobStatus.DONE.value, job.get("last_error")
    assert job["result"]["faqs"] == 1
    assert job["result"]["faq_fill"] == 18

    live = [r for r in await _faq_rows(db_engine, pid) if r[3] in _PUBLISHABLE]
    assert len(live) == 20
    assert len({r[0] for r in live}) == 20, "같은 질문이 두 번 들어갔다"

    fills = [r for r in live if r[4] == SourceType.TEMPLATE.value]
    assert len(fills) == 18
    fill_text = " ".join(r[0] for r in fills)
    for topic in ("체크인", "체크아웃", "취소", "요리", "반려동물", "보증금"):
        assert topic not in fill_text, f"★ 이미 답이 있는 주제 '{topic}' 에 문의 안내가 붙었다"
    assert all("문의" in r[1] or "확인" in r[1] for r in fills), "문의 안내가 아닌 답이 섞였다"


async def test_regeneration_replaces_fill_faqs(auth_headers, client, db_engine, monkeypatch):
    """검증: 생성을 두 번 돌린다.
    기대결과: 노출 중 FAQ 는 여전히 20건 — 옛 문의 안내는 내려가고 다시 고른다(쌓이지 않는다)."""
    _patch(monkeypatch, _copy())
    h = await auth_headers("u1")
    pid = await _place_with_facts(client, h)
    for _ in range(2):
        await client.post(f"/v1/place/{pid}/copy", headers=h, json={})
        await Worker("w", JobQueue(), build_handler(), job_deadline_sec=30).drain()

    live = [r for r in await _faq_rows(db_engine, pid) if r[3] in _PUBLISHABLE]
    assert len(live) == 20
    assert len({r[0] for r in live}) == 20


async def test_copy_requires_api_key(auth_headers, client, monkeypatch):
    """검증: GEMINI_API_KEY 없이 생성을 시도한다.
    기대결과: GENERATOR_NOT_CONFIGURED — 잡을 만들지 않는다."""
    monkeypatch.setattr(gemini_text, "is_configured", lambda: False)
    h = await auth_headers("u1")
    pid = await _place_with_facts(client, h)
    r = await client.post(f"/v1/place/{pid}/copy", headers=h, json={})
    assert r.json()["result"]["code"] == ErrorType.GENERATOR_NOT_CONFIGURED.value