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"] == {}