opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-07-21 02:28:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(api): clarify DSL import and plugin migration boundaries (#38483)

Browse Source 
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
WH-2099 2026-07-07 10:06:49 +08:00 committed by GitHub
parent b9c7199d34
commit 0a3426ea38
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 329 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
api/commands/plugin.py
View File

@ -188,13 +188,13 @@ def transform_datasource_credentials(environment: str):
firecrawl_plugin_id = "langgenius/firecrawl_datasource"
jina_plugin_id = "langgenius/jina_datasource"
if environment == "online":
notion_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(notion_plugin_id)
firecrawl_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(firecrawl_plugin_id)
jina_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(jina_plugin_id)
notion_package_identifier = plugin_migration._fetch_latest_package_identifier(notion_plugin_id)
firecrawl_package_identifier = plugin_migration._fetch_latest_package_identifier(firecrawl_plugin_id)
jina_package_identifier = plugin_migration._fetch_latest_package_identifier(jina_plugin_id)
else:
notion_plugin_unique_identifier = None
firecrawl_plugin_unique_identifier = None
jina_plugin_unique_identifier = None
notion_package_identifier = None
firecrawl_package_identifier = None
jina_package_identifier = None
oauth_credential_type = CredentialType.OAUTH2
api_key_credential_type = CredentialType.API_KEY
@ -219,9 +219,9 @@ def transform_datasource_credentials(environment: str):
installed_plugins = installer_manager.list_plugins(tenant_id)
installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins]
if notion_plugin_id not in installed_plugins_ids:
if notion_plugin_unique_identifier:
if notion_package_identifier:
# install notion plugin
PluginService.install_from_marketplace_pkg(tenant_id, [notion_plugin_unique_identifier])
PluginService.install_from_marketplace_pkg(tenant_id, [notion_package_identifier])
auth_count = 0
for notion_tenant_credential in notion_tenant_credentials:
auth_count += 1
@ -279,9 +279,9 @@ def transform_datasource_credentials(environment: str):
installed_plugins = installer_manager.list_plugins(tenant_id)
installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins]
if firecrawl_plugin_id not in installed_plugins_ids:
if firecrawl_plugin_unique_identifier:
if firecrawl_package_identifier:
# install firecrawl plugin
PluginService.install_from_marketplace_pkg(tenant_id, [firecrawl_plugin_unique_identifier])
PluginService.install_from_marketplace_pkg(tenant_id, [firecrawl_package_identifier])
auth_count = 0
for firecrawl_tenant_credential in firecrawl_tenant_credentials:
@ -343,10 +343,10 @@ def transform_datasource_credentials(environment: str):
installed_plugins = installer_manager.list_plugins(tenant_id)
installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins]
if jina_plugin_id not in installed_plugins_ids:
if jina_plugin_unique_identifier:
if jina_package_identifier:
# install jina plugin
logger.debug("Installing Jina plugin %s", jina_plugin_unique_identifier)
PluginService.install_from_marketplace_pkg(tenant_id, [jina_plugin_unique_identifier])
logger.debug("Installing Jina plugin %s", jina_package_identifier)
PluginService.install_from_marketplace_pkg(tenant_id, [jina_package_identifier])
auth_count = 0
for jina_tenant_credential in jina_tenant_credentials:

13
api/services/app_dsl_service.py
View File

@ -39,6 +39,7 @@ from libs.datetime_utils import naive_utc_now
from models import Account, App, AppMode
from models.model import AppModelConfig, AppModelConfigDict, IconType
from models.workflow import Workflow
from services.dsl_content import DSL_MAX_SIZE, dsl_content_size
from services.dsl_version import check_version_compatibility
from services.entities.dsl_entities import CheckDependenciesResult, ImportMode, ImportStatus
from services.errors.app import WorkflowNotFoundError
@ -51,7 +52,6 @@ logger = logging.getLogger(__name__)
IMPORT_INFO_REDIS_KEY_PREFIX = "app_import_info:"
CHECK_DEPENDENCIES_REDIS_KEY_PREFIX = "app_check_dependencies:"
IMPORT_INFO_REDIS_EXPIRY = 10 * 60 # 10 minutes
DSL_MAX_SIZE = 10 * 1024 * 1024 # 10MB
CURRENT_DSL_VERSION = CURRENT_APP_DSL_VERSION
@ -131,15 +131,16 @@ class AppDslService:
yaml_url = yaml_url.replace("/blob/", "/")
response = remote_fetcher.make_request("GET", yaml_url.strip(), follow_redirects=True, timeout=(10, 10))
response.raise_for_status()
content = response.content.decode()
raw_content = response.content
if len(content) > DSL_MAX_SIZE:
if dsl_content_size(raw_content) > DSL_MAX_SIZE:
return Import(
id=import_id,
status=ImportStatus.FAILED,
error="File size exceeds the limit of 10MB",
)
content = raw_content.decode("utf-8")
if not content:
return Import(
id=import_id,
@ -160,6 +161,12 @@ class AppDslService:
error="yaml_content is required when import_mode is yaml-content",
)
content = yaml_content
if dsl_content_size(content) > DSL_MAX_SIZE:
return Import(
id=import_id,
status=ImportStatus.FAILED,
error="File size exceeds the limit of 10MB",
)
# Process YAML content
try:

9
api/services/dsl_content.py Normal file
View File

@ -0,0 +1,9 @@
"""Shared DSL content size and decoding rules."""
DSL_MAX_SIZE = 10 * 1024 * 1024 # 10MB
def dsl_content_size(content: str | bytes) -> int:
if isinstance(content, bytes):
return len(content)
return len(content.encode("utf-8"))

116
api/services/plugin/plugin_migration.py
View File

@ -307,9 +307,9 @@ class PluginMigration:
return result
@classmethod
def _fetch_plugin_unique_identifier(cls, plugin_id: str) -> str | None:
def _fetch_latest_package_identifier(cls, plugin_id: str) -> str | None:
"""
Fetch plugin unique identifier using plugin id.
Fetch the latest marketplace package identifier using a plugin id.
"""
if not dify_config.MARKETPLACE_ENABLED:
return None
@ -328,7 +328,7 @@ class PluginMigration:
@classmethod
def extract_unique_plugins(cls, extracted_plugins: str) -> ExtractedPluginsDict:
plugins: dict[str, str] = {}
package_identifier_by_plugin_id: dict[str, str] = {}
plugin_ids = []
plugin_not_exist = []
logger.info("Extracting unique plugins from %s", extracted_plugins)
@ -341,19 +341,19 @@ class PluginMigration:
def fetch_plugin(plugin_id):
try:
unique_identifier = cls._fetch_plugin_unique_identifier(plugin_id)
if unique_identifier:
plugins[plugin_id] = unique_identifier
latest_package_identifier = cls._fetch_latest_package_identifier(plugin_id)
if latest_package_identifier:
package_identifier_by_plugin_id[plugin_id] = latest_package_identifier
else:
plugin_not_exist.append(plugin_id)
except Exception:
logger.exception("Failed to fetch plugin unique identifier for %s", plugin_id)
logger.exception("Failed to fetch latest package identifier for %s", plugin_id)
plugin_not_exist.append(plugin_id)
with ThreadPoolExecutor(max_workers=10) as executor:
list(tqdm.tqdm(executor.map(fetch_plugin, plugin_ids), total=len(plugin_ids)))
return {"plugins": plugins, "plugin_not_exist": plugin_not_exist}
return {"plugins": package_identifier_by_plugin_id, "plugin_not_exist": plugin_not_exist}
@classmethod
def install_plugins(cls, extracted_plugins: str, output_file: str, workers: int = 100):
@ -362,17 +362,22 @@ class PluginMigration:
"""
manager = PluginInstaller()
plugins = cls.extract_unique_plugins(extracted_plugins)
extracted = cls.extract_unique_plugins(extracted_plugins)
package_identifier_by_plugin_id = extracted["plugins"]
not_installed = []
plugin_install_failed = []
# use a fake tenant id to install all the plugins
fake_tenant_id = uuid4().hex
logger.info("Installing %s plugin instances for fake tenant %s", len(plugins["plugins"]), fake_tenant_id)
logger.info(
"Installing %s plugin instances for fake tenant %s",
len(package_identifier_by_plugin_id),
fake_tenant_id,
)
thread_pool = ThreadPoolExecutor(max_workers=workers)
response = cls.handle_plugin_instance_install(fake_tenant_id, plugins["plugins"])
response = cls.handle_plugin_instance_install(fake_tenant_id, package_identifier_by_plugin_id)
if response.get("failed"):
plugin_install_failed.extend(response.get("failed", []))
@ -384,21 +389,21 @@ class PluginMigration:
# at most 64 plugins one batch
for i in range(0, len(plugin_ids), 64):
batch_plugin_ids = plugin_ids[i : i + 64]
batch_plugin_identifiers = [
plugins["plugins"][plugin_id]
batch_package_identifiers = [
package_identifier_by_plugin_id[plugin_id]
for plugin_id in batch_plugin_ids
if plugin_id not in installed_plugins_ids and plugin_id in plugins["plugins"]
if plugin_id not in installed_plugins_ids and plugin_id in package_identifier_by_plugin_id
]
if batch_plugin_identifiers:
if batch_package_identifiers:
manager.install_from_identifiers(
tenant_id,
batch_plugin_identifiers,
batch_package_identifiers,
PluginInstallationSource.Marketplace,
metas=[
{
"plugin_unique_identifier": identifier,
"plugin_unique_identifier": package_identifier,
}
for identifier in batch_plugin_identifiers
for package_identifier in batch_package_identifiers
],
)
PluginService.invalidate_plugin_model_providers_cache(tenant_id)
@ -412,10 +417,8 @@ class PluginMigration:
tenant_id = data["tenant_id"]
plugin_ids = data["plugins"]
plugin_not_exist: list[str] = []
# get plugin unique identifier
for plugin_id in plugin_ids:
unique_identifier = plugins.get(plugin_id)
if unique_identifier:
if plugin_id not in package_identifier_by_plugin_id:
plugin_not_exist.append(plugin_id)
if plugin_not_exist:
@ -459,36 +462,44 @@ class PluginMigration:
"""
manager = PluginInstaller()
plugins = cls.extract_unique_plugins(extracted_plugins)
extracted = cls.extract_unique_plugins(extracted_plugins)
package_identifier_by_plugin_id = extracted["plugins"]
plugin_install_failed = []
# use a fake tenant id to install all the plugins
fake_tenant_id = uuid4().hex
logger.info("Installing %s plugin instances for fake tenant %s", len(plugins["plugins"]), fake_tenant_id)
logger.info(
"Installing %s plugin instances for fake tenant %s",
len(package_identifier_by_plugin_id),
fake_tenant_id,
)
thread_pool = ThreadPoolExecutor(max_workers=workers)
response = cls.handle_plugin_instance_install(fake_tenant_id, plugins["plugins"])
response = cls.handle_plugin_instance_install(fake_tenant_id, package_identifier_by_plugin_id)
if response.get("failed"):
plugin_install_failed.extend(response.get("failed", []))
def install(
tenant_id: str, plugin_ids: dict[str, str], total_success_tenant: int, total_failed_tenant: int
tenant_id: str,
package_identifier_by_plugin_id: dict[str, str],
total_success_tenant: int,
total_failed_tenant: int,
) -> None:
logger.info("Installing %s plugins for tenant %s", len(plugin_ids), tenant_id)
logger.info("Installing %s plugins for tenant %s", len(package_identifier_by_plugin_id), tenant_id)
try:
# fetch plugin already installed
installed_plugins = manager.list_plugins(tenant_id)
installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins]
# at most 64 plugins one batch
for i in range(0, len(plugin_ids), 64):
batch_plugin_ids = list(plugin_ids.keys())[i : i + 64]
batch_plugin_identifiers = [
plugin_ids[plugin_id]
for i in range(0, len(package_identifier_by_plugin_id), 64):
batch_plugin_ids = list(package_identifier_by_plugin_id.keys())[i : i + 64]
batch_package_identifiers = [
package_identifier_by_plugin_id[plugin_id]
for plugin_id in batch_plugin_ids
if plugin_id not in installed_plugins_ids and plugin_id in plugin_ids
if plugin_id not in installed_plugins_ids and plugin_id in package_identifier_by_plugin_id
]
PluginService.install_from_marketplace_pkg(tenant_id, batch_plugin_identifiers)
PluginService.install_from_marketplace_pkg(tenant_id, batch_package_identifiers)
total_success_tenant += 1
except Exception:
@ -510,7 +521,7 @@ class PluginMigration:
thread_pool.submit(
install,
tenant_id,
plugins.get("plugins", {}),
package_identifier_by_plugin_id,
total_success_tenant,
total_failed_tenant,
)
@ -542,12 +553,12 @@ class PluginMigration:
@classmethod
def handle_plugin_instance_install(
cls, tenant_id: str, plugin_identifiers_map: Mapping[str, str]
cls, tenant_id: str, package_identifier_by_plugin_id: Mapping[str, str]
) -> PluginInstallResultDict:
"""
Install plugins for a tenant.
"""
if plugin_identifiers_map and not dify_config.MARKETPLACE_ENABLED:
if package_identifier_by_plugin_id and not dify_config.MARKETPLACE_ENABLED:
raise ValueError(
"Marketplace disabled in offline mode; cannot bulk-install plugins. "
"Pre-upload plugin packages via Console first."
@ -558,17 +569,17 @@ class PluginMigration:
thread_pool = ThreadPoolExecutor(max_workers=10)
futures = []
for plugin_id, plugin_identifier in plugin_identifiers_map.items():
for plugin_id, package_identifier in package_identifier_by_plugin_id.items():
def download_and_upload(tenant_id, plugin_id, plugin_identifier):
plugin_package = marketplace.download_plugin_pkg(plugin_identifier)
def download_and_upload(tenant_id, plugin_id, package_identifier):
plugin_package = marketplace.download_plugin_pkg(package_identifier)
if not plugin_package:
raise Exception(f"Failed to download plugin {plugin_identifier}")
raise Exception(f"Failed to download plugin {package_identifier}")
# upload
manager.upload_pkg(tenant_id, plugin_package, verify_signature=True)
futures.append(thread_pool.submit(download_and_upload, tenant_id, plugin_id, plugin_identifier))
futures.append(thread_pool.submit(download_and_upload, tenant_id, plugin_id, package_identifier))
# Wait for all downloads to complete
for future in futures:
@ -578,33 +589,33 @@ class PluginMigration:
success = []
failed = []
reverse_map = {v: k for k, v in plugin_identifiers_map.items()}
plugin_id_by_package_identifier = {v: k for k, v in package_identifier_by_plugin_id.items()}
# at most 8 plugins one batch
for i in range(0, len(plugin_identifiers_map), 8):
batch_plugin_ids = list(plugin_identifiers_map.keys())[i : i + 8]
batch_plugin_identifiers = [plugin_identifiers_map[plugin_id] for plugin_id in batch_plugin_ids]
for i in range(0, len(package_identifier_by_plugin_id), 8):
batch_plugin_ids = list(package_identifier_by_plugin_id.keys())[i : i + 8]
batch_package_identifiers = [package_identifier_by_plugin_id[plugin_id] for plugin_id in batch_plugin_ids]
try:
response = manager.install_from_identifiers(
tenant_id=tenant_id,
identifiers=batch_plugin_identifiers,
identifiers=batch_package_identifiers,
source=PluginInstallationSource.Marketplace,
metas=[
{
"plugin_unique_identifier": identifier,
"plugin_unique_identifier": package_identifier,
}
for identifier in batch_plugin_identifiers
for package_identifier in batch_package_identifiers
],
)
PluginService.invalidate_plugin_model_providers_cache(tenant_id)
except Exception:
# add to failed
failed.extend(batch_plugin_identifiers)
failed.extend(batch_plugin_ids)
continue
if response.all_installed:
success.extend(batch_plugin_identifiers)
success.extend(batch_plugin_ids)
continue
task_id = response.task_id
@ -614,10 +625,13 @@ class PluginMigration:
if status.status in [PluginInstallTaskStatus.Failed, PluginInstallTaskStatus.Success]:
PluginService.invalidate_plugin_model_providers_cache(tenant_id)
for plugin in status.plugins:
plugin_id = plugin_id_by_package_identifier.get(
plugin.plugin_unique_identifier, plugin.plugin_unique_identifier.split(":", 1)[0]
)
if plugin.status == PluginInstallTaskStatus.Success:
success.append(reverse_map[plugin.plugin_unique_identifier])
success.append(plugin_id)
else:
failed.append(reverse_map[plugin.plugin_unique_identifier])
failed.append(plugin_id)
logger.error(
"Failed to install plugin %s, error: %s",
plugin.plugin_unique_identifier,

13
api/services/rag_pipeline/rag_pipeline_dsl_service.py
View File

@ -36,6 +36,7 @@ from models import Account
from models.dataset import Dataset, DatasetCollectionBinding, Pipeline
from models.enums import CollectionBindingType, DatasetRuntimeMode
from models.workflow import Workflow, WorkflowType
from services.dsl_content import DSL_MAX_SIZE, dsl_content_size
from services.dsl_version import check_version_compatibility
from services.entities.dsl_entities import CheckDependenciesResult, ImportMode, ImportStatus
from services.entities.knowledge_entities.rag_pipeline_entities import (
@ -50,7 +51,6 @@ logger = logging.getLogger(__name__)
IMPORT_INFO_REDIS_KEY_PREFIX = "app_import_info:"
CHECK_DEPENDENCIES_REDIS_KEY_PREFIX = "app_check_dependencies:"
IMPORT_INFO_REDIS_EXPIRY = 10 * 60 # 10 minutes
DSL_MAX_SIZE = 10 * 1024 * 1024 # 10MB
CURRENT_DSL_VERSION = "0.1.0"
@ -127,15 +127,16 @@ class RagPipelineDslService:
yaml_url = yaml_url.replace("/blob/", "/")
response = remote_fetcher.make_request("GET", yaml_url.strip(), follow_redirects=True, timeout=(10, 10))
response.raise_for_status()
content = response.content.decode()
raw_content = response.content
if len(content) > DSL_MAX_SIZE:
if dsl_content_size(raw_content) > DSL_MAX_SIZE:
return RagPipelineImportInfo(
id=import_id,
status=ImportStatus.FAILED,
error="File size exceeds the limit of 10MB",
)
content = raw_content.decode("utf-8")
if not content:
return RagPipelineImportInfo(
id=import_id,
@ -156,6 +157,12 @@ class RagPipelineDslService:
error="yaml_content is required when import_mode is yaml-content",
)
content = yaml_content
if dsl_content_size(content) > DSL_MAX_SIZE:
return RagPipelineImportInfo(
id=import_id,
status=ImportStatus.FAILED,
error="File size exceeds the limit of 10MB",
)
# Process YAML content
try:

20
api/services/rag_pipeline/rag_pipeline_transform_service.py
View File

@ -269,11 +269,13 @@ class RagPipelineTransformService:
installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins]
dependencies = pipeline_yaml.get("dependencies", [])
need_install_plugin_unique_identifiers = []
package_identifiers_to_install = []
for dependency in dependencies:
if dependency.get("type") == "marketplace":
plugin_unique_identifier = dependency.get("value", {}).get("plugin_unique_identifier")
plugin_id = plugin_unique_identifier.split(":")[0]
package_identifier = dependency.get("value", {}).get("plugin_unique_identifier")
if not package_identifier:
continue
plugin_id = package_identifier.split(":", 1)[0]
if plugin_id not in installed_plugins_ids:
if not dify_config.MARKETPLACE_ENABLED:
logger.warning(
@ -282,12 +284,12 @@ class RagPipelineTransformService:
plugin_id,
)
continue
plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(plugin_id) # type: ignore
if plugin_unique_identifier:
need_install_plugin_unique_identifiers.append(plugin_unique_identifier)
if need_install_plugin_unique_identifiers:
logger.debug("Installing missing pipeline plugins %s", need_install_plugin_unique_identifiers)
PluginService.install_from_marketplace_pkg(tenant_id, need_install_plugin_unique_identifiers)
latest_package_identifier = plugin_migration._fetch_latest_package_identifier(plugin_id) # type: ignore
if latest_package_identifier:
package_identifiers_to_install.append(latest_package_identifier)
if package_identifiers_to_install:
logger.debug("Installing missing pipeline plugins %s", package_identifiers_to_install)
PluginService.install_from_marketplace_pkg(tenant_id, package_identifiers_to_install)
def _transform_to_empty_pipeline(self, dataset: Dataset, session: Session):
pipeline = Pipeline(

54
api/services/snippet_dsl_service.py
View File

@ -3,12 +3,10 @@ import logging
import uuid
from collections.abc import Mapping
from datetime import UTC, datetime
from enum import StrEnum
from urllib.parse import urlparse
import yaml
from packaging import version
from pydantic import BaseModel, Field
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.orm import Session
@ -20,6 +18,9 @@ from graphon.model_runtime.utils.encoders import jsonable_encoder
from models import Account
from models.snippet import CustomizedSnippet, SnippetType
from models.workflow import Workflow
from services.dsl_content import DSL_MAX_SIZE, dsl_content_size
from services.dsl_version import check_version_compatibility
from services.entities.dsl_entities import CheckDependenciesResult, ImportMode, ImportStatus
from services.plugin.dependencies_analysis import DependenciesAnalysisService
from services.snippet_service import SNIPPET_FORBIDDEN_NODE_TYPES, SnippetService
@ -28,22 +29,9 @@ logger = logging.getLogger(__name__)
IMPORT_INFO_REDIS_KEY_PREFIX = "snippet_import_info:"
CHECK_DEPENDENCIES_REDIS_KEY_PREFIX = "snippet_check_dependencies:"
IMPORT_INFO_REDIS_EXPIRY = 10 * 60 # 10 minutes
DSL_MAX_SIZE = 10 * 1024 * 1024 # 10MB
CURRENT_DSL_VERSION = "0.1.0"
class ImportMode(StrEnum):
YAML_CONTENT = "yaml-content"
YAML_URL = "yaml-url"
class ImportStatus(StrEnum):
COMPLETED = "completed"
COMPLETED_WITH_WARNINGS = "completed-with-warnings"
PENDING = "pending"
FAILED = "failed"
class SnippetImportInfo(BaseModel):
id: str
status: ImportStatus
@ -53,32 +41,9 @@ class SnippetImportInfo(BaseModel):
error: str = ""
class CheckDependenciesResult(BaseModel):
leaked_dependencies: list[PluginDependency] = Field(default_factory=list)
def _check_version_compatibility(imported_version: str) -> ImportStatus:
"""Determine import status based on version comparison"""
try:
current_ver = version.parse(CURRENT_DSL_VERSION)
imported_ver = version.parse(imported_version)
except version.InvalidVersion:
return ImportStatus.FAILED
# If imported version is newer than current, always return PENDING
if imported_ver > current_ver:
return ImportStatus.PENDING
# If imported version is older than current's major, return PENDING
if imported_ver.major < current_ver.major:
return ImportStatus.PENDING
# If imported version is older than current's minor, return COMPLETED_WITH_WARNINGS
if imported_ver.minor < current_ver.minor:
return ImportStatus.COMPLETED_WITH_WARNINGS
# If imported version equals or is older than current's micro, return COMPLETED
return ImportStatus.COMPLETED
"""Determine import status based on version comparison."""
return check_version_compatibility(imported_version, CURRENT_DSL_VERSION)
class SnippetPendingData(BaseModel):
@ -145,13 +110,14 @@ class SnippetDslService:
status=ImportStatus.FAILED,
error=f"Failed to fetch YAML from URL: {response.status_code}",
)
content = response.text
if len(content) > DSL_MAX_SIZE:
raw_content = response.content
if dsl_content_size(raw_content) > DSL_MAX_SIZE:
return SnippetImportInfo(
id=import_id,
status=ImportStatus.FAILED,
error=f"YAML content size exceeds maximum limit of {DSL_MAX_SIZE} bytes",
)
content = raw_content.decode("utf-8")
except Exception as e:
logger.exception("Failed to fetch YAML from URL")
return SnippetImportInfo(
@ -167,7 +133,7 @@ class SnippetDslService:
error="yaml_content is required when import_mode is yaml-content",
)
content = yaml_content
if len(content) > DSL_MAX_SIZE:
if dsl_content_size(content) > DSL_MAX_SIZE:
return SnippetImportInfo(
id=import_id,
status=ImportStatus.FAILED,

58
api/tests/unit_tests/services/plugin/test_plugin_migration.py
View File

@ -1,3 +1,4 @@
import json
from unittest.mock import MagicMock, patch
import pytest
@ -8,17 +9,17 @@ from services.plugin.plugin_migration import PluginMigration
MIGRATION_MODULE = "services.plugin.plugin_migration"
def test_fetch_plugin_unique_identifier_returns_none_when_disabled(mocker: MockerFixture) -> None:
def test_fetch_latest_package_identifier_returns_none_when_disabled(mocker: MockerFixture) -> None:
mocker.patch("services.plugin.plugin_migration.dify_config.MARKETPLACE_ENABLED", False)
batch_fetch = mocker.patch("services.plugin.plugin_migration.marketplace.batch_fetch_plugin_manifests")
result = PluginMigration._fetch_plugin_unique_identifier("langgenius/openai")
result = PluginMigration._fetch_latest_package_identifier("langgenius/openai")
assert result is None
batch_fetch.assert_not_called()
def test_fetch_plugin_unique_identifier_calls_marketplace_when_enabled(mocker: MockerFixture) -> None:
def test_fetch_latest_package_identifier_calls_marketplace_when_enabled(mocker: MockerFixture) -> None:
mocker.patch("services.plugin.plugin_migration.dify_config.MARKETPLACE_ENABLED", True)
manifest = mocker.MagicMock()
manifest.latest_package_identifier = "langgenius/openai:1.0.0@abc"
@ -27,7 +28,7 @@ def test_fetch_plugin_unique_identifier_calls_marketplace_when_enabled(mocker: M
return_value=[manifest],
)
result = PluginMigration._fetch_plugin_unique_identifier("langgenius/openai")
result = PluginMigration._fetch_latest_package_identifier("langgenius/openai")
assert result == "langgenius/openai:1.0.0@abc"
@ -75,7 +76,27 @@ class TestHandlePluginInstanceInstall:
mock_marketplace.download_plugin_pkg.assert_called_once()
invalidate_cache.assert_called_once_with("tenant1")
assert "success" in result or "failed" in result
assert result["success"] == ["langgenius/openai"]
assert result["failed"] == []
def test_reports_failed_plugin_ids_when_install_batch_raises(self) -> None:
with (
patch(f"{MIGRATION_MODULE}.dify_config") as mock_cfg,
patch(f"{MIGRATION_MODULE}.marketplace") as mock_marketplace,
patch(f"{MIGRATION_MODULE}.PluginInstaller") as mock_installer_cls,
):
mock_cfg.MARKETPLACE_ENABLED = True
mock_marketplace.download_plugin_pkg.return_value = b"pkg_data"
mock_installer = MagicMock()
mock_installer_cls.return_value = mock_installer
mock_installer.install_from_identifiers.side_effect = RuntimeError("install failed")
result = PluginMigration.handle_plugin_instance_install(
"tenant1", {"langgenius/openai": "langgenius/openai:1.0.0@abc"}
)
assert result["success"] == []
assert result["failed"] == ["langgenius/openai"]
def test_install_plugins_invalidates_cache_after_direct_tenant_install(self, tmp_path) -> None:
extracted_plugins = tmp_path / "plugins.jsonl"
@ -102,3 +123,30 @@ class TestHandlePluginInstanceInstall:
mock_installer.install_from_identifiers.assert_called_once()
invalidate_cache.assert_called_once_with("tenant1")
def test_install_plugins_reports_missing_plugin_ids(self, tmp_path) -> None:
extracted_plugins = tmp_path / "plugins.jsonl"
output_file = tmp_path / "output.json"
extracted_plugins.write_text('{"tenant_id":"tenant1","plugins":["langgenius/openai","langgenius/missing"]}\n')
with (
patch(
f"{MIGRATION_MODULE}.PluginMigration.extract_unique_plugins",
return_value={
"plugins": {"langgenius/openai": "langgenius/openai:1.0.0@abc"},
"plugin_not_exist": ["langgenius/missing"],
},
),
patch(f"{MIGRATION_MODULE}.PluginMigration.handle_plugin_instance_install", return_value={}),
patch(f"{MIGRATION_MODULE}.PluginInstaller") as mock_installer_cls,
patch(f"{MIGRATION_MODULE}.PluginService.invalidate_plugin_model_providers_cache"),
):
mock_installer = MagicMock()
mock_installer.list_plugins.return_value = []
mock_installer_cls.return_value = mock_installer
PluginMigration.install_plugins(str(extracted_plugins), str(output_file), workers=1)
assert json.loads(output_file.read_text())["not_installed"] == [
{"tenant_id": "tenant1", "plugin_not_exist": ["langgenius/missing"]}
]

53
api/tests/unit_tests/services/rag_pipeline/test_rag_pipeline_dsl_service.py
View File

@ -643,6 +643,19 @@ def test_import_rag_pipeline_yaml_content_requires_mapping() -> None:
assert "content must be a mapping" in result.error
def test_import_rag_pipeline_rejects_oversized_yaml_content_by_bytes(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr("services.rag_pipeline.rag_pipeline_dsl_service.DSL_MAX_SIZE", 1)
service = RagPipelineDslService(session=Mock())
account = Mock(current_tenant_id="t1")
result = service.import_rag_pipeline(account=account, import_mode="yaml-content", yaml_content="é")
assert result.status == ImportStatus.FAILED
assert "10MB" in result.error
def test_confirm_import_returns_failed_when_pending_data_is_invalid_type(mocker: MockerFixture) -> None:
mocker.patch("services.rag_pipeline.rag_pipeline_dsl_service.redis_client.get", return_value=object())
service = RagPipelineDslService(session=Mock())
@ -901,6 +914,46 @@ def test_import_rag_pipeline_url_size_exceeds_limit(mocker: MockerFixture) -> No
assert "10MB" in result.error
def test_import_rag_pipeline_rejects_oversized_yaml_url_bytes_before_decode(
mocker: MockerFixture,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr("services.rag_pipeline.rag_pipeline_dsl_service.DSL_MAX_SIZE", 1)
response = Mock()
response.raise_for_status.return_value = None
response.content = b"\xff\xff"
mocker.patch("services.rag_pipeline.rag_pipeline_dsl_service.remote_fetcher.make_request", return_value=response)
service = RagPipelineDslService(session=Mock())
account = Mock(current_tenant_id="t1")
result = service.import_rag_pipeline(
account=account,
import_mode="yaml-url",
yaml_url="https://example.com/pipeline.yaml",
)
assert result.status == ImportStatus.FAILED
assert "10MB" in result.error
def test_import_rag_pipeline_returns_decode_error_for_invalid_yaml_url_bytes(mocker: MockerFixture) -> None:
response = Mock()
response.raise_for_status.return_value = None
response.content = b"\xff"
mocker.patch("services.rag_pipeline.rag_pipeline_dsl_service.remote_fetcher.make_request", return_value=response)
service = RagPipelineDslService(session=Mock())
account = Mock(current_tenant_id="t1")
result = service.import_rag_pipeline(
account=account,
import_mode="yaml-url",
yaml_url="https://example.com/pipeline.yaml",
)
assert result.status == ImportStatus.FAILED
assert "utf-8" in result.error
def test_import_rag_pipeline_fails_when_rag_pipeline_data_missing() -> None:
service = RagPipelineDslService(session=Mock())
account = Mock(current_tenant_id="t1")

4
api/tests/unit_tests/services/rag_pipeline/test_rag_pipeline_transform_service.py
View File

@ -89,7 +89,7 @@ def test_deal_dependencies_installs_missing_marketplace_plugins(mocker: MockerFi
installer_cls.return_value.list_plugins.return_value = [SimpleNamespace(plugin_id="installed-plugin")]
migration_cls = mocker.patch("services.rag_pipeline.rag_pipeline_transform_service.PluginMigration")
migration_cls.return_value._fetch_plugin_unique_identifier.return_value = "missing-plugin:1.0.0"
migration_cls.return_value._fetch_latest_package_identifier.return_value = "missing-plugin:1.0.0"
install_mock = mocker.patch(
"services.rag_pipeline.rag_pipeline_transform_service.PluginService.install_from_marketplace_pkg"
@ -518,7 +518,7 @@ def test_deal_dependencies_installs_when_enabled(mocker: MockerFixture) -> None:
installer = mocker.patch("services.rag_pipeline.rag_pipeline_transform_service.PluginInstaller").return_value
installer.list_plugins.return_value = []
migration = mocker.patch("services.rag_pipeline.rag_pipeline_transform_service.PluginMigration").return_value
migration._fetch_plugin_unique_identifier.return_value = "langgenius/openai:1.0.0@abc"
migration._fetch_latest_package_identifier.return_value = "langgenius/openai:1.0.0@abc"
install_call = mocker.patch(
"services.rag_pipeline.rag_pipeline_transform_service.PluginService.install_from_marketplace_pkg"
)

53
api/tests/unit_tests/services/test_app_dsl_service.py Normal file
View File

@ -0,0 +1,53 @@
from types import SimpleNamespace
from unittest.mock import Mock
from services.app_dsl_service import AppDslService, ImportStatus
def test_import_app_rejects_oversized_yaml_content_by_bytes(monkeypatch) -> None:
monkeypatch.setattr("services.app_dsl_service.DSL_MAX_SIZE", 1)
service = AppDslService(session=SimpleNamespace())
result = service.import_app(
account=SimpleNamespace(current_tenant_id="tenant-1"),
import_mode="yaml-content",
yaml_content="é",
)
assert result.status == ImportStatus.FAILED
assert "10MB" in result.error
def test_import_app_rejects_oversized_yaml_url_bytes_before_decode(monkeypatch) -> None:
monkeypatch.setattr("services.app_dsl_service.DSL_MAX_SIZE", 1)
response = Mock()
response.raise_for_status.return_value = None
response.content = b"\xff\xff"
monkeypatch.setattr("services.app_dsl_service.remote_fetcher.make_request", Mock(return_value=response))
service = AppDslService(session=SimpleNamespace())
result = service.import_app(
account=SimpleNamespace(current_tenant_id="tenant-1"),
import_mode="yaml-url",
yaml_url="https://example.com/app.yaml",
)
assert result.status == ImportStatus.FAILED
assert "10MB" in result.error
def test_import_app_returns_decode_error_for_invalid_yaml_url_bytes(monkeypatch) -> None:
response = Mock()
response.raise_for_status.return_value = None
response.content = b"\xff"
monkeypatch.setattr("services.app_dsl_service.remote_fetcher.make_request", Mock(return_value=response))
service = AppDslService(session=SimpleNamespace())
result = service.import_app(
account=SimpleNamespace(current_tenant_id="tenant-1"),
import_mode="yaml-url",
yaml_url="https://example.com/app.yaml",
)
assert result.status == ImportStatus.FAILED
assert "utf-8" in result.error

43
api/tests/unit_tests/services/test_snippet_dsl_service.py
View File

@ -95,7 +95,7 @@ def test_import_snippet_rejects_oversized_yaml_url_content(monkeypatch: pytest.M
monkeypatch.setattr("services.snippet_dsl_service.DSL_MAX_SIZE", 3)
monkeypatch.setattr(
"services.snippet_dsl_service.ssrf_proxy.get",
Mock(return_value=SimpleNamespace(status_code=200, text="too large")),
Mock(return_value=SimpleNamespace(status_code=200, content=b"too large")),
)
result = service.import_snippet(
@ -108,6 +108,43 @@ def test_import_snippet_rejects_oversized_yaml_url_content(monkeypatch: pytest.M
assert "YAML content size exceeds maximum limit" in result.error
def test_import_snippet_rejects_oversized_yaml_url_bytes_before_decode(monkeypatch: pytest.MonkeyPatch) -> None:
service = SnippetDslService(session=SimpleNamespace())
monkeypatch.setattr("services.snippet_dsl_service.DSL_MAX_SIZE", 1)
monkeypatch.setattr(
"services.snippet_dsl_service.ssrf_proxy.get",
Mock(return_value=SimpleNamespace(status_code=200, content=b"\xff\xff")),
)
result = service.import_snippet(
account=SimpleNamespace(current_tenant_id="tenant-1"),
import_mode=ImportMode.YAML_URL.value,
yaml_url="https://example.com/snippet.yaml",
)
assert result.status == ImportStatus.FAILED
assert "YAML content size exceeds maximum limit" in result.error
def test_import_snippet_returns_decode_error_for_invalid_yaml_url_bytes(
monkeypatch: pytest.MonkeyPatch,
) -> None:
service = SnippetDslService(session=SimpleNamespace())
monkeypatch.setattr(
"services.snippet_dsl_service.ssrf_proxy.get",
Mock(return_value=SimpleNamespace(status_code=200, content=b"\xff")),
)
result = service.import_snippet(
account=SimpleNamespace(current_tenant_id="tenant-1"),
import_mode=ImportMode.YAML_URL.value,
yaml_url="https://example.com/snippet.yaml",
)
assert result.status == ImportStatus.FAILED
assert "utf-8" in result.error
def test_import_snippet_returns_failed_when_yaml_url_fetch_raises(monkeypatch: pytest.MonkeyPatch) -> None:
service = SnippetDslService(session=SimpleNamespace())
monkeypatch.setattr(
@ -127,12 +164,12 @@ def test_import_snippet_returns_failed_when_yaml_url_fetch_raises(monkeypatch: p
def test_import_snippet_rejects_oversized_yaml_content(monkeypatch: pytest.MonkeyPatch) -> None:
service = SnippetDslService(session=SimpleNamespace())
monkeypatch.setattr("services.snippet_dsl_service.DSL_MAX_SIZE", 3)
monkeypatch.setattr("services.snippet_dsl_service.DSL_MAX_SIZE", 1)
result = service.import_snippet(
account=SimpleNamespace(current_tenant_id="tenant-1"),
import_mode=ImportMode.YAML_CONTENT.value,
yaml_content="too large",
yaml_content="é",
)
assert result.status == ImportStatus.FAILED