mirror of
https://github.com/OpenHands/OpenHands.git
synced 2026-03-22 13:47:19 +08:00
feat(backend): org get me route (#12760)
Co-authored-by: openhands <openhands@all-hands.dev> Co-authored-by: hieptl <hieptl.developer@gmail.com>
This commit is contained in:
@@ -1,7 +1,9 @@
|
||||
from typing import Annotated
|
||||
|
||||
from pydantic import BaseModel, EmailStr, Field, StringConstraints
|
||||
from pydantic import BaseModel, EmailStr, Field, SecretStr, StringConstraints
|
||||
from storage.org import Org
|
||||
from storage.org_member import OrgMember
|
||||
from storage.role import Role
|
||||
|
||||
|
||||
class OrgCreationError(Exception):
|
||||
@@ -51,6 +53,23 @@ class OrgNotFoundError(Exception):
|
||||
super().__init__(f'Organization with id "{org_id}" not found')
|
||||
|
||||
|
||||
class OrgMemberNotFoundError(Exception):
|
||||
"""Raised when a member is not found in an organization."""
|
||||
|
||||
def __init__(self, org_id: str, user_id: str):
|
||||
self.org_id = org_id
|
||||
self.user_id = user_id
|
||||
super().__init__(f'Member not found in organization "{org_id}"')
|
||||
|
||||
|
||||
class RoleNotFoundError(Exception):
|
||||
"""Raised when a role is not found."""
|
||||
|
||||
def __init__(self, role_id: int):
|
||||
self.role_id = role_id
|
||||
super().__init__(f'Role with id "{role_id}" not found')
|
||||
|
||||
|
||||
class OrgCreate(BaseModel):
|
||||
"""Request model for creating a new organization."""
|
||||
|
||||
@@ -196,3 +215,55 @@ class OrgMemberPage(BaseModel):
|
||||
|
||||
items: list[OrgMemberResponse]
|
||||
next_page_id: str | None = None
|
||||
|
||||
|
||||
class MeResponse(BaseModel):
|
||||
"""Response model for the current user's membership in an organization."""
|
||||
|
||||
org_id: str
|
||||
user_id: str
|
||||
email: str
|
||||
role: str
|
||||
llm_api_key: str
|
||||
max_iterations: int | None = None
|
||||
llm_model: str | None = None
|
||||
llm_api_key_for_byor: str | None = None
|
||||
llm_base_url: str | None = None
|
||||
status: str | None = None
|
||||
|
||||
@staticmethod
|
||||
def _mask_key(secret: SecretStr | None) -> str:
|
||||
"""Mask an API key, showing only last 4 characters."""
|
||||
if secret is None:
|
||||
return ''
|
||||
raw = secret.get_secret_value()
|
||||
if not raw:
|
||||
return ''
|
||||
if len(raw) <= 4:
|
||||
return '****'
|
||||
return '****' + raw[-4:]
|
||||
|
||||
@classmethod
|
||||
def from_org_member(cls, member: OrgMember, role: Role, email: str) -> 'MeResponse':
|
||||
"""Create a MeResponse from an OrgMember, Role, and user email.
|
||||
|
||||
Args:
|
||||
member: The OrgMember entity
|
||||
role: The Role entity (provides role name)
|
||||
email: The user's email address
|
||||
|
||||
Returns:
|
||||
MeResponse with masked API keys
|
||||
"""
|
||||
return cls(
|
||||
org_id=str(member.org_id),
|
||||
user_id=str(member.user_id),
|
||||
email=email,
|
||||
role=role.name,
|
||||
llm_api_key=cls._mask_key(member.llm_api_key),
|
||||
max_iterations=member.max_iterations,
|
||||
llm_model=member.llm_model,
|
||||
llm_api_key_for_byor=cls._mask_key(member.llm_api_key_for_byor) or None,
|
||||
llm_base_url=member.llm_base_url,
|
||||
status=member.status,
|
||||
)
|
||||
|
||||
@@ -5,15 +5,18 @@ from fastapi import APIRouter, Depends, HTTPException, Query, status
|
||||
from server.email_validation import get_admin_user_id
|
||||
from server.routes.org_models import (
|
||||
LiteLLMIntegrationError,
|
||||
MeResponse,
|
||||
OrgAuthorizationError,
|
||||
OrgCreate,
|
||||
OrgDatabaseError,
|
||||
OrgMemberNotFoundError,
|
||||
OrgMemberPage,
|
||||
OrgNameExistsError,
|
||||
OrgNotFoundError,
|
||||
OrgPage,
|
||||
OrgResponse,
|
||||
OrgUpdate,
|
||||
RoleNotFoundError,
|
||||
)
|
||||
from server.services.org_member_service import OrgMemberService
|
||||
from storage.org_service import OrgService
|
||||
@@ -232,6 +235,65 @@ async def get_org(
|
||||
)
|
||||
|
||||
|
||||
@org_router.get('/{org_id}/me', response_model=MeResponse)
|
||||
async def get_me(
|
||||
org_id: UUID,
|
||||
user_id: str = Depends(get_user_id),
|
||||
) -> MeResponse:
|
||||
"""Get the current user's membership record for an organization.
|
||||
|
||||
Returns the authenticated user's role, status, email, and LLM override
|
||||
fields (with masked API keys) within the specified organization.
|
||||
|
||||
Args:
|
||||
org_id: Organization ID (UUID)
|
||||
user_id: Authenticated user ID (injected by dependency)
|
||||
|
||||
Returns:
|
||||
MeResponse: The user's membership data
|
||||
|
||||
Raises:
|
||||
HTTPException: 404 if user is not a member or org doesn't exist
|
||||
HTTPException: 500 if retrieval fails
|
||||
"""
|
||||
logger.info(
|
||||
'Retrieving current member details',
|
||||
extra={'user_id': user_id, 'org_id': str(org_id)},
|
||||
)
|
||||
|
||||
try:
|
||||
user_uuid = UUID(user_id)
|
||||
return OrgMemberService.get_me(org_id, user_uuid)
|
||||
|
||||
except OrgMemberNotFoundError:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f'Organization with id "{org_id}" not found',
|
||||
)
|
||||
except RoleNotFoundError as e:
|
||||
logger.exception(
|
||||
'Role not found for org member',
|
||||
extra={
|
||||
'user_id': user_id,
|
||||
'org_id': str(org_id),
|
||||
'role_id': e.role_id,
|
||||
},
|
||||
)
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail='An unexpected error occurred',
|
||||
)
|
||||
except Exception as e:
|
||||
logger.exception(
|
||||
'Unexpected error retrieving member details',
|
||||
extra={'user_id': user_id, 'org_id': str(org_id), 'error': str(e)},
|
||||
)
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail='An unexpected error occurred',
|
||||
)
|
||||
|
||||
|
||||
@org_router.delete('/{org_id}', status_code=status.HTTP_200_OK)
|
||||
async def delete_org(
|
||||
org_id: UUID,
|
||||
|
||||
@@ -2,9 +2,16 @@
|
||||
|
||||
from uuid import UUID
|
||||
|
||||
from server.routes.org_models import OrgMemberPage, OrgMemberResponse
|
||||
from server.routes.org_models import (
|
||||
MeResponse,
|
||||
OrgMemberNotFoundError,
|
||||
OrgMemberPage,
|
||||
OrgMemberResponse,
|
||||
RoleNotFoundError,
|
||||
)
|
||||
from storage.org_member_store import OrgMemberStore
|
||||
from storage.role_store import RoleStore
|
||||
from storage.user_store import UserStore
|
||||
|
||||
from openhands.utils.async_utils import call_sync_from_async
|
||||
|
||||
@@ -16,6 +23,40 @@ ADMIN_RANK = 20
|
||||
class OrgMemberService:
|
||||
"""Service for organization member operations."""
|
||||
|
||||
@staticmethod
|
||||
def get_me(org_id: UUID, user_id: UUID) -> MeResponse:
|
||||
"""Get the current user's membership record for an organization.
|
||||
|
||||
Retrieves the authenticated user's role, status, email, and LLM override
|
||||
fields (with masked API keys) within the specified organization.
|
||||
|
||||
Args:
|
||||
org_id: Organization ID (UUID)
|
||||
user_id: User ID (UUID)
|
||||
|
||||
Returns:
|
||||
MeResponse: The user's membership data with masked API keys
|
||||
|
||||
Raises:
|
||||
OrgMemberNotFoundError: If user is not a member of the organization
|
||||
RoleNotFoundError: If the role associated with the member is not found
|
||||
"""
|
||||
# Look up the user's membership in this org
|
||||
org_member = OrgMemberStore.get_org_member(org_id, user_id)
|
||||
if org_member is None:
|
||||
raise OrgMemberNotFoundError(str(org_id), str(user_id))
|
||||
|
||||
# Resolve role name from role_id
|
||||
role = RoleStore.get_role_by_id(org_member.role_id)
|
||||
if role is None:
|
||||
raise RoleNotFoundError(org_member.role_id)
|
||||
|
||||
# Get user email
|
||||
user = UserStore.get_user_by_id(str(user_id))
|
||||
email = user.email if user and user.email else ''
|
||||
|
||||
return MeResponse.from_org_member(org_member, role, email)
|
||||
|
||||
@staticmethod
|
||||
async def get_org_members(
|
||||
org_id: UUID,
|
||||
|
||||
@@ -19,14 +19,22 @@ with patch('storage.database.engine', create=True), patch(
|
||||
from server.email_validation import get_admin_user_id
|
||||
from server.routes.org_models import (
|
||||
LiteLLMIntegrationError,
|
||||
MeResponse,
|
||||
OrgAuthorizationError,
|
||||
OrgDatabaseError,
|
||||
OrgMemberNotFoundError,
|
||||
OrgMemberPage,
|
||||
OrgMemberResponse,
|
||||
OrgNameExistsError,
|
||||
OrgNotFoundError,
|
||||
RoleNotFoundError,
|
||||
)
|
||||
from server.routes.orgs import (
|
||||
get_me,
|
||||
get_org_members,
|
||||
org_router,
|
||||
remove_org_member,
|
||||
)
|
||||
from server.routes.orgs import get_org_members, org_router, remove_org_member
|
||||
from storage.org import Org
|
||||
|
||||
from openhands.server.user_auth import get_user_id
|
||||
@@ -2289,3 +2297,341 @@ class TestRemoveOrgMemberEndpoint:
|
||||
|
||||
assert exc_info.value.status_code == status.HTTP_503_SERVICE_UNAVAILABLE
|
||||
assert exc_info.value.detail == 'Service temporarily unavailable'
|
||||
|
||||
|
||||
class TestGetMeEndpoint:
|
||||
"""Tests for GET /api/organizations/{org_id}/me endpoint.
|
||||
|
||||
This endpoint returns the current authenticated user's membership record
|
||||
for the specified organization, including role, status, email, and LLM
|
||||
override fields (with masked API key).
|
||||
|
||||
Why: The frontend useMe() hook calls this endpoint to determine the user's
|
||||
role in the org, which gates read-only mode on settings pages. Without it,
|
||||
all role-based access control on settings pages is broken (returns 404).
|
||||
"""
|
||||
|
||||
@pytest.fixture
|
||||
def test_user_id(self):
|
||||
"""Create a test user ID."""
|
||||
return str(uuid.uuid4())
|
||||
|
||||
@pytest.fixture
|
||||
def test_org_id(self):
|
||||
"""Create a test organization ID."""
|
||||
return uuid.uuid4()
|
||||
|
||||
@pytest.fixture
|
||||
def mock_me_app(self, test_user_id):
|
||||
"""Create a test FastAPI app with org routes and mocked auth."""
|
||||
app = FastAPI()
|
||||
app.include_router(org_router)
|
||||
|
||||
def mock_get_user_id():
|
||||
return test_user_id
|
||||
|
||||
app.dependency_overrides[get_user_id] = mock_get_user_id
|
||||
return app
|
||||
|
||||
def _make_me_response(
|
||||
self,
|
||||
org_id,
|
||||
user_id,
|
||||
email='test@example.com',
|
||||
role='owner',
|
||||
llm_api_key='****2345',
|
||||
llm_model='gpt-4',
|
||||
llm_base_url='https://api.example.com',
|
||||
max_iterations=50,
|
||||
llm_api_key_for_byor=None,
|
||||
status_val='active',
|
||||
):
|
||||
"""Create a MeResponse for testing."""
|
||||
return MeResponse(
|
||||
org_id=str(org_id),
|
||||
user_id=str(user_id),
|
||||
email=email,
|
||||
role=role,
|
||||
llm_api_key=llm_api_key,
|
||||
llm_model=llm_model,
|
||||
llm_base_url=llm_base_url,
|
||||
max_iterations=max_iterations,
|
||||
llm_api_key_for_byor=llm_api_key_for_byor,
|
||||
status=status_val,
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_me_success(self, mock_me_app, test_user_id, test_org_id):
|
||||
"""GIVEN: Authenticated user who is a member of the organization
|
||||
WHEN: GET /api/organizations/{org_id}/me is called
|
||||
THEN: Returns 200 with the user's membership data including role name and email
|
||||
"""
|
||||
me_response = self._make_me_response(
|
||||
org_id=test_org_id,
|
||||
user_id=test_user_id,
|
||||
email='owner@example.com',
|
||||
role='owner',
|
||||
llm_model='gpt-4',
|
||||
llm_base_url='https://api.example.com',
|
||||
max_iterations=50,
|
||||
status_val='active',
|
||||
)
|
||||
|
||||
with patch(
|
||||
'server.routes.orgs.OrgMemberService.get_me',
|
||||
return_value=me_response,
|
||||
):
|
||||
client = TestClient(mock_me_app)
|
||||
response = client.get(f'/api/organizations/{test_org_id}/me')
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data['org_id'] == str(test_org_id)
|
||||
assert data['user_id'] == test_user_id
|
||||
assert data['email'] == 'owner@example.com'
|
||||
assert data['role'] == 'owner'
|
||||
assert data['llm_model'] == 'gpt-4'
|
||||
assert data['llm_base_url'] == 'https://api.example.com'
|
||||
assert data['max_iterations'] == 50
|
||||
assert data['status'] == 'active'
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_me_masks_llm_api_key(
|
||||
self, mock_me_app, test_user_id, test_org_id
|
||||
):
|
||||
"""GIVEN: User is a member with an LLM API key set
|
||||
WHEN: GET /api/organizations/{org_id}/me is called
|
||||
THEN: The llm_api_key field is masked (not the raw secret value)
|
||||
|
||||
Why: API keys must never be returned in plaintext in API responses.
|
||||
The frontend only needs to know if a key is set, not its value.
|
||||
"""
|
||||
me_response = self._make_me_response(
|
||||
org_id=test_org_id,
|
||||
user_id=test_user_id,
|
||||
llm_api_key='****cdef', # Masked key
|
||||
)
|
||||
|
||||
with patch(
|
||||
'server.routes.orgs.OrgMemberService.get_me',
|
||||
return_value=me_response,
|
||||
):
|
||||
client = TestClient(mock_me_app)
|
||||
response = client.get(f'/api/organizations/{test_org_id}/me')
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
# The raw key must NOT appear in the response
|
||||
assert data['llm_api_key'] != 'sk-secret-real-key-abcdef'
|
||||
# Should be masked with stars
|
||||
assert '**' in data['llm_api_key']
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_me_not_a_member(self, mock_me_app, test_org_id):
|
||||
"""GIVEN: Authenticated user who is NOT a member of the organization
|
||||
WHEN: GET /api/organizations/{org_id}/me is called
|
||||
THEN: Returns 404 (to avoid leaking org existence per spec)
|
||||
"""
|
||||
with patch(
|
||||
'server.routes.orgs.OrgMemberService.get_me',
|
||||
side_effect=OrgMemberNotFoundError(str(test_org_id), 'user-id'),
|
||||
):
|
||||
client = TestClient(mock_me_app)
|
||||
response = client.get(f'/api/organizations/{test_org_id}/me')
|
||||
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_me_invalid_uuid(self, mock_me_app):
|
||||
"""GIVEN: Invalid UUID format for org_id
|
||||
WHEN: GET /api/organizations/{org_id}/me is called
|
||||
THEN: Returns 422 (FastAPI validates UUID path parameter)
|
||||
"""
|
||||
client = TestClient(mock_me_app)
|
||||
response = client.get('/api/organizations/not-a-valid-uuid/me')
|
||||
|
||||
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_me_unauthenticated(self, test_org_id):
|
||||
"""GIVEN: User is not authenticated
|
||||
WHEN: GET /api/organizations/{org_id}/me is called
|
||||
THEN: Returns 401
|
||||
"""
|
||||
app = FastAPI()
|
||||
app.include_router(org_router)
|
||||
|
||||
async def mock_unauthenticated():
|
||||
raise HTTPException(status_code=401, detail='User not authenticated')
|
||||
|
||||
app.dependency_overrides[get_user_id] = mock_unauthenticated
|
||||
|
||||
client = TestClient(app)
|
||||
response = client.get(f'/api/organizations/{test_org_id}/me')
|
||||
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_me_unexpected_error(self, mock_me_app, test_org_id):
|
||||
"""GIVEN: An unexpected error occurs during membership lookup
|
||||
WHEN: GET /api/organizations/{org_id}/me is called
|
||||
THEN: Returns 500
|
||||
"""
|
||||
with patch(
|
||||
'server.routes.orgs.OrgMemberService.get_me',
|
||||
side_effect=RuntimeError('Database connection failed'),
|
||||
):
|
||||
client = TestClient(mock_me_app)
|
||||
response = client.get(f'/api/organizations/{test_org_id}/me')
|
||||
|
||||
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_me_with_null_optional_fields(
|
||||
self, mock_me_app, test_user_id, test_org_id
|
||||
):
|
||||
"""GIVEN: User is a member with null optional fields (llm_model, llm_base_url, etc.)
|
||||
WHEN: GET /api/organizations/{org_id}/me is called
|
||||
THEN: Returns 200 with null values for optional fields
|
||||
"""
|
||||
me_response = self._make_me_response(
|
||||
org_id=test_org_id,
|
||||
user_id=test_user_id,
|
||||
llm_model=None,
|
||||
llm_base_url=None,
|
||||
max_iterations=None,
|
||||
llm_api_key='',
|
||||
)
|
||||
|
||||
with patch(
|
||||
'server.routes.orgs.OrgMemberService.get_me',
|
||||
return_value=me_response,
|
||||
):
|
||||
client = TestClient(mock_me_app)
|
||||
response = client.get(f'/api/organizations/{test_org_id}/me')
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data['llm_model'] is None
|
||||
assert data['llm_base_url'] is None
|
||||
assert data['max_iterations'] is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_me_with_admin_role(self, mock_me_app, test_user_id, test_org_id):
|
||||
"""GIVEN: User is an admin member of the organization
|
||||
WHEN: GET /api/organizations/{org_id}/me is called
|
||||
THEN: Returns correct role name 'admin'
|
||||
|
||||
Why: The frontend uses the role to determine if settings are read-only.
|
||||
Admins and owners can edit; members see read-only.
|
||||
"""
|
||||
me_response = self._make_me_response(
|
||||
org_id=test_org_id,
|
||||
user_id=test_user_id,
|
||||
role='admin',
|
||||
)
|
||||
|
||||
with patch(
|
||||
'server.routes.orgs.OrgMemberService.get_me',
|
||||
return_value=me_response,
|
||||
):
|
||||
client = TestClient(mock_me_app)
|
||||
response = client.get(f'/api/organizations/{test_org_id}/me')
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data['role'] == 'admin'
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_me_masks_byor_api_key(
|
||||
self, mock_me_app, test_user_id, test_org_id
|
||||
):
|
||||
"""GIVEN: User has an llm_api_key_for_byor set
|
||||
WHEN: GET /api/organizations/{org_id}/me is called
|
||||
THEN: The llm_api_key_for_byor field is also masked
|
||||
"""
|
||||
me_response = self._make_me_response(
|
||||
org_id=test_org_id,
|
||||
user_id=test_user_id,
|
||||
llm_api_key_for_byor='****-key', # Masked key
|
||||
)
|
||||
|
||||
with patch(
|
||||
'server.routes.orgs.OrgMemberService.get_me',
|
||||
return_value=me_response,
|
||||
):
|
||||
client = TestClient(mock_me_app)
|
||||
response = client.get(f'/api/organizations/{test_org_id}/me')
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data['llm_api_key_for_byor'] != 'sk-byor-secret-key'
|
||||
assert (
|
||||
data['llm_api_key_for_byor'] is None or '**' in data['llm_api_key_for_byor']
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_me_role_not_found_returns_500(self, mock_me_app, test_org_id):
|
||||
"""GIVEN: Role lookup fails (data integrity issue)
|
||||
WHEN: GET /api/organizations/{org_id}/me is called
|
||||
THEN: Returns 500 Internal Server Error
|
||||
"""
|
||||
with patch(
|
||||
'server.routes.orgs.OrgMemberService.get_me',
|
||||
side_effect=RoleNotFoundError(role_id=999),
|
||||
):
|
||||
client = TestClient(mock_me_app)
|
||||
response = client.get(f'/api/organizations/{test_org_id}/me')
|
||||
|
||||
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
|
||||
assert 'unexpected error' in response.json()['detail'].lower()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_me_direct_function_call_success(self, test_user_id, test_org_id):
|
||||
"""Test direct function call to get_me returns MeResponse."""
|
||||
me_response = self._make_me_response(
|
||||
org_id=test_org_id,
|
||||
user_id=test_user_id,
|
||||
email='test@example.com',
|
||||
role='owner',
|
||||
)
|
||||
|
||||
with patch(
|
||||
'server.routes.orgs.OrgMemberService.get_me',
|
||||
return_value=me_response,
|
||||
):
|
||||
result = await get_me(org_id=test_org_id, user_id=test_user_id)
|
||||
|
||||
assert isinstance(result, MeResponse)
|
||||
assert result.org_id == str(test_org_id)
|
||||
assert result.user_id == test_user_id
|
||||
assert result.role == 'owner'
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_me_direct_function_call_member_not_found(
|
||||
self, test_user_id, test_org_id
|
||||
):
|
||||
"""Test direct function call to get_me raises HTTPException on member not found."""
|
||||
with patch(
|
||||
'server.routes.orgs.OrgMemberService.get_me',
|
||||
side_effect=OrgMemberNotFoundError(str(test_org_id), test_user_id),
|
||||
):
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await get_me(org_id=test_org_id, user_id=test_user_id)
|
||||
|
||||
assert exc_info.value.status_code == status.HTTP_404_NOT_FOUND
|
||||
assert str(test_org_id) in exc_info.value.detail
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_me_direct_function_call_role_not_found(
|
||||
self, test_user_id, test_org_id
|
||||
):
|
||||
"""Test direct function call to get_me raises HTTPException on role not found."""
|
||||
with patch(
|
||||
'server.routes.orgs.OrgMemberService.get_me',
|
||||
side_effect=RoleNotFoundError(role_id=999),
|
||||
):
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await get_me(org_id=test_org_id, user_id=test_user_id)
|
||||
|
||||
assert exc_info.value.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
|
||||
|
||||
@@ -4,9 +4,16 @@ import uuid
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from pydantic import SecretStr
|
||||
from server.routes.org_models import (
|
||||
MeResponse,
|
||||
OrgMemberNotFoundError,
|
||||
RoleNotFoundError,
|
||||
)
|
||||
from server.services.org_member_service import OrgMemberService
|
||||
from storage.org_member import OrgMember
|
||||
from storage.role import Role
|
||||
from storage.user import User
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -1138,3 +1145,137 @@ class TestOrgMemberServiceIsLastOwner:
|
||||
|
||||
# Assert
|
||||
assert result is False
|
||||
|
||||
|
||||
class TestOrgMemberServiceGetMe:
|
||||
"""Test cases for OrgMemberService.get_me."""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_org_member(self, org_id, current_user_id):
|
||||
"""Create a mock OrgMember with LLM fields."""
|
||||
member = MagicMock(spec=OrgMember)
|
||||
member.org_id = org_id
|
||||
member.user_id = current_user_id
|
||||
member.role_id = 1
|
||||
member.llm_api_key = SecretStr('sk-test-key-12345')
|
||||
member.llm_api_key_for_byor = None
|
||||
member.llm_model = 'gpt-4'
|
||||
member.llm_base_url = 'https://api.example.com'
|
||||
member.max_iterations = 50
|
||||
member.status = 'active'
|
||||
return member
|
||||
|
||||
@pytest.fixture
|
||||
def mock_user(self, current_user_id):
|
||||
"""Create a mock User."""
|
||||
user = MagicMock(spec=User)
|
||||
user.id = current_user_id
|
||||
user.email = 'test@example.com'
|
||||
return user
|
||||
|
||||
def test_get_me_success_returns_me_response(
|
||||
self, org_id, current_user_id, mock_org_member, mock_user, owner_role
|
||||
):
|
||||
"""GIVEN: User is a member of the organization
|
||||
WHEN: get_me is called
|
||||
THEN: Returns MeResponse with user's membership data
|
||||
"""
|
||||
# Arrange
|
||||
with (
|
||||
patch(
|
||||
'server.services.org_member_service.OrgMemberStore.get_org_member'
|
||||
) as mock_get_member,
|
||||
patch(
|
||||
'server.services.org_member_service.RoleStore.get_role_by_id'
|
||||
) as mock_get_role,
|
||||
patch(
|
||||
'server.services.org_member_service.UserStore.get_user_by_id'
|
||||
) as mock_get_user,
|
||||
):
|
||||
mock_get_member.return_value = mock_org_member
|
||||
mock_get_role.return_value = owner_role
|
||||
mock_get_user.return_value = mock_user
|
||||
|
||||
# Act
|
||||
result = OrgMemberService.get_me(org_id, current_user_id)
|
||||
|
||||
# Assert
|
||||
assert isinstance(result, MeResponse)
|
||||
assert result.org_id == str(org_id)
|
||||
assert result.user_id == str(current_user_id)
|
||||
assert result.email == 'test@example.com'
|
||||
assert result.role == 'owner'
|
||||
assert result.llm_model == 'gpt-4'
|
||||
assert result.max_iterations == 50
|
||||
assert result.status == 'active'
|
||||
|
||||
def test_get_me_member_not_found_raises_error(self, org_id, current_user_id):
|
||||
"""GIVEN: User is not a member of the organization
|
||||
WHEN: get_me is called
|
||||
THEN: Raises OrgMemberNotFoundError
|
||||
"""
|
||||
# Arrange
|
||||
with patch(
|
||||
'server.services.org_member_service.OrgMemberStore.get_org_member'
|
||||
) as mock_get_member:
|
||||
mock_get_member.return_value = None
|
||||
|
||||
# Act & Assert
|
||||
with pytest.raises(OrgMemberNotFoundError) as exc_info:
|
||||
OrgMemberService.get_me(org_id, current_user_id)
|
||||
|
||||
assert str(org_id) in str(exc_info.value)
|
||||
|
||||
def test_get_me_role_not_found_raises_error(
|
||||
self, org_id, current_user_id, mock_org_member
|
||||
):
|
||||
"""GIVEN: Member exists but role lookup fails
|
||||
WHEN: get_me is called
|
||||
THEN: Raises RoleNotFoundError
|
||||
"""
|
||||
# Arrange
|
||||
with (
|
||||
patch(
|
||||
'server.services.org_member_service.OrgMemberStore.get_org_member'
|
||||
) as mock_get_member,
|
||||
patch(
|
||||
'server.services.org_member_service.RoleStore.get_role_by_id'
|
||||
) as mock_get_role,
|
||||
):
|
||||
mock_get_member.return_value = mock_org_member
|
||||
mock_get_role.return_value = None
|
||||
|
||||
# Act & Assert
|
||||
with pytest.raises(RoleNotFoundError) as exc_info:
|
||||
OrgMemberService.get_me(org_id, current_user_id)
|
||||
|
||||
assert exc_info.value.role_id == mock_org_member.role_id
|
||||
|
||||
def test_get_me_user_not_found_returns_empty_email(
|
||||
self, org_id, current_user_id, mock_org_member, owner_role
|
||||
):
|
||||
"""GIVEN: Member exists but user lookup returns None
|
||||
WHEN: get_me is called
|
||||
THEN: Returns MeResponse with empty email
|
||||
"""
|
||||
# Arrange
|
||||
with (
|
||||
patch(
|
||||
'server.services.org_member_service.OrgMemberStore.get_org_member'
|
||||
) as mock_get_member,
|
||||
patch(
|
||||
'server.services.org_member_service.RoleStore.get_role_by_id'
|
||||
) as mock_get_role,
|
||||
patch(
|
||||
'server.services.org_member_service.UserStore.get_user_by_id'
|
||||
) as mock_get_user,
|
||||
):
|
||||
mock_get_member.return_value = mock_org_member
|
||||
mock_get_role.return_value = owner_role
|
||||
mock_get_user.return_value = None
|
||||
|
||||
# Act
|
||||
result = OrgMemberService.get_me(org_id, current_user_id)
|
||||
|
||||
# Assert
|
||||
assert result.email == ''
|
||||
|
||||
Reference in New Issue
Block a user