mirror of
https://github.com/langgenius/dify.git
synced 2026-08-01 01:49:30 +08:00
test: exercise credential visibility with SQLite (#39827)
This commit is contained in:
Escape0707
GitHub
parent
908dc703ec
commit
676dff04c8
@ -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",
|
||||
|
||||
@ -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}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user