diff --git a/api/tests/unit_tests/services/plugin/__init__.py b/api/tests/unit_tests/services/plugin/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/api/tests/unit_tests/services/plugin/conftest.py b/api/tests/unit_tests/services/plugin/conftest.py
new file mode 100644
index 0000000000..80c6077b0c
--- /dev/null
+++ b/api/tests/unit_tests/services/plugin/conftest.py
@@ -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
diff --git a/api/tests/unit_tests/services/plugin/test_dependencies_analysis.py b/api/tests/unit_tests/services/plugin/test_dependencies_analysis.py
new file mode 100644
index 0000000000..8f0886769c
--- /dev/null
+++ b/api/tests/unit_tests/services/plugin/test_dependencies_analysis.py
@@ -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
diff --git a/api/tests/unit_tests/services/plugin/test_endpoint_service.py b/api/tests/unit_tests/services/plugin/test_endpoint_service.py
new file mode 100644
index 0000000000..ddf80c8017
--- /dev/null
+++ b/api/tests/unit_tests/services/plugin/test_endpoint_service.py
@@ -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()
diff --git a/api/tests/unit_tests/services/plugin/test_oauth_service.py b/api/tests/unit_tests/services/plugin/test_oauth_service.py
new file mode 100644
index 0000000000..27df4556bc
--- /dev/null
+++ b/api/tests/unit_tests/services/plugin/test_oauth_service.py
@@ -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)
diff --git a/api/tests/unit_tests/services/plugin/test_plugin_parameter_service.py b/api/tests/unit_tests/services/plugin/test_plugin_parameter_service.py
new file mode 100644
index 0000000000..bfa9fe976b
--- /dev/null
+++ b/api/tests/unit_tests/services/plugin/test_plugin_parameter_service.py
@@ -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"},
+            )
diff --git a/api/tests/unit_tests/services/plugin/test_plugin_service.py b/api/tests/unit_tests/services/plugin/test_plugin_service.py
new file mode 100644
index 0000000000..09b9ab498b
--- /dev/null
+++ b/api/tests/unit_tests/services/plugin/test_plugin_service.py
@@ -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()