samirshuplyakov/Test-task
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Завершить аудит и подготовить AI Lead Processor к публикации

Browse Source
This commit is contained in:
Stvponya committed 2026-11-11 17:15:52 +03:00
1 parent 634490174f
commit be36989483
14 files changed
+330 -135
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/test_classification.py
+86 -28
View File
@@ -1,14 +1,10 @@
import csv
import json
import unittest
from pathlib import Path
from unittest.mock import patch
import json
def classify(text, mode):
from src.llm_service import classify as run
return run(text, mode)
from src.llm_service import INSTRUCTIONS, classify
class ClassificationTests(unittest.TestCase):
@@ -29,7 +25,14 @@ class ClassificationTests(unittest.TestCase):
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.input_tokens,
result.output_tokens,
result.cached_tokens,
),
(0, 0, 0),
)
self.assertEqual(result, classify(row["text"], "DEMO"))
def test_invalid_inputs(self):
@@ -41,7 +44,10 @@ class ClassificationTests(unittest.TestCase):
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"):
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):
@@ -51,44 +57,94 @@ class ClassificationTests(unittest.TestCase):
def serve(request):
body = json.loads(request.content)
self.assertEqual(body["input"], [{"role": "user", "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}},
"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"}}
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)))
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):
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 = dict(category="normal", reason="Нужен ремонт плиты.", draft_reply="Уточните модель, пожалуйста.",
needs_review=False)
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.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": []},
{
"type": "output_text",
"text": '{"category": "unknown"}',
"annotations": [],
},
]
for content in responses:
with self.subTest(content=content), self.assertRaisesRegex(ValueError, "ответ"):
with (
self.subTest(content=content),
self.assertRaisesRegex(ValueError, "ответ"),
):
self.run_with_response(content)
def test_real_hides_provider_error_detail(self):
@@ -97,7 +153,9 @@ class ClassificationTests(unittest.TestCase):
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:
with self.assertRaisesRegex(ValueError, "503"):
self.run_with_response(None, status=503)
with (
patch("openai._base_client.time.sleep") as delay,
self.assertRaisesRegex(ValueError, "503"),
):
self.run_with_response(None, status=503)
delay.assert_called_once()