mirror of
http://88.130.71.182:3000/BlitTech/contexta_be.git
synced 2026-06-13 08:56:20 +00:00
123 lines
4.4 KiB
Python
123 lines
4.4 KiB
Python
"""Tests for the /chat/{id}/test endpoint."""
|
|
import pytest
|
|
from unittest.mock import MagicMock, AsyncMock, patch
|
|
|
|
AUTH = {"Authorization": "Bearer test-token"}
|
|
|
|
|
|
def _make_chatbot():
|
|
return {
|
|
"id": "cb-1",
|
|
"name": "Test Bot",
|
|
"is_published": True,
|
|
"qdrant_collection_name": "col-1",
|
|
"company_id": "company-1",
|
|
"handoff_enabled": False,
|
|
"handoff_keywords": [],
|
|
"lead_capture_enabled": False,
|
|
"lead_capture_trigger": None,
|
|
"booking_enabled": False,
|
|
"system_prompt": "",
|
|
"companies": {"name": "Acme", "logo_url": None},
|
|
}
|
|
|
|
|
|
def _make_sb(chatbot=None, is_owner=True):
|
|
sb = MagicMock()
|
|
|
|
def table_side(name):
|
|
m = MagicMock()
|
|
m.select.return_value = m
|
|
m.eq.return_value = m
|
|
m.in_.return_value = m
|
|
m.limit.return_value = m
|
|
m.order.return_value = m
|
|
|
|
if name == "chatbots":
|
|
m.execute.return_value = MagicMock(data=[chatbot or _make_chatbot()])
|
|
elif name == "companies":
|
|
cid = "company-1" if is_owner else "other-company"
|
|
m.execute.return_value = MagicMock(data=[{"id": cid, "owner_id": "owner-1"}])
|
|
else:
|
|
m.execute.return_value = MagicMock(data=[], count=0)
|
|
return m
|
|
|
|
sb.table.side_effect = table_side
|
|
sb.auth = MagicMock()
|
|
return sb
|
|
|
|
|
|
class TestChatTestEndpoint:
|
|
def _run_test(self, client, questions, chatbot=None, is_owner=True, rag_result=None):
|
|
default_rag = {
|
|
"response": "The answer is 42.",
|
|
"sources": [],
|
|
"confidence_score": 0.82,
|
|
"tokens_used": 20,
|
|
"model": "test-model",
|
|
}
|
|
with patch("app.routers.chat.get_supabase") as mock_sb, \
|
|
patch("app.routers.chat.rag_engine") as mock_rag, \
|
|
patch("app.dependencies.get_current_user") as mock_user:
|
|
mock_rag.process_query = AsyncMock(return_value=rag_result or default_rag)
|
|
mock_sb.return_value = _make_sb(chatbot=chatbot, is_owner=is_owner)
|
|
user = MagicMock()
|
|
user.id = "owner-1"
|
|
mock_user.return_value = user
|
|
return client.post(
|
|
"/api/v1/chat/cb-1/test",
|
|
json={"questions": questions},
|
|
headers=AUTH,
|
|
)
|
|
|
|
def test_returns_list_of_results(self, client):
|
|
resp = self._run_test(client, ["What is your return policy?"])
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert isinstance(body, list)
|
|
assert len(body) == 1
|
|
|
|
def test_result_shape(self, client):
|
|
resp = self._run_test(client, ["Hello?"])
|
|
result = resp.json()[0]
|
|
assert "question" in result
|
|
assert "response" in result
|
|
assert "confidence_score" in result
|
|
assert "sources" in result
|
|
assert "model_used" in result
|
|
|
|
def test_question_echoed_in_result(self, client):
|
|
resp = self._run_test(client, ["What are your hours?"])
|
|
assert resp.json()[0]["question"] == "What are your hours?"
|
|
|
|
def test_multiple_questions_all_answered(self, client):
|
|
questions = ["Q1", "Q2", "Q3"]
|
|
resp = self._run_test(client, questions)
|
|
assert len(resp.json()) == 3
|
|
returned_questions = [r["question"] for r in resp.json()]
|
|
assert returned_questions == questions
|
|
|
|
def test_requires_authentication(self, client):
|
|
resp = client.post("/api/v1/chat/cb-1/test", json={"questions": ["hi"]})
|
|
assert resp.status_code == 401
|
|
|
|
def test_rejects_more_than_10_questions(self, client):
|
|
resp = self._run_test(client, [f"Q{i}" for i in range(11)])
|
|
assert resp.status_code == 422
|
|
|
|
def test_rejects_empty_question_list(self, client):
|
|
resp = self._run_test(client, [])
|
|
assert resp.status_code == 422
|
|
|
|
def test_chatbot_without_collection_returns_400(self, client):
|
|
bot = _make_chatbot()
|
|
bot["qdrant_collection_name"] = None
|
|
resp = self._run_test(client, ["Hi"], chatbot=bot)
|
|
assert resp.status_code == 400
|
|
|
|
def test_confidence_score_passed_through(self, client):
|
|
rag_result = {"response": "Sure", "sources": [], "confidence_score": 0.73,
|
|
"tokens_used": 5, "model": "m"}
|
|
resp = self._run_test(client, ["Question?"], rag_result=rag_result)
|
|
assert resp.json()[0]["confidence_score"] == pytest.approx(0.73)
|