diff --git a/modules/datamodels/datamodelUam.py b/modules/datamodels/datamodelUam.py index 96a99fee..42659159 100644 --- a/modules/datamodels/datamodelUam.py +++ b/modules/datamodels/datamodelUam.py @@ -100,11 +100,11 @@ registerModelLabels( class UserConnection(BaseModel): id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Unique ID of the connection", json_schema_extra={"frontend_type": "text", "frontend_readonly": True, "frontend_required": False}) userId: str = Field(description="ID of the user this connection belongs to", json_schema_extra={"frontend_type": "text", "frontend_readonly": True, "frontend_required": False}) - authority: AuthAuthority = Field(description="Authentication authority", json_schema_extra={"frontend_type": "select", "frontend_readonly": True, "frontend_required": False, "frontend_options": "auth.authority"}) + authority: AuthAuthority = Field(description="Authentication authority", json_schema_extra={"frontend_type": "select", "frontend_readonly": True, "frontend_required": False, "frontend_options": "/api/connections/authorities/options"}) externalId: str = Field(description="User ID in the external system", json_schema_extra={"frontend_type": "text", "frontend_readonly": True, "frontend_required": False}) externalUsername: str = Field(description="Username in the external system", json_schema_extra={"frontend_type": "text", "frontend_readonly": False, "frontend_required": False}) externalEmail: Optional[EmailStr] = Field(None, description="Email in the external system", json_schema_extra={"frontend_type": "email", "frontend_readonly": False, "frontend_required": False}) - status: ConnectionStatus = Field(default=ConnectionStatus.ACTIVE, description="Connection status", json_schema_extra={"frontend_type": "select", "frontend_readonly": False, "frontend_required": False, "frontend_options": "connection.status"}) + status: ConnectionStatus = Field(default=ConnectionStatus.ACTIVE, description="Connection status", json_schema_extra={"frontend_type": "select", "frontend_readonly": False, "frontend_required": False, "frontend_options": "/api/connections/statuses/options"}) connectedAt: float = Field(default_factory=getUtcTimestamp, description="When the connection was established (UTC timestamp in seconds)", json_schema_extra={"frontend_type": "timestamp", "frontend_readonly": True, "frontend_required": False}) lastChecked: float = Field(default_factory=getUtcTimestamp, description="When the connection was last verified (UTC timestamp in seconds)", json_schema_extra={"frontend_type": "timestamp", "frontend_readonly": True, "frontend_required": False}) expiresAt: Optional[float] = Field(None, description="When the connection expires (UTC timestamp in seconds)", json_schema_extra={"frontend_type": "timestamp", "frontend_readonly": True, "frontend_required": False}) @@ -198,7 +198,7 @@ class User(BaseModel): authenticationAuthority: AuthAuthority = Field( default=AuthAuthority.LOCAL, description="Primary authentication authority", - json_schema_extra={"frontend_type": "select", "frontend_readonly": True, "frontend_required": False, "frontend_options": "auth.authority"} + json_schema_extra={"frontend_type": "select", "frontend_readonly": True, "frontend_required": False, "frontend_options": "/api/connections/authorities/options"} ) diff --git a/modules/features/trustee/datamodelFeatureTrustee.py b/modules/features/trustee/datamodelFeatureTrustee.py index e2d9b261..c64ec506 100644 --- a/modules/features/trustee/datamodelFeatureTrustee.py +++ b/modules/features/trustee/datamodelFeatureTrustee.py @@ -138,7 +138,7 @@ class TrusteeAccess(BaseModel): "frontend_type": "select", "frontend_readonly": False, "frontend_required": True, - "frontend_options": "TrusteeOrganisation" + "frontend_options": "/api/trustee/{instanceId}/organisations/options" } ) roleId: str = Field( @@ -147,7 +147,7 @@ class TrusteeAccess(BaseModel): "frontend_type": "select", "frontend_readonly": False, "frontend_required": True, - "frontend_options": "TrusteeRole" + "frontend_options": "/api/trustee/{instanceId}/roles/options" } ) userId: str = Field( @@ -156,7 +156,7 @@ class TrusteeAccess(BaseModel): "frontend_type": "select", "frontend_readonly": False, "frontend_required": True, - "frontend_options": "User" + "frontend_options": "/api/users/options" } ) contractId: Optional[str] = Field( @@ -166,7 +166,7 @@ class TrusteeAccess(BaseModel): "frontend_type": "select", "frontend_readonly": False, "frontend_required": False, - "frontend_options": "TrusteeContract", + "frontend_options": "/api/trustee/{instanceId}/contracts/options", "frontend_depends_on": "organisationId" } ) @@ -223,7 +223,7 @@ class TrusteeContract(BaseModel): "frontend_type": "select", "frontend_readonly": False, # Editable at creation, then readonly "frontend_required": True, - "frontend_options": "TrusteeOrganisation" + "frontend_options": "/api/trustee/{instanceId}/organisations/options" } ) label: str = Field( @@ -295,7 +295,7 @@ class TrusteeDocument(BaseModel): "frontend_type": "select", "frontend_readonly": False, "frontend_required": True, - "frontend_options": "TrusteeOrganisation" + "frontend_options": "/api/trustee/{instanceId}/organisations/options" } ) contractId: str = Field( @@ -304,7 +304,7 @@ class TrusteeDocument(BaseModel): "frontend_type": "select", "frontend_readonly": False, "frontend_required": True, - "frontend_options": "TrusteeContract", + "frontend_options": "/api/trustee/{instanceId}/contracts/options", "frontend_depends_on": "organisationId" } ) @@ -394,7 +394,7 @@ class TrusteePosition(BaseModel): "frontend_type": "select", "frontend_readonly": False, "frontend_required": True, - "frontend_options": "TrusteeOrganisation" + "frontend_options": "/api/trustee/{instanceId}/organisations/options" } ) contractId: str = Field( @@ -403,7 +403,7 @@ class TrusteePosition(BaseModel): "frontend_type": "select", "frontend_readonly": False, "frontend_required": True, - "frontend_options": "TrusteeContract", + "frontend_options": "/api/trustee/{instanceId}/contracts/options", "frontend_depends_on": "organisationId" } ) @@ -580,7 +580,7 @@ class TrusteePositionDocument(BaseModel): "frontend_type": "select", "frontend_readonly": False, "frontend_required": True, - "frontend_options": "TrusteeOrganisation" + "frontend_options": "/api/trustee/{instanceId}/organisations/options" } ) contractId: str = Field( @@ -589,7 +589,7 @@ class TrusteePositionDocument(BaseModel): "frontend_type": "select", "frontend_readonly": False, "frontend_required": True, - "frontend_options": "TrusteeContract", + "frontend_options": "/api/trustee/{instanceId}/contracts/options", "frontend_depends_on": "organisationId" } ) @@ -599,7 +599,7 @@ class TrusteePositionDocument(BaseModel): "frontend_type": "select", "frontend_readonly": False, "frontend_required": True, - "frontend_options": "TrusteeDocument", + "frontend_options": "/api/trustee/{instanceId}/documents/options", "frontend_depends_on": "contractId" } ) @@ -609,7 +609,7 @@ class TrusteePositionDocument(BaseModel): "frontend_type": "select", "frontend_readonly": False, "frontend_required": True, - "frontend_options": "TrusteePosition", + "frontend_options": "/api/trustee/{instanceId}/positions/options", "frontend_depends_on": "contractId" } ) diff --git a/modules/features/trustee/routeFeatureTrustee.py b/modules/features/trustee/routeFeatureTrustee.py index bee52513..196c8b18 100644 --- a/modules/features/trustee/routeFeatureTrustee.py +++ b/modules/features/trustee/routeFeatureTrustee.py @@ -115,6 +115,89 @@ async def _validateInstanceAccess(instanceId: str, context: RequestContext) -> s return str(instance.mandateId) +# ============================================================================ +# OPTIONS ENDPOINTS (for dropdowns) +# ============================================================================ + +@router.get("/{instanceId}/organisations/options", response_model=List[Dict[str, Any]]) +@limiter.limit("60/minute") +async def getOrganisationOptions( + request: Request, + instanceId: str = Path(..., description="Feature Instance ID"), + context: RequestContext = Depends(getRequestContext) +) -> List[Dict[str, Any]]: + """Get organisation options for select dropdowns. Returns: [{ value, label }]""" + mandateId = await _validateInstanceAccess(instanceId, context) + interface = getInterface(context.user, mandateId=mandateId, featureInstanceId=instanceId) + result = interface.getAllOrganisations(None) + items = result.items if hasattr(result, 'items') else result + return [{"value": org.id, "label": org.label or org.id} for org in items] + + +@router.get("/{instanceId}/roles/options", response_model=List[Dict[str, Any]]) +@limiter.limit("60/minute") +async def getRoleOptions( + request: Request, + instanceId: str = Path(..., description="Feature Instance ID"), + context: RequestContext = Depends(getRequestContext) +) -> List[Dict[str, Any]]: + """Get role options for select dropdowns. Returns: [{ value, label }]""" + mandateId = await _validateInstanceAccess(instanceId, context) + interface = getInterface(context.user, mandateId=mandateId, featureInstanceId=instanceId) + result = interface.getAllRoles(None) + items = result.items if hasattr(result, 'items') else result + return [{"value": role.id, "label": role.desc or role.id} for role in items] + + +@router.get("/{instanceId}/contracts/options", response_model=List[Dict[str, Any]]) +@limiter.limit("60/minute") +async def getContractOptions( + request: Request, + instanceId: str = Path(..., description="Feature Instance ID"), + context: RequestContext = Depends(getRequestContext) +) -> List[Dict[str, Any]]: + """Get contract options for select dropdowns. Returns: [{ value, label }]""" + mandateId = await _validateInstanceAccess(instanceId, context) + interface = getInterface(context.user, mandateId=mandateId, featureInstanceId=instanceId) + result = interface.getAllContracts(None) + items = result.items if hasattr(result, 'items') else result + return [{"value": c.id, "label": c.label or c.name or c.id} for c in items] + + +@router.get("/{instanceId}/documents/options", response_model=List[Dict[str, Any]]) +@limiter.limit("60/minute") +async def getDocumentOptions( + request: Request, + instanceId: str = Path(..., description="Feature Instance ID"), + context: RequestContext = Depends(getRequestContext) +) -> List[Dict[str, Any]]: + """Get document options for select dropdowns. Returns: [{ value, label }]""" + mandateId = await _validateInstanceAccess(instanceId, context) + interface = getInterface(context.user, mandateId=mandateId, featureInstanceId=instanceId) + result = interface.getAllDocuments(None) + items = result.items if hasattr(result, 'items') else result + return [{"value": d.id, "label": d.name or d.id} for d in items] + + +@router.get("/{instanceId}/positions/options", response_model=List[Dict[str, Any]]) +@limiter.limit("60/minute") +async def getPositionOptions( + request: Request, + instanceId: str = Path(..., description="Feature Instance ID"), + context: RequestContext = Depends(getRequestContext) +) -> List[Dict[str, Any]]: + """Get position options for select dropdowns. Returns: [{ value, label }]""" + mandateId = await _validateInstanceAccess(instanceId, context) + interface = getInterface(context.user, mandateId=mandateId, featureInstanceId=instanceId) + result = interface.getAllPositions(None) + items = result.items if hasattr(result, 'items') else result + return [{"value": p.id, "label": p.title or p.id} for p in items] + + +# ============================================================================ +# CRUD ENDPOINTS +# ============================================================================ + # ===== Organisation Routes ===== @router.get("/{instanceId}/organisations", response_model=PaginatedResponse[TrusteeOrganisation]) diff --git a/modules/routes/routeDataConnections.py b/modules/routes/routeDataConnections.py index f39b6638..c7af510e 100644 --- a/modules/routes/routeDataConnections.py +++ b/modules/routes/routeDataConnections.py @@ -92,6 +92,52 @@ router = APIRouter( responses={404: {"description": "Not found"}} ) + +# ============================================================================ +# OPTIONS ENDPOINTS (for dropdowns) +# ============================================================================ + +@router.get("/statuses/options", response_model=List[Dict[str, Any]]) +@limiter.limit("60/minute") +async def getConnectionStatusOptions( + request: Request, + currentUser: User = Depends(getCurrentUser) +) -> List[Dict[str, Any]]: + """ + Get connection status options for select dropdowns. + Returns standardized format: [{ value, label }] + """ + return [ + {"value": status.value, "label": status.value.capitalize()} + for status in ConnectionStatus + ] + + +@router.get("/authorities/options", response_model=List[Dict[str, Any]]) +@limiter.limit("60/minute") +async def getAuthAuthorityOptions( + request: Request, + currentUser: User = Depends(getCurrentUser) +) -> List[Dict[str, Any]]: + """ + Get authentication authority options for select dropdowns. + Returns standardized format: [{ value, label }] + """ + authorityLabels = { + "local": "Local", + "google": "Google", + "msft": "Microsoft" + } + return [ + {"value": auth.value, "label": authorityLabels.get(auth.value, auth.value)} + for auth in AuthAuthority + ] + + +# ============================================================================ +# CRUD ENDPOINTS +# ============================================================================ + @router.get("/", response_model=PaginatedResponse[UserConnection]) @limiter.limit("30/minute") async def get_connections( diff --git a/modules/routes/routeDataUsers.py b/modules/routes/routeDataUsers.py index dab8bde9..809b3521 100644 --- a/modules/routes/routeDataUsers.py +++ b/modules/routes/routeDataUsers.py @@ -145,6 +145,48 @@ router = APIRouter( responses={404: {"description": "Not found"}} ) + +# ============================================================================ +# OPTIONS ENDPOINTS (for dropdowns) +# ============================================================================ + +@router.get("/options", response_model=List[Dict[str, Any]]) +@limiter.limit("60/minute") +async def getUserOptions( + request: Request, + context: RequestContext = Depends(getRequestContext) +) -> List[Dict[str, Any]]: + """ + Get user options for select dropdowns. + MULTI-TENANT: mandateId from X-Mandate-Id header determines scope. + Returns standardized format: [{ value, label }] + """ + try: + appInterface = interfaceDbApp.getInterface(context.user) + + if context.mandateId: + result = appInterface.getUsersByMandate(str(context.mandateId), None) + users = result.items if hasattr(result, 'items') else result + elif context.isSysAdmin: + users = appInterface.getAllUsers() + else: + raise HTTPException(status_code=403, detail="Access denied") + + return [ + {"value": user.id, "label": user.fullName or user.username or user.email or user.id} + for user in users + ] + except HTTPException: + raise + except Exception as e: + logger.error(f"Error getting user options: {str(e)}") + raise HTTPException(status_code=500, detail=f"Failed to get user options: {str(e)}") + + +# ============================================================================ +# CRUD ENDPOINTS +# ============================================================================ + @router.get("/", response_model=PaginatedResponse[User]) @limiter.limit("30/minute") async def get_users( diff --git a/modules/shared/frontendOptionsTypes.py b/modules/shared/frontendOptionsTypes.py deleted file mode 100644 index 4863a04c..00000000 --- a/modules/shared/frontendOptionsTypes.py +++ /dev/null @@ -1,138 +0,0 @@ -# Copyright (c) 2025 Patrick Motsch -# All rights reserved. -""" -Type definitions and utilities for frontend_options attribute. - -The frontend_options attribute supports two formats: -1. Static List: A list of option dictionaries for static options -2. String Reference: A string identifier that references dynamic options from /api/options/{optionsName} -""" - -from typing import List, Dict, Any, Union - -try: - from typing import TypeAlias # Python 3.10+ -except ImportError: - from typing_extensions import TypeAlias # Python < 3.10 - -# Type definition for a single option item -OptionItem: TypeAlias = Dict[str, Any] -""" -Single option item format: -{ - "value": str, # The value to be stored/returned - "label": { # Multilingual labels - "en": str, - "fr": str, - ... - } -} -""" - -# Type definition for frontend_options - can be either a list or string reference -FrontendOptions: TypeAlias = Union[List[OptionItem], str] -""" -frontend_options can be either: -1. List[OptionItem]: Static list of options - Example: [{"value": "a", "label": {"en": "All", "fr": "Tous"}}] - -2. str: String reference to dynamic options API - Example: "user.role" -> Frontend fetches from /api/options/user.role -""" - - -def isStringReference(frontendOptions: FrontendOptions) -> bool: - """ - Check if frontend_options is a string reference (dynamic) or a list (static). - - Args: - frontendOptions: The frontend_options value to check - - Returns: - True if it's a string reference, False if it's a list - """ - return isinstance(frontendOptions, str) - - -def isStaticList(frontendOptions: FrontendOptions) -> bool: - """ - Check if frontend_options is a static list or a string reference. - - Args: - frontendOptions: The frontend_options value to check - - Returns: - True if it's a static list, False if it's a string reference - """ - return isinstance(frontendOptions, list) - - -def validateFrontendOptions(frontendOptions: FrontendOptions) -> bool: - """ - Validate that frontend_options is in the correct format. - - Args: - frontendOptions: The frontend_options value to validate - - Returns: - True if valid, False otherwise - """ - if isinstance(frontendOptions, str): - # String reference: should be a non-empty string - return bool(frontendOptions.strip()) - - elif isinstance(frontendOptions, list): - # Static list: should contain option dictionaries - if not frontendOptions: - return True # Empty list is valid (no options) - - for option in frontendOptions: - if not isinstance(option, dict): - return False - if "value" not in option: - return False - if "label" not in option: - return False - if not isinstance(option["label"], dict): - return False - - return True - - else: - return False - - -def getOptionsName(frontendOptions: FrontendOptions) -> str: - """ - Get the options name from a string reference. - - Args: - frontendOptions: The frontend_options value (must be a string reference) - - Returns: - The options name (e.g., "user.role") - - Raises: - ValueError: If frontendOptions is not a string reference - """ - if not isStringReference(frontendOptions): - raise ValueError(f"frontend_options is not a string reference: {type(frontendOptions)}") - return frontendOptions - - -def getStaticOptions(frontendOptions: FrontendOptions) -> List[OptionItem]: - """ - Get the static options list. - - Args: - frontendOptions: The frontend_options value (must be a static list) - - Returns: - The list of option items - - Raises: - ValueError: If frontendOptions is not a static list - """ - if not isStaticList(frontendOptions): - raise ValueError(f"frontend_options is not a static list: {type(frontendOptions)}") - return frontendOptions diff --git a/tests/integration/options/test_options_api.py b/tests/integration/options/test_options_api.py deleted file mode 100644 index 7007bf2a..00000000 --- a/tests/integration/options/test_options_api.py +++ /dev/null @@ -1,243 +0,0 @@ -# Copyright (c) 2025 Patrick Motsch -# All rights reserved. -""" -Integration tests for Options API endpoints. -Tests the actual API endpoints with real database connections. -""" - -import pytest -import secrets -from fastapi.testclient import TestClient -from modules.datamodels.datamodelUam import User -from modules.interfaces.interfaceDbApp import getRootInterface - - -@pytest.fixture -def app(): - """Create FastAPI app instance for testing.""" - from app import app as fastapi_app - return fastapi_app - - -@pytest.fixture -def testClient(app): - """Create test client for API testing.""" - return TestClient(app) - - -@pytest.fixture -def csrfToken(): - """Generate a valid CSRF token for testing.""" - # Generate a hex string between 16-64 characters (CSRF validation requirement) - return secrets.token_hex(16) # 32 character hex string - - -@pytest.fixture -def testUser() -> User: - """Create a test user for API testing.""" - # Use getRootInterface for system operations like user creation - # The root interface automatically uses the root mandate - rootInterface = getRootInterface() - user = rootInterface.createUser( - username="testuser_options", - email="testuser_options@example.com", - password="testpass123", - roleLabels=["user"] - ) - return user - - -class TestOptionsAPI: - """Test Options API endpoints.""" - - def testGetOptionsUserRole(self, testClient, testUser, csrfToken): - """Test GET /api/options/user.role endpoint.""" - # Get auth token (stored in cookie) - response = testClient.post( - "/api/local/login", - data={"username": testUser.username, "password": "testpass123"}, - headers={"X-CSRF-Token": csrfToken} - ) - assert response.status_code == 200 - - # Extract token from cookie for Bearer header - token = response.cookies.get("auth_token") - assert token is not None - - # Get options - response = testClient.get( - "/api/options/user.role", - headers={"Authorization": f"Bearer {token}"} - ) - - assert response.status_code == 200 - options = response.json() - - assert isinstance(options, list) - assert len(options) >= 4 # At least sysadmin, admin, user, viewer - - # Check structure - for option in options: - assert "value" in option - assert "label" in option - assert isinstance(option["label"], dict) - - # Check specific values - values = [opt["value"] for opt in options] - assert "sysadmin" in values - assert "admin" in values - assert "user" in values - assert "viewer" in values - - def testGetOptionsAuthAuthority(self, testClient, testUser, csrfToken): - """Test GET /api/options/auth.authority endpoint.""" - # Get auth token (stored in cookie) - response = testClient.post( - "/api/local/login", - data={"username": testUser.username, "password": "testpass123"}, - headers={"X-CSRF-Token": csrfToken} - ) - assert response.status_code == 200 - - # Extract token from cookie for Bearer header - token = response.cookies.get("auth_token") - assert token is not None - - # Get options - response = testClient.get( - "/api/options/auth.authority", - headers={"Authorization": f"Bearer {token}"} - ) - - assert response.status_code == 200 - options = response.json() - - assert isinstance(options, list) - assert len(options) == 3 # local, google, msft - - # Check structure - for option in options: - assert "value" in option - assert "label" in option - - # Check specific values - values = [opt["value"] for opt in options] - assert "local" in values - assert "google" in values - assert "msft" in values - - def testGetOptionsConnectionStatus(self, testClient, testUser, csrfToken): - """Test GET /api/options/connection.status endpoint.""" - # Get auth token (stored in cookie) - response = testClient.post( - "/api/local/login", - data={"username": testUser.username, "password": "testpass123"}, - headers={"X-CSRF-Token": csrfToken} - ) - assert response.status_code == 200 - - # Extract token from cookie for Bearer header - token = response.cookies.get("auth_token") - assert token is not None - - # Get options - response = testClient.get( - "/api/options/connection.status", - headers={"Authorization": f"Bearer {token}"} - ) - - assert response.status_code == 200 - options = response.json() - - assert isinstance(options, list) - assert len(options) >= 4 # active, inactive, expired, pending, revoked, error - - # Check structure - for option in options: - assert "value" in option - assert "label" in option - - def testGetOptionsUserConnection(self, testClient, testUser, csrfToken): - """Test GET /api/options/user.connection endpoint (context-aware).""" - # Get auth token (stored in cookie) - response = testClient.post( - "/api/local/login", - data={"username": testUser.username, "password": "testpass123"}, - headers={"X-CSRF-Token": csrfToken} - ) - assert response.status_code == 200 - - # Extract token from cookie for Bearer header - token = response.cookies.get("auth_token") - assert token is not None - - # Get options (should return empty list if no connections) - response = testClient.get( - "/api/options/user.connection", - headers={"Authorization": f"Bearer {token}"} - ) - - assert response.status_code == 200 - options = response.json() - - # Should return a list (may be empty) - assert isinstance(options, list) - - def testGetOptionsList(self, testClient, testUser, csrfToken): - """Test GET /api/options/ endpoint (list all available options).""" - # Get auth token (stored in cookie) - response = testClient.post( - "/api/local/login", - data={"username": testUser.username, "password": "testpass123"}, - headers={"X-CSRF-Token": csrfToken} - ) - assert response.status_code == 200 - - # Extract token from cookie for Bearer header - token = response.cookies.get("auth_token") - assert token is not None - - # Get available options names - response = testClient.get( - "/api/options/", - headers={"Authorization": f"Bearer {token}"} - ) - - assert response.status_code == 200 - optionsNames = response.json() - - assert isinstance(optionsNames, list) - assert "user.role" in optionsNames - assert "auth.authority" in optionsNames - assert "connection.status" in optionsNames - assert "user.connection" in optionsNames - - def testGetOptionsUnknown(self, testClient, testUser, csrfToken): - """Test GET /api/options/unknown.options endpoint (should return 400).""" - # Get auth token (stored in cookie) - response = testClient.post( - "/api/local/login", - data={"username": testUser.username, "password": "testpass123"}, - headers={"X-CSRF-Token": csrfToken} - ) - assert response.status_code == 200 - - # Extract token from cookie for Bearer header - token = response.cookies.get("auth_token") - assert token is not None - - # Get unknown options (should return error) - response = testClient.get( - "/api/options/unknown.options", - headers={"Authorization": f"Bearer {token}"} - ) - - assert response.status_code == 400 - - def testGetOptionsUnauthorized(self, testClient): - """Test GET /api/options/user.role without authentication.""" - # Try to get options without auth token - response = testClient.get("/api/options/user.role") - - # Should require authentication - assert response.status_code == 401