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

"""Gemini Vision 클라이언트 — 사진 분류·alt 생성의 계약.

실제 API 를 절대 호출하지 않는다(httpx.MockTransport). 여기서 고정하는 것:
  1. ★ 반환 길이는 항상 입력과 같다 — 실패해도 자리를 지킨다
  2. ★ 매칭은 순서가 아니라 ref 로 한다 — 순서로 하면 엉뚱한 사진에 남의 alt 가 붙는다
  3. ★ 신뢰도가 낮으면 needs_review=True — 자동 반영하지 않고 사람 확인 큐로
  4. 배치 하나가 죽어도 나머지는 산다
  5. 홍보성 형용사 금지가 프롬프트에 실제로 들어간다
"""
import base64
import json
import struct
import zlib

import httpx
import pytest

from common.enums import PlaceCategory
from services.external import gemini
from services.llm import gemini as llm
from services.external.gemini import (
    GeminiError,
    GeminiNotConfigured,
    ImageInput,
    analyze_images,
)


# ---- 도구 -----------------------------------------------------------------
def _png(rgb=(10, 20, 30), size=8) -> bytes:
    """테스트용 최소 PNG. 시그니처 판별(_sniff_mime)까지 같이 확인된다."""
    def chunk(tag, data):
        body = tag + data
        return struct.pack(">I", len(data)) + body + struct.pack(">I", zlib.crc32(body) & 0xFFFFFFFF)

    raw = b"".join(b"\x00" + bytes(rgb) * size for _ in range(size))
    return (b"\x89PNG\r\n\x1a\n"
            + chunk(b"IHDR", struct.pack(">IIBBBBB", size, size, 8, 2, 0, 0, 0))
            + chunk(b"IDAT", zlib.compress(raw))
            + chunk(b"IEND", b""))


def _inputs(n: int) -> list[ImageInput]:
    return [ImageInput(origin_url=f"https://ota.test/p{i}.png", data=_png((i * 20 % 255, 40, 60))) for i in range(n)]


def _reply(items: list[dict], *, prompt_tokens=1000, out_tokens=50) -> dict:
    """generateContent 성공 응답 흉내. 실호출에서 확인한 구조 그대로 —
    parts 에 text 와 thoughtSignature 가 같이 실린다."""
    return {
        "candidates": [{
            "content": {"parts": [{"text": json.dumps({"items": items}, ensure_ascii=False),
                                   "thoughtSignature": "xxx"}]},
            "finishReason": "STOP",
        }],
        "usageMetadata": {"promptTokenCount": prompt_tokens, "candidatesTokenCount": out_tokens},
    }


def _item(ref, label="침실", alt="침대와 협탁이 놓인 방", conf=0.95):
    return {"ref": ref, "label": label, "alt_text": alt, "confidence": conf}


def _client(handler) -> httpx.AsyncClient:
    return httpx.AsyncClient(transport=httpx.MockTransport(handler), timeout=5.0)


@pytest.fixture(autouse=True)
def _api_key(monkeypatch):
    """테스트 환경엔 키가 없다(의도된 것) — 클라이언트 경로를 타려면 넣어줘야 한다."""
    monkeypatch.setattr(llm.external_api_config, "gemini_api_key", "test-key")


# ---- 기본 경로 -------------------------------------------------------------
async def test_parses_label_alt_and_confidence():
    """검증: 정상 응답 1장.
    기대결과: label·alt_text·confidence 가 파싱되고 ok=True, 신뢰도가 높아 확인 불필요."""
    def handler(request):
        return httpx.Response(200, json=_reply([_item("img-0", "침실", "침대와 협탁이 놓인 방", 0.93)]))

    async with _client(handler) as c:
        res = await analyze_images(_inputs(1), category=PlaceCategory.LODGING, client=c)

    assert len(res) == 1
    assert res[0].ok is True
    assert res[0].label == "침실"
    assert res[0].alt_text == "침대와 협탁이 놓인 방"
    assert res[0].confidence == pytest.approx(0.93)
    assert res[0].needs_review is False


async def test_result_length_always_matches_input():
    """검증: 입력 5장인데 응답에 3장만 온다.
    기대결과: 길이 5 유지 · 빠진 2장은 ok=False, needs_review=True — ★ 조용히 사라지지 않는다."""
    def handler(request):
        return httpx.Response(200, json=_reply([_item("img-0"), _item("img-1"), _item("img-2")]))

    images = _inputs(5)
    async with _client(handler) as c:
        res = await analyze_images(images, client=c)

    assert len(res) == 5
    assert [r.origin_url for r in res] == [i.origin_url for i in images]
    assert all(r.ok for r in res[:3])
    for r in res[3:]:
        assert r.ok is False and r.needs_review is True
        assert "ref" in (r.error or "")


async def test_matches_by_ref_not_by_order():
    """검증: 모델이 순서를 뒤집어 돌려준다(img-2, img-0, img-1).
    기대결과: ★ ref 로 정확히 매칭된다 — 순서로 매칭하면 엉뚱한 사진에 남의 alt 가 붙는다."""
    def handler(request):
        return httpx.Response(200, json=_reply([
            _item("img-2", "주방", "싱크대와 조리대"),
            _item("img-0", "외관", "건물 정면"),
            _item("img-1", "욕실", "세면대와 샤워부스"),
        ]))

    images = _inputs(3)
    async with _client(handler) as c:
        res = await analyze_images(images, client=c)

    by_url = {r.origin_url: r for r in res}
    assert by_url[images[0].origin_url].label == "외관"
    assert by_url[images[1].origin_url].label == "욕실"
    assert by_url[images[2].origin_url].label == "주방"


async def test_unknown_ref_in_response_is_discarded():
    """검증: 모델이 존재하지 않는 ref(img-99)를 지어낸다.
    기대결과: 버려지고, 실제 사진은 '응답 누락'으로 표시된다."""
    def handler(request):
        return httpx.Response(200, json=_reply([_item("img-99", "침실")]))

    async with _client(handler) as c:
        res = await analyze_images(_inputs(1), client=c)

    assert len(res) == 1
    assert res[0].ok is False and res[0].needs_review is True


# ---- 신뢰도 게이트 ---------------------------------------------------------
async def test_low_confidence_goes_to_review_queue():
    """검증: 신뢰도 0.4 로 돌아온 사진(임계값 0.7).
    기대결과: ok=True 지만 needs_review=True — ★ 자동 반영하지 않고 사람이 본다."""
    def handler(request):
        return httpx.Response(200, json=_reply([_item("img-0", "침실", "흐릿한 실내", 0.4)]))

    async with _client(handler) as c:
        res = await analyze_images(_inputs(1), confidence_threshold=0.7, client=c)

    assert res[0].ok is True
    assert res[0].needs_review is True


async def test_threshold_is_configurable():
    """검증: 같은 0.4 응답에 임계값을 0.3 으로 낮춘다.
    기대결과: 확인 불필요로 내려간다 — 임계값이 실제로 파라미터로 동작한다."""
    def handler(request):
        return httpx.Response(200, json=_reply([_item("img-0", "침실", "실내", 0.4)]))

    async with _client(handler) as c:
        res = await analyze_images(_inputs(1), confidence_threshold=0.3, client=c)

    assert res[0].needs_review is False


async def test_empty_label_forces_review():
    """검증: 신뢰도는 높은데 label 이 빈 문자열이다.
    기대결과: needs_review=True — 쓸 수 없는 결과를 자동 반영하지 않는다."""
    def handler(request):
        return httpx.Response(200, json=_reply([_item("img-0", "", "설명", 0.99)]))

    async with _client(handler) as c:
        res = await analyze_images(_inputs(1), client=c)

    assert res[0].label is None
    assert res[0].needs_review is True


# ---- 배치 -----------------------------------------------------------------
async def test_splits_into_batches():
    """검증: 12장을 batch_size=5 로 보낸다.
    기대결과: 3번 호출된다(5+5+2) — 20~50장을 한 번에 밀어넣지 않는다."""
    calls = []

    def handler(request):
        body = json.loads(request.content)
        refs = [p["text"].split("]")[0][1:] for p in body["contents"][0]["parts"]
                if "text" in p and p["text"].startswith("[img-")]
        calls.append(len(refs))
        return httpx.Response(200, json=_reply([_item(r) for r in refs]))

    async with _client(handler) as c:
        res = await analyze_images(_inputs(12), batch_size=5, client=c)

    assert calls == [5, 5, 2]
    assert len(res) == 12
    assert all(r.ok for r in res)


async def test_one_failed_batch_does_not_kill_the_rest():
    """검증: 3배치 중 두 번째만 500 을 반환한다.
    기대결과: ★ 나머지 배치는 살아남고, 실패 배치의 사진만 ok=False 로 표시된다."""
    state = {"n": 0}

    def handler(request):
        state["n"] += 1
        body = json.loads(request.content)
        refs = [p["text"][1:-1] for p in body["contents"][0]["parts"]
                if "text" in p and p["text"].startswith("[img-")]
        if state["n"] == 2:
            return httpx.Response(500, text="boom")
        return httpx.Response(200, json=_reply([_item(r) for r in refs]))

    async with _client(handler) as c:
        res = await analyze_images(_inputs(6), batch_size=2, max_retries=0, client=c)

    assert len(res) == 6
    assert [r.ok for r in res] == [True, True, False, False, True, True]
    assert all(r.needs_review for r in res[2:4])


# ---- 재시도 ---------------------------------------------------------------
async def test_retries_5xx_then_succeeds():
    """검증: 첫 호출 503, 두 번째 200.
    기대결과: 재시도로 성공한다 — 일시적 장애로 사진을 버리지 않는다."""
    state = {"n": 0}

    def handler(request):
        state["n"] += 1
        if state["n"] == 1:
            return httpx.Response(503, text="unavailable")
        return httpx.Response(200, json=_reply([_item("img-0")]))

    async with _client(handler) as c:
        res = await analyze_images(_inputs(1), max_retries=2, client=c)

    assert state["n"] == 2
    assert res[0].ok is True


async def test_does_not_retry_4xx():
    """검증: 400(잘못된 요청)을 반환한다.
    기대결과: 재시도하지 않는다 — 같은 요청을 다시 보내도 결과가 같고 요금만 나간다."""
    state = {"n": 0}

    def handler(request):
        state["n"] += 1
        return httpx.Response(400, text="bad request")

    async with _client(handler) as c:
        res = await analyze_images(_inputs(1), max_retries=3, client=c)

    assert state["n"] == 1
    assert res[0].ok is False


async def test_timeout_is_retried_then_reported():
    """검증: 매번 타임아웃이 난다.
    기대결과: max_retries 만큼 시도한 뒤 해당 사진을 ok=False 로 남긴다(예외를 밖으로 안 던진다)."""
    state = {"n": 0}

    def handler(request):
        state["n"] += 1
        raise httpx.ReadTimeout("timed out", request=request)

    async with _client(handler) as c:
        res = await analyze_images(_inputs(1), max_retries=1, client=c)

    assert state["n"] == 2
    assert res[0].ok is False and res[0].needs_review is True


# ---- 오류 처리 -------------------------------------------------------------
async def test_broken_json_fails_only_that_batch():
    """검증: 구조화 출력이 깨진 JSON 으로 온다.
    기대결과: 그 배치만 실패 처리된다 — 파싱 실패가 전체를 죽이지 않는다."""
    def handler(request):
        return httpx.Response(200, json={
            "candidates": [{"content": {"parts": [{"text": "{items: [oops"}]}, "finishReason": "STOP"}]
        })

    async with _client(handler) as c:
        res = await analyze_images(_inputs(2), max_retries=0, client=c)

    assert len(res) == 2
    assert all(r.ok is False and r.needs_review for r in res)


async def test_safety_blocked_response_is_handled():
    """검증: candidates 가 비어 오는 경우(안전 필터 차단).
    기대결과: 예외가 새지 않고 ok=False 로 남는다."""
    def handler(request):
        return httpx.Response(200, json={"candidates": []})

    async with _client(handler) as c:
        res = await analyze_images(_inputs(1), max_retries=0, client=c)

    assert res[0].ok is False


async def test_missing_key_raises_not_configured(monkeypatch):
    """검증: GEMINI_API_KEY 가 비어 있다.
    기대결과: GeminiNotConfigured — ★ 서버 부팅은 막지 않고 이 어댑터만 비활성이다."""
    monkeypatch.setattr(llm.external_api_config, "gemini_api_key", "")
    assert llm.is_configured() is False
    with pytest.raises(GeminiNotConfigured):
        await analyze_images(_inputs(1))


async def test_auth_error_stops_everything():
    """검증: 401 이 돌아온다.
    기대결과: GeminiNotConfigured 로 전체 중단 — 나머지 배치를 태워봐야 똑같이 실패한다."""
    def handler(request):
        return httpx.Response(401, text="unauthorized")

    async with _client(handler) as c:
        with pytest.raises(GeminiNotConfigured):
            await analyze_images(_inputs(4), batch_size=2, client=c)


async def test_image_download_failure_is_isolated():
    """검증: 바이트 없이 URL 만 준 사진의 내려받기가 실패한다.
    기대결과: 그 사진만 ok=False, 같은 배치의 다른 사진은 정상 처리된다."""
    def handler(request):
        if request.method == "GET":
            return httpx.Response(404)
        body = json.loads(request.content)
        refs = [p["text"][1:-1] for p in body["contents"][0]["parts"]
                if "text" in p and p["text"].startswith("[img-")]
        return httpx.Response(200, json=_reply([_item(r) for r in refs]))

    images = [ImageInput(origin_url="https://ota.test/gone.png"), _inputs(1)[0]]
    async with _client(handler) as c:
        res = await analyze_images(images, client=c)

    assert len(res) == 2
    assert res[0].ok is False and "로드 실패" in res[0].error
    assert res[1].ok is True


async def test_empty_input_returns_empty():
    """검증: 빈 목록을 넘긴다.
    기대결과: 빈 목록. 호출도 하지 않는다(요금 0)."""
    assert await analyze_images([]) == []


# ---- 프롬프트 계약 ---------------------------------------------------------
async def test_prompt_forbids_promotional_adjectives():
    """검증: 실제로 보내는 요청 본문의 지시문.
    기대결과: ★ 홍보성 형용사 금지와 '지어내지 마라' 가 들어 있다 — LLM 은 사실을 만들지 않는다."""
    captured = {}

    def handler(request):
        captured["body"] = json.loads(request.content)
        return httpx.Response(200, json=_reply([_item("img-0")]))

    async with _client(handler) as c:
        await analyze_images(_inputs(1), category=PlaceCategory.LODGING, client=c)

    prompt = captured["body"]["contents"][0]["parts"][0]["text"]
    assert "아름다운" in prompt and "홍보성" in prompt
    assert "지어내지 마라" in prompt
    assert "보이는 것만" in prompt


async def test_prompt_uses_category_vocabulary():
    """검증: 업종별 라벨 어휘.
    기대결과: 숙박엔 '침실'이, 카페엔 '디저트'가 들어간다 — 라벨이 업종마다 달라야 화면에서 묶인다."""
    captured = {}

    def handler(request):
        captured.setdefault("prompts", []).append(
            json.loads(request.content)["contents"][0]["parts"][0]["text"]
        )
        return httpx.Response(200, json=_reply([_item("img-0")]))

    async with _client(handler) as c:
        await analyze_images(_inputs(1), category=PlaceCategory.LODGING, client=c)
        await analyze_images(_inputs(1), category=PlaceCategory.CAFE, client=c)

    assert "침실" in captured["prompts"][0]
    assert "디저트" in captured["prompts"][1]


async def test_unit_names_are_passed_as_hint():
    """검증: 객실 이름 후보를 넘긴다.
    기대결과: 프롬프트에 실린다 — 라벨을 units 와 맞춰야 나중에 사진이 객실에 붙는다."""
    captured = {}

    def handler(request):
        captured["prompt"] = json.loads(request.content)["contents"][0]["parts"][0]["text"]
        return httpx.Response(200, json=_reply([_item("img-0")]))

    async with _client(handler) as c:
        await analyze_images(_inputs(1), unit_names=["A동 스탠다드", "B동 복층"], client=c)

    assert "A동 스탠다드" in captured["prompt"]


async def test_structured_output_schema_is_sent():
    """검증: 요청의 generationConfig.
    기대결과: responseMimeType=application/json + responseSchema 가 실린다 — 파싱 실패를 줄이는 장치."""
    captured = {}

    def handler(request):
        captured["body"] = json.loads(request.content)
        return httpx.Response(200, json=_reply([_item("img-0")]))

    async with _client(handler) as c:
        await analyze_images(_inputs(1), client=c)

    cfg = captured["body"]["generationConfig"]
    assert cfg["responseMimeType"] == "application/json"
    assert cfg["responseSchema"]["properties"]["items"]["items"]["required"] == [
        "ref", "label", "alt_text", "confidence"
    ]
    assert cfg["temperature"] == 0


async def test_png_mime_is_sniffed_not_guessed():
    """검증: PNG 바이트를 mime_type 없이 넘긴다.
    기대결과: image/png 로 판별된다 — 수집한 URL 의 확장자는 자주 거짓말한다."""
    captured = {}

    def handler(request):
        captured["body"] = json.loads(request.content)
        return httpx.Response(200, json=_reply([_item("img-0")]))

    async with _client(handler) as c:
        await analyze_images(_inputs(1), client=c)

    inline = [p["inline_data"] for p in captured["body"]["contents"][0]["parts"] if "inline_data" in p]
    assert inline[0]["mime_type"] == "image/png"
    assert base64.b64decode(inline[0]["data"]).startswith(b"\x89PNG")


async def test_confidence_is_clamped():
    """검증: 모델이 범위 밖 신뢰도(1.7, -0.2)를 준다.
    기대결과: 0.0~1.0 으로 잘린다 — 오염된 값이 아래 로직으로 흘러가지 않는다."""
    def handler(request):
        return httpx.Response(200, json=_reply([
            _item("img-0", conf=1.7), _item("img-1", conf=-0.2),
        ]))

    async with _client(handler) as c:
        res = await analyze_images(_inputs(2), client=c)

    assert res[0].confidence == 1.0
    assert res[1].confidence == 0.0
    assert res[1].needs_review is True