137 lines
5.5 KiB
Python
137 lines
5.5 KiB
Python
"""Tests for ChatManager (backend/managers/chat_manager.py)."""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
import pytest
|
|
import requests
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from backend.managers.chat_manager import ChatManager
|
|
|
|
|
|
# ── Helpers ──────────────────────────────────────────────────────────────────
|
|
|
|
def _mock_ok(content="AI reply"):
|
|
"""Return a mocked 200 response with a single assistant choice."""
|
|
mock = MagicMock()
|
|
mock.status_code = 200
|
|
mock.json.return_value = {
|
|
"choices": [{"message": {"role": "assistant", "content": content}}]
|
|
}
|
|
mock.text = ""
|
|
return mock
|
|
|
|
|
|
# ── History management ────────────────────────────────────────────────────────
|
|
|
|
class TestHistory:
|
|
"""Tests for add_message and clear_history."""
|
|
|
|
@pytest.fixture
|
|
def cm(self):
|
|
return ChatManager()
|
|
|
|
def test_add_message_appends_correct_entry(self, cm):
|
|
cm.add_message("user", "Hello")
|
|
assert cm.chat_history == [{"role": "user", "content": "Hello"}]
|
|
|
|
def test_add_multiple_messages_preserves_order(self, cm):
|
|
cm.add_message("user", "Hi")
|
|
cm.add_message("assistant", "Hello!")
|
|
assert cm.chat_history[0]["role"] == "user"
|
|
assert cm.chat_history[1]["role"] == "assistant"
|
|
|
|
def test_clear_history_empties_list(self, cm):
|
|
cm.add_message("user", "Hi")
|
|
cm.clear_history()
|
|
assert cm.chat_history == []
|
|
|
|
|
|
# ── send_message (mocked HTTP) ────────────────────────────────────────────────
|
|
|
|
class TestSendMessage:
|
|
"""Tests for send_message: history updates, HTTP payload, error handling, and auth headers."""
|
|
|
|
@pytest.fixture
|
|
def cm(self):
|
|
return ChatManager()
|
|
|
|
def test_user_message_added_to_history(self, cm):
|
|
with patch("requests.post", return_value=_mock_ok()):
|
|
cm.send_message("Hello")
|
|
assert cm.chat_history[0] == {"role": "user", "content": "Hello"}
|
|
|
|
def test_assistant_reply_added_to_history(self, cm):
|
|
with patch("requests.post", return_value=_mock_ok("Hi there")):
|
|
cm.send_message("Hello")
|
|
assert cm.chat_history[1] == {"role": "assistant", "content": "Hi there"}
|
|
|
|
def test_returns_assistant_content_string(self, cm):
|
|
with patch("requests.post", return_value=_mock_ok("Answer")):
|
|
result = cm.send_message("Question")
|
|
assert result == "Answer"
|
|
|
|
def test_full_history_sent_in_request_payload(self, cm):
|
|
"""All prior messages must be forwarded so the model has conversation context."""
|
|
cm.add_message("system", "You are helpful.")
|
|
with patch("requests.post", return_value=_mock_ok()) as mock_post:
|
|
cm.send_message("Hello")
|
|
payload = mock_post.call_args.kwargs["json"]
|
|
assert payload["messages"][0]["role"] == "system"
|
|
assert payload["messages"][1]["role"] == "user"
|
|
|
|
def test_connection_error_raises_and_adds_error_to_history(self, cm):
|
|
with patch("requests.post", side_effect=requests.exceptions.ConnectionError("refused")):
|
|
with pytest.raises(Exception, match="Connection Error"):
|
|
cm.send_message("Hello")
|
|
assert any("Error" in msg["content"] for msg in cm.chat_history)
|
|
|
|
def test_api_error_status_raises(self, cm):
|
|
mock = MagicMock()
|
|
mock.status_code = 500
|
|
mock.text = "Internal Server Error"
|
|
with patch("requests.post", return_value=mock):
|
|
with pytest.raises(Exception, match="API Error 500"):
|
|
cm.send_message("Hello")
|
|
|
|
def test_timeout_raises(self, cm):
|
|
with patch("requests.post", side_effect=requests.exceptions.Timeout()):
|
|
with pytest.raises(Exception):
|
|
cm.send_message("Hello")
|
|
|
|
def test_empty_choices_raises(self, cm):
|
|
mock = MagicMock()
|
|
mock.status_code = 200
|
|
mock.json.return_value = {"choices": []}
|
|
with patch("requests.post", return_value=mock):
|
|
with pytest.raises(Exception, match="Invalid API response format"):
|
|
cm.send_message("Hello")
|
|
|
|
def test_api_key_included_in_header_when_set(self, cm):
|
|
cm.api_key = "test-key-123"
|
|
with patch("requests.post", return_value=_mock_ok()) as mock_post:
|
|
cm.send_message("Hello")
|
|
headers = mock_post.call_args.kwargs["headers"]
|
|
assert headers.get("Authorization") == "Bearer test-key-123"
|
|
|
|
def test_api_key_excluded_from_header_when_empty_sentinel(self, cm):
|
|
# "EMPTY" is the sentinel string the UI writes when the user leaves the key field blank.
|
|
cm.api_key = "EMPTY"
|
|
with patch("requests.post", return_value=_mock_ok()) as mock_post:
|
|
cm.send_message("Hello")
|
|
headers = mock_post.call_args.kwargs["headers"]
|
|
assert "Authorization" not in headers
|
|
|
|
def test_json_decode_error_raises(self, cm):
|
|
import json
|
|
mock = MagicMock()
|
|
mock.status_code = 200
|
|
mock.json.side_effect = json.JSONDecodeError("bad json", "", 0)
|
|
with patch("requests.post", return_value=mock):
|
|
with pytest.raises(Exception, match="JSON Decode Error"):
|
|
cm.send_message("Hello")
|
|
|
|
|