Signed-off-by: -LAN- <laipz8200@outlook.com>tags/1.3.0
| @@ -245,6 +245,13 @@ class Workflow(Base): | |||
| @property | |||
| def tool_published(self) -> bool: | |||
| """ | |||
| DEPRECATED: This property is not accurate for determining if a workflow is published as a tool. | |||
| It only checks if there's a WorkflowToolProvider for the app, not if this specific workflow version | |||
| is the one being used by the tool. | |||
| For accurate checking, use a direct query with tenant_id, app_id, and version. | |||
| """ | |||
| from models.tools import WorkflowToolProvider | |||
| return ( | |||
| @@ -28,6 +28,7 @@ from extensions.ext_database import db | |||
| from models.account import Account | |||
| from models.enums import CreatedByRole | |||
| from models.model import App, AppMode | |||
| from models.tools import WorkflowToolProvider | |||
| from models.workflow import ( | |||
| Workflow, | |||
| WorkflowNodeExecution, | |||
| @@ -523,8 +524,19 @@ class WorkflowService: | |||
| # Cannot delete a workflow that's currently in use by an app | |||
| raise WorkflowInUseError(f"Cannot delete workflow that is currently in use by app '{app.name}'") | |||
| # Check if this workflow is published as a tool | |||
| if workflow.tool_published: | |||
| # Don't use workflow.tool_published as it's not accurate for specific workflow versions | |||
| # Check if there's a tool provider using this specific workflow version | |||
| tool_provider = ( | |||
| session.query(WorkflowToolProvider) | |||
| .filter( | |||
| WorkflowToolProvider.tenant_id == workflow.tenant_id, | |||
| WorkflowToolProvider.app_id == workflow.app_id, | |||
| WorkflowToolProvider.version == workflow.version, | |||
| ) | |||
| .first() | |||
| ) | |||
| if tool_provider: | |||
| # Cannot delete a workflow that's published as a tool | |||
| raise WorkflowInUseError("Cannot delete workflow that is published as a tool") | |||
| @@ -40,6 +40,10 @@ def workflow_setup(): | |||
| def test_delete_workflow_success(workflow_setup): | |||
| # Setup mocks | |||
| # Mock the tool provider query to return None (not published as a tool) | |||
| workflow_setup["session"].query.return_value.filter.return_value.first.return_value = None | |||
| workflow_setup["session"].scalar = MagicMock( | |||
| side_effect=[workflow_setup["workflow"], None] | |||
| ) # Return workflow first, then None for app | |||
| @@ -97,7 +101,12 @@ def test_delete_workflow_in_use_by_app_error(workflow_setup): | |||
| def test_delete_workflow_published_as_tool_error(workflow_setup): | |||
| # Setup mocks | |||
| workflow_setup["workflow"].tool_published = True | |||
| from models.tools import WorkflowToolProvider | |||
| # Mock the tool provider query | |||
| mock_tool_provider = MagicMock(spec=WorkflowToolProvider) | |||
| workflow_setup["session"].query.return_value.filter.return_value.first.return_value = mock_tool_provider | |||
| workflow_setup["session"].scalar = MagicMock( | |||
| side_effect=[workflow_setup["workflow"], None] | |||
| ) # Return workflow first, then None for app | |||