From c9e6db5ef20dbc2fe426e1bff7591e5a128fd7cd Mon Sep 17 00:00:00 2001 From: WH-2099 Date: Fri, 7 Aug 2026 09:21:13 +0000 Subject: [PATCH] fix(api): scope temporary capabilities to owners (#39863) --- .../console/workspace/trigger_providers.py | 27 +- api/services/app_dsl_service.py | 21 +- api/services/entities/dsl_entities.py | 12 + .../plugin/plugin_parameter_service.py | 8 +- .../rag_pipeline/rag_pipeline_dsl_service.py | 15 +- api/services/snippet_dsl_service.py | 21 +- .../trigger_subscription_builder_service.py | 226 +++++++--------- .../services/test_app_dsl_service.py | 9 +- .../console/app/test_app_import_api.py | 11 +- .../workspace/test_trigger_provider_apis.py | 42 ++- .../plugin/test_plugin_parameter_service.py | 11 +- .../test_rag_pipeline_dsl_service.py | 24 +- .../services/test_app_dsl_service.py | 74 +++++- .../services/test_snippet_dsl_service.py | 38 ++- ...st_trigger_subscription_builder_service.py | 251 ++++++++++++++++++ 15 files changed, 618 insertions(+), 172 deletions(-) create mode 100644 api/tests/unit_tests/services/test_trigger_subscription_builder_service.py diff --git a/api/controllers/console/workspace/trigger_providers.py b/api/controllers/console/workspace/trigger_providers.py index aff800a5bc9..b228d536aa5 100644 --- a/api/controllers/console/workspace/trigger_providers.py +++ b/api/controllers/console/workspace/trigger_providers.py @@ -268,9 +268,16 @@ class TriggerSubscriptionBuilderGetApi(Resource): @edit_permission_required @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_PREFERENCES, resource_required=False) @account_initialization_required - def get(self, provider: str, subscription_builder_id: str): + @with_current_user + @with_current_tenant_id + def get(self, tenant_id: str, user: Account, provider: str, subscription_builder_id: str): """Get a subscription instance for a trigger provider""" - subscription_builder = TriggerSubscriptionBuilderService.get_subscription_builder_by_id(subscription_builder_id) + subscription_builder = TriggerSubscriptionBuilderService.get_subscription_builder_by_id( + tenant_id=tenant_id, + user_id=user.id, + provider_id=TriggerProviderID(provider), + subscription_builder_id=subscription_builder_id, + ) return subscription_builder.model_dump(mode="json") @@ -328,14 +335,16 @@ class TriggerSubscriptionBuilderUpdateApi(Resource): @edit_permission_required @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.CREDENTIAL_MANAGE, resource_required=False) @account_initialization_required + @with_current_user @with_current_tenant_id - def post(self, tenant_id: str, provider: str, subscription_builder_id: str): + def post(self, tenant_id: str, user: Account, provider: str, subscription_builder_id: str): """Update a subscription instance for a trigger provider""" payload = TriggerSubscriptionBuilderUpdatePayload.model_validate(console_ns.payload or {}) try: return TriggerSubscriptionBuilderService.update_trigger_subscription_builder( tenant_id=tenant_id, + user_id=user.id, provider_id=TriggerProviderID(provider), subscription_builder_id=subscription_builder_id, subscription_builder_updater=SubscriptionBuilderUpdater( @@ -364,11 +373,18 @@ class TriggerSubscriptionBuilderLogsApi(Resource): @edit_permission_required @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_PREFERENCES, resource_required=False) @account_initialization_required - def get(self, provider: str, subscription_builder_id: str): + @with_current_user + @with_current_tenant_id + def get(self, tenant_id: str, user: Account, provider: str, subscription_builder_id: str): """Get the request logs for a subscription instance for a trigger provider""" try: - logs = TriggerSubscriptionBuilderService.list_logs(subscription_builder_id) + logs = TriggerSubscriptionBuilderService.list_logs( + tenant_id=tenant_id, + user_id=user.id, + provider_id=TriggerProviderID(provider), + subscription_builder_id=subscription_builder_id, + ) return dump_response(TriggerSubscriptionBuilderLogsResponse, {"logs": logs}) except Exception as e: logger.exception("Error getting request logs for subscription builder", exc_info=e) @@ -652,6 +668,7 @@ class TriggerOAuthCallbackApi(Resource): # Update subscription builder TriggerSubscriptionBuilderService.update_trigger_subscription_builder( tenant_id=tenant_id, + user_id=user_id, provider_id=provider_id, subscription_builder_id=subscription_builder_id, subscription_builder_updater=SubscriptionBuilderUpdater( diff --git a/api/services/app_dsl_service.py b/api/services/app_dsl_service.py index ee69f0bc7af..460fb3e4aca 100644 --- a/api/services/app_dsl_service.py +++ b/api/services/app_dsl_service.py @@ -52,7 +52,13 @@ from services.agent.workflow_publish_service import WorkflowAgentPublishService from services.dsl_content import DSL_MAX_SIZE, dsl_content_size from services.dsl_version import check_version_compatibility from services.enterprise.rbac_service import RBACService -from services.entities.dsl_entities import CheckDependenciesResult, DslImportWarning, ImportMode, ImportStatus +from services.entities.dsl_entities import ( + CheckDependenciesResult, + DslImportWarning, + ImportMode, + ImportStatus, + PendingImportOwner, +) from services.errors.account import NoPermissionError from services.errors.app import WorkflowNotFoundError from services.plugin.dependencies_analysis import DependenciesAnalysisService @@ -80,7 +86,7 @@ class Import(BaseModel): warnings: list[DslImportWarning] = Field(default_factory=list) -class PendingData(BaseModel): +class PendingData(PendingImportOwner): import_mode: str yaml_content: str name: str | None = None @@ -241,6 +247,8 @@ class AppDslService: # If major version mismatch, store import info in Redis if status == ImportStatus.PENDING: pending_data = PendingData( + tenant_id=account.current_tenant_id, + account_id=account.id, import_mode=import_mode, yaml_content=content, name=name, @@ -346,6 +354,15 @@ class AppDslService: error="Invalid import information", ) pending_data = PendingData.model_validate_json(pending_data) + if not pending_data.is_accessible_by( + tenant_id=account.current_tenant_id, + account_id=account.id, + ): + return Import( + id=import_id, + status=ImportStatus.FAILED, + error="Import information expired or does not exist", + ) data = yaml.safe_load(pending_data.yaml_content) app = None diff --git a/api/services/entities/dsl_entities.py b/api/services/entities/dsl_entities.py index 0d5f49dc4cc..2bff3fcac24 100644 --- a/api/services/entities/dsl_entities.py +++ b/api/services/entities/dsl_entities.py @@ -18,6 +18,18 @@ class ImportStatus(StrEnum): FAILED = "failed" +class PendingImportOwner(BaseModel): + tenant_id: str | None = None + account_id: str | None = None + + def is_accessible_by(self, *, tenant_id: str | None, account_id: str) -> bool: + if tenant_id is None: + return False + owner = (self.tenant_id, self.account_id) + # Ownerless payloads come from older pods and expire after 10 minutes; #40106 removes this bridge. + return owner in ((None, None), (tenant_id, account_id)) + + class DslImportWarning(BaseModel): """Portable DSL reference that could not be restored in the target workspace.""" diff --git a/api/services/plugin/plugin_parameter_service.py b/api/services/plugin/plugin_parameter_service.py index 786c09b44e1..09d016e381a 100644 --- a/api/services/plugin/plugin_parameter_service.py +++ b/api/services/plugin/plugin_parameter_service.py @@ -12,6 +12,7 @@ from core.tools.utils.encryption import create_tool_provider_encrypter from core.trigger.entities.api_entities import TriggerProviderSubscriptionApiEntity from core.trigger.entities.entities import SubscriptionBuilder from extensions.ext_database import db +from models.provider_ids import TriggerProviderID from models.tools import BuiltinToolProvider from services.trigger.trigger_provider_service import TriggerProviderService from services.trigger.trigger_subscription_builder_service import TriggerSubscriptionBuilderService @@ -85,7 +86,12 @@ class PluginParameterService: case "trigger": subscription: TriggerProviderSubscriptionApiEntity | SubscriptionBuilder | None if credential_id: - subscription = TriggerSubscriptionBuilderService.get_subscription_builder(credential_id) + subscription = TriggerSubscriptionBuilderService.get_subscription_builder( + tenant_id=tenant_id, + user_id=user_id, + provider_id=TriggerProviderID(f"{plugin_id}/{provider}"), + subscription_builder_id=credential_id, + ) if not subscription: trigger_subscription = TriggerProviderService.get_subscription_by_id(tenant_id, credential_id) subscription = trigger_subscription.to_api_entity() if trigger_subscription else None diff --git a/api/services/rag_pipeline/rag_pipeline_dsl_service.py b/api/services/rag_pipeline/rag_pipeline_dsl_service.py index 7290123eb2e..6977c28b7a6 100644 --- a/api/services/rag_pipeline/rag_pipeline_dsl_service.py +++ b/api/services/rag_pipeline/rag_pipeline_dsl_service.py @@ -44,7 +44,7 @@ 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.dsl_entities import CheckDependenciesResult, ImportMode, ImportStatus, PendingImportOwner from services.entities.knowledge_entities.rag_pipeline_entities import ( IconInfo, KnowledgeConfiguration, @@ -70,7 +70,7 @@ class RagPipelineImportInfo(BaseModel): dataset_id: str | None = None -class RagPipelinePendingData(BaseModel): +class RagPipelinePendingData(PendingImportOwner): import_mode: str yaml_content: str pipeline_id: str | None @@ -222,6 +222,8 @@ class RagPipelineDslService: # If major version mismatch, store import info in Redis if status == ImportStatus.PENDING: pending_data = RagPipelinePendingData( + tenant_id=account.current_tenant_id, + account_id=account.id, import_mode=import_mode, yaml_content=content, pipeline_id=pipeline_id, @@ -384,6 +386,15 @@ class RagPipelineDslService: error="Invalid import information", ) pending_data = RagPipelinePendingData.model_validate_json(pending_data) + if not pending_data.is_accessible_by( + tenant_id=account.current_tenant_id, + account_id=account.id, + ): + return RagPipelineImportInfo( + id=import_id, + status=ImportStatus.FAILED, + error="Import information expired or does not exist", + ) data = yaml.safe_load(pending_data.yaml_content) pipeline = None diff --git a/api/services/snippet_dsl_service.py b/api/services/snippet_dsl_service.py index 0c694151252..22f495a2370 100644 --- a/api/services/snippet_dsl_service.py +++ b/api/services/snippet_dsl_service.py @@ -23,7 +23,13 @@ from services.agent.retirement_service import WorkflowAgentRetirementService from services.agent.workflow_publish_service import WorkflowAgentPublishService 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, DslImportWarning, ImportMode, ImportStatus +from services.entities.dsl_entities import ( + CheckDependenciesResult, + DslImportWarning, + ImportMode, + ImportStatus, + PendingImportOwner, +) from services.plugin.dependencies_analysis import DependenciesAnalysisService from services.snippet_service import SNIPPET_FORBIDDEN_NODE_TYPES, SnippetService from tasks.collect_agent_resources_task import enqueue_agent_resource_collection @@ -51,7 +57,7 @@ def _check_version_compatibility(imported_version: str) -> ImportStatus: return check_version_compatibility(imported_version, CURRENT_DSL_VERSION) -class SnippetPendingData(BaseModel): +class SnippetPendingData(PendingImportOwner): import_mode: str yaml_content: str name: str | None = None @@ -231,6 +237,8 @@ class SnippetDslService: # If major version mismatch, store import info in Redis if status == ImportStatus.PENDING: pending_data = SnippetPendingData( + tenant_id=account.current_tenant_id, + account_id=account.id, import_mode=import_mode, yaml_content=content, name=name, @@ -315,6 +323,15 @@ class SnippetDslService: pending_data_str = pending_data.decode("utf-8") if isinstance(pending_data, bytes) else pending_data pending = SnippetPendingData.model_validate_json(pending_data_str) + if not pending.is_accessible_by( + tenant_id=account.current_tenant_id, + account_id=account.id, + ): + return SnippetImportInfo( + id=import_id, + status=ImportStatus.FAILED, + error="Import information expired or does not exist", + ) data = yaml.safe_load(pending.yaml_content) if not isinstance(data, dict): diff --git a/api/services/trigger/trigger_subscription_builder_service.py b/api/services/trigger/trigger_subscription_builder_service.py index cff735b39d3..0901e5ccf7f 100644 --- a/api/services/trigger/trigger_subscription_builder_service.py +++ b/api/services/trigger/trigger_subscription_builder_service.py @@ -73,102 +73,6 @@ class TriggerSubscriptionBuilderService: with redis_client.lock(lock_key, timeout=cls.__LOCK_EXPIRE_SECONDS__): yield - @classmethod - def verify_trigger_subscription_builder( - cls, - tenant_id: str, - user_id: str, - provider_id: TriggerProviderID, - subscription_builder_id: str, - ) -> Mapping[str, Any]: - """Verify a trigger subscription builder""" - provider_controller = TriggerManager.get_trigger_provider(tenant_id, provider_id) - if not provider_controller: - raise ValueError(f"Provider {provider_id} not found") - - subscription_builder = cls.get_subscription_builder(subscription_builder_id) - if not subscription_builder: - raise ValueError(f"Subscription builder {subscription_builder_id} not found") - - if subscription_builder.credential_type == CredentialType.OAUTH2: - return {"verified": bool(subscription_builder.credentials)} - - if subscription_builder.credential_type == CredentialType.API_KEY: - credentials_to_validate = subscription_builder.credentials - try: - provider_controller.validate_credentials(user_id, credentials_to_validate) - except ToolProviderCredentialValidationError as e: - raise ValueError(f"Invalid credentials: {e}") - return {"verified": True} - - return {"verified": True} - - @classmethod - def build_trigger_subscription_builder( - cls, tenant_id: str, user_id: str, provider_id: TriggerProviderID, subscription_builder_id: str - ) -> None: - """Build a trigger subscription builder""" - provider_controller = TriggerManager.get_trigger_provider(tenant_id, provider_id) - if not provider_controller: - raise ValueError(f"Provider {provider_id} not found") - - # Acquire lock to prevent concurrent build operations - with cls.acquire_builder_lock(subscription_builder_id): - subscription_builder = cls.get_subscription_builder(subscription_builder_id) - if not subscription_builder: - raise ValueError(f"Subscription builder {subscription_builder_id} not found") - - if not subscription_builder.name: - raise ValueError("Subscription builder name is required") - - credential_type = CredentialType.of(subscription_builder.credential_type or CredentialType.UNAUTHORIZED) - if credential_type == CredentialType.UNAUTHORIZED: - # manually create - TriggerProviderService.add_trigger_subscription( - subscription_id=subscription_builder.id, - tenant_id=tenant_id, - user_id=user_id, - name=subscription_builder.name, - provider_id=provider_id, - endpoint_id=subscription_builder.endpoint_id, - parameters=subscription_builder.parameters, - properties=subscription_builder.properties, - credential_expires_at=subscription_builder.credential_expires_at or -1, - expires_at=subscription_builder.expires_at, - credentials=subscription_builder.credentials, - credential_type=credential_type, - ) - else: - # automatically create - subscription: Subscription = TriggerManager.subscribe_trigger( - tenant_id=tenant_id, - user_id=user_id, - provider_id=provider_id, - endpoint=generate_plugin_trigger_endpoint_url(subscription_builder.endpoint_id), - parameters=subscription_builder.parameters, - credentials=subscription_builder.credentials, - credential_type=credential_type, - ) - - TriggerProviderService.add_trigger_subscription( - subscription_id=subscription_builder.id, - tenant_id=tenant_id, - user_id=user_id, - name=subscription_builder.name, - provider_id=provider_id, - endpoint_id=subscription_builder.endpoint_id, - parameters=subscription_builder.parameters, - properties=subscription.properties, - credentials=subscription_builder.credentials, - credential_type=credential_type, - credential_expires_at=subscription_builder.credential_expires_at or -1, - expires_at=subscription_builder.expires_at, - ) - - # Delete the builder after successful subscription creation - cache_key = cls.encode_cache_key(subscription_builder_id) - redis_client.delete(cache_key) - @classmethod def create_trigger_subscription_builder( cls, @@ -208,6 +112,7 @@ class TriggerSubscriptionBuilderService: def update_trigger_subscription_builder( cls, tenant_id: str, + user_id: str, provider_id: TriggerProviderID, subscription_builder_id: str, subscription_builder_updater: SubscriptionBuilderUpdater, @@ -223,9 +128,12 @@ class TriggerSubscriptionBuilderService: # Acquire lock to prevent concurrent updates with cls.acquire_builder_lock(subscription_id): cache_key = cls.encode_cache_key(subscription_id) - subscription_builder_cache = cls.get_subscription_builder(subscription_builder_id) - if not subscription_builder_cache or subscription_builder_cache.tenant_id != tenant_id: - raise ValueError(f"Subscription {subscription_id} expired or not found") + subscription_builder_cache = cls._require_owned_subscription_builder( + tenant_id=tenant_id, + user_id=user_id, + provider_id=provider_id, + subscription_builder_id=subscription_builder_id, + ) subscription_builder_updater.update(subscription_builder_cache) @@ -255,9 +163,12 @@ class TriggerSubscriptionBuilderService: # Acquire lock for the entire update + verify operation with cls.acquire_builder_lock(subscription_id): cache_key = cls.encode_cache_key(subscription_id) - subscription_builder_cache = cls.get_subscription_builder(subscription_builder_id) - if not subscription_builder_cache or subscription_builder_cache.tenant_id != tenant_id: - raise ValueError(f"Subscription {subscription_id} expired or not found") + subscription_builder_cache = cls._require_owned_subscription_builder( + tenant_id=tenant_id, + user_id=user_id, + provider_id=provider_id, + subscription_builder_id=subscription_builder_id, + ) # Update subscription_builder_updater.update(subscription_builder_cache) @@ -300,20 +211,16 @@ class TriggerSubscriptionBuilderService: # Acquire lock for the entire update + build operation with cls.acquire_builder_lock(subscription_id): cache_key = cls.encode_cache_key(subscription_id) - subscription_builder_cache = cls.get_subscription_builder(subscription_builder_id) - if not subscription_builder_cache or subscription_builder_cache.tenant_id != tenant_id: - raise ValueError(f"Subscription {subscription_id} expired or not found") - - # Update - subscription_builder_updater.update(subscription_builder_cache) - redis_client.setex( - cache_key, cls.__BUILDER_CACHE_EXPIRE_SECONDS__, subscription_builder_cache.model_dump_json() + subscription_builder = cls._require_owned_subscription_builder( + tenant_id=tenant_id, + user_id=user_id, + provider_id=provider_id, + subscription_builder_id=subscription_builder_id, ) - # Re-fetch to ensure we have the latest data - subscription_builder = cls.get_subscription_builder(subscription_builder_id) - if not subscription_builder: - raise ValueError(f"Subscription builder {subscription_builder_id} not found") + # Update + subscription_builder_updater.update(subscription_builder) + redis_client.setex(cache_key, cls.__BUILDER_CACHE_EXPIRE_SECONDS__, subscription_builder.model_dump_json()) if not subscription_builder.name: raise ValueError("Subscription builder name is required") @@ -364,7 +271,6 @@ class TriggerSubscriptionBuilderService: ) # Delete the builder after successful subscription creation - cache_key = cls.encode_cache_key(subscription_builder_id) redis_client.delete(cache_key) @classmethod @@ -389,17 +295,57 @@ class TriggerSubscriptionBuilderService: ) @classmethod - def get_subscription_builder(cls, endpoint_id: str) -> SubscriptionBuilder | None: - """ - Get a trigger subscription by the endpoint ID. - """ + def _get_subscription_builder_by_endpoint_id(cls, endpoint_id: str) -> SubscriptionBuilder | None: + """Resolve the public validation capability without authenticated owner context.""" cache_key = cls.encode_cache_key(endpoint_id) subscription_cache = redis_client.get(cache_key) if subscription_cache: - return SubscriptionBuilder.model_validate_json(subscription_cache) + subscription_builder = SubscriptionBuilder.model_validate_json(subscription_cache) + if subscription_builder.endpoint_id == endpoint_id: + return subscription_builder return None + @classmethod + def get_subscription_builder( + cls, + tenant_id: str, + user_id: str, + provider_id: TriggerProviderID, + subscription_builder_id: str, + ) -> SubscriptionBuilder | None: + """Return an owned temporary builder, or None when no temporary builder exists.""" + subscription_builder = cls._get_subscription_builder_by_endpoint_id(subscription_builder_id) + if subscription_builder is None: + return None + if ( + subscription_builder.id != subscription_builder_id + or subscription_builder.tenant_id != tenant_id + or subscription_builder.user_id != user_id + or subscription_builder.provider_id != str(provider_id) + ): + raise ValueError(f"Subscription builder {subscription_builder_id} not found") + return subscription_builder + + @classmethod + def _require_owned_subscription_builder( + cls, + tenant_id: str, + user_id: str, + provider_id: TriggerProviderID, + subscription_builder_id: str, + ) -> SubscriptionBuilder: + """Return an owned temporary builder or reject an absent capability.""" + subscription_builder = cls.get_subscription_builder( + tenant_id=tenant_id, + user_id=user_id, + provider_id=provider_id, + subscription_builder_id=subscription_builder_id, + ) + if subscription_builder is None: + raise ValueError(f"Subscription builder {subscription_builder_id} not found") + return subscription_builder + @classmethod def append_log(cls, endpoint_id: str, request: Request, response: Response) -> None: """Append validation request log to Redis.""" @@ -433,9 +379,22 @@ class TriggerSubscriptionBuilderService: ) @classmethod - def list_logs(cls, endpoint_id: str) -> list[RequestLog]: + def list_logs( + cls, + tenant_id: str, + user_id: str, + provider_id: TriggerProviderID, + subscription_builder_id: str, + ) -> list[RequestLog]: """List request logs for validation endpoint.""" - key = f"trigger:subscription:builder:logs:{endpoint_id}" + subscription_builder = cls._require_owned_subscription_builder( + tenant_id=tenant_id, + user_id=user_id, + provider_id=provider_id, + subscription_builder_id=subscription_builder_id, + ) + + key = f"trigger:subscription:builder:logs:{subscription_builder.endpoint_id}" logs_json = redis_client.get(key) if not logs_json: return [] @@ -451,7 +410,7 @@ class TriggerSubscriptionBuilderService: :return: The Flask response object """ # check if validation endpoint exists - subscription_builder: SubscriptionBuilder | None = cls.get_subscription_builder(endpoint_id) + subscription_builder: SubscriptionBuilder | None = cls._get_subscription_builder_by_endpoint_id(endpoint_id) if not subscription_builder: return None @@ -482,14 +441,21 @@ class TriggerSubscriptionBuilderService: return error_response @classmethod - def get_subscription_builder_by_id(cls, subscription_builder_id: str) -> SubscriptionBuilderApiEntity: + def get_subscription_builder_by_id( + cls, + tenant_id: str, + user_id: str, + provider_id: TriggerProviderID, + subscription_builder_id: str, + ) -> SubscriptionBuilderApiEntity: """Get a trigger subscription builder API entity.""" - subscription_builder = cls.get_subscription_builder(subscription_builder_id) - if not subscription_builder: - raise ValueError(f"Subscription builder {subscription_builder_id} not found") + subscription_builder = cls._require_owned_subscription_builder( + tenant_id=tenant_id, + user_id=user_id, + provider_id=provider_id, + subscription_builder_id=subscription_builder_id, + ) return cls.builder_to_api_entity( - controller=TriggerManager.get_trigger_provider( - subscription_builder.tenant_id, TriggerProviderID(subscription_builder.provider_id) - ), + controller=TriggerManager.get_trigger_provider(tenant_id, provider_id), entity=subscription_builder, ) diff --git a/api/tests/test_containers_integration_tests/services/test_app_dsl_service.py b/api/tests/test_containers_integration_tests/services/test_app_dsl_service.py index d9bfb9decd1..3f7d9729471 100644 --- a/api/tests/test_containers_integration_tests/services/test_app_dsl_service.py +++ b/api/tests/test_containers_integration_tests/services/test_app_dsl_service.py @@ -491,6 +491,9 @@ class TestAppDslService: redis_key = f"{IMPORT_INFO_REDIS_KEY_PREFIX}{result.id}" stored = redis_client.get(redis_key) assert stored is not None + pending = PendingData.model_validate_json(stored) + assert pending.tenant_id == _DEFAULT_TENANT_ID + assert pending.account_id == _DEFAULT_ACCOUNT_ID def test_import_app_completed_uses_declared_dependencies( self, db_session_with_containers: Session, mock_external_service_dependencies @@ -606,7 +609,11 @@ class TestAppDslService: icon_background="#fff", app_id=None, ) - redis_client.setex(redis_key, IMPORT_INFO_REDIS_EXPIRY, pending.model_dump_json()) + redis_client.setex( + redis_key, + IMPORT_INFO_REDIS_EXPIRY, + pending.model_dump_json(exclude={"tenant_id", "account_id"}), + ) created_app = SimpleNamespace( id=str(uuid4()), diff --git a/api/tests/unit_tests/controllers/console/app/test_app_import_api.py b/api/tests/unit_tests/controllers/console/app/test_app_import_api.py index f0c2f9a7faa..4068ec0e7f3 100644 --- a/api/tests/unit_tests/controllers/console/app/test_app_import_api.py +++ b/api/tests/unit_tests/controllers/console/app/test_app_import_api.py @@ -58,6 +58,7 @@ def _install_features(monkeypatch: pytest.MonkeyPatch, enabled: bool) -> None: def _make_account(account_id: str = "u1") -> Account: account = Account(name="Test User", email="test@example.com") account.id = account_id + account._current_tenant = MagicMock(id="tenant-1") return account @@ -343,14 +344,13 @@ class TestAppImportConfirmApi: "current_account_with_tenant", lambda: (_make_account(), "tenant-1"), ) - monkeypatch.setattr( - app_import_module.redis_client, - "get", - lambda *_args, **_kwargs: ( + redis_get = MagicMock( + return_value=( b'{"import_mode":"yaml-content","yaml_content":"app: {}","app_id":null,' b'"name":null,"description":null,"icon_type":null,"icon":null,"icon_background":null}' - ), + ) ) + monkeypatch.setattr(app_import_module.redis_client, "get", redis_get) monkeypatch.setattr(app_import_module.dify_config, "RBAC_ENABLED", True) app_id = _install_persisting_service_result( monkeypatch, @@ -370,6 +370,7 @@ class TestAppImportConfirmApi: _assert_app_persistence(sqlite_app_engine, app_id, persisted=True) assert status == 200 assert response["permission_keys"] == ["app.acl.view_layout", "app.acl.edit"] + redis_get.assert_called_once_with("app_import_info:import-1") def test_import_confirm_does_not_attach_permission_keys_when_overwriting_existing_app( self, diff --git a/api/tests/unit_tests/controllers/console/workspace/test_trigger_provider_apis.py b/api/tests/unit_tests/controllers/console/workspace/test_trigger_provider_apis.py index 6eca14aa273..0dcad28d492 100644 --- a/api/tests/unit_tests/controllers/console/workspace/test_trigger_provider_apis.py +++ b/api/tests/unit_tests/controllers/console/workspace/test_trigger_provider_apis.py @@ -4,7 +4,7 @@ from __future__ import annotations from datetime import datetime from inspect import unwrap -from unittest.mock import MagicMock, patch +from unittest.mock import ANY, MagicMock, patch import pytest from flask import Flask @@ -175,9 +175,15 @@ class TestTriggerSubscriptionBuilderApis: patch( "controllers.console.workspace.trigger_providers.TriggerSubscriptionBuilderService.get_subscription_builder_by_id", return_value=subscription_builder(), - ), + ) as mock_get_builder, ): - assert method(api, "github", "b1")["id"] == "b1" + assert method(api, "t1", mock_user(), "github", "b1")["id"] == "b1" + mock_get_builder.assert_called_once_with( + tenant_id="t1", + user_id="u1", + provider_id=ANY, + subscription_builder_id="b1", + ) def test_verify_builder(self, app: Flask) -> None: api = TriggerSubscriptionBuilderVerifyApi() @@ -215,9 +221,16 @@ class TestTriggerSubscriptionBuilderApis: patch( "controllers.console.workspace.trigger_providers.TriggerSubscriptionBuilderService.update_trigger_subscription_builder", return_value=subscription_builder(), - ), + ) as mock_update_builder, ): - assert method(api, "t1", "github", "b1")["id"] == "b1" + assert method(api, "t1", mock_user(), "github", "b1")["id"] == "b1" + mock_update_builder.assert_called_once_with( + tenant_id="t1", + user_id="u1", + provider_id=ANY, + subscription_builder_id="b1", + subscription_builder_updater=ANY, + ) def test_logs(self, app: Flask) -> None: api = TriggerSubscriptionBuilderLogsApi() @@ -228,10 +241,16 @@ class TestTriggerSubscriptionBuilderApis: patch( "controllers.console.workspace.trigger_providers.TriggerSubscriptionBuilderService.list_logs", return_value=[request_log()], - ), + ) as mock_list_logs, ): - result = method(api, "github", "b1") + result = method(api, "t1", mock_user(), "github", "b1") assert result["logs"][0]["id"] == "log1" + mock_list_logs.assert_called_once_with( + tenant_id="t1", + user_id="u1", + provider_id=ANY, + subscription_builder_id="b1", + ) def test_build(self, app: Flask) -> None: api = TriggerSubscriptionBuilderBuildApi() @@ -377,10 +396,17 @@ class TestTriggerOAuthApis: ), patch( "controllers.console.workspace.trigger_providers.TriggerSubscriptionBuilderService.update_trigger_subscription_builder" - ), + ) as mock_update_builder, ): resp = method(api, "github") assert resp.status_code == 302 + mock_update_builder.assert_called_once_with( + tenant_id="t1", + user_id="u1", + provider_id=ANY, + subscription_builder_id="b1", + subscription_builder_updater=ANY, + ) def test_oauth_callback_no_oauth_client(self, app: Flask) -> None: api = TriggerOAuthCallbackApi() 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 index b51e13013a4..916217ce0b1 100644 --- a/api/tests/unit_tests/services/plugin/test_plugin_parameter_service.py +++ b/api/tests/unit_tests/services/plugin/test_plugin_parameter_service.py @@ -140,7 +140,7 @@ class TestGetDynamicSelectOptionsTrigger: result = PluginParameterService.get_dynamic_select_options( tenant_id="t1", user_id="u1", - plugin_id="p1", + plugin_id="org/plugin", provider="github", action="on_push", parameter="branch", @@ -149,6 +149,11 @@ class TestGetDynamicSelectOptionsTrigger: ) assert result == ["opt"] + builder_call = mock_builder_svc.get_subscription_builder.call_args.kwargs + assert builder_call["tenant_id"] == "t1" + assert builder_call["user_id"] == "u1" + assert str(builder_call["provider_id"]) == "org/plugin/github" + assert builder_call["subscription_builder_id"] == "builder-1" @patch("services.plugin.plugin_parameter_service.DynamicSelectClient") @patch("services.plugin.plugin_parameter_service.TriggerProviderService") @@ -166,7 +171,7 @@ class TestGetDynamicSelectOptionsTrigger: result = PluginParameterService.get_dynamic_select_options( tenant_id="t1", user_id="u1", - plugin_id="p1", + plugin_id="org/plugin", provider="github", action="on_push", parameter="branch", @@ -186,7 +191,7 @@ class TestGetDynamicSelectOptionsTrigger: PluginParameterService.get_dynamic_select_options( tenant_id="t1", user_id="u1", - plugin_id="p1", + plugin_id="org/plugin", provider="github", action="on_push", parameter="branch", diff --git a/api/tests/unit_tests/services/rag_pipeline/test_rag_pipeline_dsl_service.py b/api/tests/unit_tests/services/rag_pipeline/test_rag_pipeline_dsl_service.py index ab12e278125..a2cc34741ef 100644 --- a/api/tests/unit_tests/services/rag_pipeline/test_rag_pipeline_dsl_service.py +++ b/api/tests/unit_tests/services/rag_pipeline/test_rag_pipeline_dsl_service.py @@ -46,11 +46,11 @@ def service(sqlite_session: Session) -> RagPipelineDslService: return RagPipelineDslService(session=sqlite_session) -def _account(*, tenant_id: str = "tenant-1") -> Account: +def _account(*, tenant_id: str = "tenant-1", account_id: str = "account-1") -> Account: tenant = Tenant(name="Tenant") tenant.id = tenant_id account = Account(name="Account", email="account@example.com") - account.id = "account-1" + account.id = account_id account._current_tenant = tenant return account @@ -627,6 +627,10 @@ def test_import_pending_version_stores_redis(monkeypatch: pytest.MonkeyPatch, se account=_account(), import_mode=ImportMode.YAML_CONTENT.value, yaml_content=_valid_dsl(version="1.0.0") ) assert result.status == ImportStatus.PENDING + assert setex.call_args.args[0] == f"app_import_info:{result.id}" + pending = RagPipelinePendingData.model_validate_json(setex.call_args.args[2]) + assert pending.tenant_id == "tenant-1" + assert pending.account_id == "account-1" setex.assert_called_once() @@ -782,14 +786,26 @@ def test_confirm_import_updates_tenant_pipeline_and_dataset( dataset = _dataset(sqlite_session, pipeline) _workflow(sqlite_session, pipeline) pending = RagPipelinePendingData( + tenant_id="tenant-1", + account_id="account-1", import_mode=ImportMode.YAML_CONTENT.value, yaml_content=_valid_dsl(name="Confirmed"), pipeline_id=pipeline.id, ) - monkeypatch.setattr(module.redis_client, "get", Mock(return_value=pending.model_dump_json())) + redis_key = "app_import_info:import-1" + monkeypatch.setattr( + module.redis_client, + "get", + Mock(side_effect=lambda key: pending.model_dump_json() if key == redis_key else None), + ) delete = Mock() monkeypatch.setattr(module.redis_client, "delete", delete) monkeypatch.setattr(module.KnowledgeConfiguration, "model_validate", Mock(return_value=_knowledge_configuration())) + for foreign_account in (_account(tenant_id="tenant-2"), _account(account_id="account-2")): + assert service.confirm_import(import_id="import-1", account=foreign_account).status == ImportStatus.FAILED + delete.assert_not_called() + assert pipeline.name == "Pipeline" + result = service.confirm_import(import_id="import-1", account=_account()) assert result.status == ImportStatus.COMPLETED assert result.pipeline_id == pipeline.id @@ -809,7 +825,7 @@ def test_confirm_import_updates_tenant_pipeline_and_dataset( assert observed_pipeline is not None assert observed_pipeline.name == "Confirmed" - delete.assert_called_once() + delete.assert_called_once_with(redis_key) def test_export_reads_real_dataset_and_workflow_and_filters_credentials( diff --git a/api/tests/unit_tests/services/test_app_dsl_service.py b/api/tests/unit_tests/services/test_app_dsl_service.py index 403b9c582f6..e6d1b0ea939 100644 --- a/api/tests/unit_tests/services/test_app_dsl_service.py +++ b/api/tests/unit_tests/services/test_app_dsl_service.py @@ -12,7 +12,7 @@ from core.workflow.llm_environment_variable import LLMEnvironmentVariable from models import App, AppMode from models.model import AppModelConfig, AppModelConfigDict, IconType from models.workflow import Workflow -from services.app_dsl_service import AppDslService +from services.app_dsl_service import AppDslService, PendingData from services.entities.dsl_entities import ImportStatus from services.errors.account import NoPermissionError @@ -144,6 +144,78 @@ def test_import_app_returns_decode_error_for_invalid_yaml_url_bytes( assert not unbound_session.in_transaction() +def test_pending_import_is_scoped_to_its_owner(monkeypatch: pytest.MonkeyPatch, unbound_session: Session) -> None: + pending_imports: dict[str, str] = {} + monkeypatch.setattr( + "services.app_dsl_service.redis_client.setex", + lambda key, _expiry, value: pending_imports.__setitem__(key, value), + ) + service = AppDslService(session=unbound_session) + creator = Mock(id="account-1", current_tenant_id="tenant-1") + + pending = service.import_app( + account=creator, + import_mode="yaml-content", + yaml_content="version: 99.0.0\nkind: app\napp: {name: Test, mode: workflow}\n", + ) + + redis_key = f"app_import_info:{pending.id}" + assert pending.status == ImportStatus.PENDING + assert redis_key in pending_imports + pending_data = PendingData.model_validate_json(pending_imports[redis_key]) + assert pending_data.tenant_id == "tenant-1" + assert pending_data.account_id == "account-1" + + monkeypatch.setattr("services.app_dsl_service.redis_client.get", pending_imports.get) + monkeypatch.setattr("services.app_dsl_service.redis_client.delete", pending_imports.pop) + monkeypatch.setattr( + service, + "_create_or_update_app", + Mock(return_value=Mock(id="app-1", mode=AppMode.WORKFLOW)), + ) + + for other_account in ( + Mock(id="account-1", current_tenant_id="tenant-2"), + Mock(id="account-2", current_tenant_id="tenant-1"), + ): + assert service.confirm_import(import_id=pending.id, account=other_account).status == ImportStatus.FAILED + + assert service.confirm_import(import_id=pending.id, account=creator).status == ImportStatus.COMPLETED + assert redis_key not in pending_imports + + +@pytest.mark.parametrize( + ("tenant_id", "account_id", "expected"), + [ + ("tenant-1", "account-1", True), + (None, "account-1", False), + ("tenant-1", None, False), + ("tenant-2", "account-1", False), + ("tenant-1", "account-2", False), + ], +) +def test_pending_import_owner_access( + tenant_id: str | None, + account_id: str | None, + expected: bool, +) -> None: + pending = PendingData( + tenant_id=tenant_id, + account_id=account_id, + import_mode="yaml-content", + yaml_content="", + ) + + assert pending.is_accessible_by(tenant_id="tenant-1", account_id="account-1") is expected + + +def test_pending_import_owner_access_accepts_legacy_json() -> None: + pending = PendingData.model_validate_json('{"import_mode":"yaml-content","yaml_content":""}') + + assert pending.is_accessible_by(tenant_id="tenant-1", account_id="account-1") + assert not pending.is_accessible_by(tenant_id=None, account_id="account-1") + + def test_create_or_update_app_loads_existing_model_config_with_service_session( sqlite_session_factory: sessionmaker[Session], ) -> None: diff --git a/api/tests/unit_tests/services/test_snippet_dsl_service.py b/api/tests/unit_tests/services/test_snippet_dsl_service.py index 6ae44cbd588..73137dac66b 100644 --- a/api/tests/unit_tests/services/test_snippet_dsl_service.py +++ b/api/tests/unit_tests/services/test_snippet_dsl_service.py @@ -269,7 +269,7 @@ workflow: """ result = service.import_snippet( - account=SimpleNamespace(current_tenant_id="tenant-1"), + account=SimpleNamespace(id="account-1", current_tenant_id="tenant-1"), import_mode=ImportMode.YAML_CONTENT.value, yaml_content=yaml_content, name="Override", @@ -278,7 +278,10 @@ workflow: assert result.status == ImportStatus.PENDING setex.assert_called_once() + assert setex.call_args.args[0] == f"snippet_import_info:{result.id}" pending = SnippetPendingData.model_validate_json(setex.call_args.args[2]) + assert pending.tenant_id == "tenant-1" + assert pending.account_id == "account-1" assert pending.name == "Override" assert pending.description == "Override description" @@ -359,7 +362,9 @@ def test_confirm_import_returns_failed_when_pending_data_missing(monkeypatch): service = SnippetDslService(session=SimpleNamespace()) monkeypatch.setattr("services.snippet_dsl_service.redis_client.get", Mock(return_value=None)) - result = service.confirm_import(import_id="missing", account=SimpleNamespace(current_tenant_id="tenant-1")) + result = service.confirm_import( + import_id="missing", account=SimpleNamespace(id="account-1", current_tenant_id="tenant-1") + ) assert result.status == ImportStatus.FAILED assert result.error == "Import information expired or does not exist" @@ -369,13 +374,15 @@ def test_confirm_import_returns_failed_for_invalid_pending_payload(monkeypatch): service = SnippetDslService(session=SimpleNamespace()) monkeypatch.setattr("services.snippet_dsl_service.redis_client.get", Mock(return_value=object())) - result = service.confirm_import(import_id="bad", account=SimpleNamespace(current_tenant_id="tenant-1")) + result = service.confirm_import( + import_id="bad", account=SimpleNamespace(id="account-1", current_tenant_id="tenant-1") + ) assert result.status == ImportStatus.FAILED assert result.error == "Invalid import information" -def test_confirm_import_creates_snippet_from_pending_data(monkeypatch): +def test_confirm_import_is_scoped_to_its_owner(monkeypatch): service = SnippetDslService(session=SimpleNamespace(scalar=Mock(return_value=None))) account = SimpleNamespace(id="account-1", current_tenant_id="tenant-1") snippet = SimpleNamespace(id="snippet-new") @@ -391,6 +398,8 @@ workflow: edges: [] """ pending = SnippetPendingData( + tenant_id="tenant-1", + account_id="account-1", import_mode="yaml-content", yaml_content=yaml_content, name="Override name", @@ -399,10 +408,21 @@ workflow: ) create_or_update = Mock(return_value=snippet) monkeypatch.setattr(service, "_create_or_update_snippet", create_or_update) - monkeypatch.setattr("services.snippet_dsl_service.redis_client.get", Mock(return_value=pending.model_dump_json())) + redis_key = "snippet_import_info:import-1" + monkeypatch.setattr( + "services.snippet_dsl_service.redis_client.get", + Mock(side_effect=lambda key: pending.model_dump_json() if key == redis_key else None), + ) redis_delete = Mock() monkeypatch.setattr("services.snippet_dsl_service.redis_client.delete", redis_delete) + for other_account in ( + SimpleNamespace(id="account-1", current_tenant_id="tenant-2"), + SimpleNamespace(id="account-2", current_tenant_id="tenant-1"), + ): + assert service.confirm_import(import_id="import-1", account=other_account).status == ImportStatus.FAILED + + create_or_update.assert_not_called() result = service.confirm_import(import_id="import-1", account=account) assert result.status == ImportStatus.COMPLETED @@ -414,7 +434,7 @@ workflow: assert kwargs["account"] is account assert kwargs["name"] == "Override name" assert kwargs["description"] == "Override description" - redis_delete.assert_called_once_with("snippet_import_info:import-1") + redis_delete.assert_called_once_with(redis_key) def test_confirm_import_returns_failed_for_non_mapping_yaml(monkeypatch): @@ -426,7 +446,9 @@ def test_confirm_import_returns_failed_for_non_mapping_yaml(monkeypatch): ) monkeypatch.setattr("services.snippet_dsl_service.redis_client.get", Mock(return_value=pending.model_dump_json())) - result = service.confirm_import(import_id="import-1", account=SimpleNamespace(current_tenant_id="tenant-1")) + result = service.confirm_import( + import_id="import-1", account=SimpleNamespace(id="account-1", current_tenant_id="tenant-1") + ) assert result.status == ImportStatus.FAILED assert result.error == "Invalid YAML format: expected a dictionary" @@ -445,7 +467,7 @@ def test_confirm_import_returns_failed_when_create_or_update_raises(monkeypatch) result = service.confirm_import( import_id="import-1", - account=SimpleNamespace(current_tenant_id="tenant-1"), + account=SimpleNamespace(id="account-1", current_tenant_id="tenant-1"), ) assert result.status == ImportStatus.FAILED diff --git a/api/tests/unit_tests/services/test_trigger_subscription_builder_service.py b/api/tests/unit_tests/services/test_trigger_subscription_builder_service.py new file mode 100644 index 00000000000..6316de75792 --- /dev/null +++ b/api/tests/unit_tests/services/test_trigger_subscription_builder_service.py @@ -0,0 +1,251 @@ +from collections.abc import Callable +from contextlib import nullcontext +from unittest.mock import Mock, patch + +import pytest + +from core.plugin.entities.plugin_daemon import CredentialType +from core.trigger.entities.entities import SubscriptionBuilder, SubscriptionBuilderUpdater +from core.trigger.trigger_manager import TriggerManager +from models.provider_ids import TriggerProviderID +from services.trigger.trigger_subscription_builder_service import TriggerSubscriptionBuilderService + +PROVIDER_ID = TriggerProviderID("org/plugin/provider") + + +def subscription_builder() -> SubscriptionBuilder: + return SubscriptionBuilder( + id="builder-1", + name="Builder", + tenant_id="tenant-1", + user_id="user-1", + provider_id=str(PROVIDER_ID), + endpoint_id="builder-1", + parameters={}, + properties={}, + credentials={}, + credential_type=CredentialType.UNAUTHORIZED, + credential_expires_at=-1, + expires_at=-1, + ) + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("id", "other-builder"), + ("tenant_id", "other-tenant"), + ("user_id", "other-user"), + ("provider_id", "org/plugin/other"), + ], +) +def test_get_subscription_builder_rejects_non_owner(field: str, value: str) -> None: + builder = subscription_builder().model_copy(update={field: value}) + with patch.object( + TriggerSubscriptionBuilderService, + "_get_subscription_builder_by_endpoint_id", + return_value=builder, + ): + with pytest.raises(ValueError, match="not found"): + TriggerSubscriptionBuilderService.get_subscription_builder( + tenant_id="tenant-1", + user_id="user-1", + provider_id=PROVIDER_ID, + subscription_builder_id="builder-1", + ) + + +def test_get_subscription_builder_accepts_owner() -> None: + builder = subscription_builder() + with patch.object( + TriggerSubscriptionBuilderService, + "_get_subscription_builder_by_endpoint_id", + return_value=builder, + ): + assert ( + TriggerSubscriptionBuilderService.get_subscription_builder( + tenant_id="tenant-1", + user_id="user-1", + provider_id=PROVIDER_ID, + subscription_builder_id="builder-1", + ) + is builder + ) + + +def test_get_subscription_builder_returns_none_when_temporary_builder_is_absent() -> None: + with patch.object( + TriggerSubscriptionBuilderService, + "_get_subscription_builder_by_endpoint_id", + return_value=None, + ): + assert ( + TriggerSubscriptionBuilderService.get_subscription_builder( + tenant_id="tenant-1", + user_id="user-1", + provider_id=PROVIDER_ID, + subscription_builder_id="builder-1", + ) + is None + ) + + +def test_get_subscription_builder_rejects_mismatched_endpoint() -> None: + builder = subscription_builder().model_copy(update={"endpoint_id": "other-builder"}) + with patch( + "services.trigger.trigger_subscription_builder_service.redis_client.get", + return_value=builder.model_dump_json(), + ): + assert TriggerSubscriptionBuilderService._get_subscription_builder_by_endpoint_id("builder-1") is None + + +def test_get_subscription_builder_accepts_matching_endpoint() -> None: + builder = subscription_builder() + with patch( + "services.trigger.trigger_subscription_builder_service.redis_client.get", + return_value=builder.model_dump_json(), + ): + assert TriggerSubscriptionBuilderService._get_subscription_builder_by_endpoint_id("builder-1") == builder + + +@pytest.mark.parametrize( + ("operation", "needs_updater"), + [ + (TriggerSubscriptionBuilderService.update_trigger_subscription_builder, True), + (TriggerSubscriptionBuilderService.update_and_verify_builder, True), + (TriggerSubscriptionBuilderService.update_and_build_builder, True), + (TriggerSubscriptionBuilderService.list_logs, False), + (TriggerSubscriptionBuilderService.get_subscription_builder_by_id, False), + ], +) +def test_owner_scoped_operations_reject_missing_builder_before_side_effects( + operation: Callable[..., object], needs_updater: bool +) -> None: + kwargs: dict[str, object] = { + "tenant_id": "tenant-1", + "user_id": "user-1", + "provider_id": PROVIDER_ID, + "subscription_builder_id": "builder-1", + } + if needs_updater: + kwargs["subscription_builder_updater"] = SubscriptionBuilderUpdater(name="Updated") + + with ( + patch.object(TriggerManager, "get_trigger_provider", return_value=Mock()), + patch.object(TriggerSubscriptionBuilderService, "acquire_builder_lock", return_value=nullcontext()), + patch.object( + TriggerSubscriptionBuilderService, + "get_subscription_builder", + return_value=None, + ) as get_subscription_builder, + patch("services.trigger.trigger_subscription_builder_service.redis_client.setex") as setex, + patch("services.trigger.trigger_subscription_builder_service.redis_client.delete") as delete, + ): + with pytest.raises(ValueError, match="not found"): + operation(**kwargs) + + get_subscription_builder.assert_called_once_with( + tenant_id="tenant-1", + user_id="user-1", + provider_id=PROVIDER_ID, + subscription_builder_id="builder-1", + ) + setex.assert_not_called() + delete.assert_not_called() + + +def test_update_and_build_uses_the_owned_builder_without_refetching() -> None: + builder = subscription_builder() + cache_key = TriggerSubscriptionBuilderService.encode_cache_key(builder.id) + + with ( + patch.object(TriggerManager, "get_trigger_provider", return_value=Mock()), + patch.object(TriggerSubscriptionBuilderService, "acquire_builder_lock", return_value=nullcontext()), + patch.object( + TriggerSubscriptionBuilderService, + "get_subscription_builder", + return_value=builder, + ) as get_subscription_builder, + patch("services.trigger.trigger_subscription_builder_service.redis_client.setex") as setex, + patch( + "services.trigger.trigger_subscription_builder_service.TriggerProviderService.add_trigger_subscription" + ) as add_subscription, + patch("services.trigger.trigger_subscription_builder_service.redis_client.delete") as delete, + ): + TriggerSubscriptionBuilderService.update_and_build_builder( + tenant_id="tenant-1", + user_id="user-1", + provider_id=PROVIDER_ID, + subscription_builder_id=builder.id, + subscription_builder_updater=SubscriptionBuilderUpdater(name="Updated"), + ) + + get_subscription_builder.assert_called_once_with( + tenant_id="tenant-1", + user_id="user-1", + provider_id=PROVIDER_ID, + subscription_builder_id=builder.id, + ) + setex.assert_called_once_with(cache_key, 30 * 60, builder.model_dump_json()) + add_subscription.assert_called_once() + subscription_call = add_subscription.call_args.kwargs + assert subscription_call["subscription_id"] == builder.id + assert subscription_call["tenant_id"] == "tenant-1" + assert subscription_call["user_id"] == "user-1" + assert subscription_call["provider_id"] == PROVIDER_ID + assert subscription_call["endpoint_id"] == builder.endpoint_id + assert subscription_call["name"] == "Updated" + delete.assert_called_once_with(cache_key) + + +def test_list_logs_uses_the_owned_builder_endpoint() -> None: + builder = subscription_builder() + logs_key = f"trigger:subscription:builder:logs:{builder.endpoint_id}" + + with ( + patch.object(TriggerSubscriptionBuilderService, "get_subscription_builder", return_value=builder), + patch("services.trigger.trigger_subscription_builder_service.redis_client.get", return_value=None) as redis_get, + ): + assert ( + TriggerSubscriptionBuilderService.list_logs( + tenant_id="tenant-1", + user_id="user-1", + provider_id=PROVIDER_ID, + subscription_builder_id=builder.id, + ) + == [] + ) + + redis_get.assert_called_once_with(logs_key) + + +def test_process_validation_endpoint_uses_the_public_capability() -> None: + builder = subscription_builder() + request = Mock() + response = Mock() + controller = Mock() + controller.dispatch.return_value = Mock(response=response) + + with ( + patch.object( + TriggerSubscriptionBuilderService, + "_get_subscription_builder_by_endpoint_id", + return_value=builder, + ) as get_by_endpoint, + patch.object(TriggerSubscriptionBuilderService, "get_subscription_builder") as get_owned, + patch.object(TriggerManager, "get_trigger_provider", return_value=controller) as get_provider, + patch.object(TriggerSubscriptionBuilderService, "append_log") as append_log, + ): + assert ( + TriggerSubscriptionBuilderService.process_builder_validation_endpoint(builder.endpoint_id, request) + is response + ) + + get_by_endpoint.assert_called_once_with(builder.endpoint_id) + get_owned.assert_not_called() + get_provider.assert_called_once() + provider_call = get_provider.call_args.kwargs + assert provider_call["tenant_id"] == builder.tenant_id + assert str(provider_call["provider_id"]) == str(PROVIDER_ID) + controller.dispatch.assert_called_once() + append_log.assert_called_once()