SunnyQjm/dify
1
0 Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-10-04 13:47:55 +08:00
Code Issues Projects Releases Wiki Activity

test: use real account activation adapters (#43218)

Browse Source
This commit is contained in:
Asuka Minato authored and GitHub committed 2026-10-02 13:53:38 +00:00
1 parent 9f0e3a4c17
commit 1dbf027277
1 file changed
+177 -133
Download Patch File Download Diff File Expand all files Collapse all files
api/tests/unit_tests/services/test_account_activation_service.py
+177 -133
View File
@@ -1,47 +1,39 @@
from unittest.mock import Mock, call
from unittest.mock import MagicMock, call
import pytest
from pytest_mock import MockerFixture
from sqlalchemy import delete, select
from sqlalchemy.orm import Session, sessionmaker
from services.account_activation_service import (
AccountActivationEligibility,
AccountActivationRepository,
AccountActivationService,
InvitationTokenStore,
WorkspaceInvitePolicy,
WorkspaceMemberAccessSync,
WorkspaceMembershipCache,
from extensions.ext_redis import RedisClientWrapper
from models.account import Account, AccountStatus, TenantAccountJoin, TenantAccountRole
from repositories.account_activation_repository import SQLAlchemyAccountActivationRepository
from services.account.adapters import (
BillingAccountActivationEligibility,
BillingWorkspaceMembershipCache,
DeploymentWorkspaceInvitePolicy,
RBACWorkspaceMemberAccessSync,
RedisInvitationTokenStore,
)
from services.account_activation_service import AccountActivationService
from services.account_errors import AccountEmailDomainSuspendedError as EmailDomainSuspendedError
from services.account_errors import FrozenAccountError, InvalidInvitationError, InvitationAccountMismatchError
from services.entities.account_activation_entities import (
AccountInvitation,
AccountSetup,
ActivationCommand,
ActivationPersistenceResult,
InvitationLookup,
InvitationToken,
)
from tests.unit_tests.model_factories import make_account, make_tenant
def _lookup(email: str | None = "invitee@example.com") -> InvitationLookup:
return InvitationLookup(workspace_id="workspace-1", email=email, token="token-1")
def _token() -> InvitationToken:
return InvitationToken(
account_id="account-1",
email="invitee@example.com",
workspace_id="workspace-1",
role="admin",
requires_setup=True,
)
def _invitation(
*,
account_status: str = "pending",
role: str | None = "admin",
requires_setup: bool | None = True,
*, account_status: str = "pending", role: str | None = None, requires_setup: bool | None = None
) -> AccountInvitation:
return AccountInvitation(
account_id="account-1",
@@ -54,143 +46,167 @@ def _invitation(
)
def _service() -> tuple[AccountActivationService, Mock, Mock, Mock, Mock, Mock, Mock]:
tokens = Mock(spec=InvitationTokenStore)
accounts = Mock(spec=AccountActivationRepository)
policy = Mock(spec=WorkspaceInvitePolicy)
eligibility = Mock(spec=AccountActivationEligibility)
membership_cache = Mock(spec=WorkspaceMembershipCache)
member_access_sync = Mock(spec=WorkspaceMemberAccessSync)
eligibility.get_freeze_type.return_value = None
service = AccountActivationService(
tokens=tokens,
accounts=accounts,
workspace_policy=policy,
eligibility=eligibility,
membership_cache=membership_cache,
member_access_sync=member_access_sync,
@pytest.fixture
def service(
sqlite_session_factory: sessionmaker[Session], redis_transport: tuple[RedisClientWrapper, MagicMock]
) -> AccountActivationService:
redis, commands = redis_transport
commands.return_value = b"account-1"
with sqlite_session_factory.begin() as session:
session.add(make_account(email="invitee@example.com", status=AccountStatus.PENDING))
session.add(make_tenant(tenant_id="workspace-1", name="Workspace"))
return AccountActivationService(
tokens=RedisInvitationTokenStore(redis=redis),
accounts=SQLAlchemyAccountActivationRepository(sqlite_session_factory),
workspace_policy=DeploymentWorkspaceInvitePolicy(),
eligibility=BillingAccountActivationEligibility(enabled=True),
membership_cache=BillingWorkspaceMembershipCache(enabled=True),
member_access_sync=RBACWorkspaceMemberAccessSync(enabled=True),
)
return service, tokens, accounts, policy, eligibility, membership_cache, member_access_sync
@pytest.fixture
def boundaries(service: AccountActivationService, mocker: MockerFixture) -> dict[str, MagicMock]:
return {
"find": mocker.spy(service._tokens, "find"),
"revoke": mocker.spy(service._tokens, "revoke"),
"resolve": mocker.spy(service._accounts, "resolve"),
"activate": mocker.spy(service._accounts, "activate"),
"policy": mocker.patch("services.account.adapters.check_workspace_member_invite_permission"),
"freeze": mocker.patch("services.account.adapters.BillingService.get_email_freeze_type", return_value=None),
"cache": mocker.patch("services.account.adapters.BillingService.clean_billing_info_cache"),
"sync": mocker.patch(
"tasks.initialize_created_app_rbac_access_task.sync_joined_workspace_member_rbac_access_task.delay"
),
}
class TestCheckInvitation:
def test_returns_invalid_without_touching_database_when_token_is_missing(self) -> None:
service, tokens, accounts, policy, _, _, _ = _service()
tokens.find.return_value = None
def test_returns_invalid_without_touching_database_when_token_is_missing(
self,
service: AccountActivationService,
boundaries: dict[str, MagicMock],
redis_transport: tuple[RedisClientWrapper, MagicMock],
) -> None:
redis_transport[1].return_value = None
result = service.check(_lookup())
assert result.is_valid is False
assert result.data is None
accounts.resolve.assert_not_called()
policy.ensure_allowed.assert_not_called()
boundaries["resolve"].assert_not_called()
boundaries["policy"].assert_not_called()
def test_does_not_repeat_database_lookup_for_normalized_email(self) -> None:
service, tokens, accounts, policy, _, _, _ = _service()
token = _token()
tokens.find.return_value = token
accounts.resolve.return_value = None
def test_does_not_repeat_database_lookup_for_normalized_email(
self,
service: AccountActivationService,
boundaries: dict[str, MagicMock],
sqlite_session_factory: sessionmaker[Session],
) -> None:
with sqlite_session_factory.begin() as session:
session.execute(delete(Account))
result = service.check(_lookup())
assert result.is_valid is False
accounts.resolve.assert_called_once_with(token)
policy.ensure_allowed.assert_not_called()
def test_falls_back_to_normalized_email_and_applies_workspace_policy(self) -> None:
service, tokens, accounts, policy, _, _, _ = _service()
upper_case_token = InvitationToken(
account_id="account-1",
email="Invitee@Example.com",
workspace_id="workspace-1",
boundaries["resolve"].assert_called_once_with(
InvitationToken(account_id="account-1", email="invitee@example.com", workspace_id="workspace-1")
)
normalized_token = _token()
invitation = _invitation(requires_setup=None)
tokens.find.side_effect = [upper_case_token, normalized_token]
accounts.resolve.side_effect = [None, invitation]
boundaries["policy"].assert_not_called()
def test_falls_back_to_normalized_email_and_applies_workspace_policy(
self,
service: AccountActivationService,
boundaries: dict[str, MagicMock],
) -> None:
result = service.check(_lookup("Invitee@Example.com"))
assert result.is_valid is True
assert result.data is not None
assert result.data.requires_setup is True
assert result.data.account_status == "pending"
assert tokens.find.call_args_list == [
assert boundaries["find"].call_args_list == [
call(_lookup("Invitee@Example.com")),
call(_lookup("invitee@example.com")),
]
assert accounts.resolve.call_args_list == [call(upper_case_token), call(normalized_token)]
policy.ensure_allowed.assert_called_once_with("workspace-1")
assert boundaries["resolve"].call_args_list == [
call(InvitationToken(account_id="account-1", email=email, workspace_id="workspace-1"))
for email in ("Invitee@Example.com", "invitee@example.com")
]
boundaries["policy"].assert_called_once_with("workspace-1")
class TestActivateInvitation:
def test_rejects_authenticated_account_mismatch_before_side_effects(self) -> None:
service, tokens, accounts, _, eligibility, _, member_access_sync = _service()
tokens.find.return_value = _token()
accounts.resolve.return_value = _invitation()
def test_rejects_authenticated_account_mismatch_before_side_effects(
self,
service: AccountActivationService,
boundaries: dict[str, MagicMock],
) -> None:
with pytest.raises(InvitationAccountMismatchError):
service.activate(
ActivationCommand(invitation=_lookup()),
authenticated_account_id="different-account",
)
eligibility.get_freeze_type.assert_not_called()
tokens.revoke.assert_not_called()
accounts.activate.assert_not_called()
member_access_sync.sync.assert_not_called()
boundaries["freeze"].assert_not_called()
boundaries["revoke"].assert_not_called()
boundaries["activate"].assert_not_called()
boundaries["sync"].assert_not_called()
def test_rejects_frozen_account_without_consuming_token(self) -> None:
service, tokens, accounts, _, eligibility, _, member_access_sync = _service()
tokens.find.return_value = _token()
accounts.resolve.return_value = _invitation()
eligibility.get_freeze_type.return_value = "freeze"
@pytest.mark.parametrize(
("freeze_type", "error"),
[("freeze", FrozenAccountError), ("email_domain_suspended", EmailDomainSuspendedError)],
)
def test_rejects_ineligible_account_without_consuming_token(
self,
service: AccountActivationService,
boundaries: dict[str, MagicMock],
freeze_type: str,
error: type[Exception],
) -> None:
boundaries["freeze"].return_value = freeze_type
with pytest.raises(FrozenAccountError):
with pytest.raises(error):
service.activate(ActivationCommand(invitation=_lookup()), authenticated_account_id=None)
eligibility.get_freeze_type.assert_called_once_with("invitee@example.com")
tokens.revoke.assert_not_called()
accounts.activate.assert_not_called()
member_access_sync.sync.assert_not_called()
def test_requires_all_setup_fields_before_consuming_token(self) -> None:
service, tokens, accounts, _, _, _, member_access_sync = _service()
tokens.find.return_value = _token()
accounts.resolve.return_value = _invitation()
boundaries["freeze"].assert_called_once_with("invitee@example.com")
boundaries["revoke"].assert_not_called()
boundaries["activate"].assert_not_called()
boundaries["sync"].assert_not_called()
def test_requires_all_setup_fields_before_consuming_token(
self,
service: AccountActivationService,
boundaries: dict[str, MagicMock],
) -> None:
with pytest.raises(InvalidInvitationError):
service.activate(
ActivationCommand(invitation=_lookup(), name="Name"),
authenticated_account_id=None,
)
tokens.revoke.assert_not_called()
accounts.activate.assert_not_called()
member_access_sync.sync.assert_not_called()
boundaries["revoke"].assert_not_called()
boundaries["activate"].assert_not_called()
boundaries["sync"].assert_not_called()
def test_rejects_suspended_email_domain_without_consuming_token(self) -> None:
service, tokens, accounts, _, eligibility, _, member_access_sync = _service()
tokens.find.return_value = _token()
accounts.resolve.return_value = _invitation()
eligibility.get_freeze_type.return_value = "email_domain_suspended"
with pytest.raises(EmailDomainSuspendedError):
service.activate(ActivationCommand(invitation=_lookup()), authenticated_account_id=None)
eligibility.get_freeze_type.assert_called_once_with("invitee@example.com")
tokens.revoke.assert_not_called()
accounts.activate.assert_not_called()
member_access_sync.sync.assert_not_called()
def test_activates_anonymous_invitation_and_invalidates_new_membership_cache(self) -> None:
service, tokens, accounts, _, eligibility, membership_cache, member_access_sync = _service()
tokens.find.return_value = _token()
invitation = _invitation(role="owner")
accounts.resolve.return_value = invitation
accounts.activate.return_value = ActivationPersistenceResult(membership_created=True)
@pytest.mark.parametrize("role", [None, "owner"])
def test_activates_anonymous_invitation_and_invalidates_new_membership_cache(
self,
service: AccountActivationService,
boundaries: dict[str, MagicMock],
sqlite_session_factory: sessionmaker[Session],
redis_transport: tuple[RedisClientWrapper, MagicMock],
role: str | None,
) -> None:
lookup = _lookup("Invitee@Example.com")
if role == "owner":
lookup = InvitationLookup(workspace_id=None, email=None, token="token-1")
redis_transport[1].return_value = (
b'{"account_id":"account-1","email":"invitee@example.com","workspace_id":"workspace-1",'
b'"role":"owner","requires_setup":true}'
)
command = ActivationCommand(
invitation=_lookup("Invitee@Example.com"),
invitation=lookup,
name="John Doe",
interface_language="en-US",
timezone="UTC",
@@ -198,30 +214,48 @@ class TestActivateInvitation:
service.activate(command, authenticated_account_id=None)
eligibility.get_freeze_type.assert_called_once_with("invitee@example.com")
tokens.revoke.assert_called_once_with(_lookup("invitee@example.com"))
accounts.activate.assert_called_once_with(
invitation,
boundaries["freeze"].assert_called_once_with("invitee@example.com")
boundaries["revoke"].assert_called_once_with(_lookup("invitee@example.com") if role is None else lookup)
boundaries["activate"].assert_called_once_with(
_invitation(role=role, requires_setup=True if role == "owner" else None),
role="normal",
setup=AccountSetup(name="John Doe", interface_language="en-US", timezone="UTC"),
)
membership_cache.invalidate.assert_called_once_with("workspace-1")
member_access_sync.sync.assert_called_once_with("workspace-1", "account-1")
boundaries["cache"].assert_called_once_with("workspace-1")
boundaries["sync"].assert_called_once_with("workspace-1", "account-1", operator_account_id=None)
with sqlite_session_factory() as session:
account = session.get(Account, "account-1")
membership = session.scalar(select(TenantAccountJoin))
assert account is not None
assert membership is not None
assert (account.name, account.interface_language, account.timezone) == ("John Doe", "en-US", "UTC")
assert account.status == AccountStatus.ACTIVE
assert account.initialized_at is not None
assert membership.role == TenantAccountRole.NORMAL
assert membership.current is True
def test_preserves_existing_membership_cache_and_ignores_setup_fields(self) -> None:
service, tokens, accounts, _, _, membership_cache, member_access_sync = _service()
tokens.find.return_value = _token()
invitation = _invitation(
account_status="active",
role="editor",
requires_setup=False,
def test_preserves_existing_membership_cache_and_ignores_setup_fields(
self,
service: AccountActivationService,
boundaries: dict[str, MagicMock],
sqlite_session_factory: sessionmaker[Session],
redis_transport: tuple[RedisClientWrapper, MagicMock],
) -> None:
with sqlite_session_factory.begin() as session:
account = session.get(Account, "account-1")
assert account is not None
account.status = AccountStatus.ACTIVE
session.add(
TenantAccountJoin(tenant_id="workspace-1", account_id="account-1", role=TenantAccountRole.EDITOR)
)
redis_transport[1].return_value = (
b'{"account_id":"account-1","email":"invitee@example.com","workspace_id":"workspace-1",'
b'"role":"editor","requires_setup":false}'
)
accounts.resolve.return_value = invitation
accounts.activate.return_value = ActivationPersistenceResult(membership_created=False)
service.activate(
ActivationCommand(
invitation=_lookup(),
invitation=InvitationLookup(workspace_id=None, email=None, token="token-1"),
name="Ignored",
interface_language="zh-Hans",
timezone="Asia/Shanghai",
@@ -229,6 +263,16 @@ class TestActivateInvitation:
authenticated_account_id="account-1",
)
accounts.activate.assert_called_once_with(invitation, role="editor", setup=None)
membership_cache.invalidate.assert_not_called()
member_access_sync.sync.assert_called_once_with("workspace-1", "account-1")
boundaries["activate"].assert_called_once_with(
_invitation(account_status="active", role="editor", requires_setup=False), role="editor", setup=None
)
boundaries["cache"].assert_not_called()
boundaries["sync"].assert_called_once_with("workspace-1", "account-1", operator_account_id=None)
with sqlite_session_factory() as session:
account = session.get(Account, "account-1")
memberships = session.scalars(select(TenantAccountJoin)).all()
assert account is not None
assert account.name == "Test User"
assert len(memberships) == 1
assert memberships[0].role == TenantAccountRole.EDITOR
assert memberships[0].current is True