#!/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"])