162 lines
6.3 KiB
Python
162 lines
6.3 KiB
Python
import csv
|
|
import json
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from src.llm_service import INSTRUCTIONS, classify
|
|
|
|
|
|
class ClassificationTests(unittest.TestCase):
|
|
def test_demo_sample_cases(self):
|
|
path = Path(__file__).resolve().parents[1] / "samples/sample_leads.csv"
|
|
with path.open(encoding="utf-8-sig", newline="") as sample:
|
|
for row in csv.DictReader(sample):
|
|
with self.subTest(text=row["text"]):
|
|
if row["expected_category"] == "invalid":
|
|
with self.assertRaises(ValueError):
|
|
classify(row["text"], "DEMO")
|
|
continue
|
|
result = classify(row["text"], "DEMO")
|
|
self.assertEqual(result.category, row["expected_category"])
|
|
self.assertEqual(result.needs_review, row["needs_review"] == "true")
|
|
self.assertTrue(result.reason.strip())
|
|
if result.category != "spam":
|
|
self.assertTrue(result.draft_reply.strip())
|
|
else:
|
|
self.assertEqual(result.draft_reply, "")
|
|
self.assertEqual(
|
|
(
|
|
result.input_tokens,
|
|
result.output_tokens,
|
|
result.cached_tokens,
|
|
),
|
|
(0, 0, 0),
|
|
)
|
|
self.assertEqual(result, classify(row["text"], "DEMO"))
|
|
|
|
def test_invalid_inputs(self):
|
|
for text in (None, " ", "я" * 4001, "a\x00b", "a\uffff", "\udc00"):
|
|
with self.subTest(text=repr(text)[:20]), self.assertRaises(ValueError):
|
|
classify(text, "DEMO")
|
|
with self.assertRaises(ValueError):
|
|
classify("Здравствуйте", "typo")
|
|
self.assertTrue(classify("я" * 4000, "DEMO").category)
|
|
|
|
def test_real_requires_key(self):
|
|
with (
|
|
patch.dict("os.environ", {"OPENAI_API_KEY": ""}),
|
|
self.assertRaisesRegex(ValueError, "OPENAI_API_KEY"),
|
|
):
|
|
classify("Нужен счёт", "REAL AI")
|
|
|
|
def run_with_response(self, content, status=200):
|
|
"""Run the installed OpenAI parser with a local HTTP transport."""
|
|
import httpx
|
|
from openai import OpenAI
|
|
|
|
def serve(request):
|
|
body = json.loads(request.content)
|
|
self.assertEqual(body["instructions"], INSTRUCTIONS)
|
|
self.assertFalse(body["store"])
|
|
self.assertEqual(
|
|
body["input"], [{"role": "user", "content": "Сломалась плита"}]
|
|
)
|
|
self.assertTrue(body["text"]["format"]["strict"])
|
|
payload = {
|
|
"id": "resp_test",
|
|
"created_at": 1,
|
|
"object": "response",
|
|
"status": "completed",
|
|
"model": "provider-model",
|
|
"output": [
|
|
{
|
|
"id": "msg_test",
|
|
"type": "message",
|
|
"status": "completed",
|
|
"role": "assistant",
|
|
"content": [content],
|
|
}
|
|
],
|
|
"usage": {
|
|
"input_tokens": 120,
|
|
"output_tokens": 31,
|
|
"total_tokens": 151,
|
|
"input_tokens_details": {"cached_tokens": 64},
|
|
"output_tokens_details": {"reasoning_tokens": 0},
|
|
},
|
|
}
|
|
if status != 200:
|
|
payload = {
|
|
"error": {
|
|
"message": "private-request-detail",
|
|
"type": "invalid_request_error",
|
|
}
|
|
}
|
|
elif content is None:
|
|
payload = {}
|
|
return httpx.Response(status, json=payload)
|
|
|
|
def client(**settings):
|
|
return OpenAI(
|
|
**settings,
|
|
http_client=httpx.Client(transport=httpx.MockTransport(serve)),
|
|
)
|
|
|
|
with (
|
|
patch.dict("os.environ", {"OPENAI_API_KEY": "unit-test"}),
|
|
patch("src.llm_service.OpenAI", client),
|
|
):
|
|
return classify("Сломалась плита", "REAL AI")
|
|
|
|
def test_real_parses_result_and_provider_usage(self):
|
|
data = {
|
|
"category": "normal",
|
|
"reason": "Нужен ремонт плиты.",
|
|
"draft_reply": "Уточните модель, пожалуйста.",
|
|
"needs_review": False,
|
|
}
|
|
content = {"type": "output_text", "text": json.dumps(data), "annotations": []}
|
|
output = self.run_with_response(content)
|
|
self.assertEqual(
|
|
(
|
|
output.model,
|
|
output.input_tokens,
|
|
output.output_tokens,
|
|
output.cached_tokens,
|
|
),
|
|
("provider-model", 120, 31, 64),
|
|
)
|
|
self.assertEqual(output.reason, "Нужен ремонт плиты.")
|
|
self.assertTrue(output.needs_review)
|
|
|
|
def test_real_rejects_refusal_and_invalid_schema(self):
|
|
responses = [
|
|
None, # HTTP 200 с пустым объектом вместо структуры Responses API.
|
|
{"type": "refusal", "refusal": "Отклонено"},
|
|
{
|
|
"type": "output_text",
|
|
"text": '{"category": "unknown"}',
|
|
"annotations": [],
|
|
},
|
|
]
|
|
for content in responses:
|
|
with (
|
|
self.subTest(content=content),
|
|
self.assertRaisesRegex(ValueError, "ответ"),
|
|
):
|
|
self.run_with_response(content)
|
|
|
|
def test_real_hides_provider_error_detail(self):
|
|
with self.assertRaisesRegex(ValueError, "401") as caught:
|
|
self.run_with_response(None, status=401)
|
|
self.assertNotIn("private-request-detail", str(caught.exception))
|
|
|
|
def test_unavailable_service_stops_after_one_retry(self):
|
|
with (
|
|
patch("openai._base_client.time.sleep") as delay,
|
|
self.assertRaisesRegex(ValueError, "503"),
|
|
):
|
|
self.run_with_response(None, status=503)
|
|
delay.assert_called_once()
|