99 lines
3.6 KiB
Python
99 lines
3.6 KiB
Python
"""Unit tests for OrganizationService."""
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest # type: ignore[import-not-found]
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.organization import Organization
|
|
from app.models.organization_member import OrganizationMember, OrganizationRole
|
|
from app.repositories.org_repo import OrganizationRepository
|
|
from app.services.organization_service import (
|
|
OrganizationAccessDeniedError,
|
|
OrganizationContext,
|
|
OrganizationContextMissingError,
|
|
OrganizationForbiddenError,
|
|
OrganizationService,
|
|
)
|
|
|
|
|
|
class StubOrganizationRepository(OrganizationRepository):
|
|
"""Simple in-memory stand-in for OrganizationRepository."""
|
|
|
|
def __init__(self, membership: OrganizationMember | None) -> None:
|
|
super().__init__(session=MagicMock(spec=AsyncSession))
|
|
self._membership = membership
|
|
|
|
async def get_membership(self, organization_id: int, user_id: int) -> OrganizationMember | None: # pragma: no cover - helper
|
|
if (
|
|
self._membership
|
|
and self._membership.organization_id == organization_id
|
|
and self._membership.user_id == user_id
|
|
):
|
|
return self._membership
|
|
return None
|
|
|
|
|
|
def make_membership(role: OrganizationRole, *, organization_id: int = 1, user_id: int = 10) -> OrganizationMember:
|
|
organization = Organization(name="Acme Inc")
|
|
organization.id = organization_id
|
|
membership = OrganizationMember(
|
|
organization_id=organization_id,
|
|
user_id=user_id,
|
|
role=role,
|
|
)
|
|
membership.organization = organization
|
|
return membership
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_context_success() -> None:
|
|
membership = make_membership(OrganizationRole.MANAGER)
|
|
service = OrganizationService(StubOrganizationRepository(membership))
|
|
|
|
context = await service.get_context(user_id=membership.user_id, organization_id=membership.organization_id)
|
|
|
|
assert context.organization_id == membership.organization_id
|
|
assert context.role == OrganizationRole.MANAGER
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_context_missing_header() -> None:
|
|
service = OrganizationService(StubOrganizationRepository(None))
|
|
|
|
with pytest.raises(OrganizationContextMissingError):
|
|
await service.get_context(user_id=1, organization_id=None)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_context_access_denied() -> None:
|
|
service = OrganizationService(StubOrganizationRepository(None))
|
|
|
|
with pytest.raises(OrganizationAccessDeniedError):
|
|
await service.get_context(user_id=1, organization_id=99)
|
|
|
|
|
|
def test_ensure_can_manage_settings_blocks_manager() -> None:
|
|
membership = make_membership(OrganizationRole.MANAGER)
|
|
organization = membership.organization
|
|
assert organization is not None
|
|
context = OrganizationContext(organization=organization, membership=membership)
|
|
service = OrganizationService(StubOrganizationRepository(membership))
|
|
|
|
with pytest.raises(OrganizationForbiddenError):
|
|
service.ensure_can_manage_settings(context)
|
|
|
|
|
|
def test_member_must_own_entity() -> None:
|
|
membership = make_membership(OrganizationRole.MEMBER)
|
|
organization = membership.organization
|
|
assert organization is not None
|
|
context = OrganizationContext(organization=organization, membership=membership)
|
|
service = OrganizationService(StubOrganizationRepository(membership))
|
|
|
|
with pytest.raises(OrganizationForbiddenError):
|
|
service.ensure_member_owns_entity(context=context, owner_id=999)
|
|
|
|
# Same owner should pass silently.
|
|
service.ensure_member_owns_entity(context=context, owner_id=membership.user_id) |