155 lines
5.0 KiB
Python
155 lines
5.0 KiB
Python
import json
|
|
import pytest
|
|
import respx
|
|
import httpx
|
|
|
|
from app.interpret import pipeline
|
|
from app.interpret.pipeline import SajuError
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _patch_key(monkeypatch):
|
|
monkeypatch.setattr(pipeline, "ANTHROPIC_API_KEY", "test-key")
|
|
|
|
|
|
SAJU_ITEM_KEYS = [
|
|
"기질", "오행밸런스", "지지상호작용", "신살영향",
|
|
"재물운", "직업적성", "애정운", "건강운",
|
|
"현재대운", "올해세운", "인생황금기", "종합조언",
|
|
]
|
|
|
|
|
|
def _valid_saju_response():
|
|
items = []
|
|
for k in SAJU_ITEM_KEYS:
|
|
items.append({
|
|
"key": k, "title": "...", "content": "...",
|
|
"evidence": {"saju_element": "...", "reasoning": "..."}
|
|
})
|
|
return {
|
|
"items": items,
|
|
"summary": "...",
|
|
"advice": "...",
|
|
"warning": None,
|
|
"confidence": "medium",
|
|
}
|
|
|
|
|
|
def _valid_compat_response():
|
|
return {
|
|
"summary": "...",
|
|
"strengths": [
|
|
{"title": "오행 상생", "explanation": "...", "evidence": "..."},
|
|
{"title": "...", "explanation": "...", "evidence": "..."},
|
|
],
|
|
"challenges": [
|
|
{"title": "...", "explanation": "...", "evidence": "..."},
|
|
{"title": "...", "explanation": "...", "evidence": "..."},
|
|
],
|
|
"advice": "...",
|
|
"warning": None,
|
|
"confidence": "high",
|
|
}
|
|
|
|
|
|
def _claude_envelope(text: str, in_tok=200, out_tok=400):
|
|
return {
|
|
"content": [{"type": "text", "text": text}],
|
|
"usage": {"input_tokens": in_tok, "output_tokens": out_tok},
|
|
}
|
|
|
|
|
|
@respx.mock
|
|
async def test_saju_interpret_success():
|
|
respx.post("https://api.anthropic.com/v1/messages").mock(
|
|
return_value=httpx.Response(200, json=_claude_envelope(json.dumps(_valid_saju_response())))
|
|
)
|
|
result = await pipeline.interpret_saju(
|
|
saju={"day_stem": "辛"},
|
|
analysis={"element_balance": {"金": 3.0}},
|
|
daeun=[{"age": 10}],
|
|
current_year=2026,
|
|
)
|
|
assert result["reroll_count"] == 0
|
|
assert result["tokens_in"] == 200
|
|
assert result["cost_usd"] > 0
|
|
|
|
|
|
@respx.mock
|
|
async def test_saju_codeblock_stripped():
|
|
text = "```json\n" + json.dumps(_valid_saju_response()) + "\n```"
|
|
respx.post("https://api.anthropic.com/v1/messages").mock(
|
|
return_value=httpx.Response(200, json=_claude_envelope(text))
|
|
)
|
|
result = await pipeline.interpret_saju(saju={}, analysis={}, daeun=[], current_year=2026)
|
|
assert "interpretation_json" in result
|
|
|
|
|
|
@respx.mock
|
|
async def test_saju_reroll_then_success():
|
|
valid = json.dumps(_valid_saju_response())
|
|
invalid = json.dumps({"items": [], "summary": "...", "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_saju(saju={}, analysis={}, daeun=[], current_year=2026)
|
|
assert result["reroll_count"] == 1
|
|
|
|
|
|
@respx.mock
|
|
async def test_saju_reroll_fail_raises():
|
|
invalid = json.dumps({"items": [], "summary": "...", "advice": "", "confidence": "medium"})
|
|
respx.post("https://api.anthropic.com/v1/messages").mock(
|
|
return_value=httpx.Response(200, json=_claude_envelope(invalid))
|
|
)
|
|
with pytest.raises(SajuError):
|
|
await pipeline.interpret_saju(saju={}, analysis={}, daeun=[], current_year=2026)
|
|
|
|
|
|
@respx.mock
|
|
async def test_saju_http_error():
|
|
respx.post("https://api.anthropic.com/v1/messages").mock(
|
|
return_value=httpx.Response(500, text="boom")
|
|
)
|
|
with pytest.raises(SajuError):
|
|
await pipeline.interpret_saju(saju={}, analysis={}, daeun=[], current_year=2026)
|
|
|
|
|
|
@respx.mock
|
|
async def test_compat_interpret_success():
|
|
respx.post("https://api.anthropic.com/v1/messages").mock(
|
|
return_value=httpx.Response(200, json=_claude_envelope(json.dumps(_valid_compat_response())))
|
|
)
|
|
result = await pipeline.interpret_compat(
|
|
saju_a={"day_stem": "辛"}, saju_b={"day_stem": "丁"},
|
|
analysis_a={}, analysis_b={},
|
|
score=85, breakdown={"day_master_element": {"score": 25}},
|
|
)
|
|
assert result["reroll_count"] == 0
|
|
assert "interpretation_json" in result
|
|
|
|
|
|
@respx.mock
|
|
async def test_compat_reroll_then_success():
|
|
valid = json.dumps(_valid_compat_response())
|
|
invalid = json.dumps({"summary": "...", "strengths": [], "challenges": [], "advice": "", "confidence": "high"})
|
|
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_compat(
|
|
saju_a={}, saju_b={}, analysis_a={}, analysis_b={},
|
|
score=50, breakdown={},
|
|
)
|
|
assert result["reroll_count"] == 1
|
|
|
|
|
|
def test_calc_cost():
|
|
cost = pipeline.calc_cost(1_000_000, 1_000_000)
|
|
assert cost == pipeline.SAJU_COST_INPUT_PER_M + pipeline.SAJU_COST_OUTPUT_PER_M
|