from __future__ import annotations """Unit tests for ToolManager behavior with mocked providers and collaborators.""" import json import threading from types import SimpleNamespace from typing import Any from unittest.mock import Mock, patch import pytest from core.app.entities.app_invoke_entities import InvokeFrom from core.plugin.entities.plugin_daemon import CredentialType from core.tools.__base.tool_runtime import ToolRuntime from core.tools.entities.tool_entities import ( ApiProviderAuthType, ToolInvokeFrom, ToolParameter, ToolProviderType, ) from core.tools.errors import ToolProviderNotFoundError from core.tools.plugin_tool.provider import PluginToolProviderController from core.tools.tool_manager import ToolManager class _SimpleContextVar: def __init__(self): self._is_set = False self._value: Any = None def get(self): if not self._is_set: raise LookupError return self._value def set(self, value: Any): self._value = value self._is_set = True def _cm(session: Any): context = Mock() context.__enter__ = Mock(return_value=session) context.__exit__ = Mock(return_value=False) return context def _setup_list_providers_from_api_mocks( monkeypatch, *, session: Mock, hardcoded_controller: SimpleNamespace, plugin_controller: PluginToolProviderController, api_controller: SimpleNamespace, workflow_controller: SimpleNamespace, ): mock_db = Mock() mock_db.engine = object() monkeypatch.setattr("core.tools.tool_manager.db", mock_db) monkeypatch.setattr("core.tools.tool_manager.Session", lambda *args, **kwargs: _cm(session)) monkeypatch.setattr( ToolManager, "list_builtin_providers", Mock(return_value=[hardcoded_controller, plugin_controller]), ) monkeypatch.setattr( ToolManager, "list_default_builtin_providers", Mock(return_value=[SimpleNamespace(provider="hardcoded")]), ) monkeypatch.setattr("core.tools.tool_manager.is_filtered", lambda *args, **kwargs: False) monkeypatch.setattr( "core.tools.tool_manager.ToolTransformService.builtin_provider_to_user_provider", lambda **kwargs: SimpleNamespace(name=kwargs["provider_controller"].entity.identity.name), ) monkeypatch.setattr( "core.tools.tool_manager.ToolTransformService.api_provider_to_controller", Mock(side_effect=[api_controller, RuntimeError("invalid")]), ) monkeypatch.setattr( "core.tools.tool_manager.ToolTransformService.api_provider_to_user_provider", Mock(return_value=SimpleNamespace(name="api-provider")), ) monkeypatch.setattr( "core.tools.tool_manager.ToolTransformService.workflow_provider_to_controller", Mock(side_effect=[workflow_controller, RuntimeError("deleted app")]), ) monkeypatch.setattr( "core.tools.tool_manager.ToolTransformService.workflow_provider_to_user_provider", Mock(return_value=SimpleNamespace(name="workflow-provider")), ) monkeypatch.setattr( "core.tools.tool_manager.ToolLabelManager.get_tools_labels", Mock(side_effect=[{"api-1": ["search"]}, {"wf-1": ["utility"]}]), ) mock_mcp_service = Mock() mock_mcp_service.list_providers.return_value = [SimpleNamespace(name="mcp-provider")] monkeypatch.setattr("core.tools.tool_manager.MCPToolManageService", Mock(return_value=mock_mcp_service)) monkeypatch.setattr("core.tools.tool_manager.BuiltinToolProviderSort.sort", lambda providers: providers) @pytest.fixture(autouse=True) def _reset_tool_manager_state(): old_hardcoded = ToolManager._hardcoded_providers.copy() old_loaded = ToolManager._builtin_providers_loaded old_labels = ToolManager._builtin_tools_labels.copy() try: yield finally: ToolManager._hardcoded_providers = old_hardcoded ToolManager._builtin_providers_loaded = old_loaded ToolManager._builtin_tools_labels = old_labels def test_get_hardcoded_provider_loads_cache_when_empty(): provider = Mock() ToolManager._hardcoded_providers = {} def _load(): ToolManager._hardcoded_providers["weather"] = provider with patch.object(ToolManager, "load_hardcoded_providers_cache", side_effect=_load) as mock_load: assert ToolManager.get_hardcoded_provider("weather") is provider mock_load.assert_called_once() def test_get_builtin_provider_returns_plugin_for_missing_hardcoded(): hardcoded = Mock() plugin_provider = Mock() ToolManager._hardcoded_providers = {"time": hardcoded} with patch.object(ToolManager, "get_plugin_provider", return_value=plugin_provider): assert ToolManager.get_builtin_provider("time", "tenant-1") is hardcoded assert ToolManager.get_builtin_provider("plugin/time", "tenant-1") is plugin_provider def test_get_plugin_provider_uses_context_cache(): provider_context = _SimpleContextVar() lock_context = _SimpleContextVar() lock_context.set(threading.Lock()) provider_entity = SimpleNamespace(declaration=Mock(), plugin_id="pid", plugin_unique_identifier="uid") with patch("core.tools.tool_manager.contexts.plugin_tool_providers", provider_context): with patch("core.tools.tool_manager.contexts.plugin_tool_providers_lock", lock_context): with patch("core.tools.tool_manager.PluginToolManager") as mock_manager_cls: mock_manager_cls.return_value.fetch_tool_provider.return_value = provider_entity controller = SimpleNamespace(name="controller") with patch("core.tools.tool_manager.PluginToolProviderController", return_value=controller): built = ToolManager.get_plugin_provider("provider-a", "tenant-1") cached = ToolManager.get_plugin_provider("provider-a", "tenant-1") assert built is controller assert cached is controller mock_manager_cls.return_value.fetch_tool_provider.assert_called_once() def test_get_plugin_provider_raises_when_provider_missing(): provider_context = _SimpleContextVar() lock_context = _SimpleContextVar() lock_context.set(threading.Lock()) with patch("core.tools.tool_manager.contexts.plugin_tool_providers", provider_context): with patch("core.tools.tool_manager.contexts.plugin_tool_providers_lock", lock_context): with patch("core.tools.tool_manager.PluginToolManager") as mock_manager_cls: mock_manager_cls.return_value.fetch_tool_provider.return_value = None with pytest.raises(ToolProviderNotFoundError, match="plugin provider provider-a not found"): ToolManager.get_plugin_provider("provider-a", "tenant-1") def test_get_tool_runtime_builtin_without_credentials(): tool = Mock() tool.fork_tool_runtime.return_value = "runtime-tool" controller = SimpleNamespace(get_tool=Mock(return_value=tool), need_credentials=False) with patch.object(ToolManager, "get_builtin_provider", return_value=controller): result = ToolManager.get_tool_runtime( provider_type=ToolProviderType.BUILT_IN, provider_id="time", tool_name="current_time", tenant_id="tenant-1", ) assert result == "runtime-tool" runtime = tool.fork_tool_runtime.call_args.kwargs["runtime"] assert runtime.tenant_id == "tenant-1" assert runtime.credentials == {} def test_get_tool_runtime_builtin_missing_tool_raises(): controller = SimpleNamespace(get_tool=Mock(return_value=None), need_credentials=False) with patch.object(ToolManager, "get_builtin_provider", return_value=controller): with pytest.raises(ToolProviderNotFoundError, match="builtin tool missing not found"): ToolManager.get_tool_runtime( provider_type=ToolProviderType.BUILT_IN, provider_id="time", tool_name="missing", tenant_id="tenant-1", ) def test_get_tool_runtime_builtin_with_credentials_decrypts_and_forks(): tool = Mock() tool.fork_tool_runtime.return_value = "runtime-tool" controller = SimpleNamespace( get_tool=Mock(return_value=tool), need_credentials=True, get_credentials_schema_by_type=Mock(return_value=[]), ) builtin_provider = SimpleNamespace( id="cred-1", credential_type=CredentialType.API_KEY.value, credentials={"encrypted": "value"}, expires_at=-1, user_id="user-1", ) with patch.object(ToolManager, "get_builtin_provider", return_value=controller): with patch("core.helper.credential_utils.check_credential_policy_compliance"): with patch("core.tools.tool_manager.db") as mock_db: mock_db.session.scalar.return_value = builtin_provider encrypter = Mock() encrypter.decrypt.return_value = {"api_key": "secret"} cache = Mock() with patch("core.tools.tool_manager.create_provider_encrypter", return_value=(encrypter, cache)): result = ToolManager.get_tool_runtime( provider_type=ToolProviderType.BUILT_IN, provider_id="time", tool_name="weekday", tenant_id="tenant-1", ) assert result == "runtime-tool" runtime = tool.fork_tool_runtime.call_args.kwargs["runtime"] assert runtime.credentials == {"api_key": "secret"} assert runtime.credential_type == CredentialType.API_KEY @patch("core.tools.tool_manager.create_provider_encrypter") @patch("core.plugin.impl.oauth.OAuthHandler") @patch( "services.tools.builtin_tools_manage_service.BuiltinToolManageService.get_oauth_client", return_value={"client_id": "id"}, ) @patch("core.tools.tool_manager.db") @patch("core.tools.tool_manager.time.time", return_value=1000) @patch("core.helper.credential_utils.check_credential_policy_compliance") def test_get_tool_runtime_builtin_refreshes_expired_oauth_credentials( mock_check, mock_time, mock_db, mock_get_oauth_client, mock_oauth_handler_cls, mock_create_provider_encrypter, ): tool = Mock() tool.fork_tool_runtime.return_value = "runtime-tool" controller = SimpleNamespace( get_tool=Mock(return_value=tool), need_credentials=True, get_credentials_schema_by_type=Mock(return_value=[]), ) builtin_provider = SimpleNamespace( id="cred-1", credential_type=CredentialType.OAUTH2.value, credentials={"encrypted": "value"}, encrypted_credentials=None, expires_at=1, user_id="user-1", ) refreshed = SimpleNamespace(credentials={"token": "new"}, expires_at=123456) mock_db.session.scalar.return_value = builtin_provider encrypter = Mock() encrypter.decrypt.return_value = {"token": "old"} encrypter.encrypt.return_value = {"token": "encrypted"} cache = Mock() mock_create_provider_encrypter.return_value = (encrypter, cache) mock_oauth_handler_cls.return_value.refresh_credentials.return_value = refreshed with patch.object(ToolManager, "get_builtin_provider", return_value=controller): result = ToolManager.get_tool_runtime( provider_type=ToolProviderType.BUILT_IN, provider_id="time", tool_name="weekday", tenant_id="tenant-1", ) assert result == "runtime-tool" assert builtin_provider.expires_at == refreshed.expires_at assert builtin_provider.encrypted_credentials == json.dumps({"token": "encrypted"}) mock_db.session.commit.assert_called_once() cache.delete.assert_called_once() def test_get_tool_runtime_builtin_plugin_provider_deleted_raises(): plugin_controller = object.__new__(PluginToolProviderController) plugin_controller.entity = SimpleNamespace(credentials_schema=[{"name": "k"}], oauth_schema=None) plugin_controller.get_tool = Mock(return_value=Mock()) plugin_controller.get_credentials_schema_by_type = Mock(return_value=[]) with patch.object(ToolManager, "get_builtin_provider", return_value=plugin_controller): with patch("core.tools.tool_manager.is_valid_uuid", return_value=True): with patch("core.tools.tool_manager.db") as mock_db: mock_db.session.scalar.return_value = None with pytest.raises(ToolProviderNotFoundError, match="provider has been deleted"): ToolManager.get_tool_runtime( provider_type=ToolProviderType.BUILT_IN, provider_id="time", tool_name="weekday", tenant_id="tenant-1", credential_id="uuid-id", ) def test_get_tool_runtime_api_path(): api_tool = Mock() api_tool.fork_tool_runtime.return_value = "api-runtime" api_provider = Mock() api_provider.get_tool.return_value = api_tool with patch.object(ToolManager, "get_api_provider_controller", return_value=(api_provider, {"c": "enc"})): encrypter = Mock() encrypter.decrypt.return_value = {"c": "dec"} with patch("core.tools.tool_manager.create_tool_provider_encrypter", return_value=(encrypter, Mock())): assert ( ToolManager.get_tool_runtime( provider_type=ToolProviderType.API, provider_id="api-1", tool_name="search", tenant_id="tenant-1", ) == "api-runtime" ) def test_get_tool_runtime_workflow_path(): workflow_provider = SimpleNamespace(tenant_id="tenant-1") workflow_tool = Mock() workflow_tool.fork_tool_runtime.return_value = "wf-runtime" workflow_controller = Mock() workflow_controller.get_tools.return_value = [workflow_tool] session = Mock() session.begin.return_value = _cm(None) session.scalar.return_value = workflow_provider with patch("core.tools.tool_manager.db") as mock_db: mock_db.engine = object() with patch("core.tools.tool_manager.Session", return_value=_cm(session)): with patch( "core.tools.tool_manager.ToolTransformService.workflow_provider_to_controller", return_value=workflow_controller, ): assert ( ToolManager.get_tool_runtime( provider_type=ToolProviderType.WORKFLOW, provider_id="wf-1", tool_name="wf", tenant_id="tenant-1", ) == "wf-runtime" ) def test_get_tool_runtime_plugin_path(): with patch.object( ToolManager, "get_plugin_provider", return_value=SimpleNamespace(get_tool=lambda _: "plugin-tool"), ): assert ( ToolManager.get_tool_runtime( provider_type=ToolProviderType.PLUGIN, provider_id="plugin-1", tool_name="p", tenant_id="tenant-1", ) == "plugin-tool" ) def test_get_tool_runtime_mcp_path(): with patch.object( ToolManager, "get_mcp_provider_controller", return_value=SimpleNamespace(get_tool=lambda _: "mcp-tool"), ): assert ( ToolManager.get_tool_runtime( provider_type=ToolProviderType.MCP, provider_id="mcp-1", tool_name="m", tenant_id="tenant-1", ) == "mcp-tool" ) def test_get_tool_runtime_app_not_implemented(): with pytest.raises(NotImplementedError, match="app provider not implemented"): ToolManager.get_tool_runtime( provider_type=ToolProviderType.APP, provider_id="app", tool_name="x", tenant_id="tenant-1", ) def test_get_agent_runtime_apply_runtime_parameters(): parameter = ToolParameter.get_simple_instance( name="query", llm_description="query", typ=ToolParameter.ToolParameterType.STRING, required=False, ) parameter.form = ToolParameter.ToolParameterForm.FORM tool_runtime = SimpleNamespace(runtime=ToolRuntime(tenant_id="tenant-1", runtime_parameters={})) tool_runtime.get_merged_runtime_parameters = Mock(return_value=[parameter]) with patch.object(ToolManager, "get_tool_runtime", return_value=tool_runtime) as mock_get_tool_runtime: with patch.object(ToolManager, "_convert_tool_parameters_type", return_value={"query": "hello"}): manager = Mock() manager.decrypt_tool_parameters.return_value = {"query": "decrypted"} with patch("core.tools.tool_manager.ToolParameterConfigurationManager", return_value=manager): agent_tool = SimpleNamespace( provider_type=ToolProviderType.API, provider_id="api-1", tool_name="search", tool_parameters={"query": "hello"}, credential_id=None, ) result = ToolManager.get_agent_tool_runtime( tenant_id="tenant-1", app_id="app-1", agent_tool=agent_tool, user_id="user-1", invoke_from=InvokeFrom.DEBUGGER, variable_pool=None, ) assert result is tool_runtime assert tool_runtime.runtime.runtime_parameters["query"] == "decrypted" mock_get_tool_runtime.assert_called_once_with( provider_type=ToolProviderType.API, provider_id="api-1", tool_name="search", tenant_id="tenant-1", user_id="user-1", invoke_from=InvokeFrom.DEBUGGER, tool_invoke_from=ToolInvokeFrom.AGENT, credential_id=None, ) def test_get_workflow_runtime_apply_runtime_parameters(): parameter = ToolParameter.get_simple_instance( name="query", llm_description="query", typ=ToolParameter.ToolParameterType.STRING, required=False, ) parameter.form = ToolParameter.ToolParameterForm.FORM workflow_tool = SimpleNamespace( provider_type=ToolProviderType.API, provider_id="api-1", tool_name="search", tool_configurations={"query": "hello"}, credential_id=None, ) tool_runtime2 = SimpleNamespace(runtime=ToolRuntime(tenant_id="tenant-1", runtime_parameters={})) tool_runtime2.get_merged_runtime_parameters = Mock(return_value=[parameter]) with patch.object(ToolManager, "get_tool_runtime", return_value=tool_runtime2) as mock_get_tool_runtime: with patch.object(ToolManager, "_convert_tool_parameters_type", return_value={"query": "workflow"}): manager = Mock() manager.decrypt_tool_parameters.return_value = {"query": "workflow-dec"} with patch("core.tools.tool_manager.ToolParameterConfigurationManager", return_value=manager): workflow_result = ToolManager.get_workflow_tool_runtime( tenant_id="tenant-1", app_id="app-1", node_id="node-1", workflow_tool=workflow_tool, user_id="user-1", invoke_from=InvokeFrom.DEBUGGER, variable_pool=None, ) assert workflow_result is tool_runtime2 assert tool_runtime2.runtime.runtime_parameters["query"] == "workflow-dec" mock_get_tool_runtime.assert_called_once_with( provider_type=ToolProviderType.API, provider_id="api-1", tool_name="search", tenant_id="tenant-1", user_id="user-1", invoke_from=InvokeFrom.DEBUGGER, tool_invoke_from=ToolInvokeFrom.WORKFLOW, credential_id=None, ) def test_get_agent_runtime_raises_when_runtime_missing(): tool_runtime = SimpleNamespace(runtime=None, get_merged_runtime_parameters=lambda: []) agent_tool = SimpleNamespace( provider_type=ToolProviderType.API, provider_id="api-1", tool_name="search", tool_parameters={}, credential_id=None, ) with patch.object(ToolManager, "get_tool_runtime", return_value=tool_runtime): with patch.object(ToolManager, "_convert_tool_parameters_type", return_value={}): with patch("core.tools.tool_manager.ToolParameterConfigurationManager", return_value=Mock()): with pytest.raises(ValueError, match="runtime not found"): ToolManager.get_agent_tool_runtime( tenant_id="tenant-1", app_id="app-1", agent_tool=agent_tool, ) def test_get_tool_runtime_from_plugin_only_uses_form_parameters(): form_param = ToolParameter.get_simple_instance( name="q", llm_description="query", typ=ToolParameter.ToolParameterType.STRING, required=False, ) form_param.form = ToolParameter.ToolParameterForm.FORM llm_param = ToolParameter.get_simple_instance( name="llm", llm_description="llm", typ=ToolParameter.ToolParameterType.STRING, required=False, ) llm_param.form = ToolParameter.ToolParameterForm.LLM tool_entity = SimpleNamespace(runtime=ToolRuntime(tenant_id="tenant-1", runtime_parameters={})) tool_entity.get_merged_runtime_parameters = Mock(return_value=[form_param, llm_param]) with patch.object(ToolManager, "get_tool_runtime", return_value=tool_entity) as mock_get_tool_runtime: result = ToolManager.get_tool_runtime_from_plugin( tool_type=ToolProviderType.API, tenant_id="tenant-1", provider="api-1", tool_name="search", tool_parameters={"q": "hello", "llm": "ignore"}, user_id="user-1", ) assert result is tool_entity assert tool_entity.runtime.runtime_parameters == {"q": "hello"} mock_get_tool_runtime.assert_called_once_with( provider_type=ToolProviderType.API, provider_id="api-1", tool_name="search", tenant_id="tenant-1", user_id="user-1", invoke_from=InvokeFrom.SERVICE_API, tool_invoke_from=ToolInvokeFrom.PLUGIN, credential_id=None, ) def test_hardcoded_provider_icon_success(): provider = SimpleNamespace(entity=SimpleNamespace(identity=SimpleNamespace(icon="icon.svg"))) with patch.object(ToolManager, "get_hardcoded_provider", return_value=provider): with patch("core.tools.tool_manager.path.exists", return_value=True): with patch("core.tools.tool_manager.mimetypes.guess_type", return_value=("image/svg+xml", None)): icon_path, mime = ToolManager.get_hardcoded_provider_icon("time") assert icon_path.endswith("icon.svg") assert mime == "image/svg+xml" def test_hardcoded_provider_icon_missing_raises(): provider = SimpleNamespace(entity=SimpleNamespace(identity=SimpleNamespace(icon="icon.svg"))) with patch.object(ToolManager, "get_hardcoded_provider", return_value=provider): with patch("core.tools.tool_manager.path.exists", return_value=False): with pytest.raises(ToolProviderNotFoundError, match="icon not found"): ToolManager.get_hardcoded_provider_icon("time") def test_list_hardcoded_providers_cache_hit(): ToolManager._hardcoded_providers = {"p": Mock()} ToolManager._builtin_providers_loaded = True assert list(ToolManager.list_hardcoded_providers()) == list(ToolManager._hardcoded_providers.values()) def test_clear_hardcoded_providers_cache_resets(): ToolManager._hardcoded_providers = {"p": Mock()} ToolManager._builtin_providers_loaded = True ToolManager.clear_hardcoded_providers_cache() assert ToolManager._hardcoded_providers == {} assert ToolManager._builtin_providers_loaded is False def test_list_hardcoded_providers_internal_loader(): good_provider = SimpleNamespace( entity=SimpleNamespace(identity=SimpleNamespace(name="good")), get_tools=lambda: [SimpleNamespace(entity=SimpleNamespace(identity=SimpleNamespace(name="tool-a", label="A")))], ) provider_class = Mock(return_value=good_provider) with patch("core.tools.tool_manager.listdir", return_value=["good", "bad", "__skip"]): with patch("core.tools.tool_manager.path.isdir", side_effect=lambda p: "good" in p or "bad" in p): with patch( "core.tools.tool_manager.load_single_subclass_from_source", side_effect=[provider_class, RuntimeError("boom")], ): ToolManager._hardcoded_providers = {} ToolManager._builtin_tools_labels = {} providers = list(ToolManager._list_hardcoded_providers()) assert providers == [good_provider] assert ToolManager._hardcoded_providers["good"] is good_provider assert ToolManager._builtin_tools_labels["tool-a"] == "A" assert ToolManager._builtin_providers_loaded is True def test_get_tool_label_loads_cache_and_handles_missing(): ToolManager._builtin_tools_labels = {} def _load(): ToolManager._builtin_tools_labels["tool-a"] = "Label A" with patch.object(ToolManager, "load_hardcoded_providers_cache", side_effect=_load): assert ToolManager.get_tool_label("tool-a") == "Label A" assert ToolManager.get_tool_label("missing") is None def test_list_default_builtin_providers_for_postgres_and_mysql(): provider_records = [SimpleNamespace(id="id-1"), SimpleNamespace(id="id-2")] for scheme in ("postgresql", "mysql"): session = Mock() session.execute.return_value.all.return_value = [SimpleNamespace(id="id-1"), SimpleNamespace(id="id-2")] session.scalars.return_value = iter(provider_records) with patch("core.tools.tool_manager.dify_config", SimpleNamespace(SQLALCHEMY_DATABASE_URI_SCHEME=scheme)): with patch("core.tools.tool_manager.db") as mock_db: mock_db.engine = object() with patch("core.tools.tool_manager.Session", return_value=_cm(session)): providers = ToolManager.list_default_builtin_providers("tenant-1") assert providers == provider_records def test_list_providers_from_api_covers_builtin_api_workflow_and_mcp(monkeypatch: pytest.MonkeyPatch): hardcoded_controller = SimpleNamespace(entity=SimpleNamespace(identity=SimpleNamespace(name="hardcoded"))) plugin_controller = object.__new__(PluginToolProviderController) plugin_controller.entity = SimpleNamespace(identity=SimpleNamespace(name="plugin-provider")) api_db_provider_good = SimpleNamespace(id="api-1") api_db_provider_bad = SimpleNamespace(id="api-2") api_controller = SimpleNamespace(provider_id="api-1") workflow_db_provider_good = SimpleNamespace(id="wf-1") workflow_db_provider_bad = SimpleNamespace(id="wf-2") workflow_controller = SimpleNamespace(provider_id="wf-1") session = Mock() session.scalars.side_effect = [ SimpleNamespace(all=lambda: [api_db_provider_good, api_db_provider_bad]), SimpleNamespace(all=lambda: [workflow_db_provider_good, workflow_db_provider_bad]), ] _setup_list_providers_from_api_mocks( monkeypatch, session=session, hardcoded_controller=hardcoded_controller, plugin_controller=plugin_controller, api_controller=api_controller, workflow_controller=workflow_controller, ) providers = ToolManager.list_providers_from_api(user_id="user-1", tenant_id="tenant-1", typ="") names = {provider.name for provider in providers} assert {"hardcoded", "plugin-provider", "api-provider", "workflow-provider", "mcp-provider"} <= names def test_get_api_provider_controller_returns_controller_and_credentials(): provider = SimpleNamespace( id="api-1", tenant_id="tenant-1", name="api-provider", description="desc", credentials={"auth_type": "api_key_query"}, credentials_str='{"auth_type": "api_key_query", "api_key_value": "secret"}', schema_type="openapi", schema="schema", tools=[], icon='{"background": "#000", "content": "A"}', privacy_policy="privacy", custom_disclaimer="disclaimer", ) controller = Mock() with patch("core.tools.tool_manager.db") as mock_db: mock_db.session.scalar.return_value = provider with patch( "core.tools.tool_manager.ApiToolProviderController.from_db", return_value=controller ) as mock_from_db: built_controller, credentials = ToolManager.get_api_provider_controller("tenant-1", "api-1") assert built_controller is controller assert credentials == provider.credentials mock_from_db.assert_called_with(provider, ApiProviderAuthType.API_KEY_QUERY) controller.load_bundled_tools.assert_called_once_with(provider.tools) def test_user_get_api_provider_masks_credentials_and_adds_labels(): provider = SimpleNamespace( id="api-1", tenant_id="tenant-1", name="api-provider", description="desc", credentials={"auth_type": "api_key_query"}, credentials_str='{"auth_type": "api_key_query", "api_key_value": "secret"}', schema_type="openapi", schema="schema", tools=[], icon='{"background": "#000", "content": "A"}', privacy_policy="privacy", custom_disclaimer="disclaimer", ) controller = Mock() with patch("core.tools.tool_manager.db") as mock_db: mock_db.session.scalar.return_value = provider with patch("core.tools.tool_manager.ApiToolProviderController.from_db", return_value=controller): encrypter = Mock() encrypter.decrypt.return_value = {"api_key_value": "secret"} encrypter.mask_plugin_credentials.return_value = {"api_key_value": "***"} with patch("core.tools.tool_manager.create_tool_provider_encrypter", return_value=(encrypter, Mock())): with patch("core.tools.tool_manager.ToolLabelManager.get_tool_labels", return_value=["search"]): user_payload = ToolManager.user_get_api_provider("api-provider", "tenant-1") assert user_payload["credentials"]["api_key_value"] == "***" assert user_payload["labels"] == ["search"] def test_get_api_provider_controller_not_found_raises(): with patch("core.tools.tool_manager.db") as mock_db: mock_db.session.scalar.return_value = None with pytest.raises(ToolProviderNotFoundError, match="api provider missing not found"): ToolManager.get_api_provider_controller("tenant-1", "missing") def test_get_mcp_provider_controller_returns_controller(): provider_entity = SimpleNamespace(provider_icon={"background": "#111", "content": "M"}) controller = Mock() session = Mock() with patch("core.tools.tool_manager.db") as mock_db: mock_db.engine = object() with patch("core.tools.tool_manager.Session", return_value=_cm(session)): with patch("core.tools.tool_manager.MCPToolManageService") as mock_service_cls: mock_service = mock_service_cls.return_value mock_service.get_provider.return_value = provider_entity with patch("core.tools.tool_manager.MCPToolProviderController.from_db", return_value=controller): built = ToolManager.get_mcp_provider_controller("tenant-1", "mcp-1") assert built is controller def test_generate_mcp_tool_icon_url_returns_provider_icon(): provider_entity = SimpleNamespace(provider_icon={"background": "#111", "content": "M"}) session = Mock() with patch("core.tools.tool_manager.db") as mock_db: mock_db.engine = object() with patch("core.tools.tool_manager.Session", return_value=_cm(session)): with patch("core.tools.tool_manager.MCPToolManageService") as mock_service_cls: mock_service = mock_service_cls.return_value mock_service.get_provider_entity.return_value = provider_entity assert ToolManager.generate_mcp_tool_icon_url("tenant-1", "mcp-1") == provider_entity.provider_icon def test_get_mcp_provider_controller_missing_raises(): session = Mock() with patch("core.tools.tool_manager.db") as mock_db: mock_db.engine = object() with patch("core.tools.tool_manager.Session", return_value=_cm(session)): with patch("core.tools.tool_manager.MCPToolManageService") as mock_service_cls: mock_service_cls.return_value.get_provider.side_effect = ValueError("missing") with pytest.raises(ToolProviderNotFoundError, match="mcp provider mcp-1 not found"): ToolManager.get_mcp_provider_controller("tenant-1", "mcp-1") def test_generate_tool_icon_urls_for_builtin_and_plugin(): with patch("core.tools.tool_manager.dify_config.CONSOLE_API_URL", "https://console.example.com"): builtin_url = ToolManager.generate_builtin_tool_icon_url("time") plugin_url = ToolManager.generate_plugin_tool_icon_url("tenant-1", "icon.svg") assert builtin_url.endswith("/tool-provider/builtin/time/icon") assert "/plugin/icon" in plugin_url def test_generate_tool_icon_urls_for_workflow_and_api(): workflow_provider = SimpleNamespace(icon='{"background": "#222", "content": "W"}') api_provider = SimpleNamespace(icon='{"background": "#333", "content": "A"}') with patch("core.tools.tool_manager.db") as mock_db: mock_db.session.scalar.side_effect = [workflow_provider, api_provider] assert ToolManager.generate_workflow_tool_icon_url("tenant-1", "wf-1") == {"background": "#222", "content": "W"} assert ToolManager.generate_api_tool_icon_url("tenant-1", "api-1") == {"background": "#333", "content": "A"} def test_generate_tool_icon_urls_missing_workflow_and_api_use_default(): with patch("core.tools.tool_manager.db") as mock_db: mock_db.session.scalar.return_value = None assert ToolManager.generate_workflow_tool_icon_url("tenant-1", "missing")["background"] == "#252525" assert ToolManager.generate_api_tool_icon_url("tenant-1", "missing")["background"] == "#252525" def test_get_tool_icon_for_builtin_provider_variants(): plugin_provider = object.__new__(PluginToolProviderController) plugin_provider.entity = SimpleNamespace(identity=SimpleNamespace(icon="plugin.svg")) with patch.object(ToolManager, "get_builtin_provider", return_value=plugin_provider): with patch.object(ToolManager, "generate_plugin_tool_icon_url", return_value="plugin-icon"): assert ToolManager.get_tool_icon("tenant-1", ToolProviderType.BUILT_IN, "plugin-provider") == "plugin-icon" with patch.object(ToolManager, "get_builtin_provider", return_value=SimpleNamespace()): with patch.object(ToolManager, "generate_builtin_tool_icon_url", return_value="builtin-icon"): assert ToolManager.get_tool_icon("tenant-1", ToolProviderType.BUILT_IN, "time") == "builtin-icon" def test_get_tool_icon_for_api_workflow_and_mcp(): with patch.object(ToolManager, "generate_api_tool_icon_url", return_value={"background": "#000"}): assert ToolManager.get_tool_icon("tenant-1", ToolProviderType.API, "api-1") == {"background": "#000"} with patch.object(ToolManager, "generate_workflow_tool_icon_url", return_value={"background": "#111"}): assert ToolManager.get_tool_icon("tenant-1", ToolProviderType.WORKFLOW, "wf-1") == {"background": "#111"} with patch.object(ToolManager, "generate_mcp_tool_icon_url", return_value={"background": "#222"}): assert ToolManager.get_tool_icon("tenant-1", ToolProviderType.MCP, "mcp-1") == {"background": "#222"} def test_get_tool_icon_plugin_error_returns_default(): plugin_provider = object.__new__(PluginToolProviderController) plugin_provider.entity = SimpleNamespace(identity=SimpleNamespace(icon="plugin.svg")) with patch.object(ToolManager, "get_plugin_provider", return_value=plugin_provider): with patch.object(ToolManager, "generate_plugin_tool_icon_url", side_effect=RuntimeError("fail")): icon = ToolManager.get_tool_icon("tenant-1", ToolProviderType.PLUGIN, "plugin-provider") assert icon["background"] == "#252525" def test_get_tool_icon_invalid_provider_type_raises(): with pytest.raises(ValueError, match="provider type"): ToolManager.get_tool_icon("tenant-1", "invalid", "x") # type: ignore[arg-type] def test_convert_tool_parameters_type_agent_and_workflow_branches(): file_param = ToolParameter.get_simple_instance( name="file", llm_description="file", typ=ToolParameter.ToolParameterType.FILE, required=True, ) file_param.form = ToolParameter.ToolParameterForm.FORM with pytest.raises(ValueError, match="file type parameter file not supported in agent"): ToolManager._convert_tool_parameters_type( parameters=[file_param], variable_pool=None, tool_configurations={"file": "x"}, typ="agent", ) text_param = ToolParameter.get_simple_instance( name="text", llm_description="text", typ=ToolParameter.ToolParameterType.STRING, required=False, ) text_param.form = ToolParameter.ToolParameterForm.FORM plain = ToolManager._convert_tool_parameters_type( parameters=[text_param], variable_pool=None, tool_configurations={"text": "hello"}, typ="workflow", ) assert plain == {"text": "hello"} variable_pool = Mock() variable_pool.get.return_value = SimpleNamespace(value="from-variable") variable_pool.convert_template.return_value = SimpleNamespace(text="from-template") mixed = ToolManager._convert_tool_parameters_type( parameters=[text_param], variable_pool=variable_pool, tool_configurations={"text": {"type": "mixed", "value": "Hello {{name}}"}}, typ="workflow", ) assert mixed == {"text": "from-template"} variable = ToolManager._convert_tool_parameters_type( parameters=[text_param], variable_pool=variable_pool, tool_configurations={"text": {"type": "variable", "value": ["sys", "query"]}}, typ="workflow", ) assert variable == {"text": "from-variable"} def test_convert_tool_parameters_type_constant_branch(): text_param = ToolParameter.get_simple_instance( name="text", llm_description="text", typ=ToolParameter.ToolParameterType.STRING, required=False, ) text_param.form = ToolParameter.ToolParameterForm.FORM variable_pool = Mock() constant = ToolManager._convert_tool_parameters_type( parameters=[text_param], variable_pool=variable_pool, tool_configurations={"text": {"type": "constant", "value": "fixed"}}, typ="workflow", ) assert constant == {"text": "fixed"} def test_convert_tool_parameters_type_model_selector_from_legacy_top_level_config(): model_param = ToolParameter.get_simple_instance( name="vision_llm_model", llm_description="vision model", typ=ToolParameter.ToolParameterType.MODEL_SELECTOR, required=True, ) model_param.form = ToolParameter.ToolParameterForm.FORM variable_pool = Mock() runtime_parameters = ToolManager._convert_tool_parameters_type( parameters=[model_param], variable_pool=variable_pool, tool_configurations={ "vision_llm_model": { "type": "constant", "value": "", "provider": "langgenius/tongyi/tongyi", "model": "qwen3-vl-plus", "model_type": "llm", "mode": "chat", } }, typ="workflow", ) assert runtime_parameters == { "vision_llm_model": { "provider": "langgenius/tongyi/tongyi", "model": "qwen3-vl-plus", "model_type": "llm", "mode": "chat", } } def test_convert_tool_parameters_type_model_selector_from_constant_value_config(): model_param = ToolParameter.get_simple_instance( name="tts_model", llm_description="tts model", typ=ToolParameter.ToolParameterType.MODEL_SELECTOR, required=True, ) model_param.form = ToolParameter.ToolParameterForm.FORM variable_pool = Mock() runtime_parameters = ToolManager._convert_tool_parameters_type( parameters=[model_param], variable_pool=variable_pool, tool_configurations={ "tts_model": { "type": "constant", "value": { "provider": "langgenius/tongyi/tongyi", "model": "qwen3-tts-flash", "model_type": "tts", "language": "Chinese", "voice": "Cherry", }, } }, typ="workflow", ) assert runtime_parameters == { "tts_model": { "provider": "langgenius/tongyi/tongyi", "model": "qwen3-tts-flash", "model_type": "tts", "language": "Chinese", "voice": "Cherry", } }