feat(backend): develop patch /api/organizations/{orgid} api (#12470)

Co-authored-by: rohitvinodmalhotra@gmail.com <rohitvinodmalhotra@gmail.com>
Co-authored-by: openhands <openhands@all-hands.dev>
Co-authored-by: Chuck Butkus <chuck@all-hands.dev>
This commit is contained in:
Hiep Le
2026-01-23 01:29:35 +07:00
committed by GitHub
parent b685fd43dd
commit d5ee799670
5 changed files with 1192 additions and 1 deletions

View File

@@ -138,3 +138,34 @@ class OrgPage(BaseModel):
items: list[OrgResponse]
next_page_id: str | None = None
class OrgUpdate(BaseModel):
"""Request model for updating an organization."""
# Basic organization information (any authenticated user can update)
contact_name: str | None = None
contact_email: EmailStr | None = Field(default=None, strip_whitespace=True)
conversation_expiration: int | None = None
default_max_iterations: int | None = Field(default=None, gt=0)
remote_runtime_resource_factor: int | None = Field(default=None, gt=0)
billing_margin: float | None = Field(default=None, ge=0, le=1)
enable_proactive_conversation_starters: bool | None = None
sandbox_base_container_image: str | None = None
sandbox_runtime_container_image: str | None = None
mcp_config: dict | None = None
sandbox_api_key: str | None = None
max_budget_per_task: float | None = Field(default=None, gt=0)
enable_solvability_analysis: bool | None = None
v1_enabled: bool | None = None
# LLM settings (require admin/owner role)
default_llm_model: str | None = None
default_llm_api_key_for_byor: str | None = None
default_llm_base_url: str | None = None
search_api_key: str | None = None
security_analyzer: str | None = None
agent: str | None = None
confirmation_mode: bool | None = None
enable_default_condenser: bool | None = None
condenser_max_size: int | None = Field(default=None, ge=20)

View File

@@ -12,6 +12,7 @@ from server.routes.org_models import (
OrgNotFoundError,
OrgPage,
OrgResponse,
OrgUpdate,
)
from storage.org_service import OrgService
@@ -320,3 +321,82 @@ async def delete_org(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail='An unexpected error occurred',
)
@org_router.patch('/{org_id}', response_model=OrgResponse)
async def update_org(
org_id: UUID,
update_data: OrgUpdate,
user_id: str = Depends(get_user_id),
) -> OrgResponse:
"""Update an existing organization.
This endpoint allows authenticated users to update organization settings.
LLM-related settings require admin or owner role in the organization.
Args:
org_id: Organization ID to update (UUID validated by FastAPI)
update_data: Organization update data
user_id: Authenticated user ID (injected by dependency)
Returns:
OrgResponse: The updated organization details
Raises:
HTTPException: 400 if org_id is invalid UUID format (handled by FastAPI)
HTTPException: 403 if user lacks permission for LLM settings
HTTPException: 404 if organization not found
HTTPException: 422 if validation errors occur (handled by FastAPI)
HTTPException: 500 if update fails
"""
logger.info(
'Updating organization',
extra={
'user_id': user_id,
'org_id': str(org_id),
},
)
try:
# Use service layer to update organization with permission checks
updated_org = await OrgService.update_org_with_permissions(
org_id=org_id,
update_data=update_data,
user_id=user_id,
)
# Retrieve credits from LiteLLM (following same pattern as create endpoint)
credits = await OrgService.get_org_credits(user_id, updated_org.id)
return OrgResponse.from_org(updated_org, credits=credits)
except ValueError as e:
# Organization not found
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=str(e),
)
except PermissionError as e:
# User lacks permission for LLM settings
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=str(e),
)
except OrgDatabaseError as e:
logger.error(
'Database operation failed',
extra={'user_id': user_id, 'org_id': str(org_id), 'error': str(e)},
)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail='Failed to update organization',
)
except Exception as e:
logger.exception(
'Unexpected error updating organization',
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',
)

View File

@@ -13,6 +13,7 @@ from server.routes.org_models import (
OrgDatabaseError,
OrgNameExistsError,
OrgNotFoundError,
OrgUpdate,
)
from storage.lite_llm_manager import LiteLlmManager
from storage.org import Org
@@ -395,6 +396,224 @@ class OrgService:
)
return e
@staticmethod
def has_admin_or_owner_role(user_id: str, org_id: UUID) -> bool:
"""
Check if user has admin or owner role in the specified organization.
Args:
user_id: User ID to check
org_id: Organization ID to check membership in
Returns:
bool: True if user has admin or owner role, False otherwise
"""
try:
# Parse user_id as UUID for database query
user_uuid = parse_uuid(user_id)
# Get the user's membership in this organization
# Note: The type annotation says int but the actual column is UUID
org_member = OrgMemberStore.get_org_member(org_id, user_uuid)
if not org_member:
return False
# Get the role details
role = RoleStore.get_role_by_id(org_member.role_id)
if not role:
return False
# Admin and owner roles have elevated permissions
# Based on test files, both admin and owner have rank 1
return role.name in ['admin', 'owner']
except Exception as e:
logger.warning(
'Error checking user role in organization',
extra={
'user_id': user_id,
'org_id': str(org_id),
'error': str(e),
},
)
return False
@staticmethod
def is_org_member(user_id: str, org_id: UUID) -> bool:
"""
Check if user is a member of the specified organization.
Args:
user_id: User ID to check
org_id: Organization ID to check membership in
Returns:
bool: True if user is a member, False otherwise
"""
try:
user_uuid = parse_uuid(user_id)
org_member = OrgMemberStore.get_org_member(org_id, user_uuid)
return org_member is not None
except Exception as e:
logger.warning(
'Error checking user membership in organization',
extra={
'user_id': user_id,
'org_id': str(org_id),
'error': str(e),
},
)
return False
@staticmethod
def _get_llm_settings_fields() -> set[str]:
"""
Get the set of organization fields that are considered LLM settings
and require admin/owner role to update.
Returns:
set[str]: Set of field names that require elevated permissions
"""
return {
'default_llm_model',
'default_llm_api_key_for_byor',
'default_llm_base_url',
'search_api_key',
'security_analyzer',
'agent',
'confirmation_mode',
'enable_default_condenser',
'condenser_max_size',
}
@staticmethod
def _has_llm_settings_updates(update_data: OrgUpdate) -> set[str]:
"""
Check if the update contains any LLM settings fields.
Args:
update_data: The organization update data
Returns:
set[str]: Set of LLM fields being updated (empty if none)
"""
llm_fields = OrgService._get_llm_settings_fields()
update_dict = update_data.model_dump(exclude_none=True)
return llm_fields.intersection(update_dict.keys())
@staticmethod
async def update_org_with_permissions(
org_id: UUID,
update_data: OrgUpdate,
user_id: str,
) -> Org:
"""
Update organization with permission checks for LLM settings.
Args:
org_id: Organization UUID to update
update_data: Organization update data from request
user_id: ID of the user requesting the update
Returns:
Org: The updated organization object
Raises:
ValueError: If organization not found
PermissionError: If user is not a member, or lacks admin/owner role for LLM settings
OrgDatabaseError: If database update fails
"""
logger.info(
'Updating organization with permission checks',
extra={
'org_id': str(org_id),
'user_id': user_id,
'has_update_data': update_data is not None,
},
)
# Validate organization exists
existing_org = OrgStore.get_org_by_id(org_id)
if not existing_org:
raise ValueError(f'Organization with ID {org_id} not found')
# Check if user is a member of this organization
if not OrgService.is_org_member(user_id, org_id):
logger.warning(
'Non-member attempted to update organization',
extra={
'user_id': user_id,
'org_id': str(org_id),
},
)
raise PermissionError(
'User must be a member of the organization to update it'
)
# Check if update contains any LLM settings
llm_fields_being_updated = OrgService._has_llm_settings_updates(update_data)
if llm_fields_being_updated:
# Verify user has admin or owner role
has_permission = OrgService.has_admin_or_owner_role(user_id, org_id)
if not has_permission:
logger.warning(
'User attempted to update LLM settings without permission',
extra={
'user_id': user_id,
'org_id': str(org_id),
'attempted_fields': list(llm_fields_being_updated),
},
)
raise PermissionError(
'Admin or owner role required to update LLM settings'
)
logger.debug(
'User has permission to update LLM settings',
extra={
'user_id': user_id,
'org_id': str(org_id),
'llm_fields': list(llm_fields_being_updated),
},
)
# Convert to dict for OrgStore (excluding None values)
update_dict = update_data.model_dump(exclude_none=True)
if not update_dict:
logger.info(
'No fields to update',
extra={'org_id': str(org_id), 'user_id': user_id},
)
return existing_org
# Perform the update
try:
updated_org = OrgStore.update_org(org_id, update_dict)
if not updated_org:
raise OrgDatabaseError('Failed to update organization in database')
logger.info(
'Organization updated successfully',
extra={
'org_id': str(org_id),
'user_id': user_id,
'updated_fields': list(update_dict.keys()),
},
)
return updated_org
except Exception as e:
logger.error(
'Failed to update organization',
extra={
'org_id': str(org_id),
'user_id': user_id,
'error': str(e),
},
)
raise OrgDatabaseError(f'Failed to update organization: {str(e)}')
@staticmethod
async def get_org_credits(user_id: str, org_id: UUID) -> float | None:
"""

View File

@@ -7,6 +7,7 @@ Tests the POST /api/organizations endpoint with various scenarios.
import uuid
from unittest.mock import AsyncMock, patch
import httpx
import pytest
from fastapi import FastAPI, HTTPException, status
from fastapi.testclient import TestClient
@@ -1131,3 +1132,238 @@ async def test_delete_org_unauthorized():
# Assert
assert response.status_code == status.HTTP_401_UNAUTHORIZED
@pytest.fixture
def mock_update_app():
"""Create a test FastAPI app with organization routes and mocked auth for update endpoint."""
app = FastAPI()
app.include_router(org_router)
# Override the auth dependency to return a test user
async def mock_user_id():
return 'test-user-123'
app.dependency_overrides[get_user_id] = mock_user_id
return app
# Note: Success cases for update endpoint are tested in test_org_service.py
# Route handler tests focus on error handling and validation
@pytest.mark.asyncio
async def test_update_org_not_found(mock_update_app):
"""
GIVEN: Organization ID does not exist
WHEN: PATCH /api/organizations/{org_id} is called
THEN: 404 Not Found error is returned
"""
# Arrange
org_id = uuid.uuid4()
update_data = {'contact_name': 'Jane Doe'}
with patch(
'server.routes.orgs.OrgService.update_org_with_permissions',
AsyncMock(side_effect=ValueError(f'Organization with ID {org_id} not found')),
):
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=mock_update_app), base_url='http://test'
) as client:
# Act
response = await client.patch(
f'/api/organizations/{org_id}', json=update_data
)
# Assert
assert response.status_code == status.HTTP_404_NOT_FOUND
assert 'not found' in response.json()['detail'].lower()
@pytest.mark.asyncio
async def test_update_org_permission_denied_non_member(mock_update_app):
"""
GIVEN: User is not a member of the organization
WHEN: PATCH /api/organizations/{org_id} is called
THEN: 403 Forbidden error is returned
"""
# Arrange
org_id = uuid.uuid4()
update_data = {'contact_name': 'Jane Doe'}
with patch(
'server.routes.orgs.OrgService.update_org_with_permissions',
AsyncMock(
side_effect=PermissionError(
'User must be a member of the organization to update it'
)
),
):
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=mock_update_app), base_url='http://test'
) as client:
# Act
response = await client.patch(
f'/api/organizations/{org_id}', json=update_data
)
# Assert
assert response.status_code == status.HTTP_403_FORBIDDEN
assert 'member' in response.json()['detail'].lower()
@pytest.mark.asyncio
async def test_update_org_permission_denied_llm_settings(mock_update_app):
"""
GIVEN: User lacks admin/owner role but tries to update LLM settings
WHEN: PATCH /api/organizations/{org_id} is called
THEN: 403 Forbidden error is returned
"""
# Arrange
org_id = uuid.uuid4()
update_data = {'default_llm_model': 'claude-opus-4-5-20251101'}
with patch(
'server.routes.orgs.OrgService.update_org_with_permissions',
AsyncMock(
side_effect=PermissionError(
'Admin or owner role required to update LLM settings'
)
),
):
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=mock_update_app), base_url='http://test'
) as client:
# Act
response = await client.patch(
f'/api/organizations/{org_id}', json=update_data
)
# Assert
assert response.status_code == status.HTTP_403_FORBIDDEN
assert (
'admin' in response.json()['detail'].lower()
or 'owner' in response.json()['detail'].lower()
)
@pytest.mark.asyncio
async def test_update_org_database_error(mock_update_app):
"""
GIVEN: Database operation fails during update
WHEN: PATCH /api/organizations/{org_id} is called
THEN: 500 Internal Server Error is returned
"""
# Arrange
org_id = uuid.uuid4()
update_data = {'contact_name': 'Jane Doe'}
with patch(
'server.routes.orgs.OrgService.update_org_with_permissions',
AsyncMock(side_effect=OrgDatabaseError('Database connection failed')),
):
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=mock_update_app), base_url='http://test'
) as client:
# Act
response = await client.patch(
f'/api/organizations/{org_id}', json=update_data
)
# Assert
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
assert 'Failed to update organization' in response.json()['detail']
@pytest.mark.asyncio
async def test_update_org_unexpected_error(mock_update_app):
"""
GIVEN: Unexpected error occurs during update
WHEN: PATCH /api/organizations/{org_id} is called
THEN: 500 Internal Server Error is returned with generic message
"""
# Arrange
org_id = uuid.uuid4()
update_data = {'contact_name': 'Jane Doe'}
with patch(
'server.routes.orgs.OrgService.update_org_with_permissions',
AsyncMock(side_effect=RuntimeError('Unexpected system error')),
):
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=mock_update_app), base_url='http://test'
) as client:
# Act
response = await client.patch(
f'/api/organizations/{org_id}', json=update_data
)
# Assert
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
assert 'unexpected error' in response.json()['detail'].lower()
@pytest.mark.asyncio
async def test_update_org_invalid_uuid_format(mock_update_app):
"""
GIVEN: Invalid UUID format in org_id path parameter
WHEN: PATCH /api/organizations/{org_id} is called
THEN: 422 validation error is returned (handled by FastAPI)
"""
# Arrange
invalid_org_id = 'not-a-valid-uuid'
update_data = {'contact_name': 'Jane Doe'}
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=mock_update_app), base_url='http://test'
) as client:
# Act
response = await client.patch(
f'/api/organizations/{invalid_org_id}', json=update_data
)
# Assert
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
@pytest.mark.asyncio
async def test_update_org_invalid_field_values(mock_update_app):
"""
GIVEN: Update request with invalid field values (e.g., negative max_iterations)
WHEN: PATCH /api/organizations/{org_id} is called
THEN: 422 validation error is returned
"""
# Arrange
org_id = uuid.uuid4()
update_data = {'default_max_iterations': -1} # Invalid: must be > 0
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=mock_update_app), base_url='http://test'
) as client:
# Act
response = await client.patch(f'/api/organizations/{org_id}', json=update_data)
# Assert
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
@pytest.mark.asyncio
async def test_update_org_invalid_email_format(mock_update_app):
"""
GIVEN: Update request with invalid email format
WHEN: PATCH /api/organizations/{org_id} is called
THEN: 422 validation error is returned
"""
# Arrange
org_id = uuid.uuid4()
update_data = {'contact_email': 'invalid-email'} # Missing @
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=mock_update_app), base_url='http://test'
) as client:
# Act
response = await client.patch(f'/api/organizations/{org_id}', json=update_data)
# Assert
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY

View File

@@ -78,7 +78,11 @@ def test_validate_name_uniqueness_with_unique_name(session_maker):
unique_name = 'unique-org-name'
# Act & Assert - should not raise
with patch('storage.org_store.session_maker', session_maker):
with (
patch('storage.org_store.session_maker', session_maker),
patch('storage.org_member_store.session_maker', session_maker),
patch('storage.role_store.session_maker', session_maker),
):
OrgService.validate_name_uniqueness(unique_name)
@@ -1032,3 +1036,624 @@ async def test_delete_org_with_cleanup_unexpected_none_result(
await OrgService.delete_org_with_cleanup(user_id, org_id)
assert 'not found during deletion' in str(exc_info.value)
@pytest.mark.asyncio
async def test_update_org_with_permissions_success_non_llm_fields(session_maker):
"""
GIVEN: Valid organization update with non-LLM fields and user is a member
WHEN: update_org_with_permissions is called
THEN: Organization is updated successfully
"""
# Arrange
org_id = uuid.uuid4()
user_id = str(uuid.uuid4())
# Create organization and user in database
with session_maker() as session:
org = Org(
id=org_id,
name='Test Organization',
contact_name='John Doe',
contact_email='john@example.com',
org_version=5,
)
session.add(org)
user = User(id=uuid.UUID(user_id), current_org_id=org_id)
session.add(user)
role = Role(id=2, name='member', rank=2)
session.add(role)
org_member = OrgMember(
org_id=org_id,
user_id=uuid.UUID(user_id),
role_id=2,
status='active',
_llm_api_key='test-key',
)
session.add(org_member)
session.commit()
from server.routes.org_models import OrgUpdate
update_data = OrgUpdate(
contact_name='Jane Doe',
contact_email='jane@example.com',
conversation_expiration=30,
)
with (
patch('storage.org_store.session_maker', session_maker),
patch('storage.org_member_store.session_maker', session_maker),
patch('storage.role_store.session_maker', session_maker),
):
# Act
result = await OrgService.update_org_with_permissions(
org_id=org_id,
update_data=update_data,
user_id=user_id,
)
# Assert
assert result is not None
assert result.contact_name == 'Jane Doe'
assert result.contact_email == 'jane@example.com'
assert result.conversation_expiration == 30
@pytest.mark.asyncio
async def test_update_org_with_permissions_success_llm_fields_admin(session_maker):
"""
GIVEN: Valid organization update with LLM fields and user has admin role
WHEN: update_org_with_permissions is called
THEN: Organization is updated successfully
"""
# Arrange
org_id = uuid.uuid4()
user_id = str(uuid.uuid4())
# Create organization, user, and admin role in database
with session_maker() as session:
org = Org(
id=org_id,
name='Test Organization',
contact_name='John Doe',
contact_email='john@example.com',
org_version=5,
)
session.add(org)
user = User(id=uuid.UUID(user_id), current_org_id=org_id)
session.add(user)
admin_role = Role(id=1, name='admin', rank=1)
session.add(admin_role)
org_member = OrgMember(
org_id=org_id,
user_id=uuid.UUID(user_id),
role_id=1,
status='active',
_llm_api_key='test-key',
)
session.add(org_member)
session.commit()
from server.routes.org_models import OrgUpdate
update_data = OrgUpdate(
default_llm_model='claude-opus-4-5-20251101',
default_llm_base_url='https://api.anthropic.com',
)
with (
patch('storage.org_store.session_maker', session_maker),
patch('storage.org_member_store.session_maker', session_maker),
patch('storage.role_store.session_maker', session_maker),
):
# Act
result = await OrgService.update_org_with_permissions(
org_id=org_id,
update_data=update_data,
user_id=user_id,
)
# Assert
assert result is not None
assert result.default_llm_model == 'claude-opus-4-5-20251101'
assert result.default_llm_base_url == 'https://api.anthropic.com'
@pytest.mark.asyncio
async def test_update_org_with_permissions_success_llm_fields_owner(session_maker):
"""
GIVEN: Valid organization update with LLM fields and user has owner role
WHEN: update_org_with_permissions is called
THEN: Organization is updated successfully
"""
# Arrange
org_id = uuid.uuid4()
user_id = str(uuid.uuid4())
# Create organization, user, and owner role in database
with session_maker() as session:
org = Org(
id=org_id,
name='Test Organization',
contact_name='John Doe',
contact_email='john@example.com',
org_version=5,
)
session.add(org)
user = User(id=uuid.UUID(user_id), current_org_id=org_id)
session.add(user)
owner_role = Role(id=1, name='owner', rank=1)
session.add(owner_role)
org_member = OrgMember(
org_id=org_id,
user_id=uuid.UUID(user_id),
role_id=1,
status='active',
_llm_api_key='test-key',
)
session.add(org_member)
session.commit()
from server.routes.org_models import OrgUpdate
update_data = OrgUpdate(
default_llm_model='claude-opus-4-5-20251101',
security_analyzer='enabled',
)
with (
patch('storage.org_store.session_maker', session_maker),
patch('storage.org_member_store.session_maker', session_maker),
patch('storage.role_store.session_maker', session_maker),
):
# Act
result = await OrgService.update_org_with_permissions(
org_id=org_id,
update_data=update_data,
user_id=user_id,
)
# Assert
assert result is not None
assert result.default_llm_model == 'claude-opus-4-5-20251101'
assert result.security_analyzer == 'enabled'
@pytest.mark.asyncio
async def test_update_org_with_permissions_success_mixed_fields_admin(session_maker):
"""
GIVEN: Valid organization update with both LLM and non-LLM fields and user has admin role
WHEN: update_org_with_permissions is called
THEN: Organization is updated successfully
"""
# Arrange
org_id = uuid.uuid4()
user_id = str(uuid.uuid4())
# Create organization, user, and admin role in database
with session_maker() as session:
org = Org(
id=org_id,
name='Test Organization',
contact_name='John Doe',
contact_email='john@example.com',
org_version=5,
)
session.add(org)
user = User(id=uuid.UUID(user_id), current_org_id=org_id)
session.add(user)
admin_role = Role(id=1, name='admin', rank=1)
session.add(admin_role)
org_member = OrgMember(
org_id=org_id,
user_id=uuid.UUID(user_id),
role_id=1,
status='active',
_llm_api_key='test-key',
)
session.add(org_member)
session.commit()
from server.routes.org_models import OrgUpdate
update_data = OrgUpdate(
contact_name='Jane Doe',
default_llm_model='claude-opus-4-5-20251101',
conversation_expiration=30,
)
with (
patch('storage.org_store.session_maker', session_maker),
patch('storage.org_member_store.session_maker', session_maker),
patch('storage.role_store.session_maker', session_maker),
):
# Act
result = await OrgService.update_org_with_permissions(
org_id=org_id,
update_data=update_data,
user_id=user_id,
)
# Assert
assert result is not None
assert result.contact_name == 'Jane Doe'
assert result.default_llm_model == 'claude-opus-4-5-20251101'
assert result.conversation_expiration == 30
@pytest.mark.asyncio
async def test_update_org_with_permissions_empty_update(session_maker):
"""
GIVEN: Update request with no fields (all None)
WHEN: update_org_with_permissions is called
THEN: Original organization is returned unchanged
"""
# Arrange
org_id = uuid.uuid4()
user_id = str(uuid.uuid4())
# Create organization and user in database
with session_maker() as session:
org = Org(
id=org_id,
name='Test Organization',
contact_name='John Doe',
contact_email='john@example.com',
org_version=5,
)
session.add(org)
user = User(id=uuid.UUID(user_id), current_org_id=org_id)
session.add(user)
role = Role(id=2, name='member', rank=2)
session.add(role)
org_member = OrgMember(
org_id=org_id,
user_id=uuid.UUID(user_id),
role_id=2,
status='active',
_llm_api_key='test-key',
)
session.add(org_member)
session.commit()
from server.routes.org_models import OrgUpdate
update_data = OrgUpdate() # All fields None
with (
patch('storage.org_store.session_maker', session_maker),
patch('storage.org_member_store.session_maker', session_maker),
patch('storage.role_store.session_maker', session_maker),
):
# Act
result = await OrgService.update_org_with_permissions(
org_id=org_id,
update_data=update_data,
user_id=user_id,
)
# Assert
assert result is not None
assert result.name == 'Test Organization'
assert result.contact_name == 'John Doe'
@pytest.mark.asyncio
async def test_update_org_with_permissions_org_not_found(session_maker):
"""
GIVEN: Organization ID does not exist
WHEN: update_org_with_permissions is called
THEN: ValueError is raised
"""
# Arrange
org_id = uuid.uuid4()
user_id = str(uuid.uuid4())
from server.routes.org_models import OrgUpdate
update_data = OrgUpdate(contact_name='Jane Doe')
with (
patch('storage.org_store.session_maker', session_maker),
patch('storage.org_member_store.session_maker', session_maker),
patch('storage.role_store.session_maker', session_maker),
):
# Act & Assert
with pytest.raises(ValueError) as exc_info:
await OrgService.update_org_with_permissions(
org_id=org_id,
update_data=update_data,
user_id=user_id,
)
assert 'not found' in str(exc_info.value).lower()
@pytest.mark.asyncio
async def test_update_org_with_permissions_non_member(session_maker):
"""
GIVEN: User is not a member of the organization
WHEN: update_org_with_permissions is called
THEN: PermissionError is raised
"""
# Arrange
org_id = uuid.uuid4()
user_id = str(uuid.uuid4())
other_user_id = str(uuid.uuid4())
# Create organization but user is not a member
with session_maker() as session:
org = Org(
id=org_id,
name='Test Organization',
contact_name='John Doe',
contact_email='john@example.com',
org_version=5,
)
session.add(org)
user = User(id=uuid.UUID(other_user_id), current_org_id=org_id)
session.add(user)
session.commit()
from server.routes.org_models import OrgUpdate
update_data = OrgUpdate(contact_name='Jane Doe')
with (
patch('storage.org_store.session_maker', session_maker),
patch('storage.org_member_store.session_maker', session_maker),
patch('storage.role_store.session_maker', session_maker),
):
# Act & Assert
with pytest.raises(PermissionError) as exc_info:
await OrgService.update_org_with_permissions(
org_id=org_id,
update_data=update_data,
user_id=user_id,
)
assert 'member' in str(exc_info.value).lower()
@pytest.mark.asyncio
async def test_update_org_with_permissions_llm_fields_insufficient_permission(
session_maker,
):
"""
GIVEN: User is a member but lacks admin/owner role and tries to update LLM settings
WHEN: update_org_with_permissions is called
THEN: PermissionError is raised
"""
# Arrange
org_id = uuid.uuid4()
user_id = str(uuid.uuid4())
# Create organization and user with member role (not admin/owner)
with session_maker() as session:
org = Org(
id=org_id,
name='Test Organization',
contact_name='John Doe',
contact_email='john@example.com',
org_version=5,
)
session.add(org)
user = User(id=uuid.UUID(user_id), current_org_id=org_id)
session.add(user)
member_role = Role(id=2, name='member', rank=2)
session.add(member_role)
org_member = OrgMember(
org_id=org_id,
user_id=uuid.UUID(user_id),
role_id=2,
status='active',
_llm_api_key='test-key',
)
session.add(org_member)
session.commit()
from server.routes.org_models import OrgUpdate
update_data = OrgUpdate(default_llm_model='claude-opus-4-5-20251101')
with (
patch('storage.org_store.session_maker', session_maker),
patch('storage.org_member_store.session_maker', session_maker),
patch('storage.role_store.session_maker', session_maker),
):
# Act & Assert
with pytest.raises(PermissionError) as exc_info:
await OrgService.update_org_with_permissions(
org_id=org_id,
update_data=update_data,
user_id=user_id,
)
assert (
'admin' in str(exc_info.value).lower()
or 'owner' in str(exc_info.value).lower()
)
@pytest.mark.asyncio
async def test_update_org_with_permissions_database_error(session_maker):
"""
GIVEN: Database update operation fails
WHEN: update_org_with_permissions is called
THEN: OrgDatabaseError is raised
"""
# Arrange
org_id = uuid.uuid4()
user_id = str(uuid.uuid4())
# Create organization and user in database
with session_maker() as session:
org = Org(
id=org_id,
name='Test Organization',
contact_name='John Doe',
contact_email='john@example.com',
org_version=5,
)
session.add(org)
user = User(id=uuid.UUID(user_id), current_org_id=org_id)
session.add(user)
role = Role(id=2, name='member', rank=2)
session.add(role)
org_member = OrgMember(
org_id=org_id,
user_id=uuid.UUID(user_id),
role_id=2,
status='active',
_llm_api_key='test-key',
)
session.add(org_member)
session.commit()
from server.routes.org_models import OrgUpdate
update_data = OrgUpdate(contact_name='Jane Doe')
with (
patch('storage.org_store.session_maker', session_maker),
patch('storage.org_member_store.session_maker', session_maker),
patch('storage.role_store.session_maker', session_maker),
patch(
'storage.org_service.OrgStore.update_org',
return_value=None, # Simulate database failure
),
):
# Act & Assert
with pytest.raises(OrgDatabaseError) as exc_info:
await OrgService.update_org_with_permissions(
org_id=org_id,
update_data=update_data,
user_id=user_id,
)
assert 'Failed to update organization' in str(exc_info.value)
@pytest.mark.asyncio
async def test_update_org_with_permissions_only_llm_fields(session_maker):
"""
GIVEN: Update request contains only LLM fields and user has admin role
WHEN: update_org_with_permissions is called
THEN: Organization is updated successfully
"""
# Arrange
org_id = uuid.uuid4()
user_id = str(uuid.uuid4())
# Create organization, user, and admin role in database
with session_maker() as session:
org = Org(
id=org_id,
name='Test Organization',
contact_name='John Doe',
contact_email='john@example.com',
org_version=5,
)
session.add(org)
user = User(id=uuid.UUID(user_id), current_org_id=org_id)
session.add(user)
admin_role = Role(id=1, name='admin', rank=1)
session.add(admin_role)
org_member = OrgMember(
org_id=org_id,
user_id=uuid.UUID(user_id),
role_id=1,
status='active',
_llm_api_key='test-key',
)
session.add(org_member)
session.commit()
from server.routes.org_models import OrgUpdate
update_data = OrgUpdate(
default_llm_model='claude-opus-4-5-20251101',
security_analyzer='enabled',
agent='agent-mode',
)
with (
patch('storage.org_store.session_maker', session_maker),
patch('storage.org_member_store.session_maker', session_maker),
patch('storage.role_store.session_maker', session_maker),
):
# Act
result = await OrgService.update_org_with_permissions(
org_id=org_id,
update_data=update_data,
user_id=user_id,
)
# Assert
assert result is not None
assert result.default_llm_model == 'claude-opus-4-5-20251101'
assert result.security_analyzer == 'enabled'
assert result.agent == 'agent-mode'
@pytest.mark.asyncio
async def test_update_org_with_permissions_only_non_llm_fields(session_maker):
"""
GIVEN: Update request contains only non-LLM fields and user is a member
WHEN: update_org_with_permissions is called
THEN: Organization is updated successfully
"""
# Arrange
org_id = uuid.uuid4()
user_id = str(uuid.uuid4())
# Create organization and user in database
with session_maker() as session:
org = Org(
id=org_id,
name='Test Organization',
contact_name='John Doe',
contact_email='john@example.com',
org_version=5,
)
session.add(org)
user = User(id=uuid.UUID(user_id), current_org_id=org_id)
session.add(user)
role = Role(id=2, name='member', rank=2)
session.add(role)
org_member = OrgMember(
org_id=org_id,
user_id=uuid.UUID(user_id),
role_id=2,
status='active',
_llm_api_key='test-key',
)
session.add(org_member)
session.commit()
from server.routes.org_models import OrgUpdate
update_data = OrgUpdate(
contact_name='Jane Doe',
conversation_expiration=60,
enable_proactive_conversation_starters=False,
)
with (
patch('storage.org_store.session_maker', session_maker),
patch('storage.org_member_store.session_maker', session_maker),
patch('storage.role_store.session_maker', session_maker),
):
# Act
result = await OrgService.update_org_with_permissions(
org_id=org_id,
update_data=update_data,
user_id=user_id,
)
# Assert
assert result is not None
assert result.contact_name == 'Jane Doe'
assert result.conversation_expiration == 60
assert result.enable_proactive_conversation_starters is False