| @@ -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 | |||