mirror of https://github.com/langgenius/dify.git
test: add UTs for api/ services.plugin (#32588)
Co-authored-by: QuantumGhost <obelisk.reg+git@gmail.com> Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Dev Sharma
GitHub
parent
0045e387f5
commit
135b3a15a6
|
|
@ -0,0 +1,39 @@
|
|||
"""Shared fixtures for services.plugin test suite."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from services.feature_service import PluginInstallationScope
|
||||
|
||||
|
||||
def make_features(
|
||||
restrict_to_marketplace: bool = False,
|
||||
scope: PluginInstallationScope = PluginInstallationScope.ALL,
|
||||
) -> MagicMock:
|
||||
"""Create a mock FeatureService.get_system_features() result."""
|
||||
features = MagicMock()
|
||||
features.plugin_installation_permission.restrict_to_marketplace_only = restrict_to_marketplace
|
||||
features.plugin_installation_permission.plugin_installation_scope = scope
|
||||
return features
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_installer(monkeypatch):
|
||||
"""Patch PluginInstaller at the service import site."""
|
||||
mock = MagicMock()
|
||||
monkeypatch.setattr("services.plugin.plugin_service.PluginInstaller", lambda: mock)
|
||||
return mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_features():
|
||||
"""Patch FeatureService to return permissive defaults."""
|
||||
from unittest.mock import patch
|
||||
|
||||
features = make_features()
|
||||
with patch("services.plugin.plugin_service.FeatureService") as mock_fs:
|
||||
mock_fs.get_system_features.return_value = features
|
||||
yield features
|
||||
|
|
@ -0,0 +1,172 @@
|
|||
"""Tests for services.plugin.dependencies_analysis.DependenciesAnalysisService.
|
||||
|
||||
Covers: provider ID resolution, leaked dependency detection with version
|
||||
extraction, dependency generation from multiple sources, and latest
|
||||
dependencies via marketplace.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from core.plugin.entities.plugin import PluginDependency, PluginInstallationSource
|
||||
from services.plugin.dependencies_analysis import DependenciesAnalysisService
|
||||
|
||||
|
||||
class TestAnalyzeToolDependency:
|
||||
def test_valid_three_part_id(self):
|
||||
result = DependenciesAnalysisService.analyze_tool_dependency("langgenius/google/google")
|
||||
assert result == "langgenius/google"
|
||||
|
||||
def test_single_part_expands_to_langgenius(self):
|
||||
result = DependenciesAnalysisService.analyze_tool_dependency("websearch")
|
||||
assert result == "langgenius/websearch"
|
||||
|
||||
def test_invalid_format_raises(self):
|
||||
with pytest.raises(ValueError):
|
||||
DependenciesAnalysisService.analyze_tool_dependency("bad/format")
|
||||
|
||||
|
||||
class TestAnalyzeModelProviderDependency:
|
||||
def test_valid_three_part_id(self):
|
||||
result = DependenciesAnalysisService.analyze_model_provider_dependency("langgenius/openai/openai")
|
||||
assert result == "langgenius/openai"
|
||||
|
||||
def test_google_maps_to_gemini(self):
|
||||
result = DependenciesAnalysisService.analyze_model_provider_dependency("langgenius/google/google")
|
||||
assert result == "langgenius/gemini"
|
||||
|
||||
def test_single_part_expands(self):
|
||||
result = DependenciesAnalysisService.analyze_model_provider_dependency("anthropic")
|
||||
assert result == "langgenius/anthropic"
|
||||
|
||||
|
||||
class TestGetLeakedDependencies:
|
||||
def _make_dependency(self, identifier: str, dep_type=PluginDependency.Type.Marketplace):
|
||||
return PluginDependency(
|
||||
type=dep_type,
|
||||
value=PluginDependency.Marketplace(marketplace_plugin_unique_identifier=identifier),
|
||||
)
|
||||
|
||||
@patch("services.plugin.dependencies_analysis.PluginInstaller")
|
||||
def test_returns_empty_when_all_present(self, mock_installer_cls):
|
||||
mock_installer_cls.return_value.fetch_missing_dependencies.return_value = []
|
||||
deps = [self._make_dependency("org/plugin:1.0.0@hash")]
|
||||
|
||||
result = DependenciesAnalysisService.get_leaked_dependencies("t1", deps)
|
||||
|
||||
assert result == []
|
||||
|
||||
@patch("services.plugin.dependencies_analysis.PluginInstaller")
|
||||
def test_returns_missing_with_version_extracted(self, mock_installer_cls):
|
||||
missing = MagicMock()
|
||||
missing.plugin_unique_identifier = "org/plugin:1.2.3@hash"
|
||||
missing.current_identifier = "org/plugin:1.0.0@oldhash"
|
||||
mock_installer_cls.return_value.fetch_missing_dependencies.return_value = [missing]
|
||||
|
||||
deps = [self._make_dependency("org/plugin:1.2.3@hash")]
|
||||
|
||||
result = DependenciesAnalysisService.get_leaked_dependencies("t1", deps)
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].value.version == "1.2.3"
|
||||
|
||||
@patch("services.plugin.dependencies_analysis.PluginInstaller")
|
||||
def test_skips_present_dependencies(self, mock_installer_cls):
|
||||
missing = MagicMock()
|
||||
missing.plugin_unique_identifier = "org/missing:1.0.0@hash"
|
||||
missing.current_identifier = None
|
||||
mock_installer_cls.return_value.fetch_missing_dependencies.return_value = [missing]
|
||||
|
||||
deps = [
|
||||
self._make_dependency("org/present:1.0.0@hash"),
|
||||
self._make_dependency("org/missing:1.0.0@hash"),
|
||||
]
|
||||
|
||||
result = DependenciesAnalysisService.get_leaked_dependencies("t1", deps)
|
||||
|
||||
assert len(result) == 1
|
||||
|
||||
|
||||
class TestGenerateDependencies:
|
||||
def _make_installation(self, source, identifier, meta=None):
|
||||
install = MagicMock()
|
||||
install.source = source
|
||||
install.plugin_unique_identifier = identifier
|
||||
install.meta = meta or {}
|
||||
return install
|
||||
|
||||
@patch("services.plugin.dependencies_analysis.PluginInstaller")
|
||||
def test_github_source(self, mock_installer_cls):
|
||||
install = self._make_installation(
|
||||
PluginInstallationSource.Github,
|
||||
"org/plugin:1.0.0@hash",
|
||||
{"repo": "org/repo", "version": "v1.0", "package": "plugin.difypkg"},
|
||||
)
|
||||
mock_installer_cls.return_value.fetch_plugin_installation_by_ids.return_value = [install]
|
||||
|
||||
result = DependenciesAnalysisService.generate_dependencies("t1", ["p1"])
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].type == PluginDependency.Type.Github
|
||||
assert result[0].value.repo == "org/repo"
|
||||
|
||||
@patch("services.plugin.dependencies_analysis.PluginInstaller")
|
||||
def test_marketplace_source(self, mock_installer_cls):
|
||||
install = self._make_installation(PluginInstallationSource.Marketplace, "org/plugin:1.0.0@hash")
|
||||
mock_installer_cls.return_value.fetch_plugin_installation_by_ids.return_value = [install]
|
||||
|
||||
result = DependenciesAnalysisService.generate_dependencies("t1", ["p1"])
|
||||
|
||||
assert result[0].type == PluginDependency.Type.Marketplace
|
||||
|
||||
@patch("services.plugin.dependencies_analysis.PluginInstaller")
|
||||
def test_package_source(self, mock_installer_cls):
|
||||
install = self._make_installation(PluginInstallationSource.Package, "org/plugin:1.0.0@hash")
|
||||
mock_installer_cls.return_value.fetch_plugin_installation_by_ids.return_value = [install]
|
||||
|
||||
result = DependenciesAnalysisService.generate_dependencies("t1", ["p1"])
|
||||
|
||||
assert result[0].type == PluginDependency.Type.Package
|
||||
|
||||
@patch("services.plugin.dependencies_analysis.PluginInstaller")
|
||||
def test_remote_source_raises(self, mock_installer_cls):
|
||||
install = self._make_installation(PluginInstallationSource.Remote, "org/plugin:1.0.0@hash")
|
||||
mock_installer_cls.return_value.fetch_plugin_installation_by_ids.return_value = [install]
|
||||
|
||||
with pytest.raises(ValueError, match="remote plugin"):
|
||||
DependenciesAnalysisService.generate_dependencies("t1", ["p1"])
|
||||
|
||||
@patch("services.plugin.dependencies_analysis.PluginInstaller")
|
||||
def test_deduplicates_input_ids(self, mock_installer_cls):
|
||||
mock_installer_cls.return_value.fetch_plugin_installation_by_ids.return_value = []
|
||||
|
||||
DependenciesAnalysisService.generate_dependencies("t1", ["p1", "p1", "p2"])
|
||||
|
||||
call_args = mock_installer_cls.return_value.fetch_plugin_installation_by_ids.call_args[0]
|
||||
assert len(call_args[1]) == 2 # deduplicated
|
||||
|
||||
|
||||
class TestGenerateLatestDependencies:
|
||||
@patch("services.plugin.dependencies_analysis.dify_config")
|
||||
def test_returns_empty_when_marketplace_disabled(self, mock_config):
|
||||
mock_config.MARKETPLACE_ENABLED = False
|
||||
|
||||
result = DependenciesAnalysisService.generate_latest_dependencies(["p1"])
|
||||
|
||||
assert result == []
|
||||
|
||||
@patch("services.plugin.dependencies_analysis.marketplace")
|
||||
@patch("services.plugin.dependencies_analysis.dify_config")
|
||||
def test_returns_marketplace_deps_when_enabled(self, mock_config, mock_marketplace):
|
||||
mock_config.MARKETPLACE_ENABLED = True
|
||||
manifest = MagicMock()
|
||||
manifest.latest_package_identifier = "org/plugin:2.0.0@newhash"
|
||||
mock_marketplace.batch_fetch_plugin_manifests.return_value = [manifest]
|
||||
|
||||
result = DependenciesAnalysisService.generate_latest_dependencies(["p1"])
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].type == PluginDependency.Type.Marketplace
|
||||
|
|
@ -0,0 +1,41 @@
|
|||
"""Tests for services.plugin.endpoint_service.EndpointService.
|
||||
|
||||
Smoke tests to confirm delegation to PluginEndpointClient.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from services.plugin.endpoint_service import EndpointService
|
||||
|
||||
|
||||
class TestEndpointServiceDelegation:
|
||||
@patch("services.plugin.endpoint_service.PluginEndpointClient")
|
||||
def test_create_delegates_correctly(self, mock_client_cls):
|
||||
expected = MagicMock()
|
||||
mock_client_cls.return_value.create_endpoint.return_value = expected
|
||||
|
||||
result = EndpointService.create_endpoint("t1", "u1", "uid-1", "my-endpoint", {"key": "val"})
|
||||
|
||||
assert result is expected
|
||||
mock_client_cls.return_value.create_endpoint.assert_called_once_with(
|
||||
tenant_id="t1", user_id="u1", plugin_unique_identifier="uid-1", name="my-endpoint", settings={"key": "val"}
|
||||
)
|
||||
|
||||
@patch("services.plugin.endpoint_service.PluginEndpointClient")
|
||||
def test_list_delegates_correctly(self, mock_client_cls):
|
||||
expected = MagicMock()
|
||||
mock_client_cls.return_value.list_endpoints.return_value = expected
|
||||
|
||||
result = EndpointService.list_endpoints("t1", "u1", 1, 10)
|
||||
|
||||
assert result is expected
|
||||
|
||||
@patch("services.plugin.endpoint_service.PluginEndpointClient")
|
||||
def test_enable_disable_delegates(self, mock_client_cls):
|
||||
EndpointService.enable_endpoint("t1", "u1", "ep-1")
|
||||
mock_client_cls.return_value.enable_endpoint.assert_called_once()
|
||||
|
||||
EndpointService.disable_endpoint("t1", "u1", "ep-2")
|
||||
mock_client_cls.return_value.disable_endpoint.assert_called_once()
|
||||
|
|
@ -0,0 +1,90 @@
|
|||
"""Tests for services.plugin.oauth_service.OAuthProxyService.
|
||||
|
||||
Covers: CSRF proxy context creation with Redis TTL, context consumption
|
||||
with one-time use semantics, and validation error paths.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from services.plugin.oauth_service import OAuthProxyService
|
||||
|
||||
|
||||
class TestCreateProxyContext:
|
||||
def test_stores_context_in_redis_with_ttl(self):
|
||||
context_id = OAuthProxyService.create_proxy_context(
|
||||
user_id="u1", tenant_id="t1", plugin_id="p1", provider="github"
|
||||
)
|
||||
|
||||
assert context_id # non-empty UUID string
|
||||
from extensions.ext_redis import redis_client
|
||||
|
||||
redis_client.setex.assert_called_once()
|
||||
call_args = redis_client.setex.call_args
|
||||
key = call_args[0][0]
|
||||
ttl = call_args[0][1]
|
||||
stored_data = json.loads(call_args[0][2])
|
||||
|
||||
assert key.startswith("oauth_proxy_context:")
|
||||
assert ttl == 5 * 60
|
||||
assert stored_data["user_id"] == "u1"
|
||||
assert stored_data["tenant_id"] == "t1"
|
||||
assert stored_data["plugin_id"] == "p1"
|
||||
assert stored_data["provider"] == "github"
|
||||
|
||||
def test_includes_credential_id_when_provided(self):
|
||||
OAuthProxyService.create_proxy_context(
|
||||
user_id="u1", tenant_id="t1", plugin_id="p1", provider="github", credential_id="cred-1"
|
||||
)
|
||||
|
||||
from extensions.ext_redis import redis_client
|
||||
|
||||
stored_data = json.loads(redis_client.setex.call_args[0][2])
|
||||
assert stored_data["credential_id"] == "cred-1"
|
||||
|
||||
def test_excludes_credential_id_when_none(self):
|
||||
OAuthProxyService.create_proxy_context(user_id="u1", tenant_id="t1", plugin_id="p1", provider="github")
|
||||
|
||||
from extensions.ext_redis import redis_client
|
||||
|
||||
stored_data = json.loads(redis_client.setex.call_args[0][2])
|
||||
assert "credential_id" not in stored_data
|
||||
|
||||
def test_includes_extra_data(self):
|
||||
OAuthProxyService.create_proxy_context(
|
||||
user_id="u1", tenant_id="t1", plugin_id="p1", provider="github", extra_data={"scope": "repo"}
|
||||
)
|
||||
|
||||
from extensions.ext_redis import redis_client
|
||||
|
||||
stored_data = json.loads(redis_client.setex.call_args[0][2])
|
||||
assert stored_data["scope"] == "repo"
|
||||
|
||||
|
||||
class TestUseProxyContext:
|
||||
def test_raises_when_context_id_empty(self):
|
||||
with pytest.raises(ValueError, match="context_id is required"):
|
||||
OAuthProxyService.use_proxy_context("")
|
||||
|
||||
def test_raises_when_context_not_found(self):
|
||||
from extensions.ext_redis import redis_client
|
||||
|
||||
redis_client.get.return_value = None
|
||||
|
||||
with pytest.raises(ValueError, match="context_id is invalid"):
|
||||
OAuthProxyService.use_proxy_context("nonexistent-id")
|
||||
|
||||
def test_returns_data_and_deletes_key(self):
|
||||
from extensions.ext_redis import redis_client
|
||||
|
||||
stored = {"user_id": "u1", "tenant_id": "t1", "plugin_id": "p1", "provider": "github"}
|
||||
redis_client.get.return_value = json.dumps(stored).encode()
|
||||
|
||||
result = OAuthProxyService.use_proxy_context("valid-id")
|
||||
|
||||
assert result == stored
|
||||
expected_key = "oauth_proxy_context:valid-id"
|
||||
redis_client.delete.assert_called_once_with(expected_key)
|
||||
|
|
@ -0,0 +1,216 @@
|
|||
"""Tests for services.plugin.plugin_parameter_service.PluginParameterService.
|
||||
|
||||
Covers: dynamic select options via tool and trigger credential paths,
|
||||
HIDDEN_VALUE replacement, and error handling for missing records.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from services.plugin.plugin_parameter_service import PluginParameterService
|
||||
|
||||
|
||||
class TestGetDynamicSelectOptionsTool:
|
||||
@patch("services.plugin.plugin_parameter_service.DynamicSelectClient")
|
||||
@patch("services.plugin.plugin_parameter_service.ToolManager")
|
||||
def test_no_credentials_needed(self, mock_tool_mgr, mock_client_cls):
|
||||
provider_ctrl = MagicMock()
|
||||
provider_ctrl.need_credentials = False
|
||||
mock_tool_mgr.get_builtin_provider.return_value = provider_ctrl
|
||||
mock_client_cls.return_value.fetch_dynamic_select_options.return_value.options = ["opt1"]
|
||||
|
||||
result = PluginParameterService.get_dynamic_select_options(
|
||||
tenant_id="t1",
|
||||
user_id="u1",
|
||||
plugin_id="p1",
|
||||
provider="google",
|
||||
action="search",
|
||||
parameter="engine",
|
||||
credential_id=None,
|
||||
provider_type="tool",
|
||||
)
|
||||
|
||||
assert result == ["opt1"]
|
||||
call_kwargs = mock_client_cls.return_value.fetch_dynamic_select_options.call_args
|
||||
assert call_kwargs[0][5] == {} # empty credentials
|
||||
|
||||
@patch("services.plugin.plugin_parameter_service.DynamicSelectClient")
|
||||
@patch("services.plugin.plugin_parameter_service.create_tool_provider_encrypter")
|
||||
@patch("services.plugin.plugin_parameter_service.db")
|
||||
@patch("services.plugin.plugin_parameter_service.ToolManager")
|
||||
def test_fetches_credentials_with_credential_id(self, mock_tool_mgr, mock_db, mock_encrypter_fn, mock_client_cls):
|
||||
provider_ctrl = MagicMock()
|
||||
provider_ctrl.need_credentials = True
|
||||
mock_tool_mgr.get_builtin_provider.return_value = provider_ctrl
|
||||
encrypter = MagicMock()
|
||||
encrypter.decrypt.return_value = {"api_key": "decrypted"}
|
||||
mock_encrypter_fn.return_value = (encrypter, None)
|
||||
|
||||
# Mock the Session/query chain
|
||||
db_record = MagicMock()
|
||||
db_record.credentials = {"api_key": "encrypted"}
|
||||
db_record.credential_type = "api_key"
|
||||
|
||||
with patch("services.plugin.plugin_parameter_service.Session") as mock_session_cls:
|
||||
mock_session = MagicMock()
|
||||
mock_session_cls.return_value.__enter__ = MagicMock(return_value=mock_session)
|
||||
mock_session_cls.return_value.__exit__ = MagicMock(return_value=False)
|
||||
mock_session.query.return_value.where.return_value.first.return_value = db_record
|
||||
mock_client_cls.return_value.fetch_dynamic_select_options.return_value.options = ["opt1"]
|
||||
|
||||
result = PluginParameterService.get_dynamic_select_options(
|
||||
tenant_id="t1",
|
||||
user_id="u1",
|
||||
plugin_id="p1",
|
||||
provider="google",
|
||||
action="search",
|
||||
parameter="engine",
|
||||
credential_id="cred-1",
|
||||
provider_type="tool",
|
||||
)
|
||||
|
||||
assert result == ["opt1"]
|
||||
|
||||
@patch("services.plugin.plugin_parameter_service.create_tool_provider_encrypter")
|
||||
@patch("services.plugin.plugin_parameter_service.db")
|
||||
@patch("services.plugin.plugin_parameter_service.ToolManager")
|
||||
def test_raises_when_tool_provider_not_found(self, mock_tool_mgr, mock_db, mock_encrypter_fn):
|
||||
provider_ctrl = MagicMock()
|
||||
provider_ctrl.need_credentials = True
|
||||
mock_tool_mgr.get_builtin_provider.return_value = provider_ctrl
|
||||
mock_encrypter_fn.return_value = (MagicMock(), None)
|
||||
|
||||
with patch("services.plugin.plugin_parameter_service.Session") as mock_session_cls:
|
||||
mock_session = MagicMock()
|
||||
mock_session_cls.return_value.__enter__ = MagicMock(return_value=mock_session)
|
||||
mock_session_cls.return_value.__exit__ = MagicMock(return_value=False)
|
||||
mock_session.query.return_value.where.return_value.order_by.return_value.first.return_value = None
|
||||
|
||||
with pytest.raises(ValueError, match="not found"):
|
||||
PluginParameterService.get_dynamic_select_options(
|
||||
tenant_id="t1",
|
||||
user_id="u1",
|
||||
plugin_id="p1",
|
||||
provider="google",
|
||||
action="search",
|
||||
parameter="engine",
|
||||
credential_id=None,
|
||||
provider_type="tool",
|
||||
)
|
||||
|
||||
|
||||
class TestGetDynamicSelectOptionsTrigger:
|
||||
@patch("services.plugin.plugin_parameter_service.DynamicSelectClient")
|
||||
@patch("services.plugin.plugin_parameter_service.TriggerSubscriptionBuilderService")
|
||||
def test_uses_subscription_builder_when_credential_id(self, mock_builder_svc, mock_client_cls):
|
||||
sub = MagicMock()
|
||||
sub.credentials = {"token": "abc"}
|
||||
sub.credential_type = "api_key"
|
||||
mock_builder_svc.get_subscription_builder.return_value = sub
|
||||
mock_client_cls.return_value.fetch_dynamic_select_options.return_value.options = ["opt"]
|
||||
|
||||
result = PluginParameterService.get_dynamic_select_options(
|
||||
tenant_id="t1",
|
||||
user_id="u1",
|
||||
plugin_id="p1",
|
||||
provider="github",
|
||||
action="on_push",
|
||||
parameter="branch",
|
||||
credential_id="builder-1",
|
||||
provider_type="trigger",
|
||||
)
|
||||
|
||||
assert result == ["opt"]
|
||||
|
||||
@patch("services.plugin.plugin_parameter_service.DynamicSelectClient")
|
||||
@patch("services.plugin.plugin_parameter_service.TriggerProviderService")
|
||||
@patch("services.plugin.plugin_parameter_service.TriggerSubscriptionBuilderService")
|
||||
def test_falls_back_to_trigger_service(self, mock_builder_svc, mock_provider_svc, mock_client_cls):
|
||||
mock_builder_svc.get_subscription_builder.return_value = None
|
||||
trigger_sub = MagicMock()
|
||||
api_entity = MagicMock()
|
||||
api_entity.credentials = {"token": "abc"}
|
||||
api_entity.credential_type = "api_key"
|
||||
trigger_sub.to_api_entity.return_value = api_entity
|
||||
mock_provider_svc.get_subscription_by_id.return_value = trigger_sub
|
||||
mock_client_cls.return_value.fetch_dynamic_select_options.return_value.options = ["opt"]
|
||||
|
||||
result = PluginParameterService.get_dynamic_select_options(
|
||||
tenant_id="t1",
|
||||
user_id="u1",
|
||||
plugin_id="p1",
|
||||
provider="github",
|
||||
action="on_push",
|
||||
parameter="branch",
|
||||
credential_id="sub-1",
|
||||
provider_type="trigger",
|
||||
)
|
||||
|
||||
assert result == ["opt"]
|
||||
|
||||
@patch("services.plugin.plugin_parameter_service.TriggerProviderService")
|
||||
@patch("services.plugin.plugin_parameter_service.TriggerSubscriptionBuilderService")
|
||||
def test_raises_when_no_subscription_found(self, mock_builder_svc, mock_provider_svc):
|
||||
mock_builder_svc.get_subscription_builder.return_value = None
|
||||
mock_provider_svc.get_subscription_by_id.return_value = None
|
||||
|
||||
with pytest.raises(ValueError, match="not found"):
|
||||
PluginParameterService.get_dynamic_select_options(
|
||||
tenant_id="t1",
|
||||
user_id="u1",
|
||||
plugin_id="p1",
|
||||
provider="github",
|
||||
action="on_push",
|
||||
parameter="branch",
|
||||
credential_id="nonexistent",
|
||||
provider_type="trigger",
|
||||
)
|
||||
|
||||
|
||||
class TestGetDynamicSelectOptionsWithCredentials:
|
||||
@patch("services.plugin.plugin_parameter_service.DynamicSelectClient")
|
||||
@patch("services.plugin.plugin_parameter_service.TriggerProviderService")
|
||||
def test_replaces_hidden_values(self, mock_provider_svc, mock_client_cls):
|
||||
from constants import HIDDEN_VALUE
|
||||
|
||||
original = MagicMock()
|
||||
original.credentials = {"token": "real-secret", "name": "real-name"}
|
||||
original.credential_type = "api_key"
|
||||
mock_provider_svc.get_subscription_by_id.return_value = original
|
||||
mock_client_cls.return_value.fetch_dynamic_select_options.return_value.options = ["opt"]
|
||||
|
||||
result = PluginParameterService.get_dynamic_select_options_with_credentials(
|
||||
tenant_id="t1",
|
||||
user_id="u1",
|
||||
plugin_id="p1",
|
||||
provider="github",
|
||||
action="on_push",
|
||||
parameter="branch",
|
||||
credential_id="cred-1",
|
||||
credentials={"token": HIDDEN_VALUE, "name": "new-name"},
|
||||
)
|
||||
|
||||
assert result == ["opt"]
|
||||
call_args = mock_client_cls.return_value.fetch_dynamic_select_options.call_args[0]
|
||||
resolved = call_args[5]
|
||||
assert resolved["token"] == "real-secret" # replaced
|
||||
assert resolved["name"] == "new-name" # kept as-is
|
||||
|
||||
@patch("services.plugin.plugin_parameter_service.TriggerProviderService")
|
||||
def test_raises_when_subscription_not_found(self, mock_provider_svc):
|
||||
mock_provider_svc.get_subscription_by_id.return_value = None
|
||||
|
||||
with pytest.raises(ValueError, match="not found"):
|
||||
PluginParameterService.get_dynamic_select_options_with_credentials(
|
||||
tenant_id="t1",
|
||||
user_id="u1",
|
||||
plugin_id="p1",
|
||||
provider="github",
|
||||
action="on_push",
|
||||
parameter="branch",
|
||||
credential_id="nonexistent",
|
||||
credentials={"token": "val"},
|
||||
)
|
||||
|
|
@ -0,0 +1,357 @@
|
|||
"""Tests for services.plugin.plugin_service.PluginService.
|
||||
|
||||
Covers: version caching with Redis, install permission/scope gates,
|
||||
icon URL construction, asset retrieval with MIME guessing, plugin
|
||||
verification, marketplace upgrade flows, and uninstall with credential cleanup.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from core.plugin.entities.plugin import PluginInstallationSource
|
||||
from core.plugin.entities.plugin_daemon import PluginVerification
|
||||
from services.errors.plugin import PluginInstallationForbiddenError
|
||||
from services.feature_service import PluginInstallationScope
|
||||
from services.plugin.plugin_service import PluginService
|
||||
from tests.unit_tests.services.plugin.conftest import make_features
|
||||
|
||||
|
||||
class TestFetchLatestPluginVersion:
|
||||
@patch("services.plugin.plugin_service.marketplace")
|
||||
@patch("services.plugin.plugin_service.redis_client")
|
||||
def test_returns_cached_version(self, mock_redis, mock_marketplace):
|
||||
cached_json = PluginService.LatestPluginCache(
|
||||
plugin_id="p1",
|
||||
version="1.0.0",
|
||||
unique_identifier="uid-1",
|
||||
status="active",
|
||||
deprecated_reason="",
|
||||
alternative_plugin_id="",
|
||||
).model_dump_json()
|
||||
mock_redis.get.return_value = cached_json
|
||||
|
||||
result = PluginService.fetch_latest_plugin_version(["p1"])
|
||||
|
||||
assert result["p1"].version == "1.0.0"
|
||||
mock_marketplace.batch_fetch_plugin_manifests.assert_not_called()
|
||||
|
||||
@patch("services.plugin.plugin_service.marketplace")
|
||||
@patch("services.plugin.plugin_service.redis_client")
|
||||
def test_fetches_from_marketplace_on_cache_miss(self, mock_redis, mock_marketplace):
|
||||
mock_redis.get.return_value = None
|
||||
manifest = MagicMock()
|
||||
manifest.plugin_id = "p1"
|
||||
manifest.latest_version = "2.0.0"
|
||||
manifest.latest_package_identifier = "uid-2"
|
||||
manifest.status = "active"
|
||||
manifest.deprecated_reason = ""
|
||||
manifest.alternative_plugin_id = ""
|
||||
mock_marketplace.batch_fetch_plugin_manifests.return_value = [manifest]
|
||||
|
||||
result = PluginService.fetch_latest_plugin_version(["p1"])
|
||||
|
||||
assert result["p1"].version == "2.0.0"
|
||||
mock_redis.setex.assert_called_once()
|
||||
|
||||
@patch("services.plugin.plugin_service.marketplace")
|
||||
@patch("services.plugin.plugin_service.redis_client")
|
||||
def test_returns_none_for_unknown_plugin(self, mock_redis, mock_marketplace):
|
||||
mock_redis.get.return_value = None
|
||||
mock_marketplace.batch_fetch_plugin_manifests.return_value = []
|
||||
|
||||
result = PluginService.fetch_latest_plugin_version(["unknown"])
|
||||
|
||||
assert result["unknown"] is None
|
||||
|
||||
@patch("services.plugin.plugin_service.marketplace")
|
||||
@patch("services.plugin.plugin_service.redis_client")
|
||||
def test_handles_marketplace_exception_gracefully(self, mock_redis, mock_marketplace):
|
||||
mock_redis.get.return_value = None
|
||||
mock_marketplace.batch_fetch_plugin_manifests.side_effect = RuntimeError("network error")
|
||||
|
||||
result = PluginService.fetch_latest_plugin_version(["p1"])
|
||||
|
||||
assert result == {}
|
||||
|
||||
|
||||
class TestCheckMarketplaceOnlyPermission:
|
||||
@patch("services.plugin.plugin_service.FeatureService")
|
||||
def test_raises_when_restricted(self, mock_fs):
|
||||
mock_fs.get_system_features.return_value = make_features(restrict_to_marketplace=True)
|
||||
|
||||
with pytest.raises(PluginInstallationForbiddenError):
|
||||
PluginService._check_marketplace_only_permission()
|
||||
|
||||
@patch("services.plugin.plugin_service.FeatureService")
|
||||
def test_passes_when_not_restricted(self, mock_fs):
|
||||
mock_fs.get_system_features.return_value = make_features(restrict_to_marketplace=False)
|
||||
|
||||
PluginService._check_marketplace_only_permission() # should not raise
|
||||
|
||||
|
||||
class TestCheckPluginInstallationScope:
|
||||
@patch("services.plugin.plugin_service.FeatureService")
|
||||
def test_official_only_allows_langgenius(self, mock_fs):
|
||||
mock_fs.get_system_features.return_value = make_features(scope=PluginInstallationScope.OFFICIAL_ONLY)
|
||||
verification = MagicMock()
|
||||
verification.authorized_category = PluginVerification.AuthorizedCategory.Langgenius
|
||||
|
||||
PluginService._check_plugin_installation_scope(verification) # should not raise
|
||||
|
||||
@patch("services.plugin.plugin_service.FeatureService")
|
||||
def test_official_only_rejects_third_party(self, mock_fs):
|
||||
mock_fs.get_system_features.return_value = make_features(scope=PluginInstallationScope.OFFICIAL_ONLY)
|
||||
|
||||
with pytest.raises(PluginInstallationForbiddenError):
|
||||
PluginService._check_plugin_installation_scope(None)
|
||||
|
||||
@patch("services.plugin.plugin_service.FeatureService")
|
||||
def test_official_and_partners_allows_partner(self, mock_fs):
|
||||
mock_fs.get_system_features.return_value = make_features(
|
||||
scope=PluginInstallationScope.OFFICIAL_AND_SPECIFIC_PARTNERS
|
||||
)
|
||||
verification = MagicMock()
|
||||
verification.authorized_category = PluginVerification.AuthorizedCategory.Partner
|
||||
|
||||
PluginService._check_plugin_installation_scope(verification) # should not raise
|
||||
|
||||
@patch("services.plugin.plugin_service.FeatureService")
|
||||
def test_official_and_partners_rejects_none(self, mock_fs):
|
||||
mock_fs.get_system_features.return_value = make_features(
|
||||
scope=PluginInstallationScope.OFFICIAL_AND_SPECIFIC_PARTNERS
|
||||
)
|
||||
|
||||
with pytest.raises(PluginInstallationForbiddenError):
|
||||
PluginService._check_plugin_installation_scope(None)
|
||||
|
||||
@patch("services.plugin.plugin_service.FeatureService")
|
||||
def test_none_scope_always_raises(self, mock_fs):
|
||||
mock_fs.get_system_features.return_value = make_features(scope=PluginInstallationScope.NONE)
|
||||
verification = MagicMock()
|
||||
verification.authorized_category = PluginVerification.AuthorizedCategory.Langgenius
|
||||
|
||||
with pytest.raises(PluginInstallationForbiddenError):
|
||||
PluginService._check_plugin_installation_scope(verification)
|
||||
|
||||
@patch("services.plugin.plugin_service.FeatureService")
|
||||
def test_all_scope_passes_any(self, mock_fs):
|
||||
mock_fs.get_system_features.return_value = make_features(scope=PluginInstallationScope.ALL)
|
||||
|
||||
PluginService._check_plugin_installation_scope(None) # should not raise
|
||||
|
||||
|
||||
class TestGetPluginIconUrl:
|
||||
@patch("services.plugin.plugin_service.dify_config")
|
||||
def test_constructs_url_with_params(self, mock_config):
|
||||
mock_config.CONSOLE_API_URL = "https://console.example.com"
|
||||
|
||||
url = PluginService.get_plugin_icon_url("tenant-1", "icon.svg")
|
||||
|
||||
assert "tenant_id=tenant-1" in url
|
||||
assert "filename=icon.svg" in url
|
||||
assert "/plugin/icon" in url
|
||||
|
||||
|
||||
class TestGetAsset:
|
||||
@patch("services.plugin.plugin_service.PluginAssetManager")
|
||||
def test_returns_bytes_and_guessed_mime(self, mock_asset_cls):
|
||||
mock_asset_cls.return_value.fetch_asset.return_value = b"<svg/>"
|
||||
|
||||
data, mime = PluginService.get_asset("t1", "icon.svg")
|
||||
|
||||
assert data == b"<svg/>"
|
||||
assert "svg" in mime
|
||||
|
||||
@patch("services.plugin.plugin_service.PluginAssetManager")
|
||||
def test_fallback_to_octet_stream_for_unknown(self, mock_asset_cls):
|
||||
mock_asset_cls.return_value.fetch_asset.return_value = b"\x00"
|
||||
|
||||
_, mime = PluginService.get_asset("t1", "unknown_file")
|
||||
|
||||
assert mime == "application/octet-stream"
|
||||
|
||||
|
||||
class TestIsPluginVerified:
|
||||
@patch("services.plugin.plugin_service.PluginInstaller")
|
||||
def test_returns_true_when_verified(self, mock_installer_cls):
|
||||
mock_installer_cls.return_value.fetch_plugin_manifest.return_value.verified = True
|
||||
|
||||
assert PluginService.is_plugin_verified("t1", "uid-1") is True
|
||||
|
||||
@patch("services.plugin.plugin_service.PluginInstaller")
|
||||
def test_returns_false_on_exception(self, mock_installer_cls):
|
||||
mock_installer_cls.return_value.fetch_plugin_manifest.side_effect = RuntimeError("not found")
|
||||
|
||||
assert PluginService.is_plugin_verified("t1", "uid-1") is False
|
||||
|
||||
|
||||
class TestUpgradePluginWithMarketplace:
|
||||
@patch("services.plugin.plugin_service.dify_config")
|
||||
def test_raises_when_marketplace_disabled(self, mock_config):
|
||||
mock_config.MARKETPLACE_ENABLED = False
|
||||
|
||||
with pytest.raises(ValueError, match="marketplace is not enabled"):
|
||||
PluginService.upgrade_plugin_with_marketplace("t1", "old-uid", "new-uid")
|
||||
|
||||
@patch("services.plugin.plugin_service.dify_config")
|
||||
def test_raises_when_same_identifier(self, mock_config):
|
||||
mock_config.MARKETPLACE_ENABLED = True
|
||||
|
||||
with pytest.raises(ValueError, match="same plugin"):
|
||||
PluginService.upgrade_plugin_with_marketplace("t1", "same-uid", "same-uid")
|
||||
|
||||
@patch("services.plugin.plugin_service.marketplace")
|
||||
@patch("services.plugin.plugin_service.FeatureService")
|
||||
@patch("services.plugin.plugin_service.PluginInstaller")
|
||||
@patch("services.plugin.plugin_service.dify_config")
|
||||
def test_skips_download_when_already_installed(self, mock_config, mock_installer_cls, mock_fs, mock_marketplace):
|
||||
mock_config.MARKETPLACE_ENABLED = True
|
||||
mock_fs.get_system_features.return_value = make_features()
|
||||
installer = mock_installer_cls.return_value
|
||||
installer.fetch_plugin_manifest.return_value = MagicMock() # no exception = already installed
|
||||
installer.upgrade_plugin.return_value = MagicMock()
|
||||
|
||||
PluginService.upgrade_plugin_with_marketplace("t1", "old-uid", "new-uid")
|
||||
|
||||
mock_marketplace.record_install_plugin_event.assert_called_once_with("new-uid")
|
||||
installer.upgrade_plugin.assert_called_once()
|
||||
|
||||
@patch("services.plugin.plugin_service.download_plugin_pkg")
|
||||
@patch("services.plugin.plugin_service.FeatureService")
|
||||
@patch("services.plugin.plugin_service.PluginInstaller")
|
||||
@patch("services.plugin.plugin_service.dify_config")
|
||||
def test_downloads_when_not_installed(self, mock_config, mock_installer_cls, mock_fs, mock_download):
|
||||
mock_config.MARKETPLACE_ENABLED = True
|
||||
mock_fs.get_system_features.return_value = make_features()
|
||||
installer = mock_installer_cls.return_value
|
||||
installer.fetch_plugin_manifest.side_effect = RuntimeError("not found")
|
||||
mock_download.return_value = b"pkg-bytes"
|
||||
upload_resp = MagicMock()
|
||||
upload_resp.verification = None
|
||||
installer.upload_pkg.return_value = upload_resp
|
||||
installer.upgrade_plugin.return_value = MagicMock()
|
||||
|
||||
PluginService.upgrade_plugin_with_marketplace("t1", "old-uid", "new-uid")
|
||||
|
||||
mock_download.assert_called_once_with("new-uid")
|
||||
installer.upload_pkg.assert_called_once()
|
||||
|
||||
|
||||
class TestUpgradePluginWithGithub:
|
||||
@patch("services.plugin.plugin_service.FeatureService")
|
||||
@patch("services.plugin.plugin_service.PluginInstaller")
|
||||
def test_checks_marketplace_permission_and_delegates(self, mock_installer_cls, mock_fs):
|
||||
mock_fs.get_system_features.return_value = make_features()
|
||||
installer = mock_installer_cls.return_value
|
||||
installer.upgrade_plugin.return_value = MagicMock()
|
||||
|
||||
PluginService.upgrade_plugin_with_github("t1", "old-uid", "new-uid", "org/repo", "v1", "pkg.difypkg")
|
||||
|
||||
installer.upgrade_plugin.assert_called_once()
|
||||
call_args = installer.upgrade_plugin.call_args
|
||||
assert call_args[0][3] == PluginInstallationSource.Github
|
||||
|
||||
|
||||
class TestUploadPkg:
|
||||
@patch("services.plugin.plugin_service.FeatureService")
|
||||
@patch("services.plugin.plugin_service.PluginInstaller")
|
||||
def test_runs_permission_and_scope_checks(self, mock_installer_cls, mock_fs):
|
||||
mock_fs.get_system_features.return_value = make_features()
|
||||
upload_resp = MagicMock()
|
||||
upload_resp.verification = None
|
||||
mock_installer_cls.return_value.upload_pkg.return_value = upload_resp
|
||||
|
||||
result = PluginService.upload_pkg("t1", b"pkg-bytes")
|
||||
|
||||
assert result is upload_resp
|
||||
|
||||
|
||||
class TestInstallFromMarketplacePkg:
|
||||
@patch("services.plugin.plugin_service.dify_config")
|
||||
def test_raises_when_marketplace_disabled(self, mock_config):
|
||||
mock_config.MARKETPLACE_ENABLED = False
|
||||
|
||||
with pytest.raises(ValueError, match="marketplace is not enabled"):
|
||||
PluginService.install_from_marketplace_pkg("t1", ["uid-1"])
|
||||
|
||||
@patch("services.plugin.plugin_service.download_plugin_pkg")
|
||||
@patch("services.plugin.plugin_service.FeatureService")
|
||||
@patch("services.plugin.plugin_service.PluginInstaller")
|
||||
@patch("services.plugin.plugin_service.dify_config")
|
||||
def test_downloads_when_not_cached(self, mock_config, mock_installer_cls, mock_fs, mock_download):
|
||||
mock_config.MARKETPLACE_ENABLED = True
|
||||
mock_fs.get_system_features.return_value = make_features()
|
||||
installer = mock_installer_cls.return_value
|
||||
installer.fetch_plugin_manifest.side_effect = RuntimeError("not found")
|
||||
mock_download.return_value = b"pkg"
|
||||
upload_resp = MagicMock()
|
||||
upload_resp.verification = None
|
||||
upload_resp.unique_identifier = "resolved-uid"
|
||||
installer.upload_pkg.return_value = upload_resp
|
||||
installer.install_from_identifiers.return_value = "task-id"
|
||||
|
||||
result = PluginService.install_from_marketplace_pkg("t1", ["uid-1"])
|
||||
|
||||
assert result == "task-id"
|
||||
installer.install_from_identifiers.assert_called_once()
|
||||
call_args = installer.install_from_identifiers.call_args[0]
|
||||
assert call_args[1] == ["resolved-uid"] # uses response uid, not input
|
||||
|
||||
@patch("services.plugin.plugin_service.FeatureService")
|
||||
@patch("services.plugin.plugin_service.PluginInstaller")
|
||||
@patch("services.plugin.plugin_service.dify_config")
|
||||
def test_uses_cached_when_already_downloaded(self, mock_config, mock_installer_cls, mock_fs):
|
||||
mock_config.MARKETPLACE_ENABLED = True
|
||||
mock_fs.get_system_features.return_value = make_features()
|
||||
installer = mock_installer_cls.return_value
|
||||
installer.fetch_plugin_manifest.return_value = MagicMock()
|
||||
decode_resp = MagicMock()
|
||||
decode_resp.verification = None
|
||||
installer.decode_plugin_from_identifier.return_value = decode_resp
|
||||
installer.install_from_identifiers.return_value = "task-id"
|
||||
|
||||
PluginService.install_from_marketplace_pkg("t1", ["uid-1"])
|
||||
|
||||
installer.install_from_identifiers.assert_called_once()
|
||||
call_args = installer.install_from_identifiers.call_args[0]
|
||||
assert call_args[1] == ["uid-1"] # uses original uid
|
||||
|
||||
|
||||
class TestUninstall:
|
||||
@patch("services.plugin.plugin_service.PluginInstaller")
|
||||
def test_direct_uninstall_when_plugin_not_found(self, mock_installer_cls):
|
||||
installer = mock_installer_cls.return_value
|
||||
installer.list_plugins.return_value = []
|
||||
installer.uninstall.return_value = True
|
||||
|
||||
result = PluginService.uninstall("t1", "install-1")
|
||||
|
||||
assert result is True
|
||||
installer.uninstall.assert_called_once_with("t1", "install-1")
|
||||
|
||||
@patch("services.plugin.plugin_service.db")
|
||||
@patch("services.plugin.plugin_service.PluginInstaller")
|
||||
def test_cleans_credentials_when_plugin_found(self, mock_installer_cls, mock_db):
|
||||
plugin = MagicMock()
|
||||
plugin.installation_id = "install-1"
|
||||
plugin.plugin_id = "org/myplugin"
|
||||
installer = mock_installer_cls.return_value
|
||||
installer.list_plugins.return_value = [plugin]
|
||||
installer.uninstall.return_value = True
|
||||
|
||||
# Mock Session context manager
|
||||
mock_session = MagicMock()
|
||||
mock_db.engine = MagicMock()
|
||||
mock_session.scalars.return_value.all.return_value = [] # no credentials found
|
||||
|
||||
with patch("services.plugin.plugin_service.Session") as mock_session_cls:
|
||||
mock_session_cls.return_value.__enter__ = MagicMock(return_value=mock_session)
|
||||
mock_session_cls.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
result = PluginService.uninstall("t1", "install-1")
|
||||
|
||||
assert result is True
|
||||
installer.uninstall.assert_called_once()
|
||||
Loading…
Reference in New Issue