115 lines
4.0 KiB
Python
115 lines
4.0 KiB
Python
import json
|
|
import pytest
|
|
import respx
|
|
import httpx
|
|
|
|
from app import pipeline
|
|
from app.models import TarotInterpretRequest, TarotCardDraw
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _patch_key(monkeypatch):
|
|
monkeypatch.setattr(pipeline, "ANTHROPIC_API_KEY", "test-key")
|
|
|
|
|
|
def _req():
|
|
return TarotInterpretRequest(
|
|
spread_type="three_card",
|
|
category="연애",
|
|
question="Q",
|
|
cards=[
|
|
TarotCardDraw(position="과거", card_id="the-fool", reversed=False),
|
|
TarotCardDraw(position="현재", card_id="the-magician", reversed=False),
|
|
TarotCardDraw(position="미래", card_id="the-empress", reversed=True),
|
|
],
|
|
cards_reference="...",
|
|
)
|
|
|
|
|
|
def _valid_response_json():
|
|
return {
|
|
"summary": "흐름이 있음",
|
|
"cards": [
|
|
{"position": "과거", "card": "the-fool", "reversed": False,
|
|
"interpretation": "...", "advice": "...",
|
|
"evidence": {"card_meaning_used": "...", "position_logic": "...", "category_lens": "..."}},
|
|
{"position": "현재", "card": "the-magician", "reversed": False,
|
|
"interpretation": "...", "advice": "...",
|
|
"evidence": {"card_meaning_used": "...", "position_logic": "...", "category_lens": "..."}},
|
|
{"position": "미래", "card": "the-empress", "reversed": True,
|
|
"interpretation": "...", "advice": "...",
|
|
"evidence": {"card_meaning_used": "...", "position_logic": "...", "category_lens": "..."}},
|
|
],
|
|
"interactions": [{"type": "synergy", "between": ["the-fool", "the-magician"], "explanation": "..."}],
|
|
"advice": "...",
|
|
"warning": None,
|
|
"confidence": "medium",
|
|
}
|
|
|
|
|
|
def _claude_envelope(text: str, in_tok=100, out_tok=200):
|
|
return {
|
|
"content": [{"type": "text", "text": text}],
|
|
"usage": {"input_tokens": in_tok, "output_tokens": out_tok},
|
|
}
|
|
|
|
|
|
@respx.mock
|
|
async def test_interpret_success():
|
|
respx.post("https://api.anthropic.com/v1/messages").mock(
|
|
return_value=httpx.Response(200, json=_claude_envelope(json.dumps(_valid_response_json())))
|
|
)
|
|
result = await pipeline.interpret(_req())
|
|
assert result["reroll_count"] == 0
|
|
assert result["model"] == pipeline.TAROT_MODEL
|
|
assert result["tokens_in"] == 100
|
|
assert result["cost_usd"] > 0
|
|
|
|
|
|
@respx.mock
|
|
async def test_interpret_codeblock_stripped():
|
|
text = "```json\n" + json.dumps(_valid_response_json()) + "\n```"
|
|
respx.post("https://api.anthropic.com/v1/messages").mock(
|
|
return_value=httpx.Response(200, json=_claude_envelope(text))
|
|
)
|
|
result = await pipeline.interpret(_req())
|
|
assert "interpretation_json" in result
|
|
|
|
|
|
@respx.mock
|
|
async def test_interpret_reroll_then_success():
|
|
valid = json.dumps(_valid_response_json())
|
|
invalid = json.dumps({"summary": "...", "cards": [], "interactions": [], "advice": "", "confidence": "medium"})
|
|
respx.post("https://api.anthropic.com/v1/messages").mock(
|
|
side_effect=[
|
|
httpx.Response(200, json=_claude_envelope(invalid)),
|
|
httpx.Response(200, json=_claude_envelope(valid)),
|
|
]
|
|
)
|
|
result = await pipeline.interpret(_req())
|
|
assert result["reroll_count"] == 1
|
|
|
|
|
|
@respx.mock
|
|
async def test_interpret_reroll_fail_raises():
|
|
invalid = json.dumps({"summary": "...", "cards": [], "interactions": [], "advice": "", "confidence": "medium"})
|
|
respx.post("https://api.anthropic.com/v1/messages").mock(
|
|
return_value=httpx.Response(200, json=_claude_envelope(invalid))
|
|
)
|
|
with pytest.raises(pipeline.TarotError):
|
|
await pipeline.interpret(_req())
|
|
|
|
|
|
@respx.mock
|
|
async def test_interpret_http_error():
|
|
respx.post("https://api.anthropic.com/v1/messages").mock(
|
|
return_value=httpx.Response(500, text="boom")
|
|
)
|
|
with pytest.raises(pipeline.TarotError):
|
|
await pipeline.interpret(_req())
|
|
|
|
|
|
def test_calc_cost():
|
|
cost = pipeline.calc_cost(1_000_000, 1_000_000)
|
|
assert cost == pipeline.TAROT_COST_INPUT_PER_M + pipeline.TAROT_COST_OUTPUT_PER_M
|