Improve search

core/tests/test_ai_run_log.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+from asgiref.sync import async_to_sync
+from django.test import TestCase
+from unittest.mock import patch
+
+from core.messaging.ai import run_prompt
+from core.models import AI, AIRunLog, User
+
+
+class _FakeResponseMessage:
+    def __init__(self, content):
+        self.content = content
+
+
+class _FakeChoice:
+    def __init__(self, content):
+        self.message = _FakeResponseMessage(content)
+
+
+class _FakeResponse:
+    def __init__(self, content):
+        self.choices = [_FakeChoice(content)]
+
+
+class _FakeCompletionsOK:
+    async def create(self, **kwargs):
+        return _FakeResponse("ok-output")
+
+
+class _FakeChatOK:
+    def __init__(self):
+        self.completions = _FakeCompletionsOK()
+
+
+class _FakeClientOK:
+    def __init__(self, **kwargs):
+        self.chat = _FakeChatOK()
+
+
+class _FakeCompletionsFail:
+    async def create(self, **kwargs):
+        raise RuntimeError("provider boom")
+
+
+class _FakeChatFail:
+    def __init__(self):
+        self.completions = _FakeCompletionsFail()
+
+
+class _FakeClientFail:
+    def __init__(self, **kwargs):
+        self.chat = _FakeChatFail()
+
+
+class AIRunLogTests(TestCase):
+    def setUp(self):
+        self.user = User.objects.create_user(
+            username="ai-log-user",
+            email="ai-log@example.com",
+            password="x",
+        )
+        self.ai = AI.objects.create(
+            user=self.user,
+            base_url="https://example.invalid",
+            api_key="test-key",
+            model="gpt-4o-mini",
+        )
+        self.prompt = [{"role": "user", "content": "Hello world"}]
+
+    def test_run_prompt_logs_success(self):
+        with patch("core.messaging.ai.AsyncOpenAI", _FakeClientOK):
+            out = async_to_sync(run_prompt)(self.prompt, self.ai, operation="test_ok")
+        self.assertEqual("ok-output", out)
+        row = AIRunLog.objects.order_by("-id").first()
+        self.assertIsNotNone(row)
+        self.assertEqual("ok", row.status)
+        self.assertEqual("test_ok", row.operation)
+        self.assertEqual(1, row.message_count)
+        self.assertEqual(len("Hello world"), row.prompt_chars)
+        self.assertEqual(len("ok-output"), row.response_chars)
+        self.assertTrue((row.duration_ms or 0) >= 0)
+
+    def test_run_prompt_logs_failure(self):
+        with patch("core.messaging.ai.AsyncOpenAI", _FakeClientFail):
+            with self.assertRaises(RuntimeError):
+                async_to_sync(run_prompt)(self.prompt, self.ai, operation="test_fail")
+        row = AIRunLog.objects.order_by("-id").first()
+        self.assertIsNotNone(row)
+        self.assertEqual("failed", row.status)
+        self.assertEqual("test_fail", row.operation)
+        self.assertIn("provider boom", row.error)