From c0bd35594e2b311a025a49628de729a302c6ac7c Mon Sep 17 00:00:00 2001 From: NeatGuyCoding <15627489+NeatGuyCoding@users.noreply.github.com> Date: Wed, 3 Sep 2025 09:20:16 +0800 Subject: [PATCH] feat: add test containers based tests for tools manage service (#25028) --- .../test_workflow_tools_manage_service.py | 716 ++++++++++++++++++ 1 file changed, 716 insertions(+) create mode 100644 api/tests/test_containers_integration_tests/services/tools/test_workflow_tools_manage_service.py diff --git a/api/tests/test_containers_integration_tests/services/tools/test_workflow_tools_manage_service.py b/api/tests/test_containers_integration_tests/services/tools/test_workflow_tools_manage_service.py new file mode 100644 index 000000000..cb1e79d50 --- /dev/null +++ b/api/tests/test_containers_integration_tests/services/tools/test_workflow_tools_manage_service.py @@ -0,0 +1,716 @@ +import json +from unittest.mock import patch + +import pytest +from faker import Faker + +from models.tools import WorkflowToolProvider +from models.workflow import Workflow as WorkflowModel +from services.account_service import AccountService, TenantService +from services.app_service import AppService +from services.tools.workflow_tools_manage_service import WorkflowToolManageService + + +class TestWorkflowToolManageService: + """Integration tests for WorkflowToolManageService using testcontainers.""" + + @pytest.fixture + def mock_external_service_dependencies(self): + """Mock setup for external service dependencies.""" + with ( + 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, + patch( + "services.tools.workflow_tools_manage_service.WorkflowToolProviderController" + ) as mock_workflow_tool_provider_controller, + patch("services.tools.workflow_tools_manage_service.ToolLabelManager") as mock_tool_label_manager, + patch("services.tools.workflow_tools_manage_service.ToolTransformService") as mock_tool_transform_service, + ): + # 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") + + # Mock WorkflowToolProviderController + mock_workflow_tool_provider_controller.from_db.return_value = None + + # Mock ToolLabelManager + mock_tool_label_manager.update_tool_labels.return_value = None + + # Mock ToolTransformService + mock_tool_transform_service.workflow_provider_to_controller.return_value = None + + yield { + "feature_service": mock_feature_service, + "enterprise_service": mock_enterprise_service, + "model_manager": mock_model_manager, + "account_feature_service": mock_account_feature_service, + "workflow_tool_provider_controller": mock_workflow_tool_provider_controller, + "tool_label_manager": mock_tool_label_manager, + "tool_transform_service": mock_tool_transform_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, workflow) - Created app, account and workflow 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": "workflow", + "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) + + # Create workflow for the app + workflow = WorkflowModel( + tenant_id=tenant.id, + app_id=app.id, + type="workflow", + version="1.0.0", + graph=json.dumps({}), + features=json.dumps({}), + created_by=account.id, + environment_variables=[], + conversation_variables=[], + ) + + from extensions.ext_database import db + + db.session.add(workflow) + db.session.commit() + + # Update app to reference the workflow + app.workflow_id = workflow.id + db.session.commit() + + return app, account, workflow + + def _create_test_workflow_tool_parameters(self): + """Helper method to create valid workflow tool parameters.""" + return [ + { + "name": "input_text", + "description": "Input text for processing", + "form": "form", + "type": "string", + "required": True, + }, + { + "name": "output_format", + "description": "Output format specification", + "form": "form", + "type": "select", + "required": False, + }, + ] + + def test_create_workflow_tool_success(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test successful workflow tool creation with valid parameters. + + This test verifies: + - Proper workflow tool creation with all required fields + - Correct database state after creation + - Proper relationship establishment + - External service integration + - Return value correctness + """ + fake = Faker() + + # Create test data + app, account, workflow = self._create_test_app_and_account( + db_session_with_containers, mock_external_service_dependencies + ) + + # Setup workflow tool creation parameters + tool_name = fake.word() + tool_label = fake.word() + tool_icon = {"type": "emoji", "emoji": "🔧"} + tool_description = fake.text(max_nb_chars=200) + tool_parameters = self._create_test_workflow_tool_parameters() + tool_privacy_policy = fake.text(max_nb_chars=100) + tool_labels = ["automation", "workflow"] + + # Execute the method under test + result = WorkflowToolManageService.create_workflow_tool( + user_id=account.id, + tenant_id=account.current_tenant.id, + workflow_app_id=app.id, + name=tool_name, + label=tool_label, + icon=tool_icon, + description=tool_description, + parameters=tool_parameters, + privacy_policy=tool_privacy_policy, + labels=tool_labels, + ) + + # Verify the result + assert result == {"result": "success"} + + # Verify database state + from extensions.ext_database import db + + # Check if workflow tool provider was created + created_tool_provider = ( + db.session.query(WorkflowToolProvider) + .where( + WorkflowToolProvider.tenant_id == account.current_tenant.id, + WorkflowToolProvider.app_id == app.id, + ) + .first() + ) + + assert created_tool_provider is not None + assert created_tool_provider.name == tool_name + assert created_tool_provider.label == tool_label + assert created_tool_provider.icon == json.dumps(tool_icon) + assert created_tool_provider.description == tool_description + assert created_tool_provider.parameter_configuration == json.dumps(tool_parameters) + assert created_tool_provider.privacy_policy == tool_privacy_policy + assert created_tool_provider.version == workflow.version + assert created_tool_provider.user_id == account.id + assert created_tool_provider.tenant_id == account.current_tenant.id + assert created_tool_provider.app_id == app.id + + # Verify external service calls + mock_external_service_dependencies["workflow_tool_provider_controller"].from_db.assert_called_once() + mock_external_service_dependencies["tool_label_manager"].update_tool_labels.assert_called_once() + mock_external_service_dependencies[ + "tool_transform_service" + ].workflow_provider_to_controller.assert_called_once() + + def test_create_workflow_tool_duplicate_name_error( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test workflow tool creation fails when name already exists. + + This test verifies: + - Proper error handling for duplicate tool names + - Database constraint enforcement + - Correct error message + """ + fake = Faker() + + # Create test data + app, account, workflow = self._create_test_app_and_account( + db_session_with_containers, mock_external_service_dependencies + ) + + # Create first workflow tool + first_tool_name = fake.word() + first_tool_parameters = self._create_test_workflow_tool_parameters() + + WorkflowToolManageService.create_workflow_tool( + user_id=account.id, + tenant_id=account.current_tenant.id, + workflow_app_id=app.id, + name=first_tool_name, + label=fake.word(), + icon={"type": "emoji", "emoji": "🔧"}, + description=fake.text(max_nb_chars=200), + parameters=first_tool_parameters, + ) + + # Attempt to create second workflow tool with same name + second_tool_parameters = self._create_test_workflow_tool_parameters() + + with pytest.raises(ValueError) as exc_info: + WorkflowToolManageService.create_workflow_tool( + user_id=account.id, + tenant_id=account.current_tenant.id, + workflow_app_id=app.id, + name=first_tool_name, # Same name + label=fake.word(), + icon={"type": "emoji", "emoji": "⚙️"}, + description=fake.text(max_nb_chars=200), + parameters=second_tool_parameters, + ) + + # Verify error message + assert f"Tool with name {first_tool_name} or app_id {app.id} already exists" in str(exc_info.value) + + # Verify only one tool was created + from extensions.ext_database import db + + tool_count = ( + db.session.query(WorkflowToolProvider) + .where( + WorkflowToolProvider.tenant_id == account.current_tenant.id, + ) + .count() + ) + + assert tool_count == 1 + + def test_create_workflow_tool_invalid_app_error( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test workflow tool creation fails when app does not exist. + + This test verifies: + - Proper error handling for non-existent apps + - Correct error message + - No database changes when app is invalid + """ + fake = Faker() + + # Create test data + app, account, workflow = self._create_test_app_and_account( + db_session_with_containers, mock_external_service_dependencies + ) + + # Generate non-existent app ID + non_existent_app_id = fake.uuid4() + + # Attempt to create workflow tool with non-existent app + tool_parameters = self._create_test_workflow_tool_parameters() + + with pytest.raises(ValueError) as exc_info: + WorkflowToolManageService.create_workflow_tool( + user_id=account.id, + tenant_id=account.current_tenant.id, + workflow_app_id=non_existent_app_id, # Non-existent app ID + name=fake.word(), + label=fake.word(), + icon={"type": "emoji", "emoji": "🔧"}, + description=fake.text(max_nb_chars=200), + parameters=tool_parameters, + ) + + # Verify error message + assert f"App {non_existent_app_id} not found" in str(exc_info.value) + + # Verify no workflow tool was created + from extensions.ext_database import db + + tool_count = ( + db.session.query(WorkflowToolProvider) + .where( + WorkflowToolProvider.tenant_id == account.current_tenant.id, + ) + .count() + ) + + assert tool_count == 0 + + def test_create_workflow_tool_invalid_parameters_error( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test workflow tool creation fails when parameters are invalid. + + This test verifies: + - Proper error handling for invalid parameter configurations + - Parameter validation enforcement + - Correct error message + """ + fake = Faker() + + # Create test data + app, account, workflow = self._create_test_app_and_account( + db_session_with_containers, mock_external_service_dependencies + ) + + # Setup invalid workflow tool parameters (missing required fields) + invalid_parameters = [ + { + "name": "input_text", + # Missing description and form fields + "type": "string", + "required": True, + } + ] + + # Attempt to create workflow tool with invalid parameters + with pytest.raises(ValueError) as exc_info: + WorkflowToolManageService.create_workflow_tool( + user_id=account.id, + tenant_id=account.current_tenant.id, + workflow_app_id=app.id, + name=fake.word(), + label=fake.word(), + icon={"type": "emoji", "emoji": "🔧"}, + description=fake.text(max_nb_chars=200), + parameters=invalid_parameters, + ) + + # Verify error message contains validation error + assert "validation error" in str(exc_info.value).lower() + + # Verify no workflow tool was created + from extensions.ext_database import db + + tool_count = ( + db.session.query(WorkflowToolProvider) + .where( + WorkflowToolProvider.tenant_id == account.current_tenant.id, + ) + .count() + ) + + assert tool_count == 0 + + def test_create_workflow_tool_duplicate_app_id_error( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test workflow tool creation fails when app_id already exists. + + This test verifies: + - Proper error handling for duplicate app_id + - Database constraint enforcement for app_id uniqueness + - Correct error message + """ + fake = Faker() + + # Create test data + app, account, workflow = self._create_test_app_and_account( + db_session_with_containers, mock_external_service_dependencies + ) + + # Create first workflow tool + first_tool_name = fake.word() + first_tool_parameters = self._create_test_workflow_tool_parameters() + + WorkflowToolManageService.create_workflow_tool( + user_id=account.id, + tenant_id=account.current_tenant.id, + workflow_app_id=app.id, + name=first_tool_name, + label=fake.word(), + icon={"type": "emoji", "emoji": "🔧"}, + description=fake.text(max_nb_chars=200), + parameters=first_tool_parameters, + ) + + # Attempt to create second workflow tool with same app_id but different name + second_tool_name = fake.word() + second_tool_parameters = self._create_test_workflow_tool_parameters() + + with pytest.raises(ValueError) as exc_info: + WorkflowToolManageService.create_workflow_tool( + user_id=account.id, + tenant_id=account.current_tenant.id, + workflow_app_id=app.id, # Same app_id + name=second_tool_name, # Different name + label=fake.word(), + icon={"type": "emoji", "emoji": "⚙️"}, + description=fake.text(max_nb_chars=200), + parameters=second_tool_parameters, + ) + + # Verify error message + assert f"Tool with name {second_tool_name} or app_id {app.id} already exists" in str(exc_info.value) + + # Verify only one tool was created + from extensions.ext_database import db + + tool_count = ( + db.session.query(WorkflowToolProvider) + .where( + WorkflowToolProvider.tenant_id == account.current_tenant.id, + ) + .count() + ) + + assert tool_count == 1 + + def test_create_workflow_tool_workflow_not_found_error( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test workflow tool creation fails when app has no workflow. + + This test verifies: + - Proper error handling for apps without workflows + - Correct error message + - No database changes when workflow is missing + """ + fake = Faker() + + # Create test data but without workflow + app, account, _ = self._create_test_app_and_account( + db_session_with_containers, mock_external_service_dependencies + ) + + # Remove workflow reference from app + from extensions.ext_database import db + + app.workflow_id = None + db.session.commit() + + # Attempt to create workflow tool for app without workflow + tool_parameters = self._create_test_workflow_tool_parameters() + + with pytest.raises(ValueError) as exc_info: + WorkflowToolManageService.create_workflow_tool( + user_id=account.id, + tenant_id=account.current_tenant.id, + workflow_app_id=app.id, + name=fake.word(), + label=fake.word(), + icon={"type": "emoji", "emoji": "🔧"}, + description=fake.text(max_nb_chars=200), + parameters=tool_parameters, + ) + + # Verify error message + assert f"Workflow not found for app {app.id}" in str(exc_info.value) + + # Verify no workflow tool was created + tool_count = ( + db.session.query(WorkflowToolProvider) + .where( + WorkflowToolProvider.tenant_id == account.current_tenant.id, + ) + .count() + ) + + assert tool_count == 0 + + def test_update_workflow_tool_success(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test successful workflow tool update with valid parameters. + + This test verifies: + - Proper workflow tool update with all required fields + - Correct database state after update + - Proper relationship maintenance + - External service integration + - Return value correctness + """ + fake = Faker() + + # Create test data + app, account, workflow = self._create_test_app_and_account( + db_session_with_containers, mock_external_service_dependencies + ) + + # Create initial workflow tool + initial_tool_name = fake.word() + initial_tool_parameters = self._create_test_workflow_tool_parameters() + + WorkflowToolManageService.create_workflow_tool( + user_id=account.id, + tenant_id=account.current_tenant.id, + workflow_app_id=app.id, + name=initial_tool_name, + label=fake.word(), + icon={"type": "emoji", "emoji": "🔧"}, + description=fake.text(max_nb_chars=200), + parameters=initial_tool_parameters, + ) + + # Get the created tool + from extensions.ext_database import db + + created_tool = ( + db.session.query(WorkflowToolProvider) + .where( + WorkflowToolProvider.tenant_id == account.current_tenant.id, + WorkflowToolProvider.app_id == app.id, + ) + .first() + ) + + # Setup update parameters + updated_tool_name = fake.word() + updated_tool_label = fake.word() + updated_tool_icon = {"type": "emoji", "emoji": "⚙️"} + updated_tool_description = fake.text(max_nb_chars=200) + updated_tool_parameters = self._create_test_workflow_tool_parameters() + updated_tool_privacy_policy = fake.text(max_nb_chars=100) + updated_tool_labels = ["automation", "updated"] + + # Execute the update method + result = WorkflowToolManageService.update_workflow_tool( + user_id=account.id, + tenant_id=account.current_tenant.id, + workflow_tool_id=created_tool.id, + name=updated_tool_name, + label=updated_tool_label, + icon=updated_tool_icon, + description=updated_tool_description, + parameters=updated_tool_parameters, + privacy_policy=updated_tool_privacy_policy, + labels=updated_tool_labels, + ) + + # Verify the result + assert result == {"result": "success"} + + # Verify database state was updated + db.session.refresh(created_tool) + assert created_tool.name == updated_tool_name + assert created_tool.label == updated_tool_label + assert created_tool.icon == json.dumps(updated_tool_icon) + assert created_tool.description == updated_tool_description + assert created_tool.parameter_configuration == json.dumps(updated_tool_parameters) + assert created_tool.privacy_policy == updated_tool_privacy_policy + assert created_tool.version == workflow.version + assert created_tool.updated_at is not None + + # Verify external service calls + mock_external_service_dependencies["workflow_tool_provider_controller"].from_db.assert_called() + mock_external_service_dependencies["tool_label_manager"].update_tool_labels.assert_called() + mock_external_service_dependencies["tool_transform_service"].workflow_provider_to_controller.assert_called() + + def test_update_workflow_tool_not_found_error(self, db_session_with_containers, mock_external_service_dependencies): + """ + Test workflow tool update fails when tool does not exist. + + This test verifies: + - Proper error handling for non-existent tools + - Correct error message + - No database changes when tool is invalid + """ + fake = Faker() + + # Create test data + app, account, workflow = self._create_test_app_and_account( + db_session_with_containers, mock_external_service_dependencies + ) + + # Generate non-existent tool ID + non_existent_tool_id = fake.uuid4() + + # Attempt to update non-existent workflow tool + tool_parameters = self._create_test_workflow_tool_parameters() + + with pytest.raises(ValueError) as exc_info: + WorkflowToolManageService.update_workflow_tool( + user_id=account.id, + tenant_id=account.current_tenant.id, + workflow_tool_id=non_existent_tool_id, # Non-existent tool ID + name=fake.word(), + label=fake.word(), + icon={"type": "emoji", "emoji": "🔧"}, + description=fake.text(max_nb_chars=200), + parameters=tool_parameters, + ) + + # Verify error message + assert f"Tool {non_existent_tool_id} not found" in str(exc_info.value) + + # Verify no workflow tool was created + from extensions.ext_database import db + + tool_count = ( + db.session.query(WorkflowToolProvider) + .where( + WorkflowToolProvider.tenant_id == account.current_tenant.id, + ) + .count() + ) + + assert tool_count == 0 + + def test_update_workflow_tool_same_name_success( + self, db_session_with_containers, mock_external_service_dependencies + ): + """ + Test workflow tool update succeeds when keeping the same name. + + This test verifies: + - Proper handling when updating tool with same name + - Database state maintenance + - Update timestamp is set + """ + fake = Faker() + + # Create test data + app, account, workflow = self._create_test_app_and_account( + db_session_with_containers, mock_external_service_dependencies + ) + + # Create first workflow tool + first_tool_name = fake.word() + first_tool_parameters = self._create_test_workflow_tool_parameters() + + WorkflowToolManageService.create_workflow_tool( + user_id=account.id, + tenant_id=account.current_tenant.id, + workflow_app_id=app.id, + name=first_tool_name, + label=fake.word(), + icon={"type": "emoji", "emoji": "🔧"}, + description=fake.text(max_nb_chars=200), + parameters=first_tool_parameters, + ) + + # Get the created tool + from extensions.ext_database import db + + created_tool = ( + db.session.query(WorkflowToolProvider) + .where( + WorkflowToolProvider.tenant_id == account.current_tenant.id, + WorkflowToolProvider.app_id == app.id, + ) + .first() + ) + + # Attempt to update tool with same name (should not fail) + result = WorkflowToolManageService.update_workflow_tool( + user_id=account.id, + tenant_id=account.current_tenant.id, + workflow_tool_id=created_tool.id, + name=first_tool_name, # Same name + label=fake.word(), + icon={"type": "emoji", "emoji": "⚙️"}, + description=fake.text(max_nb_chars=200), + parameters=first_tool_parameters, + ) + + # Verify update was successful + assert result == {"result": "success"} + + # Verify tool still exists with the same name + db.session.refresh(created_tool) + assert created_tool.name == first_tool_name + assert created_tool.updated_at is not None