github/OpenHands
1
0
Fork 0
mirror of https://github.com/OpenHands/OpenHands.git synced 2026-03-22 05:37:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(backend): implement get /api/organizations/{orgId}/members api (#12735)

Browse Source
This commit is contained in:
Hiep Le
2026-02-06 23:47:30 +07:00
committed by GitHub
parent d43ff82534
commit 872f41e3c0
8 changed files with 1300 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
enterprise/server/routes/org_models.py
View File

@@ -178,3 +178,21 @@ class OrgUpdate(BaseModel):
confirmation_mode: bool | None = None
enable_default_condenser: bool | None = None
condenser_max_size: int | None = Field(default=None, ge=20)
class OrgMemberResponse(BaseModel):
"""Response model for a single organization member."""
user_id: str
email: str | None
role_id: int
role_name: str
role_rank: int
status: str | None
class OrgMemberPage(BaseModel):
"""Paginated response for organization members."""
items: list[OrgMemberResponse]
next_page_id: str | None = None

68
enterprise/server/routes/orgs.py
View File

@@ -8,12 +8,14 @@ from server.routes.org_models import (
OrgAuthorizationError,
OrgCreate,
OrgDatabaseError,
OrgMemberPage,
OrgNameExistsError,
OrgNotFoundError,
OrgPage,
OrgResponse,
OrgUpdate,
)
from server.services.org_member_service import OrgMemberService
from storage.org_service import OrgService
from openhands.core.logger import openhands_logger as logger
@@ -402,3 +404,69 @@ async def update_org(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail='An unexpected error occurred',
)
@org_router.get('/{org_id}/members')
async def get_org_members(
org_id: str,
page_id: Annotated[
str | None,
Query(title='Optional next_page_id from the previously returned page'),
] = None,
limit: Annotated[
int,
Query(
title='The max number of results in the page',
gt=0,
lte=100,
),
] = 100,
current_user_id: str = Depends(get_user_id),
) -> OrgMemberPage:
"""Get all members of an organization with cursor-based pagination."""
try:
success, error_code, data = await OrgMemberService.get_org_members(
org_id=UUID(org_id),
current_user_id=UUID(current_user_id),
page_id=page_id,
limit=limit,
)
if not success:
error_map = {
'not_a_member': (
status.HTTP_403_FORBIDDEN,
'You are not a member of this organization',
),
'invalid_page_id': (
status.HTTP_400_BAD_REQUEST,
'Invalid page_id format',
),
}
status_code, detail = error_map.get(
error_code, (status.HTTP_500_INTERNAL_SERVER_ERROR, 'An error occurred')
)
raise HTTPException(status_code=status_code, detail=detail)
if data is None:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail='Failed to retrieve members',
)
return data
except HTTPException:
raise
except ValueError:
logger.exception('Invalid UUID format')
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail='Invalid organization ID format',
)
except Exception:
logger.exception('Error retrieving organization members')
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail='Failed to retrieve members',
)

67
enterprise/server/services/org_member_service.py Normal file
View File

@@ -0,0 +1,67 @@
"""Service for managing organization members."""
from uuid import UUID
from server.routes.org_models import OrgMemberPage, OrgMemberResponse
from storage.org_member_store import OrgMemberStore
class OrgMemberService:
"""Service for organization member operations."""
@staticmethod
async def get_org_members(
org_id: UUID,
current_user_id: UUID,
page_id: str | None = None,
limit: int = 100,
) -> tuple[bool, str | None, OrgMemberPage | None]:
"""Get organization members with authorization check.
Returns:
Tuple of (success, error_code, data). If success is True, error_code is None.
"""
# Verify current user is a member of the organization
requester_membership = OrgMemberStore.get_org_member(org_id, current_user_id)
if not requester_membership:
return False, 'not_a_member', None
# Parse page_id to get offset (page_id is offset encoded as string)
offset = 0
if page_id is not None:
try:
offset = int(page_id)
if offset < 0:
return False, 'invalid_page_id', None
except ValueError:
return False, 'invalid_page_id', None
# Call store to get paginated members
members, has_more = await OrgMemberStore.get_org_members_paginated(
org_id=org_id, offset=offset, limit=limit
)
# Transform data to response format
items = []
for member in members:
# Access user and role relationships (eagerly loaded)
user = member.user
role = member.role
items.append(
OrgMemberResponse(
user_id=str(member.user_id),
email=user.email if user else None,
role_id=member.role_id,
role_name=role.name if role else '',
role_rank=role.rank if role else 0,
status=member.status,
)
)
# Calculate next_page_id
next_page_id = None
if has_more:
next_page_id = str(offset + limit)
return True, None, OrgMemberPage(items=items, next_page_id=next_page_id)

34
enterprise/storage/org_member_store.py
View File

@@ -6,6 +6,7 @@ from typing import Optional
from uuid import UUID
from sqlalchemy import select
from sqlalchemy.orm import joinedload
from storage.database import a_session_maker, session_maker
from storage.org_member import OrgMember
from storage.user_settings import UserSettings
@@ -135,3 +136,36 @@ class OrgMemberStore:
if (normalized := c.name.lstrip('_')) and hasattr(user_settings, normalized)
}
return kwargs
@staticmethod
async def get_org_members_paginated(
org_id: UUID,
offset: int = 0,
limit: int = 100,
) -> tuple[list[OrgMember], bool]:
"""Get paginated list of organization members with user and role info.
Returns:
Tuple of (members_list, has_more) where has_more indicates if there are more results.
"""
async with a_session_maker() as session:
# Query for limit + 1 items to determine if there are more results
# Order by user_id for consistent pagination
query = (
select(OrgMember)
.options(joinedload(OrgMember.user), joinedload(OrgMember.role))
.filter(OrgMember.org_id == org_id)
.order_by(OrgMember.user_id)
.offset(offset)
.limit(limit + 1)
)
result = await session.execute(query)
members = list(result.scalars().all())
# Check if there are more results
has_more = len(members) > limit
if has_more:
# Remove the extra item
members = members[:limit]
return members, has_more

257
enterprise/tests/unit/server/routes/test_orgs.py
View File

@@ -21,10 +21,12 @@ with patch('storage.database.engine', create=True), patch(
LiteLLMIntegrationError,
OrgAuthorizationError,
OrgDatabaseError,
OrgMemberPage,
OrgMemberResponse,
OrgNameExistsError,
OrgNotFoundError,
)
from server.routes.orgs import org_router
from server.routes.orgs import get_org_members, org_router
from storage.org import Org
from openhands.server.user_auth import get_user_id
@@ -45,6 +47,18 @@ def mock_app():
return app
@pytest.fixture
def org_id():
"""Create a test organization ID."""
return str(uuid.uuid4())
@pytest.fixture
def current_user_id():
"""Create a test current user ID."""
return str(uuid.uuid4())
@pytest.mark.asyncio
async def test_create_org_success(mock_app):
"""
@@ -1729,3 +1743,244 @@ async def test_update_org_invalid_email_format(mock_update_app):
# Assert
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
class TestGetOrgMembersEndpoint:
"""Test cases for GET /api/organizations/{org_id}/members endpoint."""
@pytest.mark.asyncio
async def test_get_members_succeeds_returns_200(self, org_id, current_user_id):
"""Test that successful retrieval returns 200 with paginated members."""
# Arrange
mock_page = OrgMemberPage(
items=[
OrgMemberResponse(
user_id=str(uuid.uuid4()),
email='user1@example.com',
role_id=1,
role_name='owner',
role_rank=10,
status='active',
)
],
next_page_id=None,
)
with patch(
'server.routes.orgs.OrgMemberService.get_org_members',
AsyncMock(return_value=(True, None, mock_page)),
) as mock_get:
# Act
result = await get_org_members(
org_id=org_id,
page_id=None,
limit=100,
current_user_id=current_user_id,
)
# Assert
assert isinstance(result, OrgMemberPage)
assert len(result.items) == 1
assert result.next_page_id is None
mock_get.assert_called_once()
@pytest.mark.asyncio
async def test_not_a_member_returns_403(self, org_id, current_user_id):
"""Test that not being a member returns 403 Forbidden."""
# Arrange
with patch(
'server.routes.orgs.OrgMemberService.get_org_members',
AsyncMock(return_value=(False, 'not_a_member', None)),
):
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
await get_org_members(
org_id=org_id,
page_id=None,
limit=100,
current_user_id=current_user_id,
)
assert exc_info.value.status_code == status.HTTP_403_FORBIDDEN
assert 'not a member of this organization' in exc_info.value.detail
@pytest.mark.asyncio
async def test_invalid_page_id_returns_400(self, org_id, current_user_id):
"""Test that invalid page_id format returns 400 Bad Request."""
# Arrange
with patch(
'server.routes.orgs.OrgMemberService.get_org_members',
AsyncMock(return_value=(False, 'invalid_page_id', None)),
):
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
await get_org_members(
org_id=org_id,
page_id='invalid',
limit=100,
current_user_id=current_user_id,
)
assert exc_info.value.status_code == status.HTTP_400_BAD_REQUEST
assert 'Invalid page_id format' in exc_info.value.detail
@pytest.mark.asyncio
async def test_invalid_org_id_format_returns_400(self, current_user_id):
"""Test that invalid org_id UUID format returns 400 Bad Request."""
# Arrange
invalid_org_id = 'not-a-uuid'
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
await get_org_members(
org_id=invalid_org_id,
page_id=None,
limit=100,
current_user_id=current_user_id,
)
assert exc_info.value.status_code == status.HTTP_400_BAD_REQUEST
assert 'Invalid organization ID format' in exc_info.value.detail
@pytest.mark.asyncio
async def test_invalid_current_user_id_format_returns_400(self, org_id):
"""Test that invalid current_user_id UUID format returns 400 Bad Request."""
# Arrange
invalid_current_user_id = 'not-a-uuid'
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
await get_org_members(
org_id=org_id,
page_id=None,
limit=100,
current_user_id=invalid_current_user_id,
)
assert exc_info.value.status_code == status.HTTP_400_BAD_REQUEST
assert 'Invalid organization ID format' in exc_info.value.detail
@pytest.mark.asyncio
async def test_data_is_none_returns_500(self, org_id, current_user_id):
"""Test that None data returns 500 Internal Server Error."""
# Arrange
with patch(
'server.routes.orgs.OrgMemberService.get_org_members',
AsyncMock(return_value=(True, None, None)),
):
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
await get_org_members(
org_id=org_id,
page_id=None,
limit=100,
current_user_id=current_user_id,
)
assert exc_info.value.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
assert 'Failed to retrieve members' in exc_info.value.detail
@pytest.mark.asyncio
async def test_unknown_error_returns_500(self, org_id, current_user_id):
"""Test that unknown error returns 500 Internal Server Error."""
# Arrange
with patch(
'server.routes.orgs.OrgMemberService.get_org_members',
AsyncMock(return_value=(False, 'unknown_error', None)),
):
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
await get_org_members(
org_id=org_id,
page_id=None,
limit=100,
current_user_id=current_user_id,
)
assert exc_info.value.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
assert 'An error occurred' in exc_info.value.detail
@pytest.mark.asyncio
async def test_service_exception_returns_500(self, org_id, current_user_id):
"""Test that service exception returns 500 Internal Server Error."""
# Arrange
with patch(
'server.routes.orgs.OrgMemberService.get_org_members',
AsyncMock(side_effect=Exception('Database connection failed')),
):
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
await get_org_members(
org_id=org_id,
page_id=None,
limit=100,
current_user_id=current_user_id,
)
assert exc_info.value.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
assert 'Failed to retrieve members' in exc_info.value.detail
@pytest.mark.asyncio
async def test_http_exception_is_re_raised(self, org_id, current_user_id):
"""Test that HTTPException from service is re-raised."""
# Arrange
original_exception = HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail='Service temporarily unavailable',
)
with patch(
'server.routes.orgs.OrgMemberService.get_org_members',
AsyncMock(side_effect=original_exception),
):
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
await get_org_members(
org_id=org_id,
page_id=None,
limit=100,
current_user_id=current_user_id,
)
assert exc_info.value.status_code == status.HTTP_503_SERVICE_UNAVAILABLE
assert exc_info.value.detail == 'Service temporarily unavailable'
@pytest.mark.asyncio
async def test_pagination_with_page_id(self, org_id, current_user_id):
"""Test that pagination works with page_id parameter."""
# Arrange
mock_page = OrgMemberPage(
items=[
OrgMemberResponse(
user_id=str(uuid.uuid4()),
email='user2@example.com',
role_id=2,
role_name='admin',
role_rank=20,
status='active',
)
],
next_page_id='200',
)
with patch(
'server.routes.orgs.OrgMemberService.get_org_members',
AsyncMock(return_value=(True, None, mock_page)),
) as mock_get:
# Act
result = await get_org_members(
org_id=org_id,
page_id='100',
limit=100,
current_user_id=current_user_id,
)
# Assert
assert isinstance(result, OrgMemberPage)
assert result.next_page_id == '200'
mock_get.assert_called_once_with(
org_id=uuid.UUID(org_id),
current_user_id=uuid.UUID(current_user_id),
page_id='100',
limit=100,
)

500
enterprise/tests/unit/server/services/test_org_member_service.py Normal file
View File

@@ -0,0 +1,500 @@
"""Tests for OrgMemberService."""
import uuid
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from server.services.org_member_service import OrgMemberService
from storage.org_member import OrgMember
from storage.role import Role
@pytest.fixture
def org_id():
"""Create a test organization ID."""
return uuid.uuid4()
@pytest.fixture
def current_user_id():
"""Create a test current user ID."""
return uuid.uuid4()
@pytest.fixture
def target_user_id():
"""Create a test target user ID."""
return uuid.uuid4()
@pytest.fixture
def owner_role():
"""Create a mock owner role."""
role = MagicMock(spec=Role)
role.id = 1
role.name = 'owner'
role.rank = 10
return role
@pytest.fixture
def admin_role():
"""Create a mock admin role."""
role = MagicMock(spec=Role)
role.id = 2
role.name = 'admin'
role.rank = 20
return role
@pytest.fixture
def user_role():
"""Create a mock user role."""
role = MagicMock(spec=Role)
role.id = 3
role.name = 'user'
role.rank = 1000
return role
@pytest.fixture
def requester_membership_owner(org_id, current_user_id, owner_role):
"""Create a mock requester membership with owner role."""
membership = MagicMock(spec=OrgMember)
membership.org_id = org_id
membership.user_id = current_user_id
membership.role_id = owner_role.id
return membership
@pytest.fixture
def requester_membership_admin(org_id, current_user_id, admin_role):
"""Create a mock requester membership with admin role."""
membership = MagicMock(spec=OrgMember)
membership.org_id = org_id
membership.user_id = current_user_id
membership.role_id = admin_role.id
return membership
@pytest.fixture
def target_membership_user(org_id, target_user_id, user_role):
"""Create a mock target membership with user role."""
membership = MagicMock(spec=OrgMember)
membership.org_id = org_id
membership.user_id = target_user_id
membership.role_id = user_role.id
return membership
@pytest.fixture
def target_membership_admin(org_id, target_user_id, admin_role):
"""Create a mock target membership with admin role."""
membership = MagicMock(spec=OrgMember)
membership.org_id = org_id
membership.user_id = target_user_id
membership.role_id = admin_role.id
return membership
class TestOrgMemberServiceGetOrgMembers:
"""Test cases for OrgMemberService.get_org_members."""
@pytest.fixture
def mock_user(self):
"""Create a mock user."""
user = MagicMock()
user.email = 'test@example.com'
return user
@pytest.fixture
def mock_role(self):
"""Create a mock role."""
role = MagicMock(spec=Role)
role.id = 1
role.name = 'owner'
role.rank = 10
return role
@pytest.fixture
def mock_org_member(self, org_id, current_user_id, mock_user, mock_role):
"""Create a mock org member with user and role."""
member = MagicMock(spec=OrgMember)
member.org_id = org_id
member.user_id = current_user_id
member.role_id = mock_role.id
member.status = 'active'
member.user = mock_user
member.role = mock_role
return member
@pytest.mark.asyncio
async def test_get_members_succeeds_returns_paginated_data(
self, org_id, current_user_id, mock_org_member, requester_membership_owner
):
"""Test that successful retrieval returns paginated member data."""
# Arrange
from server.routes.org_models import OrgMemberPage
with (
patch(
'server.services.org_member_service.OrgMemberStore.get_org_member'
) as mock_get_member,
patch(
'server.services.org_member_service.OrgMemberStore.get_org_members_paginated',
new_callable=AsyncMock,
) as mock_get_paginated,
):
mock_get_member.return_value = requester_membership_owner
mock_get_paginated.return_value = ([mock_org_member], False)
# Act
success, error_code, data = await OrgMemberService.get_org_members(
org_id=org_id,
current_user_id=current_user_id,
page_id=None,
limit=100,
)
# Assert
assert success is True
assert error_code is None
assert data is not None
assert isinstance(data, OrgMemberPage)
assert len(data.items) == 1
assert data.next_page_id is None
assert data.items[0].user_id == str(current_user_id)
assert data.items[0].email == 'test@example.com'
assert data.items[0].role_id == 1
assert data.items[0].role_name == 'owner'
assert data.items[0].role_rank == 10
assert data.items[0].status == 'active'
@pytest.mark.asyncio
async def test_user_not_a_member_returns_error(self, org_id, current_user_id):
"""Test that retrieval fails when user is not a member."""
# Arrange
with patch(
'server.services.org_member_service.OrgMemberStore.get_org_member'
) as mock_get_member:
mock_get_member.return_value = None
# Act
success, error_code, data = await OrgMemberService.get_org_members(
org_id=org_id,
current_user_id=current_user_id,
page_id=None,
limit=100,
)
# Assert
assert success is False
assert error_code == 'not_a_member'
assert data is None
@pytest.mark.asyncio
async def test_invalid_page_id_negative_returns_error(
self, org_id, current_user_id, requester_membership_owner
):
"""Test that negative page_id returns error."""
# Arrange
with patch(
'server.services.org_member_service.OrgMemberStore.get_org_member'
) as mock_get_member:
mock_get_member.return_value = requester_membership_owner
# Act
success, error_code, data = await OrgMemberService.get_org_members(
org_id=org_id,
current_user_id=current_user_id,
page_id='-1',
limit=100,
)
# Assert
assert success is False
assert error_code == 'invalid_page_id'
assert data is None
@pytest.mark.asyncio
async def test_invalid_page_id_non_integer_returns_error(
self, org_id, current_user_id, requester_membership_owner
):
"""Test that non-integer page_id returns error."""
# Arrange
with patch(
'server.services.org_member_service.OrgMemberStore.get_org_member'
) as mock_get_member:
mock_get_member.return_value = requester_membership_owner
# Act
success, error_code, data = await OrgMemberService.get_org_members(
org_id=org_id,
current_user_id=current_user_id,
page_id='not-a-number',
limit=100,
)
# Assert
assert success is False
assert error_code == 'invalid_page_id'
assert data is None
@pytest.mark.asyncio
async def test_first_page_pagination_no_page_id(
self, org_id, current_user_id, mock_org_member, requester_membership_owner
):
"""Test first page pagination when page_id is None."""
# Arrange
with (
patch(
'server.services.org_member_service.OrgMemberStore.get_org_member'
) as mock_get_member,
patch(
'server.services.org_member_service.OrgMemberStore.get_org_members_paginated',
new_callable=AsyncMock,
) as mock_get_paginated,
):
mock_get_member.return_value = requester_membership_owner
mock_get_paginated.return_value = ([mock_org_member], False)
# Act
success, error_code, data = await OrgMemberService.get_org_members(
org_id=org_id,
current_user_id=current_user_id,
page_id=None,
limit=100,
)
# Assert
assert success is True
assert data is not None
assert data.next_page_id is None
mock_get_paginated.assert_called_once_with(
org_id=org_id, offset=0, limit=100
)
@pytest.mark.asyncio
async def test_next_page_pagination_with_page_id(
self, org_id, current_user_id, mock_org_member, requester_membership_owner
):
"""Test next page pagination when page_id is provided."""
# Arrange
with (
patch(
'server.services.org_member_service.OrgMemberStore.get_org_member'
) as mock_get_member,
patch(
'server.services.org_member_service.OrgMemberStore.get_org_members_paginated',
new_callable=AsyncMock,
) as mock_get_paginated,
):
mock_get_member.return_value = requester_membership_owner
mock_get_paginated.return_value = ([mock_org_member], True)
# Act
success, error_code, data = await OrgMemberService.get_org_members(
org_id=org_id,
current_user_id=current_user_id,
page_id='100',
limit=50,
)
# Assert
assert success is True
assert data is not None
assert data.next_page_id == '150' # offset (100) + limit (50)
mock_get_paginated.assert_called_once_with(
org_id=org_id, offset=100, limit=50
)
@pytest.mark.asyncio
async def test_last_page_has_more_false(
self, org_id, current_user_id, mock_org_member, requester_membership_owner
):
"""Test last page when has_more is False."""
# Arrange
with (
patch(
'server.services.org_member_service.OrgMemberStore.get_org_member'
) as mock_get_member,
patch(
'server.services.org_member_service.OrgMemberStore.get_org_members_paginated',
new_callable=AsyncMock,
) as mock_get_paginated,
):
mock_get_member.return_value = requester_membership_owner
mock_get_paginated.return_value = ([mock_org_member], False)
# Act
success, error_code, data = await OrgMemberService.get_org_members(
org_id=org_id,
current_user_id=current_user_id,
page_id='200',
limit=100,
)
# Assert
assert success is True
assert data is not None
assert data.next_page_id is None
@pytest.mark.asyncio
async def test_empty_organization_no_members(
self, org_id, current_user_id, requester_membership_owner
):
"""Test empty organization with no members."""
# Arrange
with (
patch(
'server.services.org_member_service.OrgMemberStore.get_org_member'
) as mock_get_member,
patch(
'server.services.org_member_service.OrgMemberStore.get_org_members_paginated',
new_callable=AsyncMock,
) as mock_get_paginated,
):
mock_get_member.return_value = requester_membership_owner
mock_get_paginated.return_value = ([], False)
# Act
success, error_code, data = await OrgMemberService.get_org_members(
org_id=org_id,
current_user_id=current_user_id,
page_id=None,
limit=100,
)
# Assert
assert success is True
assert data is not None
assert len(data.items) == 0
assert data.next_page_id is None
@pytest.mark.asyncio
async def test_missing_user_relationship_handles_gracefully(
self, org_id, current_user_id, mock_role, requester_membership_owner
):
"""Test that missing user relationship is handled gracefully."""
# Arrange
member_no_user = MagicMock(spec=OrgMember)
member_no_user.org_id = org_id
member_no_user.user_id = current_user_id
member_no_user.role_id = mock_role.id
member_no_user.status = 'active'
member_no_user.user = None
member_no_user.role = mock_role
with (
patch(
'server.services.org_member_service.OrgMemberStore.get_org_member'
) as mock_get_member,
patch(
'server.services.org_member_service.OrgMemberStore.get_org_members_paginated',
new_callable=AsyncMock,
) as mock_get_paginated,
):
mock_get_member.return_value = requester_membership_owner
mock_get_paginated.return_value = ([member_no_user], False)
# Act
success, error_code, data = await OrgMemberService.get_org_members(
org_id=org_id,
current_user_id=current_user_id,
page_id=None,
limit=100,
)
# Assert
assert success is True
assert data is not None
assert len(data.items) == 1
assert data.items[0].email is None
@pytest.mark.asyncio
async def test_missing_role_relationship_handles_gracefully(
self, org_id, current_user_id, mock_user, requester_membership_owner
):
"""Test that missing role relationship is handled gracefully."""
# Arrange
member_no_role = MagicMock(spec=OrgMember)
member_no_role.org_id = org_id
member_no_role.user_id = current_user_id
member_no_role.role_id = 1
member_no_role.status = 'active'
member_no_role.user = mock_user
member_no_role.role = None
with (
patch(
'server.services.org_member_service.OrgMemberStore.get_org_member'
) as mock_get_member,
patch(
'server.services.org_member_service.OrgMemberStore.get_org_members_paginated',
new_callable=AsyncMock,
) as mock_get_paginated,
):
mock_get_member.return_value = requester_membership_owner
mock_get_paginated.return_value = ([member_no_role], False)
# Act
success, error_code, data = await OrgMemberService.get_org_members(
org_id=org_id,
current_user_id=current_user_id,
page_id=None,
limit=100,
)
# Assert
assert success is True
assert data is not None
assert len(data.items) == 1
assert data.items[0].role_name == ''
assert data.items[0].role_rank == 0
@pytest.mark.asyncio
async def test_multiple_members_returns_all(
self, org_id, current_user_id, mock_user, mock_role, requester_membership_owner
):
"""Test that multiple members are returned correctly."""
# Arrange
member1 = MagicMock(spec=OrgMember)
member1.org_id = org_id
member1.user_id = current_user_id
member1.role_id = mock_role.id
member1.status = 'active'
member1.user = mock_user
member1.role = mock_role
member2 = MagicMock(spec=OrgMember)
member2.org_id = org_id
member2.user_id = uuid.uuid4()
member2.role_id = mock_role.id
member2.status = 'active'
member2.user = mock_user
member2.role = mock_role
with (
patch(
'server.services.org_member_service.OrgMemberStore.get_org_member'
) as mock_get_member,
patch(
'server.services.org_member_service.OrgMemberStore.get_org_members_paginated',
new_callable=AsyncMock,
) as mock_get_paginated,
):
mock_get_member.return_value = requester_membership_owner
mock_get_paginated.return_value = ([member1, member2], False)
# Act
success, error_code, data = await OrgMemberService.get_org_members(
org_id=org_id,
current_user_id=current_user_id,
page_id=None,
limit=100,
)
# Assert
assert success is True
assert data is not None
assert len(data.items) == 2

355
enterprise/tests/unit/test_org_member_store.py
View File

@@ -1,8 +1,15 @@
import uuid
from unittest.mock import patch
import pytest
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.pool import StaticPool
# Mock the database module before importing OrgMemberStore
with patch('storage.database.engine'), patch('storage.database.a_engine'):
with patch('storage.database.engine', create=True), patch(
'storage.database.a_engine', create=True
):
from storage.base import Base
from storage.org import Org
from storage.org_member import OrgMember
from storage.org_member_store import OrgMemberStore
@@ -10,6 +17,31 @@ with patch('storage.database.engine'), patch('storage.database.a_engine'):
from storage.user import User
@pytest.fixture
async def async_engine():
"""Create an async SQLite engine for testing."""
engine = create_async_engine(
'sqlite+aiosqlite:///:memory:',
poolclass=StaticPool,
connect_args={'check_same_thread': False},
echo=False,
)
# Create all tables
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield engine
await engine.dispose()
@pytest.fixture
async def async_session_maker(async_engine):
"""Create an async session maker for testing."""
return async_sessionmaker(async_engine, class_=AsyncSession, expire_on_commit=False)
def test_get_org_members(session_maker):
# Test getting org_members by org ID
with session_maker() as session:
@@ -251,3 +283,324 @@ def test_remove_user_from_org_not_found(session_maker):
with patch('storage.org_member_store.session_maker', session_maker):
result = OrgMemberStore.remove_user_from_org(uuid4(), 99999)
assert result is False
@pytest.mark.asyncio
async def test_get_org_members_paginated_basic(async_session_maker):
"""Test basic pagination returns correct number of items."""
# Arrange
async with async_session_maker() as session:
org = Org(name='test-org')
session.add(org)
await session.flush()
role = Role(name='admin', rank=1)
session.add(role)
await session.flush()
# Create 5 users
users = [
User(id=uuid.uuid4(), current_org_id=org.id, email=f'user{i}@example.com')
for i in range(5)
]
session.add_all(users)
await session.flush()
# Create org members
org_members = [
OrgMember(
org_id=org.id,
user_id=user.id,
role_id=role.id,
llm_api_key=f'test-key-{i}',
status='active',
)
for i, user in enumerate(users)
]
session.add_all(org_members)
await session.commit()
org_id = org.id
# Act
with patch('storage.org_member_store.a_session_maker', async_session_maker):
members, has_more = await OrgMemberStore.get_org_members_paginated(
org_id=org_id, offset=0, limit=3
)
# Assert
assert len(members) == 3
assert has_more is True
# Verify user and role relationships are loaded
assert all(member.user is not None for member in members)
assert all(member.role is not None for member in members)
@pytest.mark.asyncio
async def test_get_org_members_paginated_no_more(async_session_maker):
"""Test pagination when there are no more results."""
# Arrange
async with async_session_maker() as session:
org = Org(name='test-org')
session.add(org)
await session.flush()
role = Role(name='admin', rank=1)
session.add(role)
await session.flush()
# Create 3 users
users = [
User(id=uuid.uuid4(), current_org_id=org.id, email=f'user{i}@example.com')
for i in range(3)
]
session.add_all(users)
await session.flush()
# Create org members
org_members = [
OrgMember(
org_id=org.id,
user_id=user.id,
role_id=role.id,
llm_api_key=f'test-key-{i}',
status='active',
)
for i, user in enumerate(users)
]
session.add_all(org_members)
await session.commit()
org_id = org.id
# Act
with patch('storage.org_member_store.a_session_maker', async_session_maker):
members, has_more = await OrgMemberStore.get_org_members_paginated(
org_id=org_id, offset=0, limit=5
)
# Assert
assert len(members) == 3
assert has_more is False
@pytest.mark.asyncio
async def test_get_org_members_paginated_exact_limit(async_session_maker):
"""Test pagination when results exactly match limit."""
# Arrange
async with async_session_maker() as session:
org = Org(name='test-org')
session.add(org)
await session.flush()
role = Role(name='admin', rank=1)
session.add(role)
await session.flush()
# Create exactly 5 users
users = [
User(id=uuid.uuid4(), current_org_id=org.id, email=f'user{i}@example.com')
for i in range(5)
]
session.add_all(users)
await session.flush()
# Create org members
org_members = [
OrgMember(
org_id=org.id,
user_id=user.id,
role_id=role.id,
llm_api_key=f'test-key-{i}',
status='active',
)
for i, user in enumerate(users)
]
session.add_all(org_members)
await session.commit()
org_id = org.id
# Act
with patch('storage.org_member_store.a_session_maker', async_session_maker):
members, has_more = await OrgMemberStore.get_org_members_paginated(
org_id=org_id, offset=0, limit=5
)
# Assert
assert len(members) == 5
assert has_more is False
@pytest.mark.asyncio
async def test_get_org_members_paginated_with_offset(async_session_maker):
"""Test pagination with offset skips correct number of items."""
# Arrange
async with async_session_maker() as session:
org = Org(name='test-org')
session.add(org)
await session.flush()
role = Role(name='admin', rank=1)
session.add(role)
await session.flush()
# Create 10 users
users = [
User(id=uuid.uuid4(), current_org_id=org.id, email=f'user{i}@example.com')
for i in range(10)
]
session.add_all(users)
await session.flush()
# Create org members
org_members = [
OrgMember(
org_id=org.id,
user_id=user.id,
role_id=role.id,
llm_api_key=f'test-key-{i}',
status='active',
)
for i, user in enumerate(users)
]
session.add_all(org_members)
await session.commit()
org_id = org.id
# Act - Get first page
with patch('storage.org_member_store.a_session_maker', async_session_maker):
first_page, has_more_first = await OrgMemberStore.get_org_members_paginated(
org_id=org_id, offset=0, limit=3
)
# Get second page
second_page, has_more_second = await OrgMemberStore.get_org_members_paginated(
org_id=org_id, offset=3, limit=3
)
# Assert
assert len(first_page) == 3
assert has_more_first is True
assert len(second_page) == 3
assert has_more_second is True
# Verify no overlap between pages
first_user_ids = {member.user_id for member in first_page}
second_user_ids = {member.user_id for member in second_page}
assert first_user_ids.isdisjoint(second_user_ids)
@pytest.mark.asyncio
async def test_get_org_members_paginated_empty_org(async_session_maker):
"""Test pagination with empty organization returns empty list."""
# Arrange
async with async_session_maker() as session:
org = Org(name='test-org')
session.add(org)
await session.commit()
org_id = org.id
# Act
with patch('storage.org_member_store.a_session_maker', async_session_maker):
members, has_more = await OrgMemberStore.get_org_members_paginated(
org_id=org_id, offset=0, limit=10
)
# Assert
assert len(members) == 0
assert has_more is False
@pytest.mark.asyncio
async def test_get_org_members_paginated_ordering(async_session_maker):
"""Test that pagination orders results by user_id."""
# Arrange
async with async_session_maker() as session:
org = Org(name='test-org')
session.add(org)
await session.flush()
role = Role(name='admin', rank=1)
session.add(role)
await session.flush()
# Create users with specific IDs to test ordering
user_ids = [uuid.uuid4() for _ in range(5)]
user_ids.sort() # Sort to verify ordering
users = [
User(id=user_id, current_org_id=org.id, email=f'user{i}@example.com')
for i, user_id in enumerate(user_ids)
]
session.add_all(users)
await session.flush()
# Create org members in reverse order to test that ordering works
org_members = [
OrgMember(
org_id=org.id,
user_id=user_id,
role_id=role.id,
llm_api_key=f'test-key-{i}',
status='active',
)
for i, user_id in enumerate(reversed(user_ids))
]
session.add_all(org_members)
await session.commit()
org_id = org.id
# Act
with patch('storage.org_member_store.a_session_maker', async_session_maker):
members, has_more = await OrgMemberStore.get_org_members_paginated(
org_id=org_id, offset=0, limit=10
)
# Assert
assert len(members) == 5
# Verify members are ordered by user_id
member_user_ids = [member.user_id for member in members]
assert member_user_ids == sorted(member_user_ids)
@pytest.mark.asyncio
async def test_get_org_members_paginated_eager_loading(async_session_maker):
"""Test that user and role relationships are eagerly loaded."""
# Arrange
async with async_session_maker() as session:
org = Org(name='test-org')
session.add(org)
await session.flush()
role = Role(name='owner', rank=10)
session.add(role)
await session.flush()
user = User(id=uuid.uuid4(), current_org_id=org.id, email='test@example.com')
session.add(user)
await session.flush()
org_member = OrgMember(
org_id=org.id,
user_id=user.id,
role_id=role.id,
llm_api_key='test-key',
status='active',
)
session.add(org_member)
await session.commit()
org_id = org.id
# Act
with patch('storage.org_member_store.a_session_maker', async_session_maker):
members, has_more = await OrgMemberStore.get_org_members_paginated(
org_id=org_id, offset=0, limit=10
)
# Assert
assert len(members) == 1
member = members[0]
# Verify relationships are loaded (not lazy)
assert member.user is not None
assert member.user.email == 'test@example.com'
assert member.role is not None
assert member.role.name == 'owner'
assert member.role.rank == 10

4
enterprise/tests/unit/test_proactive_conversation_starters.py
View File

@@ -3,7 +3,9 @@ from unittest.mock import MagicMock, patch
import pytest
# Mock the database module before importing
with patch('storage.database.engine'), patch('storage.database.a_engine'):
with patch('storage.database.engine', create=True), patch(
'storage.database.a_engine', create=True
):
from integrations.github.github_view import get_user_proactive_conversation_setting
from storage.org import Org