mirror of
https://github.com/langgenius/dify.git
synced 2026-03-07 08:35:58 +08:00
test: migrate dataset permission service SQL tests to testcontainers (#32546)
Co-authored-by: KinomotoMio <200703522+KinomotoMio@users.noreply.github.com> Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
木之本澪
GitHub
@ -0,0 +1,497 @@
|
||||
"""
|
||||
Container-backed integration tests for dataset permission services on the real SQL path.
|
||||
|
||||
This module exercises persisted DatasetPermission rows and dataset permission
|
||||
checks with testcontainers-backed infrastructure instead of database-chain mocks.
|
||||
"""
|
||||
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
|
||||
from extensions.ext_database import db
|
||||
from models import Account, Tenant, TenantAccountJoin, TenantAccountRole
|
||||
from models.dataset import (
|
||||
Dataset,
|
||||
DatasetPermission,
|
||||
DatasetPermissionEnum,
|
||||
)
|
||||
from services.dataset_service import DatasetPermissionService, DatasetService
|
||||
from services.errors.account import NoPermissionError
|
||||
|
||||
|
||||
class DatasetPermissionTestDataFactory:
|
||||
"""Create persisted entities and request payloads for dataset permission integration tests."""
|
||||
|
||||
@staticmethod
|
||||
def create_account_with_tenant(
|
||||
role: TenantAccountRole = TenantAccountRole.NORMAL,
|
||||
tenant: Tenant | None = None,
|
||||
) -> tuple[Account, Tenant]:
|
||||
"""Create a real account and tenant with specified role."""
|
||||
account = Account(
|
||||
email=f"{uuid4()}@example.com",
|
||||
name=f"user-{uuid4()}",
|
||||
interface_language="en-US",
|
||||
status="active",
|
||||
)
|
||||
if tenant is None:
|
||||
tenant = Tenant(name=f"tenant-{uuid4()}", status="normal")
|
||||
db.session.add_all([account, tenant])
|
||||
else:
|
||||
db.session.add(account)
|
||||
|
||||
db.session.flush()
|
||||
|
||||
join = TenantAccountJoin(
|
||||
tenant_id=tenant.id,
|
||||
account_id=account.id,
|
||||
role=role,
|
||||
current=True,
|
||||
)
|
||||
db.session.add(join)
|
||||
db.session.commit()
|
||||
|
||||
account.current_tenant = tenant
|
||||
return account, tenant
|
||||
|
||||
@staticmethod
|
||||
def create_dataset(
|
||||
tenant_id: str,
|
||||
created_by: str,
|
||||
permission: DatasetPermissionEnum = DatasetPermissionEnum.ONLY_ME,
|
||||
name: str = "Test Dataset",
|
||||
) -> Dataset:
|
||||
"""Create a real dataset with specified attributes."""
|
||||
dataset = Dataset(
|
||||
tenant_id=tenant_id,
|
||||
name=name,
|
||||
description="desc",
|
||||
data_source_type="upload_file",
|
||||
indexing_technique="high_quality",
|
||||
created_by=created_by,
|
||||
permission=permission,
|
||||
provider="vendor",
|
||||
retrieval_model={"top_k": 2},
|
||||
)
|
||||
db.session.add(dataset)
|
||||
db.session.commit()
|
||||
return dataset
|
||||
|
||||
@staticmethod
|
||||
def create_dataset_permission(
|
||||
dataset_id: str,
|
||||
account_id: str,
|
||||
tenant_id: str,
|
||||
has_permission: bool = True,
|
||||
) -> DatasetPermission:
|
||||
"""Create a real DatasetPermission instance."""
|
||||
permission = DatasetPermission(
|
||||
dataset_id=dataset_id,
|
||||
account_id=account_id,
|
||||
tenant_id=tenant_id,
|
||||
has_permission=has_permission,
|
||||
)
|
||||
db.session.add(permission)
|
||||
db.session.commit()
|
||||
return permission
|
||||
|
||||
@staticmethod
|
||||
def build_user_list_payload(user_ids: list[str]) -> list[dict[str, str]]:
|
||||
"""Build the request payload shape used by partial-member list updates."""
|
||||
return [{"user_id": user_id} for user_id in user_ids]
|
||||
|
||||
|
||||
class TestDatasetPermissionServiceGetPartialMemberList:
|
||||
"""Verify partial-member list reads against persisted DatasetPermission rows."""
|
||||
|
||||
def test_get_dataset_partial_member_list_with_members(self, db_session_with_containers):
|
||||
"""
|
||||
Test retrieving partial member list with multiple members.
|
||||
"""
|
||||
# Arrange
|
||||
owner, tenant = DatasetPermissionTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER)
|
||||
user_1, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
user_2, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
user_3, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
dataset = DatasetPermissionTestDataFactory.create_dataset(tenant.id, owner.id)
|
||||
|
||||
expected_account_ids = [user_1.id, user_2.id, user_3.id]
|
||||
for account_id in expected_account_ids:
|
||||
DatasetPermissionTestDataFactory.create_dataset_permission(dataset.id, account_id, tenant.id)
|
||||
|
||||
# Act
|
||||
result = DatasetPermissionService.get_dataset_partial_member_list(dataset.id)
|
||||
|
||||
# Assert
|
||||
assert set(result) == set(expected_account_ids)
|
||||
assert len(result) == 3
|
||||
|
||||
def test_get_dataset_partial_member_list_with_single_member(self, db_session_with_containers):
|
||||
"""
|
||||
Test retrieving partial member list with single member.
|
||||
"""
|
||||
# Arrange
|
||||
owner, tenant = DatasetPermissionTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER)
|
||||
user, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
dataset = DatasetPermissionTestDataFactory.create_dataset(tenant.id, owner.id)
|
||||
|
||||
expected_account_ids = [user.id]
|
||||
DatasetPermissionTestDataFactory.create_dataset_permission(dataset.id, user.id, tenant.id)
|
||||
|
||||
# Act
|
||||
result = DatasetPermissionService.get_dataset_partial_member_list(dataset.id)
|
||||
|
||||
# Assert
|
||||
assert set(result) == set(expected_account_ids)
|
||||
assert len(result) == 1
|
||||
|
||||
def test_get_dataset_partial_member_list_empty(self, db_session_with_containers):
|
||||
"""
|
||||
Test retrieving partial member list when no members exist.
|
||||
"""
|
||||
# Arrange
|
||||
owner, tenant = DatasetPermissionTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER)
|
||||
dataset = DatasetPermissionTestDataFactory.create_dataset(tenant.id, owner.id)
|
||||
|
||||
# Act
|
||||
result = DatasetPermissionService.get_dataset_partial_member_list(dataset.id)
|
||||
|
||||
# Assert
|
||||
assert result == []
|
||||
assert len(result) == 0
|
||||
|
||||
|
||||
class TestDatasetPermissionServiceUpdatePartialMemberList:
|
||||
"""Verify partial-member list updates against persisted DatasetPermission rows."""
|
||||
|
||||
def test_update_partial_member_list_add_new_members(self, db_session_with_containers):
|
||||
"""
|
||||
Test adding new partial members to a dataset.
|
||||
"""
|
||||
# Arrange
|
||||
owner, tenant = DatasetPermissionTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER)
|
||||
member_1, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
member_2, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
dataset = DatasetPermissionTestDataFactory.create_dataset(tenant.id, owner.id)
|
||||
user_list = DatasetPermissionTestDataFactory.build_user_list_payload([member_1.id, member_2.id])
|
||||
|
||||
# Act
|
||||
DatasetPermissionService.update_partial_member_list(tenant.id, dataset.id, user_list)
|
||||
|
||||
# Assert
|
||||
result = DatasetPermissionService.get_dataset_partial_member_list(dataset.id)
|
||||
assert set(result) == {member_1.id, member_2.id}
|
||||
|
||||
def test_update_partial_member_list_replace_existing(self, db_session_with_containers):
|
||||
"""
|
||||
Test replacing existing partial members with new ones.
|
||||
"""
|
||||
# Arrange
|
||||
owner, tenant = DatasetPermissionTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER)
|
||||
old_member_1, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
old_member_2, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
new_member_1, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
new_member_2, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
dataset = DatasetPermissionTestDataFactory.create_dataset(tenant.id, owner.id)
|
||||
|
||||
old_users = DatasetPermissionTestDataFactory.build_user_list_payload([old_member_1.id, old_member_2.id])
|
||||
DatasetPermissionService.update_partial_member_list(tenant.id, dataset.id, old_users)
|
||||
|
||||
new_users = DatasetPermissionTestDataFactory.build_user_list_payload([new_member_1.id, new_member_2.id])
|
||||
|
||||
# Act
|
||||
DatasetPermissionService.update_partial_member_list(tenant.id, dataset.id, new_users)
|
||||
|
||||
# Assert
|
||||
result = DatasetPermissionService.get_dataset_partial_member_list(dataset.id)
|
||||
assert set(result) == {new_member_1.id, new_member_2.id}
|
||||
|
||||
def test_update_partial_member_list_empty_list(self, db_session_with_containers):
|
||||
"""
|
||||
Test updating with empty member list (clearing all members).
|
||||
"""
|
||||
# Arrange
|
||||
owner, tenant = DatasetPermissionTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER)
|
||||
member_1, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
member_2, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
dataset = DatasetPermissionTestDataFactory.create_dataset(tenant.id, owner.id)
|
||||
users = DatasetPermissionTestDataFactory.build_user_list_payload([member_1.id, member_2.id])
|
||||
DatasetPermissionService.update_partial_member_list(tenant.id, dataset.id, users)
|
||||
|
||||
# Act
|
||||
DatasetPermissionService.update_partial_member_list(tenant.id, dataset.id, [])
|
||||
|
||||
# Assert
|
||||
result = DatasetPermissionService.get_dataset_partial_member_list(dataset.id)
|
||||
assert result == []
|
||||
|
||||
def test_update_partial_member_list_database_error_rollback(self, db_session_with_containers):
|
||||
"""
|
||||
Test error handling and rollback on database error.
|
||||
"""
|
||||
# Arrange
|
||||
owner, tenant = DatasetPermissionTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER)
|
||||
existing_member, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
replacement_member, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
dataset = DatasetPermissionTestDataFactory.create_dataset(tenant.id, owner.id)
|
||||
DatasetPermissionService.update_partial_member_list(
|
||||
tenant.id,
|
||||
dataset.id,
|
||||
DatasetPermissionTestDataFactory.build_user_list_payload([existing_member.id]),
|
||||
)
|
||||
user_list = DatasetPermissionTestDataFactory.build_user_list_payload([replacement_member.id])
|
||||
rollback_called = {"count": 0}
|
||||
original_rollback = db.session.rollback
|
||||
|
||||
# Act / Assert
|
||||
with pytest.MonkeyPatch.context() as mp:
|
||||
|
||||
def _raise_commit():
|
||||
raise Exception("Database connection error")
|
||||
|
||||
def _rollback_and_mark():
|
||||
rollback_called["count"] += 1
|
||||
original_rollback()
|
||||
|
||||
mp.setattr("services.dataset_service.db.session.commit", _raise_commit)
|
||||
mp.setattr("services.dataset_service.db.session.rollback", _rollback_and_mark)
|
||||
with pytest.raises(Exception, match="Database connection error"):
|
||||
DatasetPermissionService.update_partial_member_list(tenant.id, dataset.id, user_list)
|
||||
|
||||
# Assert
|
||||
result = DatasetPermissionService.get_dataset_partial_member_list(dataset.id)
|
||||
assert rollback_called["count"] == 1
|
||||
assert result == [existing_member.id]
|
||||
assert db_session_with_containers.query(DatasetPermission).filter_by(dataset_id=dataset.id).count() == 1
|
||||
|
||||
|
||||
class TestDatasetPermissionServiceClearPartialMemberList:
|
||||
"""Verify partial-member clearing against persisted DatasetPermission rows."""
|
||||
|
||||
def test_clear_partial_member_list_success(self, db_session_with_containers):
|
||||
"""
|
||||
Test successful clearing of partial member list.
|
||||
"""
|
||||
# Arrange
|
||||
owner, tenant = DatasetPermissionTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER)
|
||||
member_1, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
member_2, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
dataset = DatasetPermissionTestDataFactory.create_dataset(tenant.id, owner.id)
|
||||
users = DatasetPermissionTestDataFactory.build_user_list_payload([member_1.id, member_2.id])
|
||||
DatasetPermissionService.update_partial_member_list(tenant.id, dataset.id, users)
|
||||
|
||||
# Act
|
||||
DatasetPermissionService.clear_partial_member_list(dataset.id)
|
||||
|
||||
# Assert
|
||||
result = DatasetPermissionService.get_dataset_partial_member_list(dataset.id)
|
||||
assert result == []
|
||||
|
||||
def test_clear_partial_member_list_empty_list(self, db_session_with_containers):
|
||||
"""
|
||||
Test clearing partial member list when no members exist.
|
||||
"""
|
||||
# Arrange
|
||||
owner, tenant = DatasetPermissionTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER)
|
||||
dataset = DatasetPermissionTestDataFactory.create_dataset(tenant.id, owner.id)
|
||||
|
||||
# Act
|
||||
DatasetPermissionService.clear_partial_member_list(dataset.id)
|
||||
|
||||
# Assert
|
||||
result = DatasetPermissionService.get_dataset_partial_member_list(dataset.id)
|
||||
assert result == []
|
||||
|
||||
def test_clear_partial_member_list_database_error_rollback(self, db_session_with_containers):
|
||||
"""
|
||||
Test error handling and rollback on database error.
|
||||
"""
|
||||
# Arrange
|
||||
owner, tenant = DatasetPermissionTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER)
|
||||
member_1, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
member_2, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
dataset = DatasetPermissionTestDataFactory.create_dataset(tenant.id, owner.id)
|
||||
users = DatasetPermissionTestDataFactory.build_user_list_payload([member_1.id, member_2.id])
|
||||
DatasetPermissionService.update_partial_member_list(tenant.id, dataset.id, users)
|
||||
rollback_called = {"count": 0}
|
||||
original_rollback = db.session.rollback
|
||||
|
||||
# Act / Assert
|
||||
with pytest.MonkeyPatch.context() as mp:
|
||||
|
||||
def _raise_commit():
|
||||
raise Exception("Database connection error")
|
||||
|
||||
def _rollback_and_mark():
|
||||
rollback_called["count"] += 1
|
||||
original_rollback()
|
||||
|
||||
mp.setattr("services.dataset_service.db.session.commit", _raise_commit)
|
||||
mp.setattr("services.dataset_service.db.session.rollback", _rollback_and_mark)
|
||||
with pytest.raises(Exception, match="Database connection error"):
|
||||
DatasetPermissionService.clear_partial_member_list(dataset.id)
|
||||
|
||||
# Assert
|
||||
result = DatasetPermissionService.get_dataset_partial_member_list(dataset.id)
|
||||
assert rollback_called["count"] == 1
|
||||
assert set(result) == {member_1.id, member_2.id}
|
||||
assert db_session_with_containers.query(DatasetPermission).filter_by(dataset_id=dataset.id).count() == 2
|
||||
|
||||
|
||||
class TestDatasetServiceCheckDatasetPermission:
|
||||
"""Verify dataset access checks against persisted partial-member permissions."""
|
||||
|
||||
def test_check_dataset_permission_partial_members_with_permission_success(self, db_session_with_containers):
|
||||
"""
|
||||
Test that user with explicit permission can access partial_members dataset.
|
||||
"""
|
||||
# Arrange
|
||||
owner, tenant = DatasetPermissionTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER)
|
||||
user, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
|
||||
dataset = DatasetPermissionTestDataFactory.create_dataset(
|
||||
tenant.id,
|
||||
owner.id,
|
||||
permission=DatasetPermissionEnum.PARTIAL_TEAM,
|
||||
)
|
||||
DatasetPermissionTestDataFactory.create_dataset_permission(dataset.id, user.id, tenant.id)
|
||||
|
||||
# Act (should not raise)
|
||||
DatasetService.check_dataset_permission(dataset, user)
|
||||
|
||||
# Assert
|
||||
permissions = DatasetPermissionService.get_dataset_partial_member_list(dataset.id)
|
||||
assert user.id in permissions
|
||||
|
||||
def test_check_dataset_permission_partial_members_without_permission_error(self, db_session_with_containers):
|
||||
"""
|
||||
Test error when user without permission tries to access partial_members dataset.
|
||||
"""
|
||||
# Arrange
|
||||
owner, tenant = DatasetPermissionTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER)
|
||||
user, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
|
||||
dataset = DatasetPermissionTestDataFactory.create_dataset(
|
||||
tenant.id,
|
||||
owner.id,
|
||||
permission=DatasetPermissionEnum.PARTIAL_TEAM,
|
||||
)
|
||||
|
||||
# Act & Assert
|
||||
with pytest.raises(NoPermissionError, match="You do not have permission to access this dataset"):
|
||||
DatasetService.check_dataset_permission(dataset, user)
|
||||
|
||||
|
||||
class TestDatasetServiceCheckDatasetOperatorPermission:
|
||||
"""Verify operator permission checks against persisted partial-member permissions."""
|
||||
|
||||
def test_check_dataset_operator_permission_partial_members_with_permission_success(
|
||||
self, db_session_with_containers
|
||||
):
|
||||
"""
|
||||
Test that user with explicit permission can access partial_members dataset.
|
||||
"""
|
||||
# Arrange
|
||||
owner, tenant = DatasetPermissionTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER)
|
||||
user, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
|
||||
dataset = DatasetPermissionTestDataFactory.create_dataset(
|
||||
tenant.id,
|
||||
owner.id,
|
||||
permission=DatasetPermissionEnum.PARTIAL_TEAM,
|
||||
)
|
||||
DatasetPermissionTestDataFactory.create_dataset_permission(dataset.id, user.id, tenant.id)
|
||||
|
||||
# Act (should not raise)
|
||||
DatasetService.check_dataset_operator_permission(user=user, dataset=dataset)
|
||||
|
||||
# Assert
|
||||
permissions = DatasetPermissionService.get_dataset_partial_member_list(dataset.id)
|
||||
assert user.id in permissions
|
||||
|
||||
def test_check_dataset_operator_permission_partial_members_without_permission_error(
|
||||
self, db_session_with_containers
|
||||
):
|
||||
"""
|
||||
Test error when user without permission tries to access partial_members dataset.
|
||||
"""
|
||||
# Arrange
|
||||
owner, tenant = DatasetPermissionTestDataFactory.create_account_with_tenant(role=TenantAccountRole.OWNER)
|
||||
user, _ = DatasetPermissionTestDataFactory.create_account_with_tenant(
|
||||
role=TenantAccountRole.NORMAL,
|
||||
tenant=tenant,
|
||||
)
|
||||
|
||||
dataset = DatasetPermissionTestDataFactory.create_dataset(
|
||||
tenant.id,
|
||||
owner.id,
|
||||
permission=DatasetPermissionEnum.PARTIAL_TEAM,
|
||||
)
|
||||
|
||||
# Act & Assert
|
||||
with pytest.raises(NoPermissionError, match="You do not have permission to access this dataset"):
|
||||
DatasetService.check_dataset_operator_permission(user=user, dataset=dataset)
|
||||
@ -258,323 +258,6 @@ class DatasetPermissionTestDataFactory:
|
||||
return [{"user_id": user_id} for user_id in user_ids]
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Tests for get_dataset_partial_member_list
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class TestDatasetPermissionServiceGetPartialMemberList:
|
||||
"""
|
||||
Comprehensive unit tests for DatasetPermissionService.get_dataset_partial_member_list method.
|
||||
|
||||
This test class covers the retrieval of partial member lists for datasets,
|
||||
which returns a list of account IDs that have explicit permissions for
|
||||
a given dataset.
|
||||
|
||||
The get_dataset_partial_member_list method:
|
||||
1. Queries DatasetPermission table for the dataset ID
|
||||
2. Selects account_id values
|
||||
3. Returns list of account IDs
|
||||
|
||||
Test scenarios include:
|
||||
- Retrieving list with multiple members
|
||||
- Retrieving list with single member
|
||||
- Retrieving empty list (no partial members)
|
||||
- Database query validation
|
||||
"""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_db_session(self):
|
||||
"""
|
||||
Mock database session for testing.
|
||||
|
||||
Provides a mocked database session that can be used to verify
|
||||
query construction and execution.
|
||||
"""
|
||||
with patch("services.dataset_service.db.session") as mock_db:
|
||||
yield mock_db
|
||||
|
||||
def test_get_dataset_partial_member_list_with_members(self, mock_db_session):
|
||||
"""
|
||||
Test retrieving partial member list with multiple members.
|
||||
|
||||
Verifies that when a dataset has multiple partial members, all
|
||||
account IDs are returned correctly.
|
||||
|
||||
This test ensures:
|
||||
- Query is constructed correctly
|
||||
- All account IDs are returned
|
||||
- Database query is executed
|
||||
"""
|
||||
# Arrange
|
||||
dataset_id = "dataset-123"
|
||||
expected_account_ids = ["user-456", "user-789", "user-012"]
|
||||
|
||||
# Mock the scalars query to return account IDs
|
||||
mock_scalars_result = Mock()
|
||||
mock_scalars_result.all.return_value = expected_account_ids
|
||||
mock_db_session.scalars.return_value = mock_scalars_result
|
||||
|
||||
# Act
|
||||
result = DatasetPermissionService.get_dataset_partial_member_list(dataset_id)
|
||||
|
||||
# Assert
|
||||
assert result == expected_account_ids
|
||||
assert len(result) == 3
|
||||
|
||||
# Verify query was executed
|
||||
mock_db_session.scalars.assert_called_once()
|
||||
|
||||
def test_get_dataset_partial_member_list_with_single_member(self, mock_db_session):
|
||||
"""
|
||||
Test retrieving partial member list with single member.
|
||||
|
||||
Verifies that when a dataset has only one partial member, the
|
||||
single account ID is returned correctly.
|
||||
|
||||
This test ensures:
|
||||
- Query works correctly for single member
|
||||
- Result is a list with one element
|
||||
- Database query is executed
|
||||
"""
|
||||
# Arrange
|
||||
dataset_id = "dataset-123"
|
||||
expected_account_ids = ["user-456"]
|
||||
|
||||
# Mock the scalars query to return single account ID
|
||||
mock_scalars_result = Mock()
|
||||
mock_scalars_result.all.return_value = expected_account_ids
|
||||
mock_db_session.scalars.return_value = mock_scalars_result
|
||||
|
||||
# Act
|
||||
result = DatasetPermissionService.get_dataset_partial_member_list(dataset_id)
|
||||
|
||||
# Assert
|
||||
assert result == expected_account_ids
|
||||
assert len(result) == 1
|
||||
|
||||
# Verify query was executed
|
||||
mock_db_session.scalars.assert_called_once()
|
||||
|
||||
def test_get_dataset_partial_member_list_empty(self, mock_db_session):
|
||||
"""
|
||||
Test retrieving partial member list when no members exist.
|
||||
|
||||
Verifies that when a dataset has no partial members, an empty
|
||||
list is returned.
|
||||
|
||||
This test ensures:
|
||||
- Empty list is returned correctly
|
||||
- Query is executed even when no results
|
||||
- No errors are raised
|
||||
"""
|
||||
# Arrange
|
||||
dataset_id = "dataset-123"
|
||||
|
||||
# Mock the scalars query to return empty list
|
||||
mock_scalars_result = Mock()
|
||||
mock_scalars_result.all.return_value = []
|
||||
mock_db_session.scalars.return_value = mock_scalars_result
|
||||
|
||||
# Act
|
||||
result = DatasetPermissionService.get_dataset_partial_member_list(dataset_id)
|
||||
|
||||
# Assert
|
||||
assert result == []
|
||||
assert len(result) == 0
|
||||
|
||||
# Verify query was executed
|
||||
mock_db_session.scalars.assert_called_once()
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Tests for update_partial_member_list
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class TestDatasetPermissionServiceUpdatePartialMemberList:
|
||||
"""
|
||||
Comprehensive unit tests for DatasetPermissionService.update_partial_member_list method.
|
||||
|
||||
This test class covers the update of partial member lists for datasets,
|
||||
which replaces the existing partial member list with a new one.
|
||||
|
||||
The update_partial_member_list method:
|
||||
1. Deletes all existing DatasetPermission records for the dataset
|
||||
2. Creates new DatasetPermission records for each user in the list
|
||||
3. Adds all new permissions to the session
|
||||
4. Commits the transaction
|
||||
5. Rolls back on error
|
||||
|
||||
Test scenarios include:
|
||||
- Adding new partial members
|
||||
- Updating existing partial members
|
||||
- Replacing entire member list
|
||||
- Handling empty member list
|
||||
- Database transaction handling
|
||||
- Error handling and rollback
|
||||
"""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_db_session(self):
|
||||
"""
|
||||
Mock database session for testing.
|
||||
|
||||
Provides a mocked database session that can be used to verify
|
||||
database operations including queries, adds, commits, and rollbacks.
|
||||
"""
|
||||
with patch("services.dataset_service.db.session") as mock_db:
|
||||
yield mock_db
|
||||
|
||||
def test_update_partial_member_list_add_new_members(self, mock_db_session):
|
||||
"""
|
||||
Test adding new partial members to a dataset.
|
||||
|
||||
Verifies that when updating with new members, the old members
|
||||
are deleted and new members are added correctly.
|
||||
|
||||
This test ensures:
|
||||
- Old permissions are deleted
|
||||
- New permissions are created
|
||||
- All permissions are added to session
|
||||
- Transaction is committed
|
||||
"""
|
||||
# Arrange
|
||||
tenant_id = "tenant-123"
|
||||
dataset_id = "dataset-123"
|
||||
user_list = DatasetPermissionTestDataFactory.create_user_list_mock(["user-456", "user-789"])
|
||||
|
||||
# Mock the query delete operation
|
||||
mock_query = Mock()
|
||||
mock_query.where.return_value = mock_query
|
||||
mock_query.delete.return_value = None
|
||||
mock_db_session.query.return_value = mock_query
|
||||
|
||||
# Act
|
||||
DatasetPermissionService.update_partial_member_list(tenant_id, dataset_id, user_list)
|
||||
|
||||
# Assert
|
||||
# Verify old permissions were deleted
|
||||
mock_db_session.query.assert_called()
|
||||
mock_query.where.assert_called()
|
||||
|
||||
# Verify new permissions were added
|
||||
mock_db_session.add_all.assert_called_once()
|
||||
|
||||
# Verify transaction was committed
|
||||
mock_db_session.commit.assert_called_once()
|
||||
|
||||
# Verify no rollback occurred
|
||||
mock_db_session.rollback.assert_not_called()
|
||||
|
||||
def test_update_partial_member_list_replace_existing(self, mock_db_session):
|
||||
"""
|
||||
Test replacing existing partial members with new ones.
|
||||
|
||||
Verifies that when updating with a different member list, the
|
||||
old members are removed and new members are added.
|
||||
|
||||
This test ensures:
|
||||
- Old permissions are deleted
|
||||
- New permissions replace old ones
|
||||
- Transaction is committed successfully
|
||||
"""
|
||||
# Arrange
|
||||
tenant_id = "tenant-123"
|
||||
dataset_id = "dataset-123"
|
||||
user_list = DatasetPermissionTestDataFactory.create_user_list_mock(["user-999", "user-888"])
|
||||
|
||||
# Mock the query delete operation
|
||||
mock_query = Mock()
|
||||
mock_query.where.return_value = mock_query
|
||||
mock_query.delete.return_value = None
|
||||
mock_db_session.query.return_value = mock_query
|
||||
|
||||
# Act
|
||||
DatasetPermissionService.update_partial_member_list(tenant_id, dataset_id, user_list)
|
||||
|
||||
# Assert
|
||||
# Verify old permissions were deleted
|
||||
mock_db_session.query.assert_called()
|
||||
|
||||
# Verify new permissions were added
|
||||
mock_db_session.add_all.assert_called_once()
|
||||
|
||||
# Verify transaction was committed
|
||||
mock_db_session.commit.assert_called_once()
|
||||
|
||||
def test_update_partial_member_list_empty_list(self, mock_db_session):
|
||||
"""
|
||||
Test updating with empty member list (clearing all members).
|
||||
|
||||
Verifies that when updating with an empty list, all existing
|
||||
permissions are deleted and no new permissions are added.
|
||||
|
||||
This test ensures:
|
||||
- Old permissions are deleted
|
||||
- No new permissions are added
|
||||
- Transaction is committed
|
||||
"""
|
||||
# Arrange
|
||||
tenant_id = "tenant-123"
|
||||
dataset_id = "dataset-123"
|
||||
user_list = []
|
||||
|
||||
# Mock the query delete operation
|
||||
mock_query = Mock()
|
||||
mock_query.where.return_value = mock_query
|
||||
mock_query.delete.return_value = None
|
||||
mock_db_session.query.return_value = mock_query
|
||||
|
||||
# Act
|
||||
DatasetPermissionService.update_partial_member_list(tenant_id, dataset_id, user_list)
|
||||
|
||||
# Assert
|
||||
# Verify old permissions were deleted
|
||||
mock_db_session.query.assert_called()
|
||||
|
||||
# Verify add_all was called with empty list
|
||||
mock_db_session.add_all.assert_called_once_with([])
|
||||
|
||||
# Verify transaction was committed
|
||||
mock_db_session.commit.assert_called_once()
|
||||
|
||||
def test_update_partial_member_list_database_error_rollback(self, mock_db_session):
|
||||
"""
|
||||
Test error handling and rollback on database error.
|
||||
|
||||
Verifies that when a database error occurs during the update,
|
||||
the transaction is rolled back and the error is re-raised.
|
||||
|
||||
This test ensures:
|
||||
- Error is caught and handled
|
||||
- Transaction is rolled back
|
||||
- Error is re-raised
|
||||
- No commit occurs after error
|
||||
"""
|
||||
# Arrange
|
||||
tenant_id = "tenant-123"
|
||||
dataset_id = "dataset-123"
|
||||
user_list = DatasetPermissionTestDataFactory.create_user_list_mock(["user-456"])
|
||||
|
||||
# Mock the query delete operation
|
||||
mock_query = Mock()
|
||||
mock_query.where.return_value = mock_query
|
||||
mock_query.delete.return_value = None
|
||||
mock_db_session.query.return_value = mock_query
|
||||
|
||||
# Mock commit to raise an error
|
||||
database_error = Exception("Database connection error")
|
||||
mock_db_session.commit.side_effect = database_error
|
||||
|
||||
# Act & Assert
|
||||
with pytest.raises(Exception, match="Database connection error"):
|
||||
DatasetPermissionService.update_partial_member_list(tenant_id, dataset_id, user_list)
|
||||
|
||||
# Verify rollback was called
|
||||
mock_db_session.rollback.assert_called_once()
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Tests for check_permission
|
||||
# ============================================================================
|
||||
@ -776,144 +459,6 @@ class TestDatasetPermissionServiceCheckPermission:
|
||||
mock_get_partial_member_list.assert_called_once_with(dataset.id)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Tests for clear_partial_member_list
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class TestDatasetPermissionServiceClearPartialMemberList:
|
||||
"""
|
||||
Comprehensive unit tests for DatasetPermissionService.clear_partial_member_list method.
|
||||
|
||||
This test class covers the clearing of partial member lists, which removes
|
||||
all DatasetPermission records for a given dataset.
|
||||
|
||||
The clear_partial_member_list method:
|
||||
1. Deletes all DatasetPermission records for the dataset
|
||||
2. Commits the transaction
|
||||
3. Rolls back on error
|
||||
|
||||
Test scenarios include:
|
||||
- Clearing list with existing members
|
||||
- Clearing empty list (no members)
|
||||
- Database transaction handling
|
||||
- Error handling and rollback
|
||||
"""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_db_session(self):
|
||||
"""
|
||||
Mock database session for testing.
|
||||
|
||||
Provides a mocked database session that can be used to verify
|
||||
database operations including queries, deletes, commits, and rollbacks.
|
||||
"""
|
||||
with patch("services.dataset_service.db.session") as mock_db:
|
||||
yield mock_db
|
||||
|
||||
def test_clear_partial_member_list_success(self, mock_db_session):
|
||||
"""
|
||||
Test successful clearing of partial member list.
|
||||
|
||||
Verifies that when clearing a partial member list, all permissions
|
||||
are deleted and the transaction is committed.
|
||||
|
||||
This test ensures:
|
||||
- All permissions are deleted
|
||||
- Transaction is committed
|
||||
- No errors are raised
|
||||
"""
|
||||
# Arrange
|
||||
dataset_id = "dataset-123"
|
||||
|
||||
# Mock the query delete operation
|
||||
mock_query = Mock()
|
||||
mock_query.where.return_value = mock_query
|
||||
mock_query.delete.return_value = None
|
||||
mock_db_session.query.return_value = mock_query
|
||||
|
||||
# Act
|
||||
DatasetPermissionService.clear_partial_member_list(dataset_id)
|
||||
|
||||
# Assert
|
||||
# Verify query was executed
|
||||
mock_db_session.query.assert_called()
|
||||
|
||||
# Verify delete was called
|
||||
mock_query.where.assert_called()
|
||||
mock_query.delete.assert_called_once()
|
||||
|
||||
# Verify transaction was committed
|
||||
mock_db_session.commit.assert_called_once()
|
||||
|
||||
# Verify no rollback occurred
|
||||
mock_db_session.rollback.assert_not_called()
|
||||
|
||||
def test_clear_partial_member_list_empty_list(self, mock_db_session):
|
||||
"""
|
||||
Test clearing partial member list when no members exist.
|
||||
|
||||
Verifies that when clearing an already empty list, the operation
|
||||
completes successfully without errors.
|
||||
|
||||
This test ensures:
|
||||
- Operation works correctly for empty lists
|
||||
- Transaction is committed
|
||||
- No errors are raised
|
||||
"""
|
||||
# Arrange
|
||||
dataset_id = "dataset-123"
|
||||
|
||||
# Mock the query delete operation
|
||||
mock_query = Mock()
|
||||
mock_query.where.return_value = mock_query
|
||||
mock_query.delete.return_value = None
|
||||
mock_db_session.query.return_value = mock_query
|
||||
|
||||
# Act
|
||||
DatasetPermissionService.clear_partial_member_list(dataset_id)
|
||||
|
||||
# Assert
|
||||
# Verify query was executed
|
||||
mock_db_session.query.assert_called()
|
||||
|
||||
# Verify transaction was committed
|
||||
mock_db_session.commit.assert_called_once()
|
||||
|
||||
def test_clear_partial_member_list_database_error_rollback(self, mock_db_session):
|
||||
"""
|
||||
Test error handling and rollback on database error.
|
||||
|
||||
Verifies that when a database error occurs during clearing,
|
||||
the transaction is rolled back and the error is re-raised.
|
||||
|
||||
This test ensures:
|
||||
- Error is caught and handled
|
||||
- Transaction is rolled back
|
||||
- Error is re-raised
|
||||
- No commit occurs after error
|
||||
"""
|
||||
# Arrange
|
||||
dataset_id = "dataset-123"
|
||||
|
||||
# Mock the query delete operation
|
||||
mock_query = Mock()
|
||||
mock_query.where.return_value = mock_query
|
||||
mock_query.delete.return_value = None
|
||||
mock_db_session.query.return_value = mock_query
|
||||
|
||||
# Mock commit to raise an error
|
||||
database_error = Exception("Database connection error")
|
||||
mock_db_session.commit.side_effect = database_error
|
||||
|
||||
# Act & Assert
|
||||
with pytest.raises(Exception, match="Database connection error"):
|
||||
DatasetPermissionService.clear_partial_member_list(dataset_id)
|
||||
|
||||
# Verify rollback was called
|
||||
mock_db_session.rollback.assert_called_once()
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Tests for DatasetService.check_dataset_permission
|
||||
# ============================================================================
|
||||
@ -1047,72 +592,6 @@ class TestDatasetServiceCheckDatasetPermission:
|
||||
with pytest.raises(NoPermissionError, match="You do not have permission to access this dataset"):
|
||||
DatasetService.check_dataset_permission(dataset, user)
|
||||
|
||||
def test_check_dataset_permission_partial_members_with_permission_success(self, mock_db_session):
|
||||
"""
|
||||
Test that user with explicit permission can access partial_members dataset.
|
||||
|
||||
Verifies that when a user has an explicit DatasetPermission record
|
||||
for a partial_members dataset, they can access it successfully.
|
||||
|
||||
This test ensures:
|
||||
- Explicit permissions are checked correctly
|
||||
- Users with permissions can access
|
||||
- Database query is executed
|
||||
"""
|
||||
# Arrange
|
||||
user = DatasetPermissionTestDataFactory.create_user_mock(user_id="user-123", role=TenantAccountRole.NORMAL)
|
||||
dataset = DatasetPermissionTestDataFactory.create_dataset_mock(
|
||||
tenant_id="tenant-123",
|
||||
permission=DatasetPermissionEnum.PARTIAL_TEAM,
|
||||
created_by="other-user-456", # Not the creator
|
||||
)
|
||||
|
||||
# Mock permission query to return permission record
|
||||
mock_permission = DatasetPermissionTestDataFactory.create_dataset_permission_mock(
|
||||
dataset_id=dataset.id, account_id=user.id
|
||||
)
|
||||
mock_query = Mock()
|
||||
mock_query.filter_by.return_value = mock_query
|
||||
mock_query.first.return_value = mock_permission
|
||||
mock_db_session.query.return_value = mock_query
|
||||
|
||||
# Act (should not raise)
|
||||
DatasetService.check_dataset_permission(dataset, user)
|
||||
|
||||
# Assert
|
||||
# Verify permission query was executed
|
||||
mock_db_session.query.assert_called()
|
||||
|
||||
def test_check_dataset_permission_partial_members_without_permission_error(self, mock_db_session):
|
||||
"""
|
||||
Test error when user without permission tries to access partial_members dataset.
|
||||
|
||||
Verifies that when a user does not have an explicit DatasetPermission
|
||||
record for a partial_members dataset, a NoPermissionError is raised.
|
||||
|
||||
This test ensures:
|
||||
- Missing permissions are detected
|
||||
- Error message is clear
|
||||
- Error type is correct
|
||||
"""
|
||||
# Arrange
|
||||
user = DatasetPermissionTestDataFactory.create_user_mock(user_id="user-123", role=TenantAccountRole.NORMAL)
|
||||
dataset = DatasetPermissionTestDataFactory.create_dataset_mock(
|
||||
tenant_id="tenant-123",
|
||||
permission=DatasetPermissionEnum.PARTIAL_TEAM,
|
||||
created_by="other-user-456", # Not the creator
|
||||
)
|
||||
|
||||
# Mock permission query to return None (no permission)
|
||||
mock_query = Mock()
|
||||
mock_query.filter_by.return_value = mock_query
|
||||
mock_query.first.return_value = None # No permission found
|
||||
mock_db_session.query.return_value = mock_query
|
||||
|
||||
# Act & Assert
|
||||
with pytest.raises(NoPermissionError, match="You do not have permission to access this dataset"):
|
||||
DatasetService.check_dataset_permission(dataset, user)
|
||||
|
||||
def test_check_dataset_permission_partial_members_creator_success(self, mock_db_session):
|
||||
"""
|
||||
Test that creator can access partial_members dataset without explicit permission.
|
||||
@ -1311,72 +790,6 @@ class TestDatasetServiceCheckDatasetOperatorPermission:
|
||||
with pytest.raises(NoPermissionError, match="You do not have permission to access this dataset"):
|
||||
DatasetService.check_dataset_operator_permission(user=user, dataset=dataset)
|
||||
|
||||
def test_check_dataset_operator_permission_partial_members_with_permission_success(self, mock_db_session):
|
||||
"""
|
||||
Test that user with explicit permission can access partial_members dataset.
|
||||
|
||||
Verifies that when a user has an explicit DatasetPermission record
|
||||
for a partial_members dataset, they can access it successfully.
|
||||
|
||||
This test ensures:
|
||||
- Explicit permissions are checked correctly
|
||||
- Users with permissions can access
|
||||
- Database query is executed
|
||||
"""
|
||||
# Arrange
|
||||
user = DatasetPermissionTestDataFactory.create_user_mock(user_id="user-123", role=TenantAccountRole.NORMAL)
|
||||
dataset = DatasetPermissionTestDataFactory.create_dataset_mock(
|
||||
tenant_id="tenant-123",
|
||||
permission=DatasetPermissionEnum.PARTIAL_TEAM,
|
||||
created_by="other-user-456", # Not the creator
|
||||
)
|
||||
|
||||
# Mock permission query to return permission records
|
||||
mock_permission = DatasetPermissionTestDataFactory.create_dataset_permission_mock(
|
||||
dataset_id=dataset.id, account_id=user.id
|
||||
)
|
||||
mock_query = Mock()
|
||||
mock_query.filter_by.return_value = mock_query
|
||||
mock_query.all.return_value = [mock_permission] # User has permission
|
||||
mock_db_session.query.return_value = mock_query
|
||||
|
||||
# Act (should not raise)
|
||||
DatasetService.check_dataset_operator_permission(user=user, dataset=dataset)
|
||||
|
||||
# Assert
|
||||
# Verify permission query was executed
|
||||
mock_db_session.query.assert_called()
|
||||
|
||||
def test_check_dataset_operator_permission_partial_members_without_permission_error(self, mock_db_session):
|
||||
"""
|
||||
Test error when user without permission tries to access partial_members dataset.
|
||||
|
||||
Verifies that when a user does not have an explicit DatasetPermission
|
||||
record for a partial_members dataset, a NoPermissionError is raised.
|
||||
|
||||
This test ensures:
|
||||
- Missing permissions are detected
|
||||
- Error message is clear
|
||||
- Error type is correct
|
||||
"""
|
||||
# Arrange
|
||||
user = DatasetPermissionTestDataFactory.create_user_mock(user_id="user-123", role=TenantAccountRole.NORMAL)
|
||||
dataset = DatasetPermissionTestDataFactory.create_dataset_mock(
|
||||
tenant_id="tenant-123",
|
||||
permission=DatasetPermissionEnum.PARTIAL_TEAM,
|
||||
created_by="other-user-456", # Not the creator
|
||||
)
|
||||
|
||||
# Mock permission query to return empty list (no permission)
|
||||
mock_query = Mock()
|
||||
mock_query.filter_by.return_value = mock_query
|
||||
mock_query.all.return_value = [] # No permissions found
|
||||
mock_db_session.query.return_value = mock_query
|
||||
|
||||
# Act & Assert
|
||||
with pytest.raises(NoPermissionError, match="You do not have permission to access this dataset"):
|
||||
DatasetService.check_dataset_operator_permission(user=user, dataset=dataset)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Additional Documentation and Notes
|
||||
|
||||
Reference in New Issue
Block a user