mirror of
https://github.com/langgenius/dify.git
synced 2026-05-13 08:57:28 +08:00
test: migrate BillingService permission-check tests to Testcontainers integration tests (#34993)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
parent
ea41e9ab4e
commit
17da0e4146
@ -1,9 +1,13 @@
|
||||
import json
|
||||
from collections.abc import Generator
|
||||
from unittest.mock import patch
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from extensions.ext_redis import redis_client
|
||||
from models.account import Account, Tenant, TenantAccountJoin, TenantAccountRole
|
||||
from services.billing_service import BillingService
|
||||
|
||||
|
||||
@ -363,3 +367,62 @@ class TestBillingServiceGetPlanBulkWithCache:
|
||||
assert ttl_1_new <= 600
|
||||
assert ttl_2 > 0
|
||||
assert ttl_2 <= 600
|
||||
|
||||
|
||||
class TestBillingServiceIsTenantOwnerOrAdmin:
|
||||
"""
|
||||
Integration tests for BillingService.is_tenant_owner_or_admin.
|
||||
|
||||
Verifies that non-privileged roles (EDITOR, DATASET_OPERATOR) raise ValueError
|
||||
when checked against real TenantAccountJoin rows in PostgreSQL.
|
||||
"""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _auto_rollback(self, db_session_with_containers: Session) -> Generator[None, None, None]:
|
||||
yield
|
||||
db_session_with_containers.rollback()
|
||||
|
||||
def _create_account_with_tenant_role(self, db_session: Session, role: TenantAccountRole) -> tuple[Account, Tenant]:
|
||||
tenant = Tenant(name=f"Tenant {uuid4()}")
|
||||
db_session.add(tenant)
|
||||
db_session.flush()
|
||||
|
||||
account = Account(
|
||||
name=f"Account {uuid4()}",
|
||||
email=f"billing_{uuid4()}@example.com",
|
||||
password="hashed-password",
|
||||
password_salt="salt",
|
||||
interface_language="en-US",
|
||||
timezone="UTC",
|
||||
)
|
||||
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.flush()
|
||||
|
||||
# Wire up in-memory reference so current_tenant_id resolves
|
||||
account._current_tenant = tenant
|
||||
return account, tenant
|
||||
|
||||
def test_is_tenant_owner_or_admin_editor_role_raises_error(self, db_session_with_containers: Session) -> None:
|
||||
"""is_tenant_owner_or_admin raises ValueError for EDITOR role."""
|
||||
account, _ = self._create_account_with_tenant_role(db_session_with_containers, TenantAccountRole.EDITOR)
|
||||
|
||||
with pytest.raises(ValueError, match="Only team owner or team admin can perform this action"):
|
||||
BillingService.is_tenant_owner_or_admin(account)
|
||||
|
||||
def test_is_tenant_owner_or_admin_dataset_operator_raises_error(self, db_session_with_containers: Session) -> None:
|
||||
"""is_tenant_owner_or_admin raises ValueError for DATASET_OPERATOR role."""
|
||||
account, _ = self._create_account_with_tenant_role(
|
||||
db_session_with_containers, TenantAccountRole.DATASET_OPERATOR
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="Only team owner or team admin can perform this action"):
|
||||
BillingService.is_tenant_owner_or_admin(account)
|
||||
|
||||
@ -1117,42 +1117,6 @@ class TestBillingServiceEdgeCases:
|
||||
# Assert
|
||||
assert result["history_id"] == history_id
|
||||
|
||||
def test_is_tenant_owner_or_admin_editor_role_raises_error(self):
|
||||
"""Test tenant owner/admin check raises error for editor role."""
|
||||
# Arrange
|
||||
current_user = MagicMock(spec=Account)
|
||||
current_user.id = "account-123"
|
||||
current_user.current_tenant_id = "tenant-456"
|
||||
|
||||
mock_join = MagicMock(spec=TenantAccountJoin)
|
||||
mock_join.role = TenantAccountRole.EDITOR # Editor is not privileged
|
||||
|
||||
with patch("services.billing_service.db.session") as mock_session:
|
||||
mock_session.scalar.return_value = mock_join
|
||||
|
||||
# Act & Assert
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
BillingService.is_tenant_owner_or_admin(current_user)
|
||||
assert "Only team owner or team admin can perform this action" in str(exc_info.value)
|
||||
|
||||
def test_is_tenant_owner_or_admin_dataset_operator_raises_error(self):
|
||||
"""Test tenant owner/admin check raises error for dataset operator role."""
|
||||
# Arrange
|
||||
current_user = MagicMock(spec=Account)
|
||||
current_user.id = "account-123"
|
||||
current_user.current_tenant_id = "tenant-456"
|
||||
|
||||
mock_join = MagicMock(spec=TenantAccountJoin)
|
||||
mock_join.role = TenantAccountRole.DATASET_OPERATOR # Dataset operator is not privileged
|
||||
|
||||
with patch("services.billing_service.db.session") as mock_session:
|
||||
mock_session.scalar.return_value = mock_join
|
||||
|
||||
# Act & Assert
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
BillingService.is_tenant_owner_or_admin(current_user)
|
||||
assert "Only team owner or team admin can perform this action" in str(exc_info.value)
|
||||
|
||||
|
||||
class TestBillingServiceSubscriptionOperations:
|
||||
"""Unit tests for subscription operations in BillingService.
|
||||
|
||||
Loading…
Reference in New Issue
Block a user