Mina Choi
11d30bb3d1
여러 줄 주석이 설명보다 경위(예전·실측·지적)를 적고 있어 읽는 사람이 결론을 찾기 어려웠다. - ts·tsx·js·mjs·css·py 478개: 여러 줄 주석은 첫 문장 한 줄로, 과거형·날짜 문장은 삭제 - 주석 위치는 TypeScript 파서·파이썬 tokenize/ast 로 찾는다 — 문자열 안의 # · /* 는 건드리지 않는다 - eslint·ts·noqa·type: ignore 같은 지시 주석은 그대로 둔다 파이썬 275개 정리 전후 AST 동일, TS 298개 주석 뺀 토큰 동일(빈 JSX 주석 10곳만 차이). site·frontend·admin tsc, site vitest 105 passed Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
421 lines
14 KiB
Python
421 lines
14 KiB
Python
"""Gemini Vision 클라이언트 — 사진 분류·alt 생성의 계약."""
|
|
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."""
|
|
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 성공 응답 흉내."""
|
|
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장."""
|
|
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장만 온다."""
|
|
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)."""
|
|
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)."""
|
|
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 이 빈 문자열이다."""
|
|
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 로 보낸다."""
|
|
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 을 반환한다."""
|
|
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():
|
|
"""검증: 매번 타임아웃이 난다."""
|
|
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 가 비어 오는 경우(안전 필터 차단)."""
|
|
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 가 비어 있다."""
|
|
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 이 돌아온다."""
|
|
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 만 준 사진의 내려받기가 실패한다."""
|
|
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():
|
|
"""검증: 빈 목록을 넘긴다."""
|
|
assert await analyze_images([]) == []
|
|
|
|
|
|
# 프롬프트 계약
|
|
async def test_prompt_forbids_promotional_adjectives():
|
|
"""검증: 실제로 보내는 요청 본문의 지시문."""
|
|
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():
|
|
"""검증: 객실 이름 후보를 넘긴다."""
|
|
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."""
|
|
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 없이 넘긴다."""
|
|
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)를 준다."""
|
|
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
|