mirror of
https://github.com/langgenius/dify.git
synced 2026-11-13 06:41:07 +08:00
Yansong Zhang
5570cb08f5
Add revision-scoped stable saved-model inventory and source plugin admission. Preserve independent and BYOK owners, frozen contracts, actual schema RPC boundaries and non-secret source protocol facts.
424 lines
17 KiB
Python
424 lines
17 KiB
Python
"""Unit tests for core.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 collections.abc import Callable, Iterator
|
|
from typing import cast
|
|
from unittest.mock import MagicMock, patch
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from flask import Flask
|
|
from sqlalchemy import Table, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from core.plugin.entities.plugin_daemon import PluginVerification
|
|
from core.plugin.plugin_service import PluginService
|
|
from enums import DeploymentEdition
|
|
from models import ProviderType
|
|
from models.engine import db
|
|
from models.provider import Provider, ProviderCredential, TenantPreferredModelProvider
|
|
from services.entities.feature_entities import (
|
|
PluginInstallationPermissionModel,
|
|
PluginInstallationScope,
|
|
)
|
|
from services.errors.plugin import PluginInstallationForbiddenError
|
|
from tests.unit_tests.services.plugin.conftest import mock_features as mock_features # noqa: PLC0414
|
|
from tests.unit_tests.services.plugin.conftest import mock_installer as mock_installer # noqa: PLC0414
|
|
|
|
|
|
def _make_permission(
|
|
restrict_to_marketplace: bool = False,
|
|
scope: PluginInstallationScope = PluginInstallationScope.ALL,
|
|
) -> PluginInstallationPermissionModel:
|
|
return PluginInstallationPermissionModel(
|
|
restrict_to_marketplace_only=restrict_to_marketplace,
|
|
plugin_installation_scope=scope,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def plugin_db() -> Iterator[Session]:
|
|
"""Provide the production database extension with isolated SQLite plugin tables."""
|
|
app = Flask(__name__)
|
|
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
|
|
db.init_app(app)
|
|
|
|
with app.app_context():
|
|
for model in (ProviderCredential, Provider, TenantPreferredModelProvider):
|
|
table = model.__table__
|
|
assert isinstance(table, Table)
|
|
table.create(db.engine)
|
|
with Session(db.engine, expire_on_commit=False) as session:
|
|
yield session
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _plugin_config(config_overrides: Callable[..., None]) -> None:
|
|
config_overrides(
|
|
CONSOLE_API_URL="https://console.example.com",
|
|
MARKETPLACE_ENABLED=True,
|
|
DEPLOYMENT_EDITION=DeploymentEdition.COMMUNITY,
|
|
)
|
|
|
|
|
|
class TestFetchLatestPluginVersion:
|
|
@patch("core.plugin.plugin_service.marketplace")
|
|
@patch("core.plugin.plugin_service.redis_client")
|
|
def test_returns_cached_version(self, mock_redis: MagicMock, mock_marketplace: MagicMock) -> None:
|
|
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"])
|
|
|
|
cached = result["p1"]
|
|
assert cached is not None
|
|
assert cached.version == "1.0.0"
|
|
mock_marketplace.batch_fetch_plugin_manifests.assert_not_called()
|
|
|
|
@patch("core.plugin.plugin_service.marketplace")
|
|
@patch("core.plugin.plugin_service.redis_client")
|
|
def test_fetches_from_marketplace_on_cache_miss(self, mock_redis: MagicMock, mock_marketplace: MagicMock) -> None:
|
|
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"])
|
|
|
|
fetched = result["p1"]
|
|
assert fetched is not None
|
|
assert fetched.version == "2.0.0"
|
|
mock_redis.setex.assert_called_once()
|
|
|
|
@patch("core.plugin.plugin_service.marketplace")
|
|
@patch("core.plugin.plugin_service.redis_client")
|
|
def test_returns_none_for_unknown_plugin(self, mock_redis: MagicMock, mock_marketplace: MagicMock) -> None:
|
|
mock_redis.get.return_value = None
|
|
no_manifests: list[object] = []
|
|
mock_marketplace.batch_fetch_plugin_manifests.return_value = no_manifests
|
|
|
|
result = PluginService.fetch_latest_plugin_version(["unknown"])
|
|
|
|
assert result["unknown"] is None
|
|
|
|
@patch("core.plugin.plugin_service.marketplace")
|
|
@patch("core.plugin.plugin_service.redis_client")
|
|
def test_handles_marketplace_exception_gracefully(self, mock_redis: MagicMock, mock_marketplace: MagicMock) -> None:
|
|
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("core.plugin.plugin_service.SystemFeatureService")
|
|
def test_raises_when_restricted(self, mock_fs: MagicMock) -> None:
|
|
mock_fs.get_plugin_installation_permission.return_value = _make_permission(restrict_to_marketplace=True)
|
|
|
|
with pytest.raises(PluginInstallationForbiddenError):
|
|
PluginService._check_marketplace_only_permission()
|
|
|
|
@patch("core.plugin.plugin_service.SystemFeatureService")
|
|
def test_passes_when_not_restricted(self, mock_fs: MagicMock) -> None:
|
|
mock_fs.get_plugin_installation_permission.return_value = _make_permission(restrict_to_marketplace=False)
|
|
|
|
PluginService._check_marketplace_only_permission() # should not raise
|
|
|
|
@patch("core.plugin.plugin_service.SystemFeatureService")
|
|
def test_raises_when_scope_denies_all(self, mock_fs: MagicMock) -> None:
|
|
mock_fs.get_plugin_installation_permission.return_value = _make_permission(scope=PluginInstallationScope.NONE)
|
|
|
|
with pytest.raises(PluginInstallationForbiddenError, match="not allowed"):
|
|
PluginService._check_marketplace_only_permission()
|
|
|
|
|
|
class TestCheckPluginInstallationScope:
|
|
@patch("core.plugin.plugin_service.SystemFeatureService")
|
|
def test_official_only_allows_langgenius(self, mock_fs: MagicMock) -> None:
|
|
mock_fs.get_plugin_installation_permission.return_value = _make_permission(
|
|
scope=PluginInstallationScope.OFFICIAL_ONLY
|
|
)
|
|
verification = MagicMock()
|
|
verification.authorized_category = PluginVerification.AuthorizedCategory.Langgenius
|
|
|
|
PluginService._check_plugin_installation_scope(verification) # should not raise
|
|
|
|
@patch("core.plugin.plugin_service.SystemFeatureService")
|
|
def test_official_only_rejects_third_party(self, mock_fs: MagicMock) -> None:
|
|
mock_fs.get_plugin_installation_permission.return_value = _make_permission(
|
|
scope=PluginInstallationScope.OFFICIAL_ONLY
|
|
)
|
|
|
|
with pytest.raises(PluginInstallationForbiddenError):
|
|
PluginService._check_plugin_installation_scope(None)
|
|
|
|
@patch("core.plugin.plugin_service.SystemFeatureService")
|
|
def test_official_and_partners_allows_partner(self, mock_fs: MagicMock) -> None:
|
|
mock_fs.get_plugin_installation_permission.return_value = _make_permission(
|
|
scope=PluginInstallationScope.OFFICIAL_AND_SPECIFIC_PARTNERS
|
|
)
|
|
verification = MagicMock()
|
|
verification.authorized_category = PluginVerification.AuthorizedCategory.Partner
|
|
|
|
PluginService._check_plugin_installation_scope(verification) # should not raise
|
|
|
|
@patch("core.plugin.plugin_service.SystemFeatureService")
|
|
def test_official_and_partners_rejects_none(self, mock_fs: MagicMock) -> None:
|
|
mock_fs.get_plugin_installation_permission.return_value = _make_permission(
|
|
scope=PluginInstallationScope.OFFICIAL_AND_SPECIFIC_PARTNERS
|
|
)
|
|
|
|
with pytest.raises(PluginInstallationForbiddenError):
|
|
PluginService._check_plugin_installation_scope(None)
|
|
|
|
@patch("core.plugin.plugin_service.SystemFeatureService")
|
|
def test_none_scope_always_raises(self, mock_fs: MagicMock) -> None:
|
|
mock_fs.get_plugin_installation_permission.return_value = _make_permission(scope=PluginInstallationScope.NONE)
|
|
verification = MagicMock()
|
|
verification.authorized_category = PluginVerification.AuthorizedCategory.Langgenius
|
|
|
|
with pytest.raises(PluginInstallationForbiddenError):
|
|
PluginService._check_plugin_installation_scope(verification)
|
|
|
|
@patch("core.plugin.plugin_service.SystemFeatureService")
|
|
def test_all_scope_passes_any(self, mock_fs: MagicMock) -> None:
|
|
mock_fs.get_plugin_installation_permission.return_value = _make_permission(scope=PluginInstallationScope.ALL)
|
|
|
|
PluginService._check_plugin_installation_scope(None) # should not raise
|
|
|
|
@patch("core.plugin.plugin_service.SystemFeatureService")
|
|
def test_unknown_scope_always_raises(self, mock_fs: MagicMock) -> None:
|
|
permission = _make_permission()
|
|
permission.plugin_installation_scope = cast(PluginInstallationScope, "unknown-scope")
|
|
mock_fs.get_plugin_installation_permission.return_value = permission
|
|
|
|
with pytest.raises(PluginInstallationForbiddenError, match="policy is invalid"):
|
|
PluginService._check_plugin_installation_scope(None)
|
|
|
|
|
|
class TestGetPluginIconUrl:
|
|
def test_constructs_url_with_params(self) -> None:
|
|
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("core.plugin.plugin_service.PluginAssetManager")
|
|
def test_returns_bytes_and_guessed_mime(self, mock_asset_cls: MagicMock) -> None:
|
|
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("core.plugin.plugin_service.PluginAssetManager")
|
|
def test_fallback_to_octet_stream_for_unknown(self, mock_asset_cls: MagicMock) -> None:
|
|
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("core.plugin.plugin_service.PluginInstaller")
|
|
def test_returns_true_when_verified(self, mock_installer_cls: MagicMock) -> None:
|
|
mock_installer_cls.return_value.fetch_plugin_manifest.return_value.verified = True
|
|
|
|
assert PluginService.is_plugin_verified("t1", "uid-1") is True
|
|
|
|
@patch("core.plugin.plugin_service.PluginInstaller")
|
|
def test_returns_false_on_exception(self, mock_installer_cls: MagicMock) -> None:
|
|
mock_installer_cls.return_value.fetch_plugin_manifest.side_effect = RuntimeError("not found")
|
|
|
|
assert PluginService.is_plugin_verified("t1", "uid-1") is False
|
|
|
|
|
|
class TestUploadPkg:
|
|
@patch("core.plugin.plugin_service.SystemFeatureService")
|
|
@patch("core.plugin.plugin_service.PluginInstaller")
|
|
def test_runs_permission_and_scope_checks(self, mock_installer_cls: MagicMock, mock_fs: MagicMock) -> None:
|
|
mock_fs.get_plugin_installation_permission.return_value = _make_permission()
|
|
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 TestUninstall:
|
|
@patch("core.plugin.plugin_service.PluginInstaller")
|
|
def test_direct_uninstall_when_plugin_not_found(self, mock_installer_cls: MagicMock) -> None:
|
|
installer = mock_installer_cls.return_value
|
|
no_plugins: list[object] = []
|
|
installer.list_plugins.return_value = no_plugins
|
|
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("core.plugin.plugin_service.PluginInstaller")
|
|
def test_cleans_credentials_when_plugin_found(self, mock_installer_cls: MagicMock, plugin_db: Session) -> None:
|
|
tenant_id = str(uuid4())
|
|
plugin_id = "org/myplugin"
|
|
provider_name = f"{plugin_id}/model-provider"
|
|
|
|
credential = ProviderCredential(
|
|
tenant_id=tenant_id,
|
|
provider_name=provider_name,
|
|
credential_name="default",
|
|
encrypted_config="{}",
|
|
)
|
|
plugin_db.add(credential)
|
|
plugin_db.flush()
|
|
credential_id = credential.id
|
|
|
|
provider = Provider(
|
|
tenant_id=tenant_id,
|
|
provider_name=provider_name,
|
|
credential_id=credential_id,
|
|
)
|
|
plugin_db.add(provider)
|
|
plugin_db.flush()
|
|
provider_id = provider.id
|
|
|
|
pref = TenantPreferredModelProvider(
|
|
tenant_id=tenant_id,
|
|
provider_name=provider_name,
|
|
preferred_provider_type=ProviderType.CUSTOM,
|
|
)
|
|
plugin_db.add(pref)
|
|
plugin_db.commit()
|
|
|
|
plugin = MagicMock()
|
|
plugin.installation_id = "install-1"
|
|
plugin.plugin_id = plugin_id
|
|
installer = mock_installer_cls.return_value
|
|
installer.list_plugins.return_value = [plugin]
|
|
installer.uninstall.return_value = True
|
|
|
|
result = PluginService.uninstall(tenant_id, "install-1")
|
|
|
|
assert result is True
|
|
installer.uninstall.assert_called_once()
|
|
|
|
plugin_db.expire_all()
|
|
|
|
remaining_creds = plugin_db.scalars(
|
|
select(ProviderCredential).where(ProviderCredential.id == credential_id)
|
|
).all()
|
|
assert len(remaining_creds) == 0
|
|
|
|
updated_provider = plugin_db.get(Provider, provider_id)
|
|
assert updated_provider is not None
|
|
assert updated_provider.credential_id is None
|
|
|
|
remaining_prefs = plugin_db.scalars(
|
|
select(TenantPreferredModelProvider).where(
|
|
TenantPreferredModelProvider.tenant_id == tenant_id,
|
|
TenantPreferredModelProvider.provider_name == provider_name,
|
|
)
|
|
).all()
|
|
assert len(remaining_prefs) == 0
|
|
|
|
@patch("core.plugin.plugin_service.PluginInstaller")
|
|
def test_preserves_credentials_when_replacing_plugin(
|
|
self, mock_installer_cls: MagicMock, plugin_db: Session
|
|
) -> None:
|
|
tenant_id = str(uuid4())
|
|
plugin_id = "org/myplugin"
|
|
provider_name = f"{plugin_id}/model-provider"
|
|
|
|
credential = ProviderCredential(
|
|
tenant_id=tenant_id,
|
|
provider_name=provider_name,
|
|
credential_name="default",
|
|
encrypted_config="{}",
|
|
)
|
|
plugin_db.add(credential)
|
|
plugin_db.flush()
|
|
|
|
provider = Provider(
|
|
tenant_id=tenant_id,
|
|
provider_name=provider_name,
|
|
credential_id=credential.id,
|
|
)
|
|
plugin_db.add(provider)
|
|
|
|
preferred_provider = TenantPreferredModelProvider(
|
|
tenant_id=tenant_id,
|
|
provider_name=provider_name,
|
|
preferred_provider_type=ProviderType.CUSTOM,
|
|
)
|
|
plugin_db.add(preferred_provider)
|
|
plugin_db.commit()
|
|
|
|
plugin = MagicMock(installation_id="install-1", plugin_id=plugin_id)
|
|
installer = mock_installer_cls.return_value
|
|
installer.list_plugins.return_value = [plugin]
|
|
installer.uninstall.return_value = True
|
|
|
|
result = PluginService.uninstall(tenant_id, "install-1", preserve_credentials=True)
|
|
|
|
assert result is True
|
|
plugin_db.expire_all()
|
|
assert plugin_db.get(ProviderCredential, credential.id) is not None
|
|
stored_provider = plugin_db.get(Provider, provider.id)
|
|
assert stored_provider is not None
|
|
assert stored_provider.credential_id == credential.id
|
|
assert plugin_db.get(TenantPreferredModelProvider, preferred_provider.id) is not None
|
|
|
|
@patch("core.plugin.plugin_service.PluginInstaller")
|
|
def test_preserves_credentials_when_daemon_uninstall_fails(
|
|
self, mock_installer_cls: MagicMock, plugin_db: Session
|
|
) -> None:
|
|
tenant_id = str(uuid4())
|
|
plugin_id = "org/myplugin"
|
|
credential = ProviderCredential(
|
|
tenant_id=tenant_id,
|
|
provider_name=f"{plugin_id}/model-provider",
|
|
credential_name="default",
|
|
encrypted_config="{}",
|
|
)
|
|
plugin_db.add(credential)
|
|
plugin_db.commit()
|
|
|
|
plugin = MagicMock(installation_id="install-1", plugin_id=plugin_id)
|
|
installer = mock_installer_cls.return_value
|
|
installer.list_plugins.return_value = [plugin]
|
|
installer.uninstall.return_value = False
|
|
|
|
result = PluginService.uninstall(tenant_id, "install-1")
|
|
|
|
assert result is False
|
|
plugin_db.expire_all()
|
|
assert plugin_db.get(ProviderCredential, credential.id) is not None
|