diff --git a/server/admin_management/api/app/schemas.py b/server/admin_management/api/app/schemas.py index 8502ddf82..1b2f4faaa 100644 --- a/server/admin_management/api/app/schemas.py +++ b/server/admin_management/api/app/schemas.py @@ -205,15 +205,6 @@ class FamAccessControlPrivilegeCreateDto(BaseModel): model_config = ConfigDict(from_attributes=True) -class FamAccessControlPrivilegeCreateErrorDto(BaseModel): - user_id: int - user: FamUserInfoDto - forest_client_number: Annotated[str, StringConstraints(max_length=8)] - parent_role: FamRoleBase - - model_config = ConfigDict(from_attributes=True) - - class FamAccessControlPrivilegeGetResponse(BaseModel): access_control_privilege_id: int user_id: int @@ -226,9 +217,7 @@ class FamAccessControlPrivilegeGetResponse(BaseModel): class FamAccessControlPrivilegeCreateResponse(BaseModel): status_code: int - detail: Union[ - FamAccessControlPrivilegeGetResponse, FamAccessControlPrivilegeCreateErrorDto - ] = None + detail: FamAccessControlPrivilegeGetResponse error_message: Optional[str] = None model_config = ConfigDict(from_attributes=True) @@ -237,12 +226,15 @@ class FamAccessControlPrivilegeCreateResponse(BaseModel): # ------------------------------------- FAM Admin User Access ---------------------------------------- # class FamApplicationDto(BaseModel): id: int = Field(validation_alias="application_id") - name: Annotated[str, StringConstraints(max_length=100)] = \ - Field(validation_alias="application_name") - description: Annotated[Optional[str], StringConstraints(max_length=200)] = \ - Field(default=None, validation_alias="application_description") - env: Optional[famConstants.AppEnv] = \ - Field(validation_alias="app_environment", default=None) + name: Annotated[str, StringConstraints(max_length=100)] = Field( + validation_alias="application_name" + ) + description: Annotated[Optional[str], StringConstraints(max_length=200)] = Field( + default=None, validation_alias="application_description" + ) + env: Optional[famConstants.AppEnv] = Field( + validation_alias="app_environment", default=None + ) model_config = ConfigDict(from_attributes=True) @@ -251,8 +243,9 @@ class FamRoleDto(BaseModel): # Note, this "id" for role can either be concrete role's or abstract role's id. # In abstract role with this id, forest_clients should be present. id: int = Field(validation_alias="role_id") - name: Annotated[str, StringConstraints(max_length=100)] = \ - Field(validation_alias="role_name") + name: Annotated[str, StringConstraints(max_length=100)] = Field( + validation_alias="role_name" + ) type_code: famConstants.RoleType = Field(validation_alias="role_type_code") forest_clients: Optional[List[str]] = Field(default=None) diff --git a/server/admin_management/api/app/services/access_control_privilege_service.py b/server/admin_management/api/app/services/access_control_privilege_service.py index cea2548ef..b9767ae42 100644 --- a/server/admin_management/api/app/services/access_control_privilege_service.py +++ b/server/admin_management/api/app/services/access_control_privilege_service.py @@ -6,8 +6,9 @@ from api.app import schemas from api.app.integration.forest_client_integration import ForestClientService -from api.app.repositories.access_control_privilege_repository import \ - AccessControlPrivilegeRepository +from api.app.repositories.access_control_privilege_repository import ( + AccessControlPrivilegeRepository, +) from api.app.services.role_service import RoleService from api.app.services.user_service import UserService from api.app.utils import utils @@ -81,51 +82,29 @@ def create_access_control_privilege_many( for forest_client_number in request.forest_client_numbers: # validate the forest client number validator = ForestClientValidator(forest_client_number) - error_msg = "" if not validator.forest_client_number_exists(): error_msg = ( - "Invalid access control privilege request. " + "Invalid role assignment request. " + f"Forest client number {forest_client_number} does not exist." ) - elif not validator.forest_client_active(): + utils.raise_http_exception(HTTPStatus.BAD_REQUEST, error_msg) + + if not validator.forest_client_active(): error_msg = ( - "Invalid access control privilege request. " + "Invalid role assignment request. " + f"Forest client number {forest_client_number} is not in active status: " - + f"{validator.get_forest_client()[famConstants.FOREST_CLIENT_STATUS['KEY']]}" + + f"{validator.get_forest_client()[famConstants.FOREST_CLIENT_STATUS['KEY']]}." ) + utils.raise_http_exception(HTTPStatus.BAD_REQUEST, error_msg) - if error_msg != "": - # raise error when adding privilege for only one forest client number - if len(request.forest_client_numbers) == 1: - utils.raise_http_exception(HTTPStatus.BAD_REQUEST, error_msg) - else: - create_return_list.append( - schemas.FamAccessControlPrivilegeCreateResponse( - **{ - "status_code": HTTPStatus.BAD_REQUEST, - "detail": schemas.FamAccessControlPrivilegeCreateErrorDto( - **{ - "user_id": fam_user.user_id, - "user": fam_user, - "forest_client_number": forest_client_number, - "parent_role": fam_role, - } - ), - "error_message": error_msg, - } - ) - ) - else: - # Check if child role exists or add a new child role - child_role = ( - self.role_service.find_or_create_forest_client_child_role( - forest_client_number, fam_role, requester - ) - ) - handle_create_return = self.grant_privilege( - fam_user.user_id, child_role.role_id, requester - ) - create_return_list.append(handle_create_return) + # Check if child role exists or add a new child role + child_role = self.role_service.find_or_create_forest_client_child_role( + forest_client_number, fam_role, requester + ) + handle_create_return = self.grant_privilege( + fam_user.user_id, child_role.role_id, requester + ) + create_return_list.append(handle_create_return) else: handle_create_return = self.grant_privilege( diff --git a/server/admin_management/tests/routers/test_router_access_control_privilege.py b/server/admin_management/tests/routers/test_router_access_control_privilege.py index ea890e9cb..e27fe490b 100644 --- a/server/admin_management/tests/routers/test_router_access_control_privilege.py +++ b/server/admin_management/tests/routers/test_router_access_control_privilege.py @@ -4,17 +4,21 @@ from api.app.main import apiPrefix from api.app.jwt_validation import ERROR_PERMISSION_REQUIRED -from api.app.routers.router_guards import ERROR_INVALID_ROLE_ID, ERROR_INVALID_APPLICATION_ID +from api.app.routers.router_guards import ( + ERROR_INVALID_ROLE_ID, + ERROR_INVALID_APPLICATION_ID, +) from tests.constants import ( TEST_FOREST_CLIENT_NUMBER, TEST_FOREST_CLIENT_NUMBER_TWO, + TEST_INACTIVE_FOREST_CLIENT_NUMBER, TEST_FOM_DEV_ADMIN_ROLE, TEST_NOT_EXIST_ROLE_ID, TEST_ACCESS_CONTROL_PRIVILEGE_CREATE_REQUEST, TEST_FAM_ADMIN_ROLE, TEST_NOT_EXIST_APPLICATION_ID, INVALID_APPLICATION_ID, - TEST_APPLICATION_ID_FOM_DEV + TEST_APPLICATION_ID_FOM_DEV, ) import tests.jwt_utils as jwt_utils @@ -85,6 +89,22 @@ def test_create_access_control_privilege_many( assert data[0].get("status_code") == HTTPStatus.CONFLICT assert data[1].get("status_code") == HTTPStatus.OK + # test create access control privilege with invalid forest client numbers + response = test_client_fixture.post( + f"{endPoint}", + json={ + **TEST_ACCESS_CONTROL_PRIVILEGE_CREATE_REQUEST, + "forest_client_numbers": [ + TEST_FOREST_CLIENT_NUMBER, + TEST_INACTIVE_FOREST_CLIENT_NUMBER, + ], + }, + headers=jwt_utils.headers(token), + ) + assert response.status_code == HTTPStatus.BAD_REQUEST + assert response.json() is not None + assert str(response.json()["detail"]).find("is not in active status") != -1 + def test_get_access_control_privileges_by_application_id( test_client_fixture: starlette.testclient.TestClient, test_rsa_key diff --git a/server/admin_management/tests/services/test_access_control_privilege_service.py b/server/admin_management/tests/services/test_access_control_privilege_service.py index 240e9a7eb..8a20f447f 100644 --- a/server/admin_management/tests/services/test_access_control_privilege_service.py +++ b/server/admin_management/tests/services/test_access_control_privilege_service.py @@ -297,7 +297,7 @@ def test_create_access_control_privilege_many_active_and_inactive_forest_client( ): # test create access control privilege for abstract parent role with 3 forest client numbers, # one is inactive, one is active, one is invalid - return_result = ( + with pytest.raises(HTTPException) as e: access_control_privilege_service.create_access_control_privilege_many( schemas.FamAccessControlPrivilegeCreateRequest( **{ @@ -311,35 +311,9 @@ def test_create_access_control_privilege_many_active_and_inactive_forest_client( ), TEST_CREATOR, ) - ) - assert len(return_result) == 3 - assert return_result[0].status_code == HTTPStatus.OK - assert return_result[ - 0 - ].detail.user.user_name == TEST_ACCESS_CONTROL_PRIVILEGE_CREATE_REQUEST.get( - "user_name" - ) - assert return_result[1].status_code == HTTPStatus.BAD_REQUEST - assert return_result[ - 1 - ].detail.user.user_name == TEST_ACCESS_CONTROL_PRIVILEGE_CREATE_REQUEST.get( - "user_name" - ) assert ( - str(return_result[1].error_message).find( + str(e._excinfo).find( f"Forest client number {TEST_INACTIVE_FOREST_CLIENT_NUMBER} is not in active status" ) != -1 ) - assert return_result[2].status_code == HTTPStatus.BAD_REQUEST - assert return_result[ - 2 - ].detail.user.user_name == TEST_ACCESS_CONTROL_PRIVILEGE_CREATE_REQUEST.get( - "user_name" - ) - assert ( - str(return_result[2].error_message).find( - f"Forest client number {TEST_NON_EXIST_FOREST_CLIENT_NUMBER} does not exist" - ) - != -1 - )