From 15fe5980731ea2b8301351756f8519ffe7bf2aaa Mon Sep 17 00:00:00 2001 From: Ida Dittrich Date: Thu, 26 Feb 2026 16:06:10 +0100 Subject: [PATCH] removed chatbot tests --- tests/functional/chatbot/__init__.py | 3 - tests/functional/chatbot/test_chatbot.py | 217 ----------------------- 2 files changed, 220 deletions(-) delete mode 100644 tests/functional/chatbot/__init__.py delete mode 100644 tests/functional/chatbot/test_chatbot.py diff --git a/tests/functional/chatbot/__init__.py b/tests/functional/chatbot/__init__.py deleted file mode 100644 index c51f6418..00000000 --- a/tests/functional/chatbot/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -# Copyright (c) 2025 Patrick Motsch -# All rights reserved. -"""Chatbot functional tests.""" diff --git a/tests/functional/chatbot/test_chatbot.py b/tests/functional/chatbot/test_chatbot.py deleted file mode 100644 index a87e8933..00000000 --- a/tests/functional/chatbot/test_chatbot.py +++ /dev/null @@ -1,217 +0,0 @@ -#!/usr/bin/env python3 -# Copyright (c) 2025 Patrick Motsch -# All rights reserved. -""" -Chatbot Functional Tests - -Tests the chatbot implementation to ensure: -1. Chatbot initialization works correctly -2. Streaming events are emitted properly -3. Tool calls execute correctly -4. Messages are stored in database -5. No infinite loops occur -""" - -import asyncio -import os -import sys -from pathlib import Path - -# Add the gateway to path (go up 2 levels from tests/functional/chatbot/) -_gateway_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..")) -if _gateway_path not in sys.path: - sys.path.insert(0, _gateway_path) - -import pytest -from modules.features.chatbot.chatbot import Chatbot -from modules.features.chatbot.chatbotAIBridge import AICenterChatModel -from modules.features.chatbot.chatbotMemory import DatabaseCheckpointer -from modules.features.chatbot.config import load_chatbot_config_from_dict -from modules.features.chatbot.streamingHelper import ChatStreamingHelper -from modules.datamodels.datamodelUam import User -from modules.datamodels.datamodelAi import OperationTypeEnum, ProcessingModeEnum - - -class TestChatbot: - """Test suite for chatbot functionality.""" - - @pytest.fixture - def test_user(self): - """Create a test user.""" - return User( - id="test_user_chatbot", - username="test_chatbot", - email="test@example.com", - fullName="Test Chatbot User", - language="de", - mandateId="test_mandate", - ) - - @pytest.fixture - def workflow_id(self): - """Generate a test workflow ID.""" - import uuid - return str(uuid.uuid4()) - - @pytest.mark.asyncio - async def test_chatbot_initialization(self, test_user, workflow_id): - """Test that chatbot can be initialized correctly.""" - # Load config - config = load_chatbot_config_from_dict({}, config_id="test") - - # Create system prompt - from datetime import datetime - system_prompt = config.systemPrompt.replace( - "{{DATE}}", - datetime.now().strftime("%d.%m.%Y") - ) - - # Create AI center model - operation_type = OperationTypeEnum[config.model.operationType] - processing_mode = ProcessingModeEnum[config.model.processingMode] - - model = AICenterChatModel( - user=test_user, - operation_type=operation_type, - processing_mode=processing_mode - ) - - # Create memory/checkpointer - memory = DatabaseCheckpointer(user=test_user, workflow_id=workflow_id) - - # Create chatbot instance - chatbot = await Chatbot.create( - model=model, - memory=memory, - system_prompt=system_prompt, - workflow_id=workflow_id - ) - - assert chatbot is not None - assert chatbot.model is not None - assert chatbot.memory is not None - assert chatbot.app is not None - assert chatbot.system_prompt == system_prompt - - @pytest.mark.asyncio - async def test_streaming_helper_role_from_message(self): - """Test ChatStreamingHelper.role_from_message.""" - from langchain_core.messages import HumanMessage, AIMessage, SystemMessage - - human_msg = HumanMessage(content="Hello") - assert ChatStreamingHelper.role_from_message(msg=human_msg) == "user" - - ai_msg = AIMessage(content="Hi there") - assert ChatStreamingHelper.role_from_message(msg=ai_msg) == "assistant" - - system_msg = SystemMessage(content="You are a helpful assistant") - assert ChatStreamingHelper.role_from_message(msg=system_msg) == "system" - - @pytest.mark.asyncio - async def test_streaming_helper_flatten_content(self): - """Test ChatStreamingHelper.flatten_content.""" - # Test string - assert ChatStreamingHelper.flatten_content(content="Hello") == "Hello" - - # Test list - content_list = [{"type": "text", "text": "Hello"}, {"type": "text", "text": "World"}] - result = ChatStreamingHelper.flatten_content(content=content_list) - assert "Hello" in result - assert "World" in result - - # Test dict - content_dict = {"text": "Simple message"} - assert ChatStreamingHelper.flatten_content(content=content_dict) == "Simple message" - - # Test None - assert ChatStreamingHelper.flatten_content(content=None) == "" - - @pytest.mark.asyncio - async def test_streaming_helper_message_to_dict(self): - """Test ChatStreamingHelper.message_to_dict.""" - from langchain_core.messages import HumanMessage - - msg = HumanMessage(content="Hello there") - result = ChatStreamingHelper.message_to_dict(msg=msg) - - assert result["role"] == "user" - assert result["content"] == "Hello there" - - @pytest.mark.asyncio - async def test_streaming_helper_extract_messages_from_output(self): - """Test ChatStreamingHelper.extract_messages_from_output.""" - # Test dict with messages - output_dict = {"messages": [{"role": "user", "content": "Hello"}]} - messages = ChatStreamingHelper.extract_messages_from_output(output_obj=output_dict) - assert len(messages) == 1 - - # Test None - messages = ChatStreamingHelper.extract_messages_from_output(output_obj=None) - assert len(messages) == 0 - - # Test object with messages attribute - class MockOutput: - def __init__(self): - self.messages = [{"role": "assistant", "content": "Hi"}] - - mock_output = MockOutput() - messages = ChatStreamingHelper.extract_messages_from_output(output_obj=mock_output) - assert len(messages) == 1 - - @pytest.mark.asyncio - async def test_chatbot_should_continue_logic(self, test_user, workflow_id): - """Test that should_continue logic works correctly (no infinite loops).""" - # Load config - config = load_chatbot_config_from_dict({}, config_id="test") - - # Create system prompt - from datetime import datetime - system_prompt = config.systemPrompt.replace( - "{{DATE}}", - datetime.now().strftime("%d.%m.%Y") - ) - - # Create AI center model - operation_type = OperationTypeEnum[config.model.operationType] - processing_mode = ProcessingModeEnum[config.model.processingMode] - - model = AICenterChatModel( - user=test_user, - operation_type=operation_type, - processing_mode=processing_mode - ) - - # Create memory/checkpointer - memory = DatabaseCheckpointer(user=test_user, workflow_id=workflow_id) - - # Create chatbot instance - chatbot = await Chatbot.create( - model=model, - memory=memory, - system_prompt=system_prompt, - workflow_id=workflow_id - ) - - # The should_continue logic is internal to the workflow - # We can test that the workflow compiles successfully - assert chatbot.app is not None - - # Test that we can invoke the workflow (this will test should_continue internally) - # Use a simple message that shouldn't cause infinite loops - try: - result = await chatbot.chat( - message="Hallo", - chat_id=workflow_id - ) - # Should return messages without infinite loop - assert result is not None - assert isinstance(result, list) - except Exception as e: - # If there's an error, it shouldn't be an infinite loop error - # (infinite loops would timeout or hit max iterations) - assert "infinite" not in str(e).lower() - assert "loop" not in str(e).lower() - - -if __name__ == "__main__": - pytest.main([__file__, "-v"])