855c4edde4
test_execution_engine, test_mcp_server_code_execution test_mcp_server_files_search test_mcp_server_web_search müssen noch gemacht werden
225 lines
8.7 KiB
Python
225 lines
8.7 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
|
|
|
|
|
|
# ── Initialization ────────────────────────────────────────────────────────────
|
|
|
|
class TestChatManagerInit:
|
|
def test_chat_history_starts_empty(self):
|
|
cm = ChatManager()
|
|
assert cm.chat_history == []
|
|
|
|
def test_api_url_contains_endpoint(self):
|
|
cm = ChatManager()
|
|
assert isinstance(cm.api_url, str)
|
|
assert "/v1/chat/completions" in cm.api_url
|
|
|
|
|
|
# ── History management ────────────────────────────────────────────────────────
|
|
|
|
class TestHistory:
|
|
@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_get_history_returns_internal_list(self, cm):
|
|
cm.add_message("user", "Hi")
|
|
assert cm.get_history() is cm.chat_history
|
|
|
|
def test_clear_history_empties_list(self, cm):
|
|
cm.add_message("user", "Hi")
|
|
cm.clear_history()
|
|
assert cm.chat_history == []
|
|
|
|
def test_clear_history_on_empty_is_safe(self, cm):
|
|
cm.clear_history()
|
|
assert cm.chat_history == []
|
|
|
|
|
|
# ── send_message (mocked HTTP) ────────────────────────────────────────────────
|
|
|
|
class TestSendMessage:
|
|
@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_history_grows_by_two_per_call(self, cm):
|
|
with patch("requests.post", return_value=_mock_ok()):
|
|
cm.send_message("First")
|
|
cm.send_message("Second")
|
|
assert len(cm.chat_history) == 4
|
|
|
|
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_missing_choices_key_raises(self, cm):
|
|
mock = MagicMock()
|
|
mock.status_code = 200
|
|
mock.json.return_value = {}
|
|
with patch("requests.post", return_value=mock):
|
|
with pytest.raises(Exception):
|
|
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(self, cm):
|
|
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")
|
|
|
|
|
|
# ── get_chat_display ──────────────────────────────────────────────────────────
|
|
|
|
class TestGetChatDisplay:
|
|
@pytest.fixture
|
|
def cm(self):
|
|
return ChatManager()
|
|
|
|
def test_empty_history_returns_empty_list(self, cm):
|
|
assert cm.get_chat_display() == []
|
|
|
|
def test_display_has_role_and_content_keys(self, cm):
|
|
cm.add_message("user", "Hello")
|
|
entry = cm.get_chat_display()[0]
|
|
assert "role" in entry
|
|
assert "content" in entry
|
|
|
|
def test_display_preserves_message_order(self, cm):
|
|
cm.add_message("user", "First")
|
|
cm.add_message("assistant", "Second")
|
|
display = cm.get_chat_display()
|
|
assert display[0]["role"] == "user"
|
|
assert display[1]["role"] == "assistant"
|
|
|
|
def test_display_returns_copy_not_reference(self, cm):
|
|
"""Mutating the returned list must not corrupt internal history."""
|
|
cm.add_message("user", "Hi")
|
|
display = cm.get_chat_display()
|
|
display.clear()
|
|
assert len(cm.chat_history) == 1
|
|
|
|
def test_system_messages_included_in_display(self, cm):
|
|
cm.add_message("system", "Be helpful.")
|
|
assert cm.get_chat_display()[0]["role"] == "system"
|
|
|
|
|
|
# ── Integration (skipped when API unreachable) ────────────────────────────────
|
|
|
|
class TestSendMessageIntegration:
|
|
@pytest.fixture
|
|
def cm(self):
|
|
return ChatManager()
|
|
|
|
def test_real_api_returns_non_empty_string(self, cm):
|
|
try:
|
|
response = cm.send_message("Reply with exactly the word PONG.")
|
|
assert isinstance(response, str)
|
|
assert len(response) > 0
|
|
assert len(cm.chat_history) == 2
|
|
except Exception as e:
|
|
pytest.skip(f"API not reachable: {e}")
|
|
|
|
def test_real_api_multi_turn_history_grows(self, cm):
|
|
try:
|
|
cm.send_message("Remember the number 42.")
|
|
cm.send_message("What number did I ask you to remember?")
|
|
assert len(cm.chat_history) == 4
|
|
except Exception as e:
|
|
pytest.skip(f"API not reachable: {e}")
|