From 676dff04c833a082660afc404f6c18912a327cf7 Mon Sep 17 00:00:00 2001 From: Escape0707 Date: Fri, 31 Jul 2026 12:20:21 +0900 Subject: [PATCH] test: exercise credential visibility with SQLite (#39827) --- api/tests/unit_tests/pyrefly.toml | 1 - .../test_credential_permission_service.py | 150 +++++++++++++----- 2 files changed, 106 insertions(+), 45 deletions(-) diff --git a/api/tests/unit_tests/pyrefly.toml b/api/tests/unit_tests/pyrefly.toml index 027947d39c0..6181dae49c6 100644 --- a/api/tests/unit_tests/pyrefly.toml +++ b/api/tests/unit_tests/pyrefly.toml @@ -836,7 +836,6 @@ project-excludes = [ "services/test_clear_free_plan_tenant_expired_logs.py", "services/test_code_based_extension_service.py", "services/test_conversation_service.py", - "services/test_credential_permission_service.py", "services/test_credit_pool_service.py", "services/test_dataset_service_dataset.py", "services/test_dataset_service_document.py", diff --git a/api/tests/unit_tests/services/test_credential_permission_service.py b/api/tests/unit_tests/services/test_credential_permission_service.py index d4e8596b14c..cc416969bc9 100644 --- a/api/tests/unit_tests/services/test_credential_permission_service.py +++ b/api/tests/unit_tests/services/test_credential_permission_service.py @@ -5,42 +5,66 @@ and admin bypass behavior. """ from types import SimpleNamespace -from unittest.mock import MagicMock +from typing import cast from uuid import uuid4 import pytest from sqlalchemy import select from sqlalchemy.orm import Session +from core.plugin.entities.plugin_daemon import CredentialType as TriggerCredentialType +from models.account import Account from models.credential_permission import CredentialPermission, CredentialType +from models.enums import PermissionEnum +from models.trigger import TriggerSubscription from services.credential_permission_service import CredentialPermissionService -pytestmark = [ - pytest.mark.usefixtures("sqlite_session"), - pytest.mark.parametrize("sqlite_session", [(CredentialPermission,)], indirect=True), -] - @pytest.fixture -def tenant_id(): +def tenant_id() -> str: return str(uuid4()) @pytest.fixture -def user_id(): +def user_id() -> str: return str(uuid4()) @pytest.fixture -def other_user_id(): +def other_user_id() -> str: return str(uuid4()) @pytest.fixture -def credential_id(): +def credential_id() -> str: return str(uuid4()) +def _subscription( + *, + tenant_id: str, + owner_id: str, + name: str, + visibility: PermissionEnum, +) -> TriggerSubscription: + return TriggerSubscription( + name=name, + tenant_id=tenant_id, + user_id=owner_id, + provider_id="test/provider", + endpoint_id=f"{name}-endpoint", + parameters={}, + properties={}, + credentials={}, + credential_type=TriggerCredentialType.API_KEY, + visibility=visibility, + ) + + +def _user(user_id: str, *, is_admin: bool) -> Account: + return cast(Account, SimpleNamespace(id=user_id, is_admin_or_owner=is_admin)) + + class TestGetPartialMemberList: def test_returns_empty_when_no_permissions( self, sqlite_session: Session, credential_id: str, tenant_id: str, user_id: str @@ -89,50 +113,88 @@ class TestGetPartialMemberList: class TestApplyVisibilityFilter: - """Test the visibility filter logic using mock model columns.""" + def test_admin_does_not_bypass_personal_visibility( + self, + sqlite_session: Session, + tenant_id: str, + user_id: str, + other_user_id: str, + ) -> None: + private_subscription = _subscription( + tenant_id=tenant_id, + owner_id=other_user_id, + name="private", + visibility=PermissionEnum.ONLY_ME, + ) + sqlite_session.add(private_subscription) + sqlite_session.commit() - def _make_mock_columns(self): - """Create mock model columns for testing.""" - model_id = MagicMock(name="id_column") - model_user_id = MagicMock(name="user_id_column") - model_visibility = MagicMock(name="visibility_column") - return model_id, model_user_id, model_visibility - - def _make_user(self, user_id: str, is_admin: bool): - return SimpleNamespace(id=user_id, is_admin_or_owner=is_admin) - - def test_admin_gets_filtered_too(self, user_id): - """Admin should NOT bypass visibility — personal credentials are private regardless of role.""" - from models.trigger import TriggerSubscription - - query = select(TriggerSubscription) - result = CredentialPermissionService.apply_visibility_filter( - query, + query = CredentialPermissionService.apply_visibility_filter( + select(TriggerSubscription).where(TriggerSubscription.tenant_id == tenant_id), model_id_column=TriggerSubscription.id, model_user_id_column=TriggerSubscription.user_id, model_visibility_column=TriggerSubscription.visibility, credential_type=CredentialType.TRIGGER_SUBSCRIPTION, - user=self._make_user(user_id, is_admin=True), + user=_user(user_id, is_admin=True), ) - # No admin bypass: query should have WHERE clause - compiled = str(result.compile(compile_kwargs={"literal_binds": True})) - assert "WHERE" in compiled - def test_non_admin_adds_filter_on_real_model(self, user_id): - """Non-admin should get a filtered query when using real SQLAlchemy columns.""" - from models.trigger import TriggerSubscription + assert sqlite_session.scalars(query).all() == [] - query = select(TriggerSubscription) - result = CredentialPermissionService.apply_visibility_filter( - query, + def test_non_admin_sees_team_owned_and_partial_member_subscriptions( + self, + sqlite_session: Session, + tenant_id: str, + user_id: str, + other_user_id: str, + ) -> None: + team_subscription = _subscription( + tenant_id=tenant_id, + owner_id=other_user_id, + name="team", + visibility=PermissionEnum.ALL_TEAM, + ) + owned_subscription = _subscription( + tenant_id=tenant_id, + owner_id=user_id, + name="owned", + visibility=PermissionEnum.ONLY_ME, + ) + shared_subscription = _subscription( + tenant_id=tenant_id, + owner_id=other_user_id, + name="shared", + visibility=PermissionEnum.PARTIAL_TEAM, + ) + private_subscription = _subscription( + tenant_id=tenant_id, + owner_id=other_user_id, + name="private", + visibility=PermissionEnum.ONLY_ME, + ) + sqlite_session.add_all( + [ + team_subscription, + owned_subscription, + shared_subscription, + private_subscription, + CredentialPermission( + credential_id=shared_subscription.id, + credential_type=CredentialType.TRIGGER_SUBSCRIPTION, + account_id=user_id, + tenant_id=tenant_id, + ), + ] + ) + sqlite_session.commit() + + query = CredentialPermissionService.apply_visibility_filter( + select(TriggerSubscription).where(TriggerSubscription.tenant_id == tenant_id), model_id_column=TriggerSubscription.id, model_user_id_column=TriggerSubscription.user_id, model_visibility_column=TriggerSubscription.visibility, credential_type=CredentialType.TRIGGER_SUBSCRIPTION, - user=self._make_user(user_id, is_admin=False), + user=_user(user_id, is_admin=False), ) - # The compiled SQL should include a WHERE clause referencing user_id and visibility - compiled = str(result.compile(compile_kwargs={"literal_binds": True})) - assert "WHERE" in compiled - assert "visibility" in compiled - assert "user_id" in compiled + + visible_ids = {subscription.id for subscription in sqlite_session.scalars(query)} + assert visible_ids == {team_subscription.id, owned_subscription.id, shared_subscription.id}