"""Tests for /api/v1/llm/draft-description. Anthropic SDK streaming mock'lanır. Gerçek API çağrısı yok. """ from __future__ import annotations import json from collections.abc import AsyncIterator from unittest.mock import AsyncMock, MagicMock, patch import pytest from httpx import ASGITransport, AsyncClient from app.config import get_settings from app.main import app def _parse_sse(body: str) -> list[tuple[str, dict]]: events: list[tuple[str, dict]] = [] current_event = "" current_data: list[str] = [] for line in body.split("\n"): if line.startswith("event: "): current_event = line.removeprefix("event: ").strip() elif line.startswith("data: "): current_data.append(line.removeprefix("data: ")) elif line == "" and current_event: events.append((current_event, json.loads("\n".join(current_data)))) current_event = "" current_data = [] return events def _sample_payload() -> dict: return { "title": "Post-kuantum dirençli ağ geçidi", "jurisdiction": "TR", "reference_code": "P-2026-010", "intake": { "technical_field": "Veri haberleşme güvenliği ve kriptografi.", "problem": "Klasik TLS el sıkışması kuantum bilgisayarla kırılabilir.", "prior_art": "Cloudflare X25519+Kyber768 hibriti denemiştir.", "solution": "ML-DSA imzalı, X25519 tabanlı anahtar paylaşımı.", "novelty": "Tek handshake içinde çift algoritma zinciri.", "advantages": "Kuantum dirençli, <10ms gecikme.", "industrial_applicability": "Bankacılık, e-devlet.", "variations": "X448 veya ML-DSA-87 kullanımı mümkün.", "terminology": "ML-DSA: Module-Lattice DSA.", "example_use": "Bankanın mobil uygulaması handshake'te hibriti kullanır.", }, } @pytest.mark.asyncio async def test_draft_description_returns_503_without_api_key() -> None: get_settings.cache_clear() with patch.dict("os.environ", {"ANTHROPIC_API_KEY": ""}, clear=False): get_settings.cache_clear() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.post("/api/v1/llm/draft-description", json=_sample_payload()) assert response.status_code == 503 get_settings.cache_clear() @pytest.mark.asyncio async def test_draft_description_streams_and_passes_system_prompt() -> None: """Mock client: SSE event sırası + system prompt cache_control + Opus 4.7 model.""" get_settings.cache_clear() # Mock streaming events — content_block_delta text_delta class FakeDelta: def __init__(self, text: str) -> None: self.type = "text_delta" self.text = text class FakeEvent: def __init__(self, event_type: str, delta: FakeDelta | None = None) -> None: self.type = event_type self.delta = delta async def fake_stream_iter() -> AsyncIterator[FakeEvent]: yield FakeEvent("content_block_delta", FakeDelta("## 1. Buluşun")) yield FakeEvent("content_block_delta", FakeDelta(" Ait Olduğu Teknik Alan\n")) fake_final = MagicMock() fake_final.stop_reason = "end_turn" fake_final.usage.input_tokens = 5000 fake_final.usage.output_tokens = 1200 fake_final.usage.cache_creation_input_tokens = 4370 fake_final.usage.cache_read_input_tokens = 0 fake_stream = MagicMock() fake_stream.__aiter__ = MagicMock(return_value=fake_stream_iter()) fake_stream.get_final_message = AsyncMock(return_value=fake_final) fake_ctx = MagicMock() fake_ctx.__aenter__ = AsyncMock(return_value=fake_stream) fake_ctx.__aexit__ = AsyncMock(return_value=None) fake_client = MagicMock() fake_client.messages.stream = MagicMock(return_value=fake_ctx) with patch.dict("os.environ", {"ANTHROPIC_API_KEY": "sk-ant-test"}, clear=False): get_settings.cache_clear() with patch("app.api.drafting.get_anthropic_client", return_value=fake_client): transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.post( "/api/v1/llm/draft-description", json=_sample_payload() ) assert response.status_code == 200 events = _parse_sse(response.text) event_types = [e[0] for e in events] assert event_types == ["start", "delta", "delta", "end"] start_data = events[0][1] assert start_data["model"] == "claude-opus-4-7" assert start_data["jurisdiction"] == "TR" assert start_data["title"] == "Post-kuantum dirençli ağ geçidi" assert events[1][1]["text"] == "## 1. Buluşun" assert events[2][1]["text"] == " Ait Olduğu Teknik Alan\n" end_data = events[3][1] assert end_data["stop_reason"] == "end_turn" assert end_data["usage"]["cache_creation_input_tokens"] == 4370 # Opus 4.7 + cache_control + adaptive thinking parametreleri verify call_kwargs = fake_client.messages.stream.call_args.kwargs assert call_kwargs["model"] == "claude-opus-4-7" assert call_kwargs["thinking"] == {"type": "adaptive"} assert call_kwargs["system"][0]["cache_control"] == {"type": "ephemeral"} assert "TürkPatent" in call_kwargs["system"][0]["text"] assert ( "8-bölümlü" in call_kwargs["messages"][0]["content"] or "tarifname" in call_kwargs["messages"][0]["content"] ) # Intake değerleri kullanıcı mesajında görünmeli assert "X25519" in call_kwargs["messages"][0]["content"] assert "ML-DSA" in call_kwargs["messages"][0]["content"] get_settings.cache_clear() @pytest.mark.asyncio async def test_draft_description_handles_empty_intake_fields() -> None: """Boş intake alanı → placeholder, uydurma yapılmaz.""" payload = _sample_payload() payload["intake"]["prior_art"] = "" payload["intake"]["variations"] = "" # Empty stream + final — amacımız sadece stream() kwargs'ını yakalamak async def empty_iter() -> AsyncIterator: if False: yield # pragma: no cover fake_final = MagicMock() fake_final.stop_reason = "end_turn" fake_final.usage.input_tokens = 100 fake_final.usage.output_tokens = 0 fake_final.usage.cache_creation_input_tokens = 0 fake_final.usage.cache_read_input_tokens = 0 fake_stream = MagicMock() fake_stream.__aiter__ = MagicMock(return_value=empty_iter()) fake_stream.get_final_message = AsyncMock(return_value=fake_final) fake_ctx = MagicMock() fake_ctx.__aenter__ = AsyncMock(return_value=fake_stream) fake_ctx.__aexit__ = AsyncMock(return_value=None) fake_client = MagicMock() fake_client.messages.stream = MagicMock(return_value=fake_ctx) with patch.dict("os.environ", {"ANTHROPIC_API_KEY": "sk-ant-test"}, clear=False): get_settings.cache_clear() with patch("app.api.drafting.get_anthropic_client", return_value=fake_client): transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: await client.post("/api/v1/llm/draft-description", json=payload) call_kwargs = fake_client.messages.stream.call_args.kwargs user_content = call_kwargs["messages"][0]["content"] assert "[kullanıcı tarafından tamamlanacaktır]" in user_content # Dolu alanlar placeholder ile karışmamalı assert "X25519" in user_content get_settings.cache_clear() @pytest.mark.asyncio async def test_draft_description_validates_title_length() -> None: payload = _sample_payload() payload["title"] = "ab" # min_length=3 with patch.dict("os.environ", {"ANTHROPIC_API_KEY": "sk-ant-test"}, clear=False): get_settings.cache_clear() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.post("/api/v1/llm/draft-description", json=payload) assert response.status_code == 422 get_settings.cache_clear() @pytest.mark.asyncio async def test_draft_description_rejects_invalid_jurisdiction() -> None: payload = _sample_payload() payload["jurisdiction"] = "JP" # pattern TR|EP|US|PCT with patch.dict("os.environ", {"ANTHROPIC_API_KEY": "sk-ant-test"}, clear=False): get_settings.cache_clear() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.post("/api/v1/llm/draft-description", json=payload) assert response.status_code == 422 get_settings.cache_clear()