From 1cf7c3430a6743714873271e190eae92f4a315d6 Mon Sep 17 00:00:00 2001 From: NeatGuyCoding <15627489+NeatGuyCoding@users.noreply.github.com> Date: Wed, 13 Aug 2025 11:21:32 +0800 Subject: [PATCH] Add more comprehensive Test Containers Based Tests for advanced prompt service (#23850) --- .../test_advanced_prompt_template_service.py | 885 ++++++++++++++ .../services/test_agent_service.py | 1033 +++++++++++++++++ 2 files changed, 1918 insertions(+) create mode 100644 api/tests/test_containers_integration_tests/services/test_advanced_prompt_template_service.py create mode 100644 api/tests/test_containers_integration_tests/services/test_agent_service.py diff --git a/api/tests/test_containers_integration_tests/services/test_advanced_prompt_template_service.py b/api/tests/test_containers_integration_tests/services/test_advanced_prompt_template_service.py new file mode 100644 index 000000000..9ed9008af --- /dev/null +++ b/api/tests/test_containers_integration_tests/services/test_advanced_prompt_template_service.py @@ -0,0 +1,885 @@ +import copy + +import pytest +from faker import Faker + +from core.prompt.prompt_templates.advanced_prompt_templates import ( + BAICHUAN_CHAT_APP_CHAT_PROMPT_CONFIG, + BAICHUAN_CHAT_APP_COMPLETION_PROMPT_CONFIG, + BAICHUAN_COMPLETION_APP_CHAT_PROMPT_CONFIG, + BAICHUAN_COMPLETION_APP_COMPLETION_PROMPT_CONFIG, + BAICHUAN_CONTEXT, + CHAT_APP_CHAT_PROMPT_CONFIG, + CHAT_APP_COMPLETION_PROMPT_CONFIG, + COMPLETION_APP_CHAT_PROMPT_CONFIG, + COMPLETION_APP_COMPLETION_PROMPT_CONFIG, + CONTEXT, +) +from models.model import AppMode +from services.advanced_prompt_template_service import AdvancedPromptTemplateService + + +class TestAdvancedPromptTemplateService: + """Integration tests for AdvancedPromptTemplateService using testcontainers.""" + + @pytest.fixture + def mock_external_service_dependencies(self): + """Mock setup for external service dependencies.""" + # This service doesn't have external dependencies, but we keep the pattern + # for consistency with other test files + return {} + + def test_get_prompt_baichuan_model_success(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test successful prompt generation for Baichuan model. + + This test verifies: + - Proper prompt generation for Baichuan models + - Correct model detection logic + - Appropriate prompt template selection + """ + fake = Faker() + + # Test data for Baichuan model + args = { + "app_mode": AppMode.CHAT.value, + "model_mode": "completion", + "model_name": "baichuan-13b-chat", + "has_context": "true", + } + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_prompt(args) + + # Assert: Verify the expected outcomes + assert result is not None + assert "completion_prompt_config" in result + assert "prompt" in result["completion_prompt_config"] + assert "text" in result["completion_prompt_config"]["prompt"] + + # Verify context is included for Baichuan model + prompt_text = result["completion_prompt_config"]["prompt"]["text"] + assert BAICHUAN_CONTEXT in prompt_text + assert "{{#pre_prompt#}}" in prompt_text + assert "{{#histories#}}" in prompt_text + assert "{{#query#}}" in prompt_text + + def test_get_prompt_common_model_success(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test successful prompt generation for common models. + + This test verifies: + - Proper prompt generation for non-Baichuan models + - Correct model detection logic + - Appropriate prompt template selection + """ + fake = Faker() + + # Test data for common model + args = { + "app_mode": AppMode.CHAT.value, + "model_mode": "completion", + "model_name": "gpt-3.5-turbo", + "has_context": "true", + } + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_prompt(args) + + # Assert: Verify the expected outcomes + assert result is not None + assert "completion_prompt_config" in result + assert "prompt" in result["completion_prompt_config"] + assert "text" in result["completion_prompt_config"]["prompt"] + + # Verify context is included for common model + prompt_text = result["completion_prompt_config"]["prompt"]["text"] + assert CONTEXT in prompt_text + assert "{{#pre_prompt#}}" in prompt_text + assert "{{#histories#}}" in prompt_text + assert "{{#query#}}" in prompt_text + + def test_get_prompt_case_insensitive_baichuan_detection( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test Baichuan model detection is case insensitive. + + This test verifies: + - Model name detection works regardless of case + - Proper prompt template selection for different case variations + """ + fake = Faker() + + # Test different case variations + test_cases = ["Baichuan-13B-Chat", "BAICHUAN-13B-CHAT", "baichuan-13b-chat", "BaiChuan-13B-Chat"] + + for model_name in test_cases: + args = { + "app_mode": AppMode.CHAT.value, + "model_mode": "completion", + "model_name": model_name, + "has_context": "true", + } + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_prompt(args) + + # Assert: Verify Baichuan template is used + assert result is not None + prompt_text = result["completion_prompt_config"]["prompt"]["text"] + assert BAICHUAN_CONTEXT in prompt_text + + def test_get_common_prompt_chat_app_completion_mode( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test common prompt generation for chat app with completion mode. + + This test verifies: + - Correct prompt template selection for chat app + completion mode + - Proper context integration + - Template structure validation + """ + fake = Faker() + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_common_prompt(AppMode.CHAT.value, "completion", "true") + + # Assert: Verify the expected outcomes + assert result is not None + assert "completion_prompt_config" in result + assert "prompt" in result["completion_prompt_config"] + assert "text" in result["completion_prompt_config"]["prompt"] + assert "conversation_histories_role" in result["completion_prompt_config"] + assert "stop" in result + + # Verify context is included + prompt_text = result["completion_prompt_config"]["prompt"]["text"] + assert CONTEXT in prompt_text + assert "{{#pre_prompt#}}" in prompt_text + assert "{{#histories#}}" in prompt_text + assert "{{#query#}}" in prompt_text + + def test_get_common_prompt_chat_app_chat_mode(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test common prompt generation for chat app with chat mode. + + This test verifies: + - Correct prompt template selection for chat app + chat mode + - Proper context integration + - Template structure validation + """ + fake = Faker() + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_common_prompt(AppMode.CHAT.value, "chat", "true") + + # Assert: Verify the expected outcomes + assert result is not None + assert "chat_prompt_config" in result + assert "prompt" in result["chat_prompt_config"] + assert len(result["chat_prompt_config"]["prompt"]) > 0 + assert "role" in result["chat_prompt_config"]["prompt"][0] + assert "text" in result["chat_prompt_config"]["prompt"][0] + + # Verify context is included + prompt_text = result["chat_prompt_config"]["prompt"][0]["text"] + assert CONTEXT in prompt_text + assert "{{#pre_prompt#}}" in prompt_text + + def test_get_common_prompt_completion_app_completion_mode( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test common prompt generation for completion app with completion mode. + + This test verifies: + - Correct prompt template selection for completion app + completion mode + - Proper context integration + - Template structure validation + """ + fake = Faker() + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_common_prompt(AppMode.COMPLETION.value, "completion", "true") + + # Assert: Verify the expected outcomes + assert result is not None + assert "completion_prompt_config" in result + assert "prompt" in result["completion_prompt_config"] + assert "text" in result["completion_prompt_config"]["prompt"] + assert "stop" in result + + # Verify context is included + prompt_text = result["completion_prompt_config"]["prompt"]["text"] + assert CONTEXT in prompt_text + assert "{{#pre_prompt#}}" in prompt_text + + def test_get_common_prompt_completion_app_chat_mode( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test common prompt generation for completion app with chat mode. + + This test verifies: + - Correct prompt template selection for completion app + chat mode + - Proper context integration + - Template structure validation + """ + fake = Faker() + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_common_prompt(AppMode.COMPLETION.value, "chat", "true") + + # Assert: Verify the expected outcomes + assert result is not None + assert "chat_prompt_config" in result + assert "prompt" in result["chat_prompt_config"] + assert len(result["chat_prompt_config"]["prompt"]) > 0 + assert "role" in result["chat_prompt_config"]["prompt"][0] + assert "text" in result["chat_prompt_config"]["prompt"][0] + + # Verify context is included + prompt_text = result["chat_prompt_config"]["prompt"][0]["text"] + assert CONTEXT in prompt_text + assert "{{#pre_prompt#}}" in prompt_text + + def test_get_common_prompt_no_context(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test common prompt generation without context. + + This test verifies: + - Correct handling when has_context is "false" + - Context is not included in prompt + - Template structure remains intact + """ + fake = Faker() + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_common_prompt(AppMode.CHAT.value, "completion", "false") + + # Assert: Verify the expected outcomes + assert result is not None + assert "completion_prompt_config" in result + assert "prompt" in result["completion_prompt_config"] + assert "text" in result["completion_prompt_config"]["prompt"] + + # Verify context is NOT included + prompt_text = result["completion_prompt_config"]["prompt"]["text"] + assert CONTEXT not in prompt_text + assert "{{#pre_prompt#}}" in prompt_text + assert "{{#histories#}}" in prompt_text + assert "{{#query#}}" in prompt_text + + def test_get_common_prompt_unsupported_app_mode( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test common prompt generation with unsupported app mode. + + This test verifies: + - Proper handling of unsupported app modes + - Default empty dict return + """ + fake = Faker() + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_common_prompt("unsupported_mode", "completion", "true") + + # Assert: Verify empty dict is returned + assert result == {} + + def test_get_common_prompt_unsupported_model_mode( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test common prompt generation with unsupported model mode. + + This test verifies: + - Proper handling of unsupported model modes + - Default empty dict return + """ + fake = Faker() + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_common_prompt(AppMode.CHAT.value, "unsupported_mode", "true") + + # Assert: Verify empty dict is returned + assert result == {} + + def test_get_completion_prompt_with_context(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test completion prompt generation with context. + + This test verifies: + - Proper context integration in completion prompts + - Template structure preservation + - Context placement at the beginning + """ + fake = Faker() + + # Create test prompt template + prompt_template = copy.deepcopy(CHAT_APP_COMPLETION_PROMPT_CONFIG) + original_text = prompt_template["completion_prompt_config"]["prompt"]["text"] + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_completion_prompt(prompt_template, "true", CONTEXT) + + # Assert: Verify the expected outcomes + assert result is not None + assert "completion_prompt_config" in result + assert "prompt" in result["completion_prompt_config"] + assert "text" in result["completion_prompt_config"]["prompt"] + + # Verify context is prepended to original text + result_text = result["completion_prompt_config"]["prompt"]["text"] + assert result_text.startswith(CONTEXT) + assert original_text in result_text + assert result_text == CONTEXT + original_text + + def test_get_completion_prompt_without_context( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test completion prompt generation without context. + + This test verifies: + - Original template is preserved when no context + - No modification to prompt text + """ + fake = Faker() + + # Create test prompt template + prompt_template = copy.deepcopy(CHAT_APP_COMPLETION_PROMPT_CONFIG) + original_text = prompt_template["completion_prompt_config"]["prompt"]["text"] + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_completion_prompt(prompt_template, "false", CONTEXT) + + # Assert: Verify the expected outcomes + assert result is not None + assert "completion_prompt_config" in result + assert "prompt" in result["completion_prompt_config"] + assert "text" in result["completion_prompt_config"]["prompt"] + + # Verify original text is unchanged + result_text = result["completion_prompt_config"]["prompt"]["text"] + assert result_text == original_text + assert CONTEXT not in result_text + + def test_get_chat_prompt_with_context(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test chat prompt generation with context. + + This test verifies: + - Proper context integration in chat prompts + - Template structure preservation + - Context placement at the beginning of first message + """ + fake = Faker() + + # Create test prompt template + prompt_template = copy.deepcopy(CHAT_APP_CHAT_PROMPT_CONFIG) + original_text = prompt_template["chat_prompt_config"]["prompt"][0]["text"] + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_chat_prompt(prompt_template, "true", CONTEXT) + + # Assert: Verify the expected outcomes + assert result is not None + assert "chat_prompt_config" in result + assert "prompt" in result["chat_prompt_config"] + assert len(result["chat_prompt_config"]["prompt"]) > 0 + assert "text" in result["chat_prompt_config"]["prompt"][0] + + # Verify context is prepended to original text + result_text = result["chat_prompt_config"]["prompt"][0]["text"] + assert result_text.startswith(CONTEXT) + assert original_text in result_text + assert result_text == CONTEXT + original_text + + def test_get_chat_prompt_without_context(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test chat prompt generation without context. + + This test verifies: + - Original template is preserved when no context + - No modification to prompt text + """ + fake = Faker() + + # Create test prompt template + prompt_template = copy.deepcopy(CHAT_APP_CHAT_PROMPT_CONFIG) + original_text = prompt_template["chat_prompt_config"]["prompt"][0]["text"] + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_chat_prompt(prompt_template, "false", CONTEXT) + + # Assert: Verify the expected outcomes + assert result is not None + assert "chat_prompt_config" in result + assert "prompt" in result["chat_prompt_config"] + assert len(result["chat_prompt_config"]["prompt"]) > 0 + assert "text" in result["chat_prompt_config"]["prompt"][0] + + # Verify original text is unchanged + result_text = result["chat_prompt_config"]["prompt"][0]["text"] + assert result_text == original_text + assert CONTEXT not in result_text + + def test_get_baichuan_prompt_chat_app_completion_mode( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test Baichuan prompt generation for chat app with completion mode. + + This test verifies: + - Correct Baichuan prompt template selection for chat app + completion mode + - Proper Baichuan context integration + - Template structure validation + """ + fake = Faker() + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_baichuan_prompt(AppMode.CHAT.value, "completion", "true") + + # Assert: Verify the expected outcomes + assert result is not None + assert "completion_prompt_config" in result + assert "prompt" in result["completion_prompt_config"] + assert "text" in result["completion_prompt_config"]["prompt"] + assert "conversation_histories_role" in result["completion_prompt_config"] + assert "stop" in result + + # Verify Baichuan context is included + prompt_text = result["completion_prompt_config"]["prompt"]["text"] + assert BAICHUAN_CONTEXT in prompt_text + assert "{{#pre_prompt#}}" in prompt_text + assert "{{#histories#}}" in prompt_text + assert "{{#query#}}" in prompt_text + + def test_get_baichuan_prompt_chat_app_chat_mode( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test Baichuan prompt generation for chat app with chat mode. + + This test verifies: + - Correct Baichuan prompt template selection for chat app + chat mode + - Proper Baichuan context integration + - Template structure validation + """ + fake = Faker() + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_baichuan_prompt(AppMode.CHAT.value, "chat", "true") + + # Assert: Verify the expected outcomes + assert result is not None + assert "chat_prompt_config" in result + assert "prompt" in result["chat_prompt_config"] + assert len(result["chat_prompt_config"]["prompt"]) > 0 + assert "role" in result["chat_prompt_config"]["prompt"][0] + assert "text" in result["chat_prompt_config"]["prompt"][0] + + # Verify Baichuan context is included + prompt_text = result["chat_prompt_config"]["prompt"][0]["text"] + assert BAICHUAN_CONTEXT in prompt_text + assert "{{#pre_prompt#}}" in prompt_text + + def test_get_baichuan_prompt_completion_app_completion_mode( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test Baichuan prompt generation for completion app with completion mode. + + This test verifies: + - Correct Baichuan prompt template selection for completion app + completion mode + - Proper Baichuan context integration + - Template structure validation + """ + fake = Faker() + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_baichuan_prompt(AppMode.COMPLETION.value, "completion", "true") + + # Assert: Verify the expected outcomes + assert result is not None + assert "completion_prompt_config" in result + assert "prompt" in result["completion_prompt_config"] + assert "text" in result["completion_prompt_config"]["prompt"] + assert "stop" in result + + # Verify Baichuan context is included + prompt_text = result["completion_prompt_config"]["prompt"]["text"] + assert BAICHUAN_CONTEXT in prompt_text + assert "{{#pre_prompt#}}" in prompt_text + + def test_get_baichuan_prompt_completion_app_chat_mode( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test Baichuan prompt generation for completion app with chat mode. + + This test verifies: + - Correct Baichuan prompt template selection for completion app + chat mode + - Proper Baichuan context integration + - Template structure validation + """ + fake = Faker() + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_baichuan_prompt(AppMode.COMPLETION.value, "chat", "true") + + # Assert: Verify the expected outcomes + assert result is not None + assert "chat_prompt_config" in result + assert "prompt" in result["chat_prompt_config"] + assert len(result["chat_prompt_config"]["prompt"]) > 0 + assert "role" in result["chat_prompt_config"]["prompt"][0] + assert "text" in result["chat_prompt_config"]["prompt"][0] + + # Verify Baichuan context is included + prompt_text = result["chat_prompt_config"]["prompt"][0]["text"] + assert BAICHUAN_CONTEXT in prompt_text + assert "{{#pre_prompt#}}" in prompt_text + + def test_get_baichuan_prompt_no_context(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test Baichuan prompt generation without context. + + This test verifies: + - Correct handling when has_context is "false" + - Baichuan context is not included in prompt + - Template structure remains intact + """ + fake = Faker() + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_baichuan_prompt(AppMode.CHAT.value, "completion", "false") + + # Assert: Verify the expected outcomes + assert result is not None + assert "completion_prompt_config" in result + assert "prompt" in result["completion_prompt_config"] + assert "text" in result["completion_prompt_config"]["prompt"] + + # Verify Baichuan context is NOT included + prompt_text = result["completion_prompt_config"]["prompt"]["text"] + assert BAICHUAN_CONTEXT not in prompt_text + assert "{{#pre_prompt#}}" in prompt_text + assert "{{#histories#}}" in prompt_text + assert "{{#query#}}" in prompt_text + + def test_get_baichuan_prompt_unsupported_app_mode( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test Baichuan prompt generation with unsupported app mode. + + This test verifies: + - Proper handling of unsupported app modes + - Default empty dict return + """ + fake = Faker() + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_baichuan_prompt("unsupported_mode", "completion", "true") + + # Assert: Verify empty dict is returned + assert result == {} + + def test_get_baichuan_prompt_unsupported_model_mode( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test Baichuan prompt generation with unsupported model mode. + + This test verifies: + - Proper handling of unsupported model modes + - Default empty dict return + """ + fake = Faker() + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_baichuan_prompt(AppMode.CHAT.value, "unsupported_mode", "true") + + # Assert: Verify empty dict is returned + assert result == {} + + def test_get_prompt_all_app_modes_common_model( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test prompt generation for all app modes with common model. + + This test verifies: + - All app modes work correctly with common models + - Proper template selection for each combination + """ + fake = Faker() + + # Test all app modes + app_modes = [AppMode.CHAT.value, AppMode.COMPLETION.value] + model_modes = ["completion", "chat"] + + for app_mode in app_modes: + for model_mode in model_modes: + args = { + "app_mode": app_mode, + "model_mode": model_mode, + "model_name": "gpt-3.5-turbo", + "has_context": "true", + } + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_prompt(args) + + # Assert: Verify result is not empty + assert result is not None + assert result != {} + + def test_get_prompt_all_app_modes_baichuan_model( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test prompt generation for all app modes with Baichuan model. + + This test verifies: + - All app modes work correctly with Baichuan models + - Proper template selection for each combination + """ + fake = Faker() + + # Test all app modes + app_modes = [AppMode.CHAT.value, AppMode.COMPLETION.value] + model_modes = ["completion", "chat"] + + for app_mode in app_modes: + for model_mode in model_modes: + args = { + "app_mode": app_mode, + "model_mode": model_mode, + "model_name": "baichuan-13b-chat", + "has_context": "true", + } + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_prompt(args) + + # Assert: Verify result is not empty + assert result is not None + assert result != {} + + def test_get_prompt_edge_cases(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test prompt generation with edge cases. + + This test verifies: + - Handling of edge case inputs + - Proper error handling + - Consistent behavior with unusual inputs + """ + fake = Faker() + + # Test edge cases + edge_cases = [ + {"app_mode": "", "model_mode": "completion", "model_name": "gpt-3.5-turbo", "has_context": "true"}, + {"app_mode": AppMode.CHAT.value, "model_mode": "", "model_name": "gpt-3.5-turbo", "has_context": "true"}, + {"app_mode": AppMode.CHAT.value, "model_mode": "completion", "model_name": "", "has_context": "true"}, + { + "app_mode": AppMode.CHAT.value, + "model_mode": "completion", + "model_name": "gpt-3.5-turbo", + "has_context": "", + }, + ] + + for args in edge_cases: + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_prompt(args) + + # Assert: Verify method handles edge cases gracefully + # Should either return a valid result or empty dict, but not crash + assert result is not None + + def test_template_immutability(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test that original templates are not modified. + + This test verifies: + - Original template constants are not modified + - Deep copy is used properly + - Template immutability is maintained + """ + fake = Faker() + + # Store original templates + original_chat_completion = copy.deepcopy(CHAT_APP_COMPLETION_PROMPT_CONFIG) + original_chat_chat = copy.deepcopy(CHAT_APP_CHAT_PROMPT_CONFIG) + original_completion_completion = copy.deepcopy(COMPLETION_APP_COMPLETION_PROMPT_CONFIG) + original_completion_chat = copy.deepcopy(COMPLETION_APP_CHAT_PROMPT_CONFIG) + + # Test with context + args = { + "app_mode": AppMode.CHAT.value, + "model_mode": "completion", + "model_name": "gpt-3.5-turbo", + "has_context": "true", + } + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_prompt(args) + + # Assert: Verify original templates are unchanged + assert original_chat_completion == CHAT_APP_COMPLETION_PROMPT_CONFIG + assert original_chat_chat == CHAT_APP_CHAT_PROMPT_CONFIG + assert original_completion_completion == COMPLETION_APP_COMPLETION_PROMPT_CONFIG + assert original_completion_chat == COMPLETION_APP_CHAT_PROMPT_CONFIG + + def test_baichuan_template_immutability(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test that original Baichuan templates are not modified. + + This test verifies: + - Original Baichuan template constants are not modified + - Deep copy is used properly + - Template immutability is maintained + """ + fake = Faker() + + # Store original templates + original_baichuan_chat_completion = copy.deepcopy(BAICHUAN_CHAT_APP_COMPLETION_PROMPT_CONFIG) + original_baichuan_chat_chat = copy.deepcopy(BAICHUAN_CHAT_APP_CHAT_PROMPT_CONFIG) + original_baichuan_completion_completion = copy.deepcopy(BAICHUAN_COMPLETION_APP_COMPLETION_PROMPT_CONFIG) + original_baichuan_completion_chat = copy.deepcopy(BAICHUAN_COMPLETION_APP_CHAT_PROMPT_CONFIG) + + # Test with context + args = { + "app_mode": AppMode.CHAT.value, + "model_mode": "completion", + "model_name": "baichuan-13b-chat", + "has_context": "true", + } + + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_prompt(args) + + # Assert: Verify original templates are unchanged + assert original_baichuan_chat_completion == BAICHUAN_CHAT_APP_COMPLETION_PROMPT_CONFIG + assert original_baichuan_chat_chat == BAICHUAN_CHAT_APP_CHAT_PROMPT_CONFIG + assert original_baichuan_completion_completion == BAICHUAN_COMPLETION_APP_COMPLETION_PROMPT_CONFIG + assert original_baichuan_completion_chat == BAICHUAN_COMPLETION_APP_CHAT_PROMPT_CONFIG + + def test_context_integration_consistency(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test consistency of context integration across different scenarios. + + This test verifies: + - Context is always prepended correctly + - Context integration is consistent across different templates + - No context duplication or corruption + """ + fake = Faker() + + # Test different scenarios + test_scenarios = [ + { + "app_mode": AppMode.CHAT.value, + "model_mode": "completion", + "model_name": "gpt-3.5-turbo", + "has_context": "true", + }, + { + "app_mode": AppMode.CHAT.value, + "model_mode": "chat", + "model_name": "gpt-3.5-turbo", + "has_context": "true", + }, + { + "app_mode": AppMode.COMPLETION.value, + "model_mode": "completion", + "model_name": "gpt-3.5-turbo", + "has_context": "true", + }, + { + "app_mode": AppMode.COMPLETION.value, + "model_mode": "chat", + "model_name": "gpt-3.5-turbo", + "has_context": "true", + }, + ] + + for args in test_scenarios: + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_prompt(args) + + # Assert: Verify context integration is consistent + assert result is not None + assert result != {} + + # Check that context is properly integrated + if "completion_prompt_config" in result: + prompt_text = result["completion_prompt_config"]["prompt"]["text"] + assert prompt_text.startswith(CONTEXT) + elif "chat_prompt_config" in result: + prompt_text = result["chat_prompt_config"]["prompt"][0]["text"] + assert prompt_text.startswith(CONTEXT) + + def test_baichuan_context_integration_consistency( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test consistency of Baichuan context integration across different scenarios. + + This test verifies: + - Baichuan context is always prepended correctly + - Context integration is consistent across different templates + - No context duplication or corruption + """ + fake = Faker() + + # Test different scenarios + test_scenarios = [ + { + "app_mode": AppMode.CHAT.value, + "model_mode": "completion", + "model_name": "baichuan-13b-chat", + "has_context": "true", + }, + { + "app_mode": AppMode.CHAT.value, + "model_mode": "chat", + "model_name": "baichuan-13b-chat", + "has_context": "true", + }, + { + "app_mode": AppMode.COMPLETION.value, + "model_mode": "completion", + "model_name": "baichuan-13b-chat", + "has_context": "true", + }, + { + "app_mode": AppMode.COMPLETION.value, + "model_mode": "chat", + "model_name": "baichuan-13b-chat", + "has_context": "true", + }, + ] + + for args in test_scenarios: + # Act: Execute the method under test + result = AdvancedPromptTemplateService.get_prompt(args) + + # Assert: Verify context integration is consistent + assert result is not None + assert result != {} + + # Check that Baichuan context is properly integrated + if "completion_prompt_config" in result: + prompt_text = result["completion_prompt_config"]["prompt"]["text"] + assert prompt_text.startswith(BAICHUAN_CONTEXT) + elif "chat_prompt_config" in result: + prompt_text = result["chat_prompt_config"]["prompt"][0]["text"] + assert prompt_text.startswith(BAICHUAN_CONTEXT) diff --git a/api/tests/test_containers_integration_tests/services/test_agent_service.py b/api/tests/test_containers_integration_tests/services/test_agent_service.py new file mode 100644 index 000000000..d63b188b1 --- /dev/null +++ b/api/tests/test_containers_integration_tests/services/test_agent_service.py @@ -0,0 +1,1033 @@ +import json +from unittest.mock import MagicMock, patch + +import pytest +from faker import Faker + +from core.plugin.impl.exc import PluginDaemonClientSideError +from models.model import AppModelConfig, Conversation, EndUser, Message, MessageAgentThought +from services.account_service import AccountService, TenantService +from services.agent_service import AgentService +from services.app_service import AppService + + +class TestAgentService: + """Integration tests for AgentService using testcontainers.""" + + @pytest.fixture + def mock_external_service_dependencies(self): + """Mock setup for external service dependencies.""" + with ( + patch("services.agent_service.PluginAgentClient") as mock_plugin_agent_client, + patch("services.agent_service.ToolManager") as mock_tool_manager, + patch("services.agent_service.AgentConfigManager") as mock_agent_config_manager, + patch("services.agent_service.current_user") as mock_current_user, + patch("services.app_service.FeatureService") as mock_feature_service, + patch("services.app_service.EnterpriseService") as mock_enterprise_service, + patch("services.app_service.ModelManager") as mock_model_manager, + patch("services.account_service.FeatureService") as mock_account_feature_service, + ): + # Setup default mock returns for agent service + mock_plugin_agent_client_instance = mock_plugin_agent_client.return_value + mock_plugin_agent_client_instance.fetch_agent_strategy_providers.return_value = [ + MagicMock( + plugin_id="test_plugin", + declaration=MagicMock( + identity=MagicMock(name="test_provider"), + strategies=[MagicMock(identity=MagicMock(name="test_strategy"))], + ), + ) + ] + mock_plugin_agent_client_instance.fetch_agent_strategy_provider.return_value = MagicMock( + plugin_id="test_plugin", + declaration=MagicMock( + identity=MagicMock(name="test_provider"), + strategies=[MagicMock(identity=MagicMock(name="test_strategy"))], + ), + ) + + # Setup ToolManager mocks + mock_tool_manager.get_tool_icon.return_value = "test_icon" + mock_tool_manager.get_tool_label.return_value = MagicMock( + to_dict=lambda: {"en_US": "Test Tool", "zh_Hans": "测试工具"} + ) + + # Setup AgentConfigManager mocks + mock_agent_config = MagicMock() + mock_agent_config.tools = [ + MagicMock(tool_name="test_tool", provider_type="test_provider", provider_id="test_id") + ] + mock_agent_config_manager.convert.return_value = mock_agent_config + + # Setup current_user mock + mock_current_user.timezone = "UTC" + + # Setup default mock returns for app service + mock_feature_service.get_system_features.return_value.webapp_auth.enabled = False + mock_enterprise_service.WebAppAuth.update_app_access_mode.return_value = None + mock_enterprise_service.WebAppAuth.cleanup_webapp.return_value = None + + # Setup default mock returns for account service + mock_account_feature_service.get_system_features.return_value.is_allow_register = True + + # Mock ModelManager for model configuration + mock_model_instance = mock_model_manager.return_value + mock_model_instance.get_default_model_instance.return_value = None + mock_model_instance.get_default_provider_model_name.return_value = ("openai", "gpt-3.5-turbo") + + yield { + "plugin_agent_client": mock_plugin_agent_client, + "tool_manager": mock_tool_manager, + "agent_config_manager": mock_agent_config_manager, + "current_user": mock_current_user, + "feature_service": mock_feature_service, + "enterprise_service": mock_enterprise_service, + "model_manager": mock_model_manager, + "account_feature_service": mock_account_feature_service, + } + + def _create_test_app_and_account(self, db_session_with_containers, mock_external_service_dependencies): + """ + Helper method to create a test app and account for testing. + + Args: + db_session_with_containers: Database session from testcontainers infrastructure + mock_external_service_dependencies: Mock dependencies + + Returns: + tuple: (app, account) - Created app and account instances + """ + fake = Faker() + + # Setup mocks for account creation + mock_external_service_dependencies[ + "account_feature_service" + ].get_system_features.return_value.is_allow_register = True + + # Create account and tenant + account = AccountService.create_account( + email=fake.email(), + name=fake.name(), + interface_language="en-US", + password=fake.password(length=12), + ) + TenantService.create_owner_tenant_if_not_exist(account, name=fake.company()) + tenant = account.current_tenant + + # Create app with realistic data + app_args = { + "name": fake.company(), + "description": fake.text(max_nb_chars=100), + "mode": "agent-chat", + "icon_type": "emoji", + "icon": "🤖", + "icon_background": "#FF6B6B", + "api_rph": 100, + "api_rpm": 10, + } + + app_service = AppService() + app = app_service.create_app(tenant.id, app_args, account) + + # Update the app model config to set agent_mode for agent-chat mode + if app.mode == "agent-chat" and app.app_model_config: + app.app_model_config.agent_mode = json.dumps({"enabled": True, "strategy": "react", "tools": []}) + from extensions.ext_database import db + + db.session.commit() + + return app, account + + def _create_test_conversation_and_message(self, db_session_with_containers, app, account): + """ + Helper method to create a test conversation and message with agent thoughts. + + Args: + db_session_with_containers: Database session from testcontainers infrastructure + app: App instance + account: Account instance + + Returns: + tuple: (conversation, message) - Created conversation and message instances + """ + fake = Faker() + + from extensions.ext_database import db + + # Create conversation + conversation = Conversation( + id=fake.uuid4(), + app_id=app.id, + from_account_id=account.id, + from_end_user_id=None, + name=fake.sentence(), + inputs={}, + status="normal", + mode="chat", + from_source="api", + ) + db.session.add(conversation) + db.session.commit() + + # Create app model config + app_model_config = AppModelConfig( + id=fake.uuid4(), + app_id=app.id, + provider="openai", + model_id="gpt-3.5-turbo", + configs={}, + model="gpt-3.5-turbo", + agent_mode=json.dumps({"enabled": True, "strategy": "react", "tools": []}), + ) + db.session.add(app_model_config) + db.session.commit() + + # Update conversation with app model config + conversation.app_model_config_id = app_model_config.id + db.session.commit() + + # Create message + message = Message( + id=fake.uuid4(), + conversation_id=conversation.id, + app_id=app.id, + from_account_id=account.id, + from_end_user_id=None, + inputs={}, + query=fake.text(max_nb_chars=100), + message=[{"role": "user", "text": fake.text(max_nb_chars=100)}], + answer=fake.text(max_nb_chars=200), + message_tokens=100, + message_unit_price=0.001, + answer_tokens=200, + answer_unit_price=0.001, + provider_response_latency=1.5, + currency="USD", + from_source="api", + ) + db.session.add(message) + db.session.commit() + + return conversation, message + + def _create_test_agent_thoughts(self, db_session_with_containers, message): + """ + Helper method to create test agent thoughts for a message. + + Args: + db_session_with_containers: Database session from testcontainers infrastructure + message: Message instance + + Returns: + list: Created agent thoughts + """ + fake = Faker() + + from extensions.ext_database import db + + agent_thoughts = [] + + # Create first agent thought + thought1 = MessageAgentThought( + id=fake.uuid4(), + message_id=message.id, + position=1, + thought="I need to analyze the user's request", + tool="test_tool", + tool_labels_str=json.dumps({"test_tool": {"en_US": "Test Tool", "zh_Hans": "测试工具"}}), + tool_meta_str=json.dumps( + { + "test_tool": { + "error": None, + "time_cost": 0.5, + "tool_config": {"tool_provider_type": "test_provider", "tool_provider": "test_id"}, + "tool_parameters": {}, + } + } + ), + tool_input=json.dumps({"test_tool": {"input": "test_input"}}), + observation=json.dumps({"test_tool": {"output": "test_output"}}), + tokens=50, + created_by_role="account", + created_by=message.from_account_id, + ) + db.session.add(thought1) + agent_thoughts.append(thought1) + + # Create second agent thought + thought2 = MessageAgentThought( + id=fake.uuid4(), + message_id=message.id, + position=2, + thought="Based on the analysis, I can provide a response", + tool="dataset_tool", + tool_labels_str=json.dumps({"dataset_tool": {"en_US": "Dataset Tool", "zh_Hans": "数据集工具"}}), + tool_meta_str=json.dumps( + { + "dataset_tool": { + "error": None, + "time_cost": 0.3, + "tool_config": {"tool_provider_type": "dataset-retrieval", "tool_provider": "dataset_id"}, + "tool_parameters": {}, + } + } + ), + tool_input=json.dumps({"dataset_tool": {"query": "test_query"}}), + observation=json.dumps({"dataset_tool": {"results": "test_results"}}), + tokens=30, + created_by_role="account", + created_by=message.from_account_id, + ) + db.session.add(thought2) + agent_thoughts.append(thought2) + + db.session.commit() + + return agent_thoughts + + def test_get_agent_logs_success(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test successful retrieval of agent logs with complete data. + """ + fake = Faker() + + # Create test data + app, account = self._create_test_app_and_account(db_session_with_containers, mock_external_service_dependencies) + conversation, message = self._create_test_conversation_and_message(db_session_with_containers, app, account) + agent_thoughts = self._create_test_agent_thoughts(db_session_with_containers, message) + + # Execute the method under test + result = AgentService.get_agent_logs(app, str(conversation.id), str(message.id)) + + # Verify the result structure + assert result is not None + assert "meta" in result + assert "iterations" in result + assert "files" in result + + # Verify meta information + meta = result["meta"] + assert meta["status"] == "success" + assert meta["executor"] == account.name + assert meta["iterations"] == 2 + assert meta["agent_mode"] == "react" + assert meta["total_tokens"] == 300 # 100 + 200 + assert meta["elapsed_time"] == 1.5 + + # Verify iterations + iterations = result["iterations"] + assert len(iterations) == 2 + + # Verify first iteration + first_iteration = iterations[0] + assert first_iteration["tokens"] == 50 + assert first_iteration["thought"] == "I need to analyze the user's request" + assert len(first_iteration["tool_calls"]) == 1 + + tool_call = first_iteration["tool_calls"][0] + assert tool_call["tool_name"] == "test_tool" + assert tool_call["tool_label"] == {"en_US": "Test Tool", "zh_Hans": "测试工具"} + assert tool_call["status"] == "success" + assert tool_call["time_cost"] == 0.5 + assert tool_call["tool_icon"] == "test_icon" + + # Verify second iteration + second_iteration = iterations[1] + assert second_iteration["tokens"] == 30 + assert second_iteration["thought"] == "Based on the analysis, I can provide a response" + assert len(second_iteration["tool_calls"]) == 1 + + dataset_tool_call = second_iteration["tool_calls"][0] + assert dataset_tool_call["tool_name"] == "dataset_tool" + assert dataset_tool_call["tool_label"] == {"en_US": "Dataset Tool", "zh_Hans": "数据集工具"} + assert dataset_tool_call["status"] == "success" + assert dataset_tool_call["time_cost"] == 0.3 + assert dataset_tool_call["tool_icon"] == "" # dataset-retrieval tools have empty icon + + def test_get_agent_logs_conversation_not_found( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test error handling when conversation is not found. + """ + fake = Faker() + + # Create test data + app, account = self._create_test_app_and_account(db_session_with_containers, mock_external_service_dependencies) + + # Execute the method under test with non-existent conversation + with pytest.raises(ValueError, match="Conversation not found"): + AgentService.get_agent_logs(app, fake.uuid4(), fake.uuid4()) + + def test_get_agent_logs_message_not_found(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test error handling when message is not found. + """ + fake = Faker() + + # Create test data + app, account = self._create_test_app_and_account(db_session_with_containers, mock_external_service_dependencies) + conversation, message = self._create_test_conversation_and_message(db_session_with_containers, app, account) + + # Execute the method under test with non-existent message + with pytest.raises(ValueError, match="Message not found"): + AgentService.get_agent_logs(app, str(conversation.id), fake.uuid4()) + + def test_get_agent_logs_with_end_user(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test agent logs retrieval when conversation is from end user. + """ + fake = Faker() + + # Create test data + app, account = self._create_test_app_and_account(db_session_with_containers, mock_external_service_dependencies) + + from extensions.ext_database import db + + # Create end user + end_user = EndUser( + id=fake.uuid4(), + tenant_id=app.tenant_id, + app_id=app.id, + type="web_app", + is_anonymous=False, + session_id=fake.uuid4(), + name=fake.name(), + ) + db.session.add(end_user) + db.session.commit() + + # Create conversation with end user + conversation = Conversation( + id=fake.uuid4(), + app_id=app.id, + from_account_id=None, + from_end_user_id=end_user.id, + name=fake.sentence(), + inputs={}, + status="normal", + mode="chat", + from_source="api", + ) + db.session.add(conversation) + db.session.commit() + + # Create app model config + app_model_config = AppModelConfig( + id=fake.uuid4(), + app_id=app.id, + provider="openai", + model_id="gpt-3.5-turbo", + configs={}, + model="gpt-3.5-turbo", + agent_mode=json.dumps({"enabled": True, "strategy": "react", "tools": []}), + ) + db.session.add(app_model_config) + db.session.commit() + + # Update conversation with app model config + conversation.app_model_config_id = app_model_config.id + db.session.commit() + + # Create message + message = Message( + id=fake.uuid4(), + conversation_id=conversation.id, + app_id=app.id, + from_account_id=None, + from_end_user_id=end_user.id, + inputs={}, + query=fake.text(max_nb_chars=100), + message=[{"role": "user", "text": fake.text(max_nb_chars=100)}], + answer=fake.text(max_nb_chars=200), + message_tokens=100, + message_unit_price=0.001, + answer_tokens=200, + answer_unit_price=0.001, + provider_response_latency=1.5, + currency="USD", + from_source="api", + ) + db.session.add(message) + db.session.commit() + + # Execute the method under test + result = AgentService.get_agent_logs(app, str(conversation.id), str(message.id)) + + # Verify the result + assert result is not None + assert result["meta"]["executor"] == end_user.name + + def test_get_agent_logs_with_unknown_executor(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test agent logs retrieval when executor is unknown. + """ + fake = Faker() + + # Create test data + app, account = self._create_test_app_and_account(db_session_with_containers, mock_external_service_dependencies) + + from extensions.ext_database import db + + # Create conversation with non-existent account + conversation = Conversation( + id=fake.uuid4(), + app_id=app.id, + from_account_id=fake.uuid4(), # Non-existent account + from_end_user_id=None, + name=fake.sentence(), + inputs={}, + status="normal", + mode="chat", + from_source="api", + ) + db.session.add(conversation) + db.session.commit() + + # Create app model config + app_model_config = AppModelConfig( + id=fake.uuid4(), + app_id=app.id, + provider="openai", + model_id="gpt-3.5-turbo", + configs={}, + model="gpt-3.5-turbo", + agent_mode=json.dumps({"enabled": True, "strategy": "react", "tools": []}), + ) + db.session.add(app_model_config) + db.session.commit() + + # Update conversation with app model config + conversation.app_model_config_id = app_model_config.id + db.session.commit() + + # Create message + message = Message( + id=fake.uuid4(), + conversation_id=conversation.id, + app_id=app.id, + from_account_id=fake.uuid4(), # Non-existent account + from_end_user_id=None, + inputs={}, + query=fake.text(max_nb_chars=100), + message=[{"role": "user", "text": fake.text(max_nb_chars=100)}], + answer=fake.text(max_nb_chars=200), + message_tokens=100, + message_unit_price=0.001, + answer_tokens=200, + answer_unit_price=0.001, + provider_response_latency=1.5, + currency="USD", + from_source="api", + ) + db.session.add(message) + db.session.commit() + + # Execute the method under test + result = AgentService.get_agent_logs(app, str(conversation.id), str(message.id)) + + # Verify the result + assert result is not None + assert result["meta"]["executor"] == "Unknown" + + def test_get_agent_logs_with_tool_error(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test agent logs retrieval with tool errors. + """ + fake = Faker() + + # Create test data + app, account = self._create_test_app_and_account(db_session_with_containers, mock_external_service_dependencies) + conversation, message = self._create_test_conversation_and_message(db_session_with_containers, app, account) + + from extensions.ext_database import db + + # Create agent thought with tool error + thought_with_error = MessageAgentThought( + id=fake.uuid4(), + message_id=message.id, + position=1, + thought="I need to analyze the user's request", + tool="error_tool", + tool_labels_str=json.dumps({"error_tool": {"en_US": "Error Tool", "zh_Hans": "错误工具"}}), + tool_meta_str=json.dumps( + { + "error_tool": { + "error": "Tool execution failed", + "time_cost": 0.5, + "tool_config": {"tool_provider_type": "test_provider", "tool_provider": "test_id"}, + "tool_parameters": {}, + } + } + ), + tool_input=json.dumps({"error_tool": {"input": "test_input"}}), + observation=json.dumps({"error_tool": {"output": "error_output"}}), + tokens=50, + created_by_role="account", + created_by=message.from_account_id, + ) + db.session.add(thought_with_error) + db.session.commit() + + # Execute the method under test + result = AgentService.get_agent_logs(app, str(conversation.id), str(message.id)) + + # Verify the result + assert result is not None + iterations = result["iterations"] + assert len(iterations) == 1 + + tool_call = iterations[0]["tool_calls"][0] + assert tool_call["status"] == "error" + assert tool_call["error"] == "Tool execution failed" + + def test_get_agent_logs_without_agent_thoughts( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test agent logs retrieval when message has no agent thoughts. + """ + fake = Faker() + + # Create test data + app, account = self._create_test_app_and_account(db_session_with_containers, mock_external_service_dependencies) + conversation, message = self._create_test_conversation_and_message(db_session_with_containers, app, account) + + # Execute the method under test + result = AgentService.get_agent_logs(app, str(conversation.id), str(message.id)) + + # Verify the result + assert result is not None + assert result["meta"]["iterations"] == 0 + assert len(result["iterations"]) == 0 + + def test_get_agent_logs_app_model_config_not_found( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test error handling when app model config is not found. + """ + fake = Faker() + + # Create test data + app, account = self._create_test_app_and_account(db_session_with_containers, mock_external_service_dependencies) + + from extensions.ext_database import db + + # Remove app model config to test error handling + app.app_model_config_id = None + db.session.commit() + + # Create conversation without app model config + conversation = Conversation( + id=fake.uuid4(), + app_id=app.id, + from_account_id=account.id, + from_end_user_id=None, + name=fake.sentence(), + inputs={}, + status="normal", + mode="chat", + from_source="api", + app_model_config_id=None, # Explicitly set to None + ) + db.session.add(conversation) + db.session.commit() + + # Create message + message = Message( + id=fake.uuid4(), + conversation_id=conversation.id, + app_id=app.id, + from_account_id=account.id, + from_end_user_id=None, + inputs={}, + query=fake.text(max_nb_chars=100), + message=[{"role": "user", "text": fake.text(max_nb_chars=100)}], + answer=fake.text(max_nb_chars=200), + message_tokens=100, + message_unit_price=0.001, + answer_tokens=200, + answer_unit_price=0.001, + provider_response_latency=1.5, + currency="USD", + from_source="api", + ) + db.session.add(message) + db.session.commit() + + # Execute the method under test + with pytest.raises(ValueError, match="App model config not found"): + AgentService.get_agent_logs(app, str(conversation.id), str(message.id)) + + def test_get_agent_logs_agent_config_not_found( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test error handling when agent config is not found. + """ + fake = Faker() + + # Create test data + app, account = self._create_test_app_and_account(db_session_with_containers, mock_external_service_dependencies) + conversation, message = self._create_test_conversation_and_message(db_session_with_containers, app, account) + + # Mock AgentConfigManager to return None + mock_external_service_dependencies["agent_config_manager"].convert.return_value = None + + # Execute the method under test + with pytest.raises(ValueError, match="Agent config not found"): + AgentService.get_agent_logs(app, str(conversation.id), str(message.id)) + + def test_list_agent_providers_success(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test successful listing of agent providers. + """ + fake = Faker() + + # Create test data + app, account = self._create_test_app_and_account(db_session_with_containers, mock_external_service_dependencies) + + # Execute the method under test + result = AgentService.list_agent_providers(str(account.id), str(app.tenant_id)) + + # Verify the result + assert result is not None + assert len(result) == 1 + assert result[0].plugin_id == "test_plugin" + + # Verify the mock was called correctly + mock_plugin_client = mock_external_service_dependencies["plugin_agent_client"].return_value + mock_plugin_client.fetch_agent_strategy_providers.assert_called_once_with(str(app.tenant_id)) + + def test_get_agent_provider_success(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test successful retrieval of specific agent provider. + """ + fake = Faker() + + # Create test data + app, account = self._create_test_app_and_account(db_session_with_containers, mock_external_service_dependencies) + + provider_name = "test_provider" + + # Execute the method under test + result = AgentService.get_agent_provider(str(account.id), str(app.tenant_id), provider_name) + + # Verify the result + assert result is not None + assert result.plugin_id == "test_plugin" + + # Verify the mock was called correctly + mock_plugin_client = mock_external_service_dependencies["plugin_agent_client"].return_value + mock_plugin_client.fetch_agent_strategy_provider.assert_called_once_with(str(app.tenant_id), provider_name) + + def test_get_agent_provider_plugin_error(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test error handling when plugin daemon client raises an error. + """ + fake = Faker() + + # Create test data + app, account = self._create_test_app_and_account(db_session_with_containers, mock_external_service_dependencies) + + provider_name = "test_provider" + error_message = "Plugin not found" + + # Mock PluginAgentClient to raise an error + mock_plugin_client = mock_external_service_dependencies["plugin_agent_client"].return_value + mock_plugin_client.fetch_agent_strategy_provider.side_effect = PluginDaemonClientSideError(error_message) + + # Execute the method under test + with pytest.raises(ValueError, match=error_message): + AgentService.get_agent_provider(str(account.id), str(app.tenant_id), provider_name) + + def test_get_agent_logs_with_complex_tool_data( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test agent logs retrieval with complex tool data and multiple tools. + """ + fake = Faker() + + # Create test data + app, account = self._create_test_app_and_account(db_session_with_containers, mock_external_service_dependencies) + conversation, message = self._create_test_conversation_and_message(db_session_with_containers, app, account) + + from extensions.ext_database import db + + # Create agent thought with multiple tools + complex_thought = MessageAgentThought( + id=fake.uuid4(), + message_id=message.id, + position=1, + thought="I need to use multiple tools to complete this task", + tool="tool1;tool2;tool3", + tool_labels_str=json.dumps( + { + "tool1": {"en_US": "First Tool", "zh_Hans": "第一个工具"}, + "tool2": {"en_US": "Second Tool", "zh_Hans": "第二个工具"}, + "tool3": {"en_US": "Third Tool", "zh_Hans": "第三个工具"}, + } + ), + tool_meta_str=json.dumps( + { + "tool1": { + "error": None, + "time_cost": 0.5, + "tool_config": {"tool_provider_type": "test_provider", "tool_provider": "test_id"}, + "tool_parameters": {"param1": "value1"}, + }, + "tool2": { + "error": "Tool 2 failed", + "time_cost": 0.3, + "tool_config": {"tool_provider_type": "another_provider", "tool_provider": "another_id"}, + "tool_parameters": {"param2": "value2"}, + }, + "tool3": { + "error": None, + "time_cost": 0.7, + "tool_config": {"tool_provider_type": "dataset-retrieval", "tool_provider": "dataset_id"}, + "tool_parameters": {"param3": "value3"}, + }, + } + ), + tool_input=json.dumps( + {"tool1": {"input1": "data1"}, "tool2": {"input2": "data2"}, "tool3": {"input3": "data3"}} + ), + observation=json.dumps( + {"tool1": {"output1": "result1"}, "tool2": {"output2": "result2"}, "tool3": {"output3": "result3"}} + ), + tokens=100, + created_by_role="account", + created_by=message.from_account_id, + ) + db.session.add(complex_thought) + db.session.commit() + + # Execute the method under test + result = AgentService.get_agent_logs(app, str(conversation.id), str(message.id)) + + # Verify the result + assert result is not None + iterations = result["iterations"] + assert len(iterations) == 1 + + tool_calls = iterations[0]["tool_calls"] + assert len(tool_calls) == 3 + + # Verify first tool + assert tool_calls[0]["tool_name"] == "tool1" + assert tool_calls[0]["tool_label"] == {"en_US": "First Tool", "zh_Hans": "第一个工具"} + assert tool_calls[0]["status"] == "success" + assert tool_calls[0]["tool_parameters"] == {"param1": "value1"} + + # Verify second tool (with error) + assert tool_calls[1]["tool_name"] == "tool2" + assert tool_calls[1]["tool_label"] == {"en_US": "Second Tool", "zh_Hans": "第二个工具"} + assert tool_calls[1]["status"] == "error" + assert tool_calls[1]["error"] == "Tool 2 failed" + + # Verify third tool (dataset tool) + assert tool_calls[2]["tool_name"] == "tool3" + assert tool_calls[2]["tool_label"] == {"en_US": "Third Tool", "zh_Hans": "第三个工具"} + assert tool_calls[2]["status"] == "success" + assert tool_calls[2]["tool_icon"] == "" # dataset-retrieval tools have empty icon + + def test_get_agent_logs_with_files(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test agent logs retrieval with message files and agent thought files. + """ + fake = Faker() + + # Create test data + app, account = self._create_test_app_and_account(db_session_with_containers, mock_external_service_dependencies) + conversation, message = self._create_test_conversation_and_message(db_session_with_containers, app, account) + + from core.file import FileTransferMethod, FileType + from extensions.ext_database import db + from models.enums import CreatorUserRole + + # Add files to message + from models.model import MessageFile + + message_file1 = MessageFile( + message_id=message.id, + type=FileType.IMAGE, + transfer_method=FileTransferMethod.REMOTE_URL, + url="http://example.com/file1.jpg", + belongs_to="user", + created_by_role=CreatorUserRole.ACCOUNT, + created_by=message.from_account_id, + ) + message_file2 = MessageFile( + message_id=message.id, + type=FileType.IMAGE, + transfer_method=FileTransferMethod.REMOTE_URL, + url="http://example.com/file2.png", + belongs_to="user", + created_by_role=CreatorUserRole.ACCOUNT, + created_by=message.from_account_id, + ) + db.session.add(message_file1) + db.session.add(message_file2) + db.session.commit() + + # Create agent thought with files + thought_with_files = MessageAgentThought( + id=fake.uuid4(), + message_id=message.id, + position=1, + thought="I need to process some files", + tool="file_tool", + tool_labels_str=json.dumps({"file_tool": {"en_US": "File Tool", "zh_Hans": "文件工具"}}), + tool_meta_str=json.dumps( + { + "file_tool": { + "error": None, + "time_cost": 0.5, + "tool_config": {"tool_provider_type": "test_provider", "tool_provider": "test_id"}, + "tool_parameters": {}, + } + } + ), + tool_input=json.dumps({"file_tool": {"input": "test_input"}}), + observation=json.dumps({"file_tool": {"output": "test_output"}}), + message_files=json.dumps(["file1", "file2"]), + tokens=50, + created_by_role="account", + created_by=message.from_account_id, + ) + db.session.add(thought_with_files) + db.session.commit() + + # Execute the method under test + result = AgentService.get_agent_logs(app, str(conversation.id), str(message.id)) + + # Verify the result + assert result is not None + assert len(result["files"]) == 2 + + iterations = result["iterations"] + assert len(iterations) == 1 + assert len(iterations[0]["files"]) == 2 + assert "file1" in iterations[0]["files"] + assert "file2" in iterations[0]["files"] + + def test_get_agent_logs_with_different_timezone( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test agent logs retrieval with different timezone settings. + """ + fake = Faker() + + # Create test data + app, account = self._create_test_app_and_account(db_session_with_containers, mock_external_service_dependencies) + conversation, message = self._create_test_conversation_and_message(db_session_with_containers, app, account) + + # Mock current_user with different timezone + mock_external_service_dependencies["current_user"].timezone = "Asia/Shanghai" + + # Execute the method under test + result = AgentService.get_agent_logs(app, str(conversation.id), str(message.id)) + + # Verify the result + assert result is not None + assert "start_time" in result["meta"] + + # Verify the timezone conversion + start_time = result["meta"]["start_time"] + assert "T" in start_time # ISO format + assert "+08:00" in start_time or "Z" in start_time # Timezone offset + + def test_get_agent_logs_with_empty_tool_data(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test agent logs retrieval with empty tool data. + """ + fake = Faker() + + # Create test data + app, account = self._create_test_app_and_account(db_session_with_containers, mock_external_service_dependencies) + conversation, message = self._create_test_conversation_and_message(db_session_with_containers, app, account) + + from extensions.ext_database import db + + # Create agent thought with empty tool data + empty_thought = MessageAgentThought( + id=fake.uuid4(), + message_id=message.id, + position=1, + thought="I need to analyze the user's request", + tool="", # Empty tool + tool_labels_str="{}", # Empty labels + tool_meta_str="{}", # Empty meta + tool_input="", # Empty input + observation="", # Empty observation + tokens=50, + created_by_role="account", + created_by=message.from_account_id, + ) + db.session.add(empty_thought) + db.session.commit() + + # Execute the method under test + result = AgentService.get_agent_logs(app, str(conversation.id), str(message.id)) + + # Verify the result + assert result is not None + iterations = result["iterations"] + assert len(iterations) == 1 + + # Verify empty tool calls + tool_calls = iterations[0]["tool_calls"] + assert len(tool_calls) == 0 # No tools to process + + def test_get_agent_logs_with_malformed_json(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test agent logs retrieval with malformed JSON data in tool fields. + """ + fake = Faker() + + # Create test data + app, account = self._create_test_app_and_account(db_session_with_containers, mock_external_service_dependencies) + conversation, message = self._create_test_conversation_and_message(db_session_with_containers, app, account) + + from extensions.ext_database import db + + # Create agent thought with malformed JSON + malformed_thought = MessageAgentThought( + id=fake.uuid4(), + message_id=message.id, + position=1, + thought="I need to analyze the user's request", + tool="test_tool", + tool_labels_str="invalid json", # Malformed JSON + tool_meta_str="invalid json", # Malformed JSON + tool_input="invalid json", # Malformed JSON + observation="invalid json", # Malformed JSON + tokens=50, + created_by_role="account", + created_by=message.from_account_id, + ) + db.session.add(malformed_thought) + db.session.commit() + + # Execute the method under test + result = AgentService.get_agent_logs(app, str(conversation.id), str(message.id)) + + # Verify the result - should handle malformed JSON gracefully + assert result is not None + iterations = result["iterations"] + assert len(iterations) == 1 + + tool_calls = iterations[0]["tool_calls"] + assert len(tool_calls) == 1 + + # Verify default values for malformed JSON + tool_call = tool_calls[0] + assert tool_call["tool_name"] == "test_tool" + assert tool_call["tool_label"] == "test_tool" # Default to tool name + assert tool_call["tool_input"] == {} + assert tool_call["tool_output"] == "invalid json" # Raw observation value + assert tool_call["tool_parameters"] == {}