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

"""Gemini 텍스트 생성 — ★ LLM 이 사실을 만들지 못하게 막는 게 전부다.

프롬프트로 "지어내지 마라" 라고 부탁하는 것만으로는 부족하다. 모델은 종종 어긴다.
그래서 ground_check 가 결과를 코드로 검증하고, 통과 못 한 문장은 버린다.
이 파일은 그 검증이 실제로 무엇을 잡아내는지 고정한다.
"""
import json

import httpx
import pytest

from common.enums import PlaceCategory
# ★ 겹마다 사는 곳이 다르다(services/llm/__init__.py 의 설명 참조).
#   근거 검증 규칙 → grounding, HTTP 호출·키 → llm, 조립 → external.
from services.external import gemini_text as gt
from services.grounding import copy as grounding_copy
from services.llm import gemini as llm

FACTS = [
    grounding_copy.FactInput("check_in_time", "체크인 시간", "15:00"),
    grounding_copy.FactInput("check_out_time", "체크아웃 시간", "11:00"),
    grounding_copy.FactInput("pet_allowed", "반려동물 동반", "false"),
    grounding_copy.FactInput("cooking_allowed", "취사 가능", "true"),
    grounding_copy.FactInput("bbq_fee", "바비큐 이용료", "20000", "원"),
    grounding_copy.FactInput("parking", "주차 가능", "true"),
]


def _payload(intro="", meta="", faqs=None, intro_keys=None):
    body = {
        "intro": intro,
        "intro_fact_keys": intro_keys if intro_keys is not None else ["check_in_time"],
        "meta_description": meta,
        "faqs": faqs or [],
    }
    return {
        "candidates": [{
            "content": {"parts": [
                {"thoughtSignature": "..."},          # text 없는 파트가 섞여 온다(실호출 확인)
                {"text": json.dumps(body, ensure_ascii=False)},
            ]},
            "finishReason": "STOP",
        }],
        "usageMetadata": {"promptTokenCount": 800, "candidatesTokenCount": 200},
    }


def _client(handler):
    return httpx.AsyncClient(transport=httpx.MockTransport(handler))


def _ok(payload, calls=None):
    def handler(request):
        if calls is not None:
            calls.append(request)
        return httpx.Response(200, json=payload)
    return handler


@pytest.fixture(autouse=True)
def _configured(monkeypatch):
    """모든 테스트에서 키가 있는 것으로 둔다(실 API 는 MockTransport 가 막는다)."""
    monkeypatch.setattr(llm.external_api_config, "gemini_api_key", "test-key")


# ── 정상 경로 ─────────────────────────────────────────────────────────────
async def test_generates_intro_meta_and_faqs():
    """검증: 근거 있는 문장만 담긴 정상 응답.
    기대결과: 소개문·메타·FAQ 가 그대로 통과하고 근거 key 가 살아 있다."""
    payload = _payload(
        intro="체크인은 15시, 체크아웃은 11시입니다. 취사가 가능하며 주차도 하실 수 있습니다.",
        meta="체크인 15시, 취사 가능한 숙소입니다.",
        faqs=[{"question": "체크인 시간은 언제인가요?", "answer": "15시부터입니다.",
               "fact_keys": ["check_in_time"]}],
    )
    async with _client(_ok(payload)) as c:
        res = await gt.generate_copy("하조대펜션", PlaceCategory.LODGING, FACTS, client=c)

    assert res.intro and "15시" in res.intro
    assert res.meta_description
    assert len(res.faqs) == 1
    assert res.faqs[0].fact_keys == ["check_in_time"]
    assert res.rejected == []


async def test_request_body_pins_structured_output():
    """검증: 요청 본문.
    기대결과: responseSchema 로 출력 구조가 고정되고 fact 가 프롬프트에 실린다."""
    calls = []
    async with _client(_ok(_payload(intro="체크인은 15시입니다."), calls)) as c:
        await gt.generate_copy("하조대펜션", PlaceCategory.LODGING, FACTS, client=c)

    body = json.loads(calls[0].content)
    assert body["generationConfig"]["responseMimeType"] == "application/json"
    assert "faqs" in body["generationConfig"]["responseSchema"]["properties"]
    assert calls[0].headers["x-goog-api-key"] == "test-key"
    prompt = body["contents"][0]["parts"][0]["text"]
    assert "check_in_time" in prompt and "15:00" in prompt
    assert "지어내지" in prompt or "쓰지 마라" in prompt


# ── ★ 환각 차단 ───────────────────────────────────────────────────────────
async def test_ungrounded_number_is_rejected():
    """검증: fact 는 15:00 인데 모델이 '14시' 라고 썼다.
    기대결과: ★ 소개문이 버려지고 rejected 에 사유가 남는다 — 틀린 시각이 사이트로 나가면 클레임이다."""
    payload = _payload(intro="체크인은 14시부터 가능합니다.", meta="체크인 15시.")
    async with _client(_ok(payload)) as c:
        res = await gt.generate_copy("하조대펜션", PlaceCategory.LODGING, FACTS, client=c)

    assert res.intro is None
    assert any("14" in reason for _text, reason in res.rejected)


async def test_ungrounded_facility_is_rejected():
    """검증: fact 에 없는 시설(수영장)을 언급했다.
    기대결과: ★ 반려 — 없는 시설을 보고 온 손님은 헛걸음한다."""
    payload = _payload(intro="수영장을 갖춘 숙소입니다. 체크인은 15시입니다.", meta="체크인 15시.")
    async with _client(_ok(payload)) as c:
        res = await gt.generate_copy("하조대펜션", PlaceCategory.LODGING, FACTS, client=c)

    assert res.intro is None
    assert any("수영장" in reason for _t, reason in res.rejected)


async def test_contradicting_boolean_fact_is_rejected():
    """검증: pet_allowed=false 인데 '반려동물 동반 가능' 이라고 썼다.
    기대결과: ★ 반려 — 사실을 뒤집은 문장이다."""
    payload = _payload(intro="반려동물 동반이 가능한 숙소입니다.", meta="체크인 15시.")
    async with _client(_ok(payload)) as c:
        res = await gt.generate_copy("하조대펜션", PlaceCategory.LODGING, FACTS, client=c)

    assert res.intro is None
    assert any("반려동물" in reason for _t, reason in res.rejected)


async def test_negated_mention_of_false_fact_passes():
    """검증: pet_allowed=false 를 '불가' 라고 바르게 썼다.
    기대결과: 통과 — 부정 표현까지 막으면 사실을 못 알린다."""
    payload = _payload(intro="반려동물 동반은 불가합니다. 체크인은 15시입니다.", meta="체크인 15시.")
    async with _client(_ok(payload)) as c:
        res = await gt.generate_copy("하조대펜션", PlaceCategory.LODGING, FACTS, client=c)

    assert res.intro is not None


async def test_promotional_language_is_rejected():
    """검증: 최상급·홍보성 표현.
    기대결과: ★ 반려 — 이 서비스의 글은 광고문이 아니라 사실 전달이다."""
    payload = _payload(intro="국내 최고의 완벽한 숙소입니다.", meta="체크인 15시.")
    async with _client(_ok(payload)) as c:
        res = await gt.generate_copy("하조대펜션", PlaceCategory.LODGING, FACTS, client=c)

    assert res.intro is None
    assert any("홍보성" in reason for _t, reason in res.rejected)


async def test_scaled_number_matches_fact():
    """검증: fact 는 20000 인데 문장은 '2만원' 이다.
    기대결과: 통과 — 같은 값을 다르게 쓴 것뿐이다(과잉 반려를 막는다)."""
    payload = _payload(intro="바비큐 이용료는 2만원입니다. 체크인은 15시입니다.", meta="체크인 15시.")
    async with _client(_ok(payload)) as c:
        res = await gt.generate_copy("하조대펜션", PlaceCategory.LODGING, FACTS, client=c)

    assert res.intro is not None


async def test_faq_without_fact_keys_is_dropped():
    """검증: 근거 key 를 못 대는 FAQ.
    기대결과: ★ 버려진다 — 사실인지 확인할 방법이 없는 문답이다."""
    payload = _payload(
        intro="체크인은 15시입니다.", meta="체크인 15시.",
        faqs=[
            {"question": "근처에 볼거리가 있나요?", "answer": "많습니다.", "fact_keys": []},
            {"question": "체크인은 언제인가요?", "answer": "15시입니다.", "fact_keys": ["check_in_time"]},
        ],
    )
    async with _client(_ok(payload)) as c:
        res = await gt.generate_copy("하조대펜션", PlaceCategory.LODGING, FACTS, client=c)

    assert len(res.faqs) == 1
    assert res.faqs[0].question == "체크인은 언제인가요?"
    assert any("fact_keys" in reason for _t, reason in res.rejected)


async def test_invented_fact_key_is_stripped():
    """검증: 모델이 존재하지 않는 fact key 를 근거로 적었다.
    기대결과: 그 key 만 걸러진다 — 없는 근거를 있는 것처럼 두면 추적이 깨진다."""
    payload = _payload(
        intro="체크인은 15시입니다.", meta="체크인 15시.",
        intro_keys=["check_in_time", "swimming_pool", "spa_open"],
    )
    async with _client(_ok(payload)) as c:
        res = await gt.generate_copy("하조대펜션", PlaceCategory.LODGING, FACTS, client=c)

    assert res.intro_fact_keys == ["check_in_time"]


async def test_bad_faq_does_not_kill_good_ones():
    """검증: FAQ 3개 중 1개만 근거가 틀렸다.
    기대결과: 그 항목만 버려지고 나머지는 산다 — 하나 때문에 전부 잃으면 안 된다."""
    payload = _payload(
        intro="체크인은 15시입니다.", meta="체크인 15시.",
        faqs=[
            {"question": "체크인?", "answer": "15시입니다.", "fact_keys": ["check_in_time"]},
            {"question": "수영장 있나요?", "answer": "네 있습니다.", "fact_keys": ["parking"]},
            {"question": "주차 되나요?", "answer": "가능합니다.", "fact_keys": ["parking"]},
        ],
    )
    async with _client(_ok(payload)) as c:
        res = await gt.generate_copy("하조대펜션", PlaceCategory.LODGING, FACTS, client=c)

    assert len(res.faqs) == 2
    assert all("수영장" not in f.question for f in res.faqs)


async def test_faq_question_about_unavailable_facility_survives():
    """검증: 'X 가능한가요?' 라 묻고 '불가합니다' 라 답한 FAQ.
    기대결과: ★ 통과 — 질문은 사실을 주장하지 않는다. (실호출에서 멀쩡한 FAQ 가 버려진 오탐을 고친 것)"""
    payload = _payload(
        intro="체크인은 15시입니다.", meta="체크인 15시.",
        faqs=[{"question": "반려동물 동반이 가능한가요?", "answer": "아니요, 반려동물 동반은 불가합니다.",
               "fact_keys": ["pet_allowed"]}],
    )
    async with _client(_ok(payload)) as c:
        res = await gt.generate_copy("하조대펜션", PlaceCategory.LODGING, FACTS, client=c)

    assert len(res.faqs) == 1, f"멀쩡한 FAQ 가 버려졌다: {res.rejected}"


async def test_faq_answer_contradicting_fact_is_rejected():
    """검증: 'X 가능한가요?' 에 사실과 반대로 '네 가능합니다' 라 답했다.
    기대결과: ★ 반려 — 질문을 주장에서 뺀 빈틈을 답변 극성 검사가 막는다."""
    payload = _payload(
        intro="체크인은 15시입니다.", meta="체크인 15시.",
        faqs=[{"question": "반려동물 동반이 가능한가요?", "answer": "네, 가능합니다.",
               "fact_keys": ["pet_allowed"]}],
    )
    async with _client(_ok(payload)) as c:
        res = await gt.generate_copy("하조대펜션", PlaceCategory.LODGING, FACTS, client=c)

    assert res.faqs == []
    assert any("부정하지 않는다" in reason for _t, reason in res.rejected)


def test_question_mentioning_missing_facility_is_still_rejected():
    """검증: 없는 시설을 묻는 질문("수영장 있나요?").
    기대결과: 반려 — 의문문 예외는 값-반대 판정에만 적용되고, 없는 시설 언급은 그대로 잡는다."""
    ok, reasons = grounding_copy.ground_check("수영장이 있나요? 네 있습니다.", FACTS)
    assert ok is False
    assert any("수영장" in r for r in reasons)


# ── 입력 가드 ─────────────────────────────────────────────────────────────
async def test_no_facts_means_no_call_and_no_output():
    """검증: 근거 fact 가 하나도 없다.
    기대결과: ★ API 를 호출조차 하지 않고 빈 결과 — 근거 없이 쓰면 그게 환각이다."""
    calls = []
    async with _client(_ok(_payload(intro="아무거나"), calls)) as c:
        res = await gt.generate_copy("하조대펜션", PlaceCategory.LODGING, [], client=c)

    assert calls == [], "근거가 없는데 유료 API 를 호출했다"
    assert res.intro is None and res.faqs == []
    assert res.rejected and "근거 fact" in res.rejected[0][1]


async def test_missing_api_key_raises():
    """검증: GEMINI_API_KEY 미설정.
    기대결과: GeminiTextNotConfigured — 부팅은 막지 않고 이 어댑터만 비활성이다."""
    llm.external_api_config.gemini_api_key = ""
    with pytest.raises(llm.GeminiNotConfigured):
        await gt.generate_copy("x", PlaceCategory.LODGING, FACTS)


# ── 네트워크 ──────────────────────────────────────────────────────────────
async def test_retries_on_5xx_then_succeeds():
    """검증: 첫 호출이 503, 두 번째가 200.
    기대결과: 재시도해서 성공한다 — 일시적 장애로 생성을 포기하지 않는다."""
    calls = []

    def handler(request):
        calls.append(request)
        if len(calls) == 1:
            return httpx.Response(503, text="unavailable")
        return httpx.Response(200, json=_payload(intro="체크인은 15시입니다.", meta="체크인 15시."))

    async with _client(handler) as c:
        res = await gt.generate_copy("하조대펜션", PlaceCategory.LODGING, FACTS, client=c, max_retries=2)

    assert len(calls) == 2
    assert res.intro is not None


async def test_does_not_retry_on_4xx():
    """검증: 400 응답.
    기대결과: 재시도하지 않는다 — 잘못된 요청은 다시 보내도 같고 요금만 나간다."""
    calls = []

    def handler(request):
        calls.append(request)
        return httpx.Response(400, text="bad request")

    async with _client(handler) as c:
        with pytest.raises(llm.GeminiError):
            await gt.generate_copy("하조대펜션", PlaceCategory.LODGING, FACTS, client=c, max_retries=2)

    assert len(calls) == 1


async def test_auth_failure_is_not_configured():
    """검증: 401 응답.
    기대결과: GeminiTextNotConfigured — 재시도가 무의미한 설정 문제다."""
    async with _client(lambda r: httpx.Response(401, text="unauthorized")) as c:
        with pytest.raises(llm.GeminiNotConfigured):
            await gt.generate_copy("하조대펜션", PlaceCategory.LODGING, FACTS, client=c)


async def test_broken_json_raises_invalid_output():
    """검증: 구조화 출력이 JSON 이 아니다.
    기대결과: GeminiTextInvalidOutput — 조용히 빈 결과로 넘기지 않는다."""
    payload = {"candidates": [{"content": {"parts": [{"text": "{깨진 json"}]}, "finishReason": "STOP"}]}
    async with _client(_ok(payload)) as c:
        with pytest.raises(llm.GeminiInvalidOutput):
            await gt.generate_copy("하조대펜션", PlaceCategory.LODGING, FACTS, client=c)


async def test_empty_candidates_raises():
    """검증: 안전 필터 등으로 candidates 가 비어 왔다.
    기대결과: GeminiTextInvalidOutput."""
    async with _client(_ok({"candidates": []})) as c:
        with pytest.raises(llm.GeminiInvalidOutput):
            await gt.generate_copy("하조대펜션", PlaceCategory.LODGING, FACTS, client=c)


# ── ground_check 단위 ─────────────────────────────────────────────────────
def test_ground_check_reports_every_reason():
    """검증: 한 문장에 문제가 여러 개.
    기대결과: 사유가 전부 모여 나온다 — 운영자가 무엇이 문제인지 다 봐야 한다."""
    ok, reasons = grounding_copy.ground_check("국내 최고의 수영장을 갖춘 3층 건물입니다.", FACTS)
    assert ok is False
    assert len(reasons) >= 3


def test_ground_check_rejects_empty_text():
    """검증: 빈 문장.
    기대결과: 반려."""
    ok, reasons = grounding_copy.ground_check("   ", FACTS)
    assert ok is False and reasons == ["빈 문장"]


def test_ground_check_allows_facility_word_present_in_fact_value():
    """검증: fact 값 안에 그 낱말이 그대로 있다(대표 메뉴 = '수영장 뷰 라떼').
    기대결과: 통과 — 근거가 fact 에 실제로 있다."""
    facts = FACTS + [grounding_copy.FactInput("signature_menu", "대표 메뉴", "수영장 뷰 라떼")]
    ok, _reasons = grounding_copy.ground_check("대표 메뉴는 수영장 뷰 라떼입니다.", facts)
    assert ok is True


def test_ground_check_grounds_numbers_from_unit_summaries():
    """검증: 객실 요약에만 있는 숫자(최대 4명)를 문장이 인용했다.
    기대결과: 통과 — generate_copy 가 객실 요약도 근거로 펼쳐 넘긴다."""
    grounding = FACTS + gt._unit_facts([{"name": "A동", "facts": {"max_capacity": "4"}}])
    ok, _reasons = grounding_copy.ground_check("A동은 최대 4명까지 이용하실 수 있습니다.", grounding)
    assert ok is True


# ── 요약(summarize_text) ──────────────────────────────────────────────────
@pytest.fixture(autouse=True)
def _clear_summary_cache():
    """모듈 전역 캐시가 테스트끼리 새어 들어가지 않게 매번 비운다."""
    gt._SUMMARY_CACHE.clear()


async def test_summarize_returns_shortened_text():
    """검증: 긴 문장을 요약 API 로 축약한다.
    기대결과: 응답 텍스트가 그대로 반환되고, 원문이 요청 프롬프트에 실린다."""
    payload = {
        "candidates": [{"content": {"parts": [{"text": "짧게 줄인 문장입니다."}]}, "finishReason": "STOP"}],
        "usageMetadata": {"promptTokenCount": 300, "candidatesTokenCount": 20},
    }
    calls = []
    async with _client(_ok(payload, calls)) as c:
        result = await gt.summarize_text("첫 번째 테스트용 원문입니다. " * 20, client=c)

    assert result == "짧게 줄인 문장입니다."
    prompt = json.loads(calls[0].content)["contents"][0]["parts"][0]["text"]
    assert "첫 번째 테스트용 원문입니다." in prompt


async def test_summarize_skips_when_not_configured():
    """검증: GEMINI_API_KEY 미설정.
    기대결과: 호출 자체를 안 하고 None — 캔버스는 원문으로 폴백한다."""
    llm.external_api_config.gemini_api_key = ""
    result = await gt.summarize_text("두 번째 테스트용 원문입니다. " * 20)
    assert result is None


async def test_summarize_caches_repeated_calls():
    """검증: 같은 원문을 두 번 요약 요청한다.
    기대결과: 두 번째는 API 를 다시 부르지 않고 캐시된 값을 돌려준다."""
    payload = {"candidates": [{"content": {"parts": [{"text": "캐시 확인용 요약"}]}, "finishReason": "STOP"}]}
    calls = []
    text = "세 번째 테스트용 원문입니다. " * 20
    async with _client(_ok(payload, calls)) as c:
        first = await gt.summarize_text(text, client=c)
        second = await gt.summarize_text(text, client=c)

    assert first == second == "캐시 확인용 요약"
    assert len(calls) == 1, "같은 원문인데 API 를 두 번 불렀다"


async def test_summarize_returns_none_on_repeated_failure():
    """검증: 재시도까지 전부 5xx 로 실패한다.
    기대결과: 예외를 올리지 않고 None — 요약 실패가 캔버스를 깨면 안 된다."""
    async with _client(lambda r: httpx.Response(503, text="unavailable")) as c:
        result = await gt.summarize_text("네 번째 테스트용 원문입니다. " * 20, client=c, max_retries=1)

    assert result is None


async def test_summarize_empty_text_returns_none():
    """검증: 빈 문자열.
    기대결과: 호출 없이 None."""
    result = await gt.summarize_text("   ")
    assert result is None