From 56200015e6313cd049458191cc1c8d60028057a1 Mon Sep 17 00:00:00 2001 From: Weike Zhang <66246918+weike-zhang@users.noreply.github.com> Date: Wed, 23 Sep 2026 11:27:54 +0000 Subject: [PATCH 1/2] fix(api): keep creator access to datasets created via console upload and RAG pipeline (#42507) Co-authored-by: weike-zhang Co-authored-by: Crazywoola <100913391+crazywoola@users.noreply.github.com> (cherry picked from commit f832cff60c48ae2c69cee2405687d7fedfde4227) --- .../console/datasets/datasets_document.py | 6 +++ .../rag_pipeline/rag_pipeline_datasets.py | 6 +++ .../test_rag_pipeline_datasets.py | 52 +++++++++++++++++++ .../datasets/test_datasets_document.py | 40 ++++++++++++++ 4 files changed, 104 insertions(+) diff --git a/api/controllers/console/datasets/datasets_document.py b/api/controllers/console/datasets/datasets_document.py index 3299462037d3f2..661c70d32e2e98 100644 --- a/api/controllers/console/datasets/datasets_document.py +++ b/api/controllers/console/datasets/datasets_document.py @@ -692,6 +692,12 @@ def post(self, session: Session, current_tenant_id: str, current_user: Account): dataset.id, enterprise_rbac_service.ReplaceMemberBindings(automatic_include_workspace_members=False), ) + enterprise_rbac_service.try_sync_creator_access_policy_member_bindings( + current_tenant_id, + current_user.id, + enterprise_rbac_service.RBACResourceType.DATASET, + dataset.id, + ) return dump_response( DatasetAndDocumentResponse, diff --git a/api/controllers/console/datasets/rag_pipeline/rag_pipeline_datasets.py b/api/controllers/console/datasets/rag_pipeline/rag_pipeline_datasets.py index ed0a7b0b1bccda..73ca05aba37f59 100644 --- a/api/controllers/console/datasets/rag_pipeline/rag_pipeline_datasets.py +++ b/api/controllers/console/datasets/rag_pipeline/rag_pipeline_datasets.py @@ -97,6 +97,12 @@ def post( dataset_id, enterprise_rbac_service.ReplaceMemberBindings(automatic_include_workspace_members=False), ) + enterprise_rbac_service.try_sync_creator_access_policy_member_bindings( + current_tenant_id, + current_user.id, + enterprise_rbac_service.RBACResourceType.DATASET, + dataset_id, + ) return dump_response(RagPipelineImportResponse, import_info), 201 diff --git a/api/tests/unit_tests/controllers/console/datasets/rag_pipeline/test_rag_pipeline_datasets.py b/api/tests/unit_tests/controllers/console/datasets/rag_pipeline/test_rag_pipeline_datasets.py index 3e2c9cefd043de..1f38ccbe05dc57 100644 --- a/api/tests/unit_tests/controllers/console/datasets/rag_pipeline/test_rag_pipeline_datasets.py +++ b/api/tests/unit_tests/controllers/console/datasets/rag_pipeline/test_rag_pipeline_datasets.py @@ -18,7 +18,9 @@ RagPipelineDatasetImportPayload, ) from models.account import Account, TenantAccountRole +from services.enterprise import rbac_service as enterprise_rbac_service from services.entities.dsl_entities import ImportStatus +from tests.unit_tests.config_override import config_overrides_context def _account(*, editor: bool) -> Account: @@ -71,6 +73,56 @@ def test_post_success(self, app: Flask) -> None: "error": "", } + def test_post_success_syncs_creator_under_rbac(self, app: Flask) -> None: + """RBAC: RAG-pipeline dataset create must keep the creator bound to their own dataset. + + Regression guard for the #42430 flip to automatic_include_workspace_members=False: + this path bypasses DatasetService.create_empty_dataset (which syncs the creator + internally), so the controller has to sync the creator explicitly or the creator + gets 403 on their own dataset. + """ + api = CreateRagPipelineDatasetApi() + method = unwrap(api.post) + + payload = self._valid_payload() + user = _account(editor=True) + import_info = { + "id": "import-1", + "status": ImportStatus.COMPLETED, + "dataset_id": "ds-1", + "pipeline_id": "pipeline-1", + "current_dsl_version": "0.1.0", + "imported_dsl_version": "0.1.0", + "error": "", + } + + mock_service = MagicMock() + mock_service.create_rag_pipeline_dataset.return_value = import_info + + with ( + app.test_request_context("/", json=payload), + patch.object(type(console_ns), "payload", payload), + config_overrides_context(RBAC_ENABLED=True), + patch( + "controllers.console.datasets.rag_pipeline.rag_pipeline_datasets.RagPipelineDslService", + return_value=mock_service, + ), + patch( + "controllers.console.datasets.rag_pipeline.rag_pipeline_datasets.enterprise_rbac_service.RBACService.DatasetAccess.replace_whitelist" + ), + patch( + "controllers.console.datasets.rag_pipeline.rag_pipeline_datasets.enterprise_rbac_service.try_sync_creator_access_policy_member_bindings" + ) as sync_creator, + ): + response, status = method(api, RagPipelineDatasetImportPayload.model_validate(payload), "tenant-1", user) + + assert status == 201 + assert sync_creator.called, "RAG pipeline create must sync the creator's own access" + sync_creator.assert_called_once_with( + "tenant-1", user.id, enterprise_rbac_service.RBACResourceType.DATASET, "ds-1" + ) + assert response["dataset_id"] == "ds-1" + def test_post_forbidden_non_editor(self, app: Flask) -> None: api = CreateRagPipelineDatasetApi() method = unwrap(api.post) diff --git a/api/tests/unit_tests/controllers/console/datasets/test_datasets_document.py b/api/tests/unit_tests/controllers/console/datasets/test_datasets_document.py index e2795b74d64e5d..d51c559b56d10c 100644 --- a/api/tests/unit_tests/controllers/console/datasets/test_datasets_document.py +++ b/api/tests/unit_tests/controllers/console/datasets/test_datasets_document.py @@ -497,6 +497,46 @@ def test_post_success_serializes_created_dataset_and_documents(self, app: Flask, assert response["batch"] == "batch-init" assert created_dataset.permission == DatasetPermissionEnum.ALL_TEAM + def test_post_success_syncs_creator_access_on_upload_create(self, app: Flask, patch_tenant): + """RBAC: the console upload create must keep the creator bound to their own dataset. + + Regression guard for the #42430 flip to automatic_include_workspace_members=False: + unlike POST /console/api/datasets (DatasetService.create_empty_dataset syncs the + creator internally), this path bypasses that helper, so the controller has to sync + the creator explicitly or the creator gets 403 on their own dataset. + """ + api = DatasetInitApi() + method = unwrap(api.post) + user, tenant_id = patch_tenant + payload = {"indexing_technique": "economy"} + created_dataset = make_dataset() + created_document = make_document(id="doc-init") + session = self.session + captured = {} + with ( + app.test_request_context("/", json=payload), + patch.object(type(console_ns), "payload", payload), + config_overrides_context(RBAC_ENABLED=True), + patch( + "controllers.console.datasets.datasets_document.DocumentService.document_create_args_validate", + return_value=None, + ), + patch( + "controllers.console.datasets.datasets_document.DocumentService.save_document_without_dataset_id", + return_value=(created_dataset, [created_document], "batch-init"), + ), + patch( + "controllers.console.datasets.datasets_document.enterprise_rbac_service.RBACService.DatasetAccess.replace_whitelist" + ), + patch( + "controllers.console.datasets.datasets_document.enterprise_rbac_service.try_sync_creator_access_policy_member_bindings" + ) as sync_creator, + ): + response = method(api, session, tenant_id, user) + captured["sync_calls"] = sync_creator.call_args_list + assert sync_creator.called, "upload create must sync the creator's own access byte-through" + assert response["dataset"]["id"] == "ds-1" + class TestDocumentResource(_UsesSQLiteSession): def test_get_document_resolves_owner_chain(self, dataset): From 55798f5367025e9036bd154e29733bd5614459f7 Mon Sep 17 00:00:00 2001 From: XiaoBa Date: Tue, 13 Oct 2026 16:56:49 +0000 Subject: [PATCH 2/2] fix(api): preserve creator RBAC access across app and RAG pipeline creation (#43941) Adapted to LTS 1.17.x: App import, confirm and copy initialize creator access in the existing controllers and AppDslService instead of the console gateway that only exists on main. (cherry picked from commit 51a8dba933f383bf04d4a18b35b680b9ced38beb) Amp-Thread-ID: https://ampcode.com/threads/T-01a19f37-375f-7645-9260-69c071966f14 --- api/controllers/console/app/app_import.py | 35 ++--- .../rag_pipeline/rag_pipeline_import.py | 27 ++++ api/controllers/inner_api/app/dsl.py | 3 +- api/controllers/openapi/app_dsl.py | 4 +- api/services/app_dsl_service.py | 27 ++++ api/services/dataset_service.py | 9 ++ api/services/enterprise/rbac_service.py | 41 +++++- .../rag_pipeline/rag_pipeline_dsl_service.py | 10 +- .../console/app/test_app_import_api.py | 51 ++++--- .../rag_pipeline/test_rag_pipeline_import.py | 109 +++++++++++++- .../services/enterprise/test_rbac_service.py | 136 ++++++++++++++++++ .../test_rag_pipeline_dsl_service.py | 83 ++++++++++- .../services/test_app_dsl_service.py | 102 ++++++++++++- .../services/test_dataset_service_dataset.py | 65 +++++++++ 14 files changed, 648 insertions(+), 54 deletions(-) diff --git a/api/controllers/console/app/app_import.py b/api/controllers/console/app/app_import.py index 18de98f8d9aca2..e9104bcbea6486 100644 --- a/api/controllers/console/app/app_import.py +++ b/api/controllers/console/app/app_import.py @@ -5,7 +5,7 @@ from configs import dify_config from controllers.common.rbac import PlainApp, RBACCheck, Workspace -from controllers.common.schema import register_enum_models, register_schema_models +from controllers.common.schema import register_enum_models, register_response_schema_models, register_schema_models from controllers.console.app.wraps import get_app_model from controllers.console.wraps import ( RBACPermission, @@ -18,16 +18,10 @@ with_current_user, ) from extensions.ext_database import db -from extensions.ext_redis import redis_client from libs.login import current_account_with_tenant, login_required from models.account import Account from models.model import App -from services.app_dsl_service import ( - IMPORT_INFO_REDIS_KEY_PREFIX, - AppDslService, - Import, - PendingData, -) +from services.app_dsl_service import AppDslService, Import, initialize_imported_app_creator_access from services.enterprise.enterprise_service import EnterpriseService from services.entities.dsl_entities import CheckDependenciesResult, ImportStatus from services.errors.account import NoPermissionError @@ -50,7 +44,8 @@ class AppImportPayload(BaseModel): register_enum_models(console_ns, ImportStatus) -register_schema_models(console_ns, AppImportPayload, Import, CheckDependenciesResult) +register_schema_models(console_ns, AppImportPayload, CheckDependenciesResult) +register_response_schema_models(console_ns, Import) def _current_user_and_tenant_id(current_user: Account | None) -> tuple[Account, str | None]: @@ -114,13 +109,14 @@ def post(self, req_data: AppImportPayload, current_user: Account | None = None): else: session.commit() - is_created_app = req_data.app_id is None and result.status in { + is_created_app = result.app_created and result.status in { ImportStatus.COMPLETED, ImportStatus.COMPLETED_WITH_WARNINGS, } if dify_config.RBAC_ENABLED and is_created_app and result.app_id: current_user, current_tenant_id = _current_user_and_tenant_id(current_user) if current_tenant_id: + initialize_imported_app_creator_access(result, tenant_id=current_tenant_id, account_id=current_user.id) result.permission_keys = get_app_permission_keys( current_tenant_id, current_user.id, @@ -153,11 +149,6 @@ class AppImportConfirmApi(Resource): @with_current_user def post(self, current_user: Account | None = None, import_id: str = ""): current_user = current_user if current_user is not None else _current_user_and_tenant_id(None)[0] - redis_key = f"{IMPORT_INFO_REDIS_KEY_PREFIX}{import_id}" - pending_data_raw = redis_client.get(redis_key) - pending_data: PendingData | None = None - if pending_data_raw: - pending_data = PendingData.model_validate_json(pending_data_raw) with Session(db.engine, expire_on_commit=False) as session: import_service = AppDslService(session) @@ -172,18 +163,14 @@ def post(self, current_user: Account | None = None, import_id: str = ""): else: session.commit() - is_created_app = bool( - pending_data - and pending_data.app_id is None - and result.status - in { - ImportStatus.COMPLETED, - ImportStatus.COMPLETED_WITH_WARNINGS, - } - ) + is_created_app = result.app_created and result.status in { + ImportStatus.COMPLETED, + ImportStatus.COMPLETED_WITH_WARNINGS, + } if dify_config.RBAC_ENABLED and is_created_app and result.app_id: current_user, current_tenant_id = _current_user_and_tenant_id(current_user) if current_tenant_id: + initialize_imported_app_creator_access(result, tenant_id=current_tenant_id, account_id=current_user.id) result.permission_keys = get_app_permission_keys( current_tenant_id, current_user.id, diff --git a/api/controllers/console/datasets/rag_pipeline/rag_pipeline_import.py b/api/controllers/console/datasets/rag_pipeline/rag_pipeline_import.py index 5af74390ab1aec..2b23d1d28a6dc6 100644 --- a/api/controllers/console/datasets/rag_pipeline/rag_pipeline_import.py +++ b/api/controllers/console/datasets/rag_pipeline/rag_pipeline_import.py @@ -28,6 +28,7 @@ from libs.login import login_required from models.account import Account from models.dataset import Pipeline +from services.enterprise import rbac_service as enterprise_rbac_service from services.entities.dsl_entities import ImportStatus from services.rag_pipeline.rag_pipeline_dsl_service import RagPipelineDslService @@ -107,6 +108,19 @@ def post(self, req_data: RagPipelineImportPayload, current_user: Account) -> Jso else: session.commit() + if ( + result.status in {ImportStatus.COMPLETED, ImportStatus.COMPLETED_WITH_WARNINGS} + and result.dataset_created + and result.dataset_id + ): + assert current_user.current_tenant_id is not None + enterprise_rbac_service.try_sync_creator_access_policy_member_bindings( + current_user.current_tenant_id, + current_user.id, + enterprise_rbac_service.RBACResourceType.DATASET, + result.dataset_id, + ) + # Return appropriate status code based on result status = result.status match status: @@ -138,6 +152,19 @@ def post(self, current_user: Account, import_id: str) -> JsonResponseWithStatus: else: session.commit() + if ( + result.status in {ImportStatus.COMPLETED, ImportStatus.COMPLETED_WITH_WARNINGS} + and result.dataset_created + and result.dataset_id + ): + assert current_user.current_tenant_id is not None + enterprise_rbac_service.try_sync_creator_access_policy_member_bindings( + current_user.current_tenant_id, + current_user.id, + enterprise_rbac_service.RBACResourceType.DATASET, + result.dataset_id, + ) + # Return appropriate status code based on result if result.status == ImportStatus.FAILED: return dump_response(RagPipelineImportResponse, result), 400 diff --git a/api/controllers/inner_api/app/dsl.py b/api/controllers/inner_api/app/dsl.py index 80e514855ee10f..365e4a318e0d69 100644 --- a/api/controllers/inner_api/app/dsl.py +++ b/api/controllers/inner_api/app/dsl.py @@ -20,7 +20,7 @@ from extensions.ext_database import db from models import Account, App from models.account import AccountStatus -from services.app_dsl_service import AppDslService +from services.app_dsl_service import AppDslService, initialize_imported_app_creator_access from services.entities.dsl_entities import ImportMode, ImportStatus from services.errors.app import IsDraftWorkflowError, WorkflowNotFoundError @@ -83,6 +83,7 @@ def post(self, args: InnerAppDSLImportPayload, workspace_id: str): session.rollback() else: session.commit() + initialize_imported_app_creator_access(result, tenant_id=workspace_id, account_id=account.id) if result.status == ImportStatus.FAILED: return result.model_dump(mode="json"), 400 diff --git a/api/controllers/openapi/app_dsl.py b/api/controllers/openapi/app_dsl.py index 0978c2c628fba9..ecd390e119d14d 100644 --- a/api/controllers/openapi/app_dsl.py +++ b/api/controllers/openapi/app_dsl.py @@ -16,7 +16,7 @@ from libs.oauth_bearer import Scope, TokenType from models import Account, App from models.account import TenantAccountRole -from services.app_dsl_service import AppDslService, Import +from services.app_dsl_service import AppDslService, Import, initialize_imported_app_creator_access from services.entities.dsl_entities import CheckDependenciesResult, ImportStatus from services.errors.account import NoPermissionError from services.errors.app import WorkflowNotFoundError @@ -70,6 +70,7 @@ def post(self, workspace_id: str, *, auth_data: AuthData, body: AppDslImportPayl session.rollback() else: session.commit() + initialize_imported_app_creator_access(result, tenant_id=workspace_id, account_id=account.id) match result.status: case ImportStatus.FAILED: @@ -113,6 +114,7 @@ def post(self, workspace_id: str, import_id: str, *, auth_data: AuthData): session.rollback() else: session.commit() + initialize_imported_app_creator_access(result, tenant_id=workspace_id, account_id=account.id) if result.status == ImportStatus.FAILED: return result, 400 diff --git a/api/services/app_dsl_service.py b/api/services/app_dsl_service.py index 4c9f10729a07d0..7d40e89891d48a 100644 --- a/api/services/app_dsl_service.py +++ b/api/services/app_dsl_service.py @@ -52,6 +52,7 @@ 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 import rbac_service from services.enterprise.enterprise_service import EnterpriseService from services.enterprise.rbac_service import RBACService from services.entities.dsl_entities import ( @@ -86,6 +87,7 @@ class Import(BaseModel): status: ImportStatus app_id: str | None = None app_mode: str | None = None + app_created: bool = Field(default=False, exclude=True) permission_keys: list[str] = Field(default_factory=list) current_dsl_version: str = CURRENT_DSL_VERSION imported_dsl_version: str = "" @@ -93,6 +95,26 @@ class Import(BaseModel): warnings: list[DslImportWarning] = Field(default_factory=list) +def initialize_imported_app_creator_access(result: Import, *, tenant_id: str, account_id: str) -> None: + """Initialize ordinary App creator access only after the caller commits the import.""" + if ( + result.status in (ImportStatus.COMPLETED, ImportStatus.COMPLETED_WITH_WARNINGS) + and result.app_created + and result.app_id + and result.app_mode + in ( + AppMode.CHAT, + AppMode.ADVANCED_CHAT, + AppMode.COMPLETION, + AppMode.WORKFLOW, + AppMode.AGENT_CHAT, + ) + ): + rbac_service.try_sync_creator_access_policy_member_bindings( + tenant_id, account_id, rbac_service.RBACResourceType.APP, result.app_id + ) + + class PendingData(PendingImportOwner): import_mode: str yaml_content: str @@ -152,6 +174,7 @@ def copy_app( self._session.rollback() return result, None self._session.commit() + initialize_imported_app_creator_access(result, tenant_id=tenant_id, account_id=account.id) if result.app_id and SystemFeatureService.is_webapp_auth_enabled(): try: @@ -345,6 +368,7 @@ def import_app( ) # Create or update app + app_created = app is None app = self._create_or_update_app( app=app, data=data, @@ -365,6 +389,7 @@ def import_app( id=import_id, status=result_status, app_id=app.id, + app_created=app_created, app_mode=app.mode, imported_dsl_version=imported_version, warnings=self._warnings, @@ -433,6 +458,7 @@ def confirm_import(self, *, import_id: str, account: Account) -> Import: ) # Create or update app + app_created = app is None app = self._create_or_update_app( app=app, data=data, @@ -451,6 +477,7 @@ def confirm_import(self, *, import_id: str, account: Account) -> Import: id=import_id, status=self._status_with_warnings(ImportStatus.COMPLETED), app_id=app.id, + app_created=app_created, app_mode=app.mode, current_dsl_version=CURRENT_DSL_VERSION, imported_dsl_version=data.get("version", "0.1.0"), diff --git a/api/services/dataset_service.py b/api/services/dataset_service.py index 822642083b939c..dd95694dfb64f9 100644 --- a/api/services/dataset_service.py +++ b/api/services/dataset_service.py @@ -542,7 +542,16 @@ def create_empty_rag_pipeline_dataset( pipeline_id=pipeline.id, ) session.add(dataset) + session.flush() + dataset_id = dataset.id + creator_id = current_user.id session.commit() + enterprise_rbac_service.try_sync_creator_access_policy_member_bindings( + tenant_id, + creator_id, + enterprise_rbac_service.RBACResourceType.DATASET, + dataset_id, + ) return dataset @staticmethod diff --git a/api/services/enterprise/rbac_service.py b/api/services/enterprise/rbac_service.py index 0edeb9485aff3a..6210bb6ec2aeb7 100644 --- a/api/services/enterprise/rbac_service.py +++ b/api/services/enterprise/rbac_service.py @@ -925,15 +925,54 @@ def try_sync_creator_access_policy_member_bindings( resource_type: RBACResourceType | str, resource_id: str, ) -> None: + """Best-effort creator bindings, including app and dataset whitelist visibility gates.""" if not dify_config.RBAC_ENABLED: return try: - RBACService.AccessPolicies.sync_creator_access_policy_member_bindings( + bindings = RBACService.AccessPolicies.sync_creator_access_policy_member_bindings( tenant_id, account_id, resource_type=resource_type, resource_id=resource_id, ) + if resource_type in (RBACResourceType.APP, RBACResourceType.DATASET): + policy_ids = sorted( + { + binding.access_policy_id + for binding in bindings + if binding.tenant_id == tenant_id + and binding.resource_type == resource_type + and binding.resource_id == resource_id + and binding.account_id == account_id + and binding.access_policy_id + } + ) + if not policy_ids: + raise ValueError("Creator sync returned no matching resource policy binding") + # Policy bindings alone do not satisfy the resource whitelist gate. + # Append the returned creator policies without replacing other members or policies. + if resource_type == RBACResourceType.APP: + RBACService.AppAccess.append_whitelist_members_batch( + tenant_id, + account_id, + [ + AppendAppWhitelistMembersBatchItem( + app_id=resource_id, account_ids=[account_id], policy_id=policy_id + ) + for policy_id in policy_ids + ], + ) + else: + RBACService.DatasetAccess.append_whitelist_members_batch( + tenant_id, + account_id, + [ + AppendDatasetWhitelistMembersBatchItem( + dataset_id=resource_id, account_ids=[account_id], policy_id=policy_id + ) + for policy_id in policy_ids + ], + ) except Exception: logger.warning( "Failed to sync creator access policy member binding for " diff --git a/api/services/rag_pipeline/rag_pipeline_dsl_service.py b/api/services/rag_pipeline/rag_pipeline_dsl_service.py index 1e655e510db09e..51fe0b71b08518 100644 --- a/api/services/rag_pipeline/rag_pipeline_dsl_service.py +++ b/api/services/rag_pipeline/rag_pipeline_dsl_service.py @@ -13,7 +13,7 @@ from Crypto.Cipher import AES from Crypto.Util.Padding import pad, unpad from flask_login import current_user -from pydantic import BaseModel +from pydantic import BaseModel, Field from sqlalchemy import select from sqlalchemy.orm import Session @@ -68,6 +68,8 @@ class RagPipelineImportInfo(BaseModel): imported_dsl_version: str = "" error: str = "" dataset_id: str | None = None + # Internal creation outcome for post-commit authorization; never part of the DSL response. + dataset_created: bool = Field(default=False, exclude=True) class RagPipelinePendingData(PendingImportOwner): @@ -274,6 +276,7 @@ def import_rag_pipeline( graph = workflow.get("graph", {}) nodes = graph.get("nodes", []) dataset_id = None + dataset_created = False for node in nodes: if node.get("data", {}).get("type") == KNOWLEDGE_INDEX_NODE_TYPE: knowledge_configuration = KnowledgeConfiguration.model_validate(node.get("data", {})) @@ -284,6 +287,7 @@ def import_rag_pipeline( ): raise ValueError("Chunk structure is not compatible with the published pipeline") if not dataset: + dataset_created = True datasets = self._session.scalars( select(Dataset).where(Dataset.tenant_id == account.current_tenant_id) ).all() @@ -349,6 +353,7 @@ def import_rag_pipeline( status=status, pipeline_id=pipeline.id, dataset_id=dataset_id, + dataset_created=dataset_created, imported_dsl_version=imported_version, ) @@ -427,10 +432,12 @@ def confirm_import(self, *, import_id: str, account: Account) -> RagPipelineImpo graph = workflow.get("graph", {}) nodes = graph.get("nodes", []) dataset_id = None + dataset_created = False for node in nodes: if node.get("data", {}).get("type") == KNOWLEDGE_INDEX_NODE_TYPE: knowledge_configuration = KnowledgeConfiguration.model_validate(node.get("data", {})) if not dataset: + dataset_created = True dataset = Dataset( tenant_id=account.current_tenant_id, name=name, @@ -499,6 +506,7 @@ def confirm_import(self, *, import_id: str, account: Account) -> RagPipelineImpo status=ImportStatus.COMPLETED, pipeline_id=pipeline.id, dataset_id=dataset_id, + dataset_created=dataset_created, current_dsl_version=CURRENT_DSL_VERSION, imported_dsl_version=data.get("version", "0.1.0"), ) 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 96739b27060c11..c2886ee308290a 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 @@ -17,7 +17,8 @@ from models.base import TypeBase from models.engine import db from models.model import App, AppMode -from services.app_dsl_service import ImportStatus +from services.app_dsl_service import Import, ImportStatus +from services.enterprise import rbac_service from services.entities.dsl_entities import CheckDependenciesResult from tests.unit_tests.config_override import apply_config_overrides @@ -37,9 +38,12 @@ def __init__( status: ImportStatus, app_id: str | None = "app-1", permission_keys: list[str] | None = None, + app_created: bool = False, ): self.status = status self.app_id = app_id + self.app_mode = AppMode.WORKFLOW + self.app_created = app_created self.permission_keys = permission_keys or [] def model_dump(self, mode: str = "json"): @@ -128,6 +132,17 @@ def _return_result(import_service: app_import_module.AppDslService, *_args, **_k return app_id +def _install_creator_sync(monkeypatch: pytest.MonkeyPatch, transaction_events: TransactionEvents) -> list[tuple]: + """Record creator sync calls with the number of commits observed at call time.""" + calls: list[tuple] = [] + + def _sync(*args) -> None: + calls.append((*args, transaction_events.commits)) + + monkeypatch.setattr(rbac_service, "try_sync_creator_access_policy_member_bindings", _sync) + return calls + + def _assert_app_persistence(sqlite_app_engine: Engine, app_id: str, *, persisted: bool) -> None: with Session(sqlite_app_engine) as session: assert (session.get(App, app_id) is not None) is persisted @@ -239,8 +254,9 @@ def test_import_post_attaches_permission_keys_when_creating_new_app_and_rbac_ena app_id = _install_persisting_service_result( monkeypatch, method_name="import_app", - result=_Result(ImportStatus.COMPLETED, app_id="app-123"), + result=_Result(ImportStatus.COMPLETED, app_id="app-123", app_created=True), ) + sync_calls = _install_creator_sync(monkeypatch, transaction_events) monkeypatch.setattr( app_import_module, "get_app_permission_keys", @@ -251,6 +267,7 @@ def test_import_post_attaches_permission_keys_when_creating_new_app_and_rbac_ena response, status = method(app_import_module.AppImportPayload(mode="yaml-content")) assert transaction_events.commits == 1 + assert sync_calls == [("tenant-1", "u1", rbac_service.RBACResourceType.APP, "app-123", 1)] _assert_app_persistence(sqlite_app_engine, app_id, persisted=True) assert status == 200 assert response["permission_keys"] == ["app.acl.view_layout", "app.acl.edit"] @@ -277,6 +294,7 @@ def test_import_post_does_not_attach_permission_keys_when_overwriting_existing_a method_name="import_app", result=_Result(ImportStatus.COMPLETED, app_id="app-123"), ) + sync_calls = _install_creator_sync(monkeypatch, transaction_events) monkeypatch.setattr( app_import_module, "get_app_permission_keys", @@ -291,6 +309,7 @@ def test_import_post_does_not_attach_permission_keys_when_overwriting_existing_a response, status = method(app_import_module.AppImportPayload(mode="yaml-content", app_id="existing-app")) assert transaction_events.commits == 1 + assert sync_calls == [] _assert_app_persistence(sqlite_app_engine, app_id, persisted=True) assert status == 200 assert response["permission_keys"] == [] @@ -341,20 +360,13 @@ def test_import_confirm_attaches_permission_keys_when_creating_new_app_and_rbac_ "current_account_with_tenant", lambda: (_make_account(), "tenant-1"), ) - redis_get = MagicMock( - return_value=( - b'{"tenant_id":"tenant-1","account_id":"u1","import_mode":"yaml-content",' - b'"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) apply_config_overrides(monkeypatch, RBAC_ENABLED=True) app_id = _install_persisting_service_result( monkeypatch, method_name="confirm_import", - result=_Result(ImportStatus.COMPLETED, app_id="app-456"), + result=_Result(ImportStatus.COMPLETED, app_id="app-456", app_created=True), ) + sync_calls = _install_creator_sync(monkeypatch, transaction_events) monkeypatch.setattr( app_import_module, "get_app_permission_keys", @@ -368,7 +380,7 @@ def test_import_confirm_attaches_permission_keys_when_creating_new_app_and_rbac_ _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") + assert sync_calls == [("tenant-1", "u1", rbac_service.RBACResourceType.APP, "app-456", 1)] def test_import_confirm_does_not_attach_permission_keys_when_overwriting_existing_app( self, @@ -385,21 +397,13 @@ def test_import_confirm_does_not_attach_permission_keys_when_overwriting_existin "current_account_with_tenant", lambda: (_make_account(), "tenant-1"), ) - monkeypatch.setattr( - app_import_module.redis_client, - "get", - lambda *_args, **_kwargs: ( - b'{"tenant_id":"tenant-1","account_id":"u1","import_mode":"yaml-content",' - b'"yaml_content":"app: {}","app_id":"existing-app",' - b'"name":null,"description":null,"icon_type":null,"icon":null,"icon_background":null}' - ), - ) apply_config_overrides(monkeypatch, RBAC_ENABLED=True) app_id = _install_persisting_service_result( monkeypatch, method_name="confirm_import", result=_Result(ImportStatus.COMPLETED, app_id="app-456"), ) + sync_calls = _install_creator_sync(monkeypatch, transaction_events) monkeypatch.setattr( app_import_module, "get_app_permission_keys", @@ -410,6 +414,7 @@ def test_import_confirm_does_not_attach_permission_keys_when_overwriting_existin response, status = method(import_id="import-1") assert transaction_events.commits == 1 + assert sync_calls == [] _assert_app_persistence(sqlite_app_engine, app_id, persisted=True) assert status == 200 assert response["permission_keys"] == [] @@ -434,3 +439,7 @@ def test_import_check_dependencies_returns_result( assert status == 200 assert response["leaked_dependencies"] == [] + + +def test_import_response_schema_hides_creation_marker() -> None: + assert "app_created" not in app_import_module.console_ns.models[Import.__name__].__schema__["properties"] diff --git a/api/tests/unit_tests/controllers/console/datasets/rag_pipeline/test_rag_pipeline_import.py b/api/tests/unit_tests/controllers/console/datasets/rag_pipeline/test_rag_pipeline_import.py index c2e80c24bd5c49..be7296d5eeee03 100644 --- a/api/tests/unit_tests/controllers/console/datasets/rag_pipeline/test_rag_pipeline_import.py +++ b/api/tests/unit_tests/controllers/console/datasets/rag_pipeline/test_rag_pipeline_import.py @@ -2,13 +2,15 @@ from __future__ import annotations -from collections.abc import Iterator +from collections.abc import Callable, Iterator from inspect import unwrap from unittest.mock import MagicMock, patch import pytest from flask import Flask +from sqlalchemy.orm import Session +from controllers.common.schema import JsonResponseWithStatus from controllers.console import console_ns from controllers.console.datasets.rag_pipeline.rag_pipeline_import import ( IncludeSecretQuery, @@ -19,9 +21,10 @@ RagPipelineImportPayload, ) from core.plugin.entities.plugin import PluginDependency, PluginDependencyType -from models.account import Account +from models.account import Account, Tenant from models.dataset import Pipeline from models.engine import db +from services.enterprise import rbac_service from services.entities.dsl_entities import CheckDependenciesResult, ImportStatus from services.rag_pipeline.rag_pipeline_dsl_service import RagPipelineImportInfo @@ -300,3 +303,105 @@ def test_get_with_include_secret(self, app: Flask) -> None: assert status == 200 assert response == {"data": "yaml: data"} + + +@pytest.mark.parametrize("confirm", [False, True]) +@pytest.mark.parametrize( + ("import_status", "created", "dataset_id", "rbac_enabled", "commit_fails", "expected_calls"), + [ + (ImportStatus.COMPLETED, True, "dataset-1", True, False, 1), + (ImportStatus.COMPLETED_WITH_WARNINGS, True, "dataset-1", True, False, 1), + (ImportStatus.COMPLETED, False, "dataset-1", True, False, 0), + (ImportStatus.PENDING, True, "dataset-1", True, False, 0), + (ImportStatus.FAILED, True, "dataset-1", True, False, 0), + (ImportStatus.COMPLETED, True, None, True, False, 0), + (ImportStatus.COMPLETED, True, "dataset-1", False, False, 0), + (ImportStatus.COMPLETED, True, "dataset-1", True, True, 0), + ], +) +def test_creator_binding_only_after_successful_new_dataset_commit( + app: Flask, + config_overrides: Callable[..., None], + confirm: bool, + import_status: ImportStatus, + created: bool, + dataset_id: str | None, + rbac_enabled: bool, + commit_fails: bool, + expected_calls: int, +) -> None: + config_overrides(RBAC_ENABLED=rbac_enabled) + user = _account() + user.id = "creator-1" + tenant = Tenant(name="Test Tenant") + tenant.id = "tenant-1" + user._current_tenant = tenant + result = RagPipelineImportInfo.model_validate( + { + "id": "import-1", + "status": import_status, + "dataset_id": dataset_id, + "dataset_created": created, + } + ) + service = MagicMock() + service.import_rag_pipeline.return_value = result + service.confirm_import.return_value = result + events: list[str] = [] + session = Session(db.engine, expire_on_commit=False) + commit = session.commit + close = session.close + + def commit_session() -> None: + if commit_fails: + raise RuntimeError("commit failed") + commit() + events.append("commit") + + def close_session() -> None: + close() + events.append("close") + + def sync_binding(*_args: object, **_kwargs: object) -> list[object]: + events.append("sync") + return [] + + with ( + app.test_request_context("/"), + patch("controllers.console.datasets.rag_pipeline.rag_pipeline_import.Session", return_value=session), + patch( + "controllers.console.datasets.rag_pipeline.rag_pipeline_import.RagPipelineDslService", return_value=service + ), + patch.object(session, "commit", side_effect=commit_session), + patch.object(session, "close", side_effect=close_session), + patch.object( + rbac_service.RBACService.AccessPolicies, + "sync_creator_access_policy_member_bindings", + side_effect=sync_binding, + ) as sync, + patch.object(rbac_service.RBACService.DatasetAccess, "replace_whitelist") as replace, + ): + + def invoke() -> JsonResponseWithStatus: + if confirm: + api = RagPipelineImportConfirmApi() + return unwrap(api.post)(api, user, "import-1") + api = RagPipelineImportApi() + return unwrap(api.post)(api, RagPipelineImportPayload(mode="yaml-content"), user) + + if commit_fails: + with pytest.raises(RuntimeError, match="commit failed"): + invoke() + else: + response, _ = invoke() + assert "dataset_created" not in response + assert sync.call_count == expected_calls + if expected_calls: + assert events == ["commit", "close", "sync"] + sync.assert_called_once_with( + "tenant-1", + "creator-1", + resource_type=rbac_service.RBACResourceType.DATASET, + resource_id="dataset-1", + ) + replace.assert_not_called() diff --git a/api/tests/unit_tests/services/enterprise/test_rbac_service.py b/api/tests/unit_tests/services/enterprise/test_rbac_service.py index 1d76cc79ff2ad3..e16480486412f5 100644 --- a/api/tests/unit_tests/services/enterprise/test_rbac_service.py +++ b/api/tests/unit_tests/services/enterprise/test_rbac_service.py @@ -8,6 +8,7 @@ from __future__ import annotations +from collections.abc import Callable from types import SimpleNamespace from unittest.mock import MagicMock, patch @@ -1168,3 +1169,138 @@ def test_legacy_workspace_keys_no_longer_carry_agent_manage(self): ): assert "agent.manage" not in keys assert {"agent.acl.preview", "agent.acl.access_point_view"} <= set(keys) + + +class TestCreatorDatasetAccess: + @staticmethod + def binding(**overrides: str) -> dict[str, str]: + return { + "id": "binding-1", + "tenant_id": "tenant-1", + "access_policy_id": "creator-policy", + "resource_type": "dataset", + "resource_id": "dataset-1", + "account_id": "creator-1", + **overrides, + } + + def test_dataset_creator_is_appended_with_returned_policy( + self, mock_send: MagicMock, config_overrides: Callable[..., None] + ) -> None: + config_overrides(RBAC_ENABLED=True) + mock_send.side_effect = [{"data": [self.binding()]}, dict[str, str]()] + + svc.try_sync_creator_access_policy_member_bindings("tenant-1", "creator-1", "dataset", "dataset-1") + + assert mock_send.call_count == 2 + sync, append = mock_send.call_args_list + assert sync.args == ("PUT", "/rbac/access-policies/creator-member-bindings") + assert append.args == ("POST", "/rbac/datasets/whitelist/members/batch") + assert append.kwargs["tenant_id"] == "tenant-1" + assert append.kwargs["account_id"] == "creator-1" + assert append.kwargs["json"] == { + "data": [{"dataset_id": "dataset-1", "account_ids": ["creator-1"], "policy_id": "creator-policy"}] + } + + @pytest.mark.parametrize( + "overrides", + [ + {"tenant_id": "other"}, + {"resource_id": "other"}, + {"account_id": "other"}, + {"resource_type": "app"}, + {"access_policy_id": ""}, + ], + ) + def test_mismatched_binding_does_not_grant_access( + self, + mock_send: MagicMock, + config_overrides: Callable[..., None], + overrides: dict[str, str], + caplog: pytest.LogCaptureFixture, + ) -> None: + config_overrides(RBAC_ENABLED=True) + mock_send.return_value = {"data": [self.binding(**overrides)]} + svc.try_sync_creator_access_policy_member_bindings("tenant-1", "creator-1", "dataset", "dataset-1") + mock_send.assert_called_once() + assert "Failed to sync creator access" in caplog.text + + def test_empty_response_does_not_guess_a_policy( + self, mock_send: MagicMock, config_overrides: Callable[..., None], caplog: pytest.LogCaptureFixture + ) -> None: + config_overrides(RBAC_ENABLED=True) + mock_send.return_value = {"data": list[dict[str, str]]()} + svc.try_sync_creator_access_policy_member_bindings("tenant-1", "creator-1", "dataset", "dataset-1") + mock_send.assert_called_once() + assert "Failed to sync creator access" in caplog.text + + @pytest.mark.parametrize("resource_type", ["agent"]) + def test_other_resource_types_keep_existing_behavior( + self, mock_send: MagicMock, config_overrides: Callable[..., None], resource_type: str + ) -> None: + config_overrides(RBAC_ENABLED=True) + mock_send.return_value = {"data": list[dict[str, str]]()} + svc.try_sync_creator_access_policy_member_bindings("tenant-1", "creator-1", resource_type, "resource-1") + mock_send.assert_called_once() + + def test_disabled_rbac_makes_no_request(self, mock_send: MagicMock, config_overrides: Callable[..., None]) -> None: + config_overrides(RBAC_ENABLED=False) + svc.try_sync_creator_access_policy_member_bindings("tenant-1", "creator-1", "dataset", "dataset-1") + mock_send.assert_not_called() + + @pytest.mark.parametrize("failure_stage", ["sync", "append"]) + def test_remote_failures_are_logged( + self, + mock_send: MagicMock, + config_overrides: Callable[..., None], + caplog: pytest.LogCaptureFixture, + failure_stage: str, + ) -> None: + config_overrides(RBAC_ENABLED=True) + error = RuntimeError("RBAC unavailable") + mock_send.side_effect = [error] if failure_stage == "sync" else [{"data": [self.binding()]}, error] + svc.try_sync_creator_access_policy_member_bindings("tenant-1", "creator-1", "dataset", "dataset-1") + assert mock_send.call_count == (1 if failure_stage == "sync" else 2) + assert "Failed to sync creator access" in caplog.text + + def test_only_matching_policies_are_appended_once( + self, mock_send: MagicMock, config_overrides: Callable[..., None] + ) -> None: + config_overrides(RBAC_ENABLED=True) + mock_send.side_effect = [ + { + "data": [ + self.binding(), + self.binding(), + self.binding(access_policy_id="another-creator-policy"), + self.binding(account_id="another-account", access_policy_id="unrelated-policy"), + ] + }, + dict[str, str](), + ] + svc.try_sync_creator_access_policy_member_bindings( + "tenant-1", "creator-1", svc.RBACResourceType.DATASET, "dataset-1" + ) + assert mock_send.call_count == 2 + assert mock_send.call_args.kwargs["json"] == { + "data": [ + {"dataset_id": "dataset-1", "account_ids": ["creator-1"], "policy_id": policy} + for policy in ["another-creator-policy", "creator-policy"] + ] + } + + +class TestCreatorAppAccess: + binding = staticmethod(TestCreatorDatasetAccess.binding) + + def test_app_creator_is_appended_with_returned_policy( + self, mock_send: MagicMock, config_overrides: Callable[..., None] + ) -> None: + config_overrides(RBAC_ENABLED=True) + mock_send.side_effect = [{"data": [self.binding(resource_type="app", resource_id="app-1")]}, dict[str, str]()] + svc.try_sync_creator_access_policy_member_bindings("tenant-1", "creator-1", "app", "app-1") + assert mock_send.call_count == 2 + assert mock_send.call_args.args == ("POST", "/rbac/apps/whitelist/members/batch") + assert mock_send.call_args.kwargs["json"] == { + "data": [{"app_id": "app-1", "account_ids": ["creator-1"], "policy_id": "creator-policy"}] + } 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 44859fb7e2d309..a8773208ff61f7 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 @@ -8,7 +8,7 @@ from __future__ import annotations import json -from collections.abc import Generator +from collections.abc import Generator, Sequence from contextlib import contextmanager from types import SimpleNamespace from typing import Any @@ -656,6 +656,7 @@ 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 result.dataset_created is False 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" @@ -1035,3 +1036,83 @@ def test_create_dataset_name_is_tenant_scoped_and_ignores_template_rows( permission="only_me", ), ) + + +@pytest.mark.parametrize("confirm", [False, True]) +@pytest.mark.parametrize("existing", ["none", "pipeline", "dataset"]) +def test_import_reports_actual_dataset_creation( + monkeypatch: pytest.MonkeyPatch, + service: RagPipelineDslService, + sqlite_session: Session, + confirm: bool, + existing: str, +) -> None: + pipeline_id = None + if existing != "none": + pipeline = _pipeline(sqlite_session) + if existing == "dataset": + _dataset(sqlite_session, pipeline) + _workflow(sqlite_session, pipeline) + pipeline_id = pipeline.id + monkeypatch.setattr(module.KnowledgeConfiguration, "model_validate", Mock(return_value=_knowledge_configuration())) + if confirm: + pending = RagPipelinePendingData( + tenant_id="tenant-1", + account_id="account-1", + import_mode=ImportMode.YAML_CONTENT.value, + yaml_content=_valid_dsl(), + pipeline_id=pipeline_id, + ) + monkeypatch.setattr(module.redis_client, "get", Mock(return_value=pending.model_dump_json())) + monkeypatch.setattr(module.redis_client, "delete", Mock()) + result = service.confirm_import(import_id="import-1", account=_account()) + else: + result = service.import_rag_pipeline( + account=_account(), + import_mode=ImportMode.YAML_CONTENT.value, + yaml_content=_valid_dsl(), + pipeline_id=pipeline_id, + ) + assert result.status == ImportStatus.COMPLETED + assert result.dataset_created is (existing != "dataset") + assert "dataset_created" not in result.model_dump() + assert "dataset_created" not in json.loads(result.model_dump_json()) + assert sqlite_session.in_transaction() + + +@pytest.mark.parametrize("confirm", [False, True]) +def test_failed_import_does_not_report_created_dataset( + monkeypatch: pytest.MonkeyPatch, + service: RagPipelineDslService, + sqlite_session: Session, + confirm: bool, +) -> None: + monkeypatch.setattr(module.KnowledgeConfiguration, "model_validate", Mock(return_value=_knowledge_configuration())) + flush = sqlite_session.flush + + def fail_after_dataset_flush(objects: Sequence[object] | None = None) -> None: + created_dataset = any(isinstance(obj, Dataset) for obj in sqlite_session.new) + flush(objects) + if created_dataset: + raise RuntimeError("dataset flush failed") + + monkeypatch.setattr(sqlite_session, "flush", fail_after_dataset_flush) + if confirm: + pending = RagPipelinePendingData( + tenant_id="tenant-1", + account_id="account-1", + import_mode=ImportMode.YAML_CONTENT.value, + yaml_content=_valid_dsl(), + pipeline_id=None, + ) + monkeypatch.setattr(module.redis_client, "get", Mock(return_value=pending.model_dump_json())) + result = service.confirm_import(import_id="import-1", account=_account()) + else: + result = service.import_rag_pipeline( + account=_account(), + import_mode=ImportMode.YAML_CONTENT.value, + yaml_content=_valid_dsl(), + ) + assert result.status == ImportStatus.FAILED + assert result.error == "dataset flush failed" + assert result.dataset_created is False 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 86a410c665e5ca..d74ca635a0bf7b 100644 --- a/api/tests/unit_tests/services/test_app_dsl_service.py +++ b/api/tests/unit_tests/services/test_app_dsl_service.py @@ -9,6 +9,7 @@ from sqlalchemy import event, select from sqlalchemy.orm import Session, sessionmaker +from constants.dsl_version import CURRENT_APP_DSL_VERSION from core.rbac import RBACPermission, RBACResourceScope from core.workflow.llm_environment_variable import LLMEnvironmentVariable from models import Account, App, AppMode, Tenant @@ -144,7 +145,10 @@ def test_copy_app_commits_before_inheriting_access( def import_app(**_kwargs: object) -> Import: sqlite_session.add(_app(app_id=copied_id, tenant_id=_TENANT_ID)) - return Import(id="import-1", status=status, app_id=copied_id) + return Import(id="import-1", status=status, app_id=copied_id, app_mode=AppMode.WORKFLOW, app_created=True) + + def sync_creator(*_args: object) -> None: + assert not sqlite_session.in_transaction() def get_access_mode(app_id: str) -> SimpleNamespace: assert not sqlite_session.in_transaction() @@ -163,13 +167,18 @@ def update_access_mode(app_id: str, access_mode: str) -> None: monkeypatch.setattr(EnterpriseService.WebAppAuth, "get_app_access_mode_by_id", get_access_mode) update_access = Mock(side_effect=update_access_mode) monkeypatch.setattr(EnterpriseService.WebAppAuth, "update_app_access_mode", update_access) + sync = Mock(side_effect=sync_creator) + monkeypatch.setattr("services.enterprise.rbac_service.try_sync_creator_access_policy_member_bindings", sync) - result, copied = service.copy_app(app_model=original, account=_account(tenant_id=_TENANT_ID), tenant_id=_TENANT_ID) + result, copied = service.copy_app( + app_model=original, account=_account(account_id=_CALLER_ID, tenant_id=_TENANT_ID), tenant_id=_TENANT_ID + ) assert result.status == status assert copied is not None assert copied.id == copied_id update_access.assert_called_once() + sync.assert_called_once_with(_TENANT_ID, _CALLER_ID, "app", copied_id) def test_copy_app_scopes_result_to_current_tenant(sqlite_session: Session, monkeypatch: pytest.MonkeyPatch) -> None: @@ -795,3 +804,92 @@ def test_append_workflow_export_data_reports_missing_selected_workflow( session=unbound_session, workflow_id=workflow_id, ) + + +@pytest.mark.parametrize("confirmed", [False, True]) +@pytest.mark.parametrize("overwrite", [False, True]) +def test_import_creation_marker_tracks_persisted_app( + sqlite_session: Session, monkeypatch: pytest.MonkeyPatch, confirmed: bool, overwrite: bool +) -> None: + account = _account(account_id=_CALLER_ID, tenant_id=_TENANT_ID) + target = _persist_overwrite_target(sqlite_session) if overwrite else None + data = yaml.safe_dump( + { + "version": CURRENT_APP_DSL_VERSION, + "kind": "app", + "app": {"name": "Imported", "mode": "workflow"}, + "workflow": {"graph": {"nodes": [], "edges": []}, "features": {}}, + } + ) + monkeypatch.setattr("services.app_dsl_service.WorkflowService", Mock()) + service = AppDslService(sqlite_session) + if confirmed: + pending = PendingData( + tenant_id=_TENANT_ID, + account_id=_CALLER_ID, + import_mode="yaml-content", + yaml_content=data, + app_id=_OVERWRITE_APP_ID if overwrite else None, + ) + monkeypatch.setattr("services.app_dsl_service.redis_client.get", lambda _: pending.model_dump_json()) + result = service.confirm_import(import_id="pending", account=account) + else: + result = service.import_app( + account=account, + import_mode="yaml-content", + yaml_content=data, + app_id=_OVERWRITE_APP_ID if overwrite else None, + ) + assert result.status == ImportStatus.COMPLETED, result.error + assert result.app_created is (not overwrite) + assert sqlite_session.get(App, result.app_id) is not None + assert "app_created" not in result.model_dump() + assert "app_created" not in result.model_json_schema(mode="serialization")["properties"] + + +@pytest.mark.parametrize("mode", list(AppMode) + [None]) +def test_import_creator_access_only_initializes_ordinary_apps( + mode: AppMode | None, monkeypatch: pytest.MonkeyPatch +) -> None: + from services.app_dsl_service import initialize_imported_app_creator_access + + sync = Mock() + monkeypatch.setattr("services.enterprise.rbac_service.try_sync_creator_access_policy_member_bindings", sync) + result = Import(id="import", status=ImportStatus.COMPLETED, app_id="app", app_mode=mode, app_created=True) + initialize_imported_app_creator_access(result, tenant_id="tenant", account_id="creator") + assert sync.call_count == ( + 1 + if mode in {AppMode.CHAT, AppMode.COMPLETION, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT, AppMode.WORKFLOW} + else 0 + ) + + +@pytest.mark.parametrize("confirmed", [False, True]) +def test_failed_import_does_not_report_creation( + sqlite_session: Session, monkeypatch: pytest.MonkeyPatch, confirmed: bool +) -> None: + data = "version: " + CURRENT_APP_DSL_VERSION + "\nkind: app\napp: {name: Broken, mode: chat}\n" + service = AppDslService(sqlite_session) + if confirmed: + pending = PendingData( + tenant_id="tenant-1", account_id="account-1", import_mode="yaml-content", yaml_content=data + ) + monkeypatch.setattr("services.app_dsl_service.redis_client.get", lambda _: pending.model_dump_json()) + result = service.confirm_import(import_id="pending", account=_account()) + else: + result = service.import_app(account=_account(), import_mode="yaml-content", yaml_content=data) + assert result.status == ImportStatus.FAILED + assert not result.app_created + + +def test_pending_import_does_not_report_creation_or_cache_marker( + unbound_session: Session, monkeypatch: pytest.MonkeyPatch +) -> None: + cache = Mock() + monkeypatch.setattr("services.app_dsl_service.redis_client.setex", cache) + result = AppDslService(unbound_session).import_app( + account=_account(), import_mode="yaml-content", yaml_content=_PENDING_WORKFLOW_DSL + ) + assert result.status == ImportStatus.PENDING + assert not result.app_created + assert "app_created" not in cache.call_args.args[2] diff --git a/api/tests/unit_tests/services/test_dataset_service_dataset.py b/api/tests/unit_tests/services/test_dataset_service_dataset.py index b1c7608c2862f9..a5449fc8c354e5 100644 --- a/api/tests/unit_tests/services/test_dataset_service_dataset.py +++ b/api/tests/unit_tests/services/test_dataset_service_dataset.py @@ -30,6 +30,7 @@ Pipeline, ) from services.dataset_service import DatasetCollectionBindingService, DatasetPermissionService, DatasetService +from services.enterprise import rbac_service from services.entities.knowledge_entities.rag_pipeline_entities import ( IconInfo, RagPipelineDatasetCreateEntity, @@ -1213,3 +1214,67 @@ def fail_execute(_orm_execute_state) -> None: event.remove(sqlite_session, "do_orm_execute", fail_execute) assert transaction_events == [] + + +@pytest.mark.parametrize("rbac_enabled", [False, True]) +@pytest.mark.parametrize("commit_fails", [False, True]) +@pytest.mark.parametrize("expire_on_commit", [False, True]) +def test_empty_pipeline_syncs_creator_after_commit( + sqlite_session: Session, + sqlite_session_factory: sessionmaker[Session], + config_overrides: Callable[..., None], + rbac_enabled: bool, + commit_fails: bool, + expire_on_commit: bool, +) -> None: + config_overrides(RBAC_ENABLED=rbac_enabled) + sqlite_session.expire_on_commit = expire_on_commit + account = _account() + entity = RagPipelineDatasetCreateEntity( + name="New pipeline", + description="", + permission=DatasetPermissionEnum.ONLY_ME, + icon_info=IconInfo(icon_type="emoji", icon="📚", icon_background="#FFFFFF", icon_url=None), + ) + commit = sqlite_session.commit + observed: list[tuple[str, str, object]] = [] + transactions_at_sync: list[bool] = [] + + def commit_session() -> None: + if commit_fails: + raise RuntimeError("commit failed") + commit() + + def sync_binding(_tenant_id: str, _account_id: str, *, resource_type: object, resource_id: str) -> list[object]: + transactions_at_sync.append(sqlite_session.in_transaction()) + with sqlite_session_factory() as observer: + stored = observer.get(Dataset, resource_id) + if stored is not None: + observed.append((stored.tenant_id, stored.created_by, resource_type)) + return [] + + with ( + patch("services.dataset_service.current_user", account), + patch.object(sqlite_session, "commit", side_effect=commit_session), + patch.object( + rbac_service.RBACService.AccessPolicies, + "sync_creator_access_policy_member_bindings", + side_effect=sync_binding, + ) as sync, + ): + if commit_fails: + with pytest.raises(RuntimeError, match="commit failed"): + DatasetService.create_empty_rag_pipeline_dataset("tenant-1", entity, sqlite_session) + else: + dataset = DatasetService.create_empty_rag_pipeline_dataset("tenant-1", entity, sqlite_session) + if rbac_enabled: + assert transactions_at_sync == [False] + assert observed == [("tenant-1", account.id, rbac_service.RBACResourceType.DATASET)] + sync.assert_called_once_with( + "tenant-1", + account.id, + resource_type=rbac_service.RBACResourceType.DATASET, + resource_id=dataset.id, + ) + if commit_fails or not rbac_enabled: + sync.assert_not_called()