55 lines
1.8 KiB
Python
55 lines
1.8 KiB
Python
import io
|
|
from urllib import request
|
|
|
|
import pytest
|
|
|
|
from review_agent.llm import build_messages, stream_reply
|
|
from review_agent.models import Conversation
|
|
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
class FakeStreamingResponse:
|
|
def __iter__(self):
|
|
return iter(
|
|
[
|
|
b'data: {"choices":[{"delta":{"content":"A"}}]}\n\n',
|
|
b"data: not-json\n\n",
|
|
b'data: {"choices":[{"delta":{"content":"B"}}]}\n\n',
|
|
b"data: [DONE]\n\n",
|
|
]
|
|
)
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, traceback):
|
|
return False
|
|
|
|
|
|
def test_stream_reply_skips_malformed_sse_data(monkeypatch, settings, django_user_model):
|
|
settings.LLM_API_KEY = "key"
|
|
settings.LLM_MODEL = "model"
|
|
settings.LLM_BASE_URL = "https://example.test/v1"
|
|
monkeypatch.setattr(request, "urlopen", lambda req, timeout: FakeStreamingResponse())
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
conversation = Conversation.objects.create(user=user, title="会话")
|
|
|
|
chunks = list(stream_reply(conversation, "你好"))
|
|
|
|
assert chunks == ["A", "B"]
|
|
|
|
|
|
def test_build_messages_includes_knowledge_context(django_user_model):
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
conversation = Conversation.objects.create(user=user, title="会话")
|
|
|
|
messages = build_messages(conversation, "孙之烨是谁", knowledge_context="来源：简历\n孙之烨负责审核智能体项目。")
|
|
|
|
assert messages[0]["role"] == "system"
|
|
assert messages[1]["role"] == "system"
|
|
assert "全局知识库" in messages[1]["content"]
|
|
assert "孙之烨负责审核智能体项目" in messages[1]["content"]
|
|
assert messages[-1] == {"role": "user", "content": "孙之烨是谁"}
|