Permissions settings for enforcements (#636)

* requirement source title related fix

* permissions settings for enforcements

Author: dinesh-aot

compliance-api/src/compliance_api/schemas/inspection_approval.py

@@ -6,6 +6,7 @@
 
 from compliance_api.models.inspection_record_approval import InspectionRecordApproval as InspectionRecordApprovalModel
 from compliance_api.models.inspection_record_approval import IRApprovalStatusEnum
+from compliance_api.schemas.common import KeyValueSchema
 from compliance_api.utils.constant import INPUT_DATE_TIME_FORMAT
 
 from .base_schema import AutoSchemaBase, BaseSchema

compliance-api/src/compliance_api/services/administrative_penalty.py

@@ -182,11 +182,12 @@ def update_administrative_penalty(cls, administrative_penalty_id, update_data):
             inspection_id=update_data.get("inspection_id")
             or administrative_penalty.inspection_id
         )
-        ServiceUtils.access_check_update_for_inspection(inspection)
-        # If administrative penalty is closed, then check the inspection status
-        # No more modification possible once the AP and IR is closed
+        #  Primary or super user can update ap if it is not closed
+        if not administrative_penalty.is_closed:
+            ServiceUtils.access_check_update_for_inspection(inspection)
+        #  Super user can update ap if it is closed
         if administrative_penalty.is_closed:
-            ServiceUtils.inspection_status_check(inspection)
+            ServiceUtils.access_check_for_super_user()
 
         requirement_ids = update_data.get("inspection_requirement_ids", [])
         if requirement_ids:

compliance-api/src/compliance_api/services/charge_recommendation.py

@@ -87,11 +87,12 @@ def update_charge_recommendation(
         inspection = ServiceUtils.inspection_exist_check(
             inspection_id=update_data.get("inspection_id")
         )
-        ServiceUtils.access_check_update_for_inspection(inspection)
-        # If charge recommendation is closed, then check the inspection status
-        # No more modification possible once the CR and IR is closed
+        # Primary or super user can update CR if it is not closed
+        if not charge_recommendation.is_closed:
+            ServiceUtils.access_check_update_for_inspection(inspection)
+        # Super user can update CR if it is closed
         if charge_recommendation.is_closed:
-            ServiceUtils.inspection_status_check(inspection)
+            ServiceUtils.access_check_for_super_user()
         requirement_ids = update_data.get("inspection_requirement_ids", [])
         with session_scope() as session:
             extracted_update_data = _extract_cr_data(update_data)

compliance-api/src/compliance_api/services/inspection.py

@@ -497,14 +497,12 @@ def generate_inspections_excel(cls, args):
         return output
 
     @classmethod
-    def get_pending_items(cls, inspection_id: int):
+    def get_pending_items(cls, inspection_id: int):  # pylint: disable=too-many-locals
         """Get pending items for an inspection (optimized with single query).
 
         Returns a list of items (enforcement actions, inspection records) that are mapped to requirements
         or the inspection but not yet created or not in proper status.
         """
-        from sqlalchemy.orm import aliased
-
         pending_items = []
         seen_item_numbers = set()
 
@@ -521,7 +519,7 @@ def get_pending_items(cls, inspection_id: int):
         order_map = aliased(OrderInspectionRequirementMapModel)
         order = aliased(OrderModel)
         wl_map = aliased(WarningLetterInspectionRequirementMapModel)
-        wl = aliased(WarningLetterModel)
+        warning_letter = aliased(WarningLetterModel)
         ap_map = aliased(AdministrativePenaltyInspectionRequirementMapModel)
         vt_map = aliased(ViolationTicketInspectionRequirementMapModel)
         cr_map = aliased(ChargeRecommendationInspectionRequirementMapModel)
@@ -535,8 +533,8 @@ def get_pending_items(cls, inspection_id: int):
                 enf_action.name.label("enforcement_action_name"),
                 order.order_number.label("order_number"),
                 order.order_status.label("order_status"),
-                wl.warning_letter_number.label("warning_letter_number"),
-                wl.status.label("warning_letter_status"),
+                warning_letter.warning_letter_number.label("warning_letter_number"),
+                warning_letter.status.label("warning_letter_status"),
                 order_map.id.label("order_map_id"),
                 wl_map.id.label("wl_map_id"),
                 ap_map.id.label("ap_map_id"),
@@ -574,7 +572,7 @@ def get_pending_items(cls, inspection_id: int):
                     wl_map.is_deleted.is_(False),
                 ),
             )
-            .outerjoin(wl, wl.id == wl_map.warning_letter_id)
+            .outerjoin(warning_letter, warning_letter.id == wl_map.warning_letter_id)
             .outerjoin(
                 ap_map,
                 and_(
@@ -1199,7 +1197,7 @@ def _set_charge_recommendation_enforcement_action_object(
 
 def _build_query_for_enforcement_actions_and_requirement_details(inspection_ids):
     """Build the query for bulk fetching enforcement actions and requirement details."""
-    # pylint: disable=invalid-name
+    # pylint: disable=invalid-name,too-many-locals
     # Create aliases for mapping tables
     order_map_alias = aliased(OrderInspectionRequirementMapModel)
     warning_letter_map_alias = aliased(WarningLetterInspectionRequirementMapModel)
@@ -1339,7 +1337,9 @@ def _build_query_for_enforcement_actions_and_requirement_details(inspection_ids)
     return results
 
 
-def _bulk_fetch_enforcement_actions_and_requirement_details(inspection_ids):
+def _bulk_fetch_enforcement_actions_and_requirement_details(
+    inspection_ids,
+):  # pylint: disable=too-many-locals,too-many-branches
     """Bulk fetch all enforcement actions and build requirement details in a single optimized query.
 
     Returns a dict with two keys per inspection:
@@ -1504,7 +1504,7 @@ def _bulk_fetch_enforcement_actions_and_requirement_details(inspection_ids):
     return result_data
 
 
-def _process_enforcement_status(
+def _process_enforcement_status(  # pylint: disable=too-many-return-statements,too-many-branches
     enforcement_action_id: int,
     order_map_id,
     order_number,
@@ -1544,7 +1544,7 @@ def _process_enforcement_status(
             return {"is_created": True, "item_number": order_number, "is_issued": False}
         return None
 
-    elif enforcement_action_id == EnforcementActionOptionEnum.WARNING_LETTER.value:
+    if enforcement_action_id == EnforcementActionOptionEnum.WARNING_LETTER.value:
         if not wl_map_id:
             return {"is_created": False, "item_number": None}
         if not warning_letter_number:
@@ -1557,20 +1557,20 @@ def _process_enforcement_status(
             }
         return None
 
-    elif (
+    if (
         enforcement_action_id
         == EnforcementActionOptionEnum.ADMINISTRATIVE_PENALTY_RECOMMENDATION.value
     ):
         if not ap_map_id:
             return {"is_created": False, "item_number": None}
         return None
 
-    elif enforcement_action_id == EnforcementActionOptionEnum.VIOLATION_TICKET.value:
+    if enforcement_action_id == EnforcementActionOptionEnum.VIOLATION_TICKET.value:
         if not vt_map_id:
             return {"is_created": False, "item_number": None}
         return None
 
-    elif (
+    if (
         enforcement_action_id == EnforcementActionOptionEnum.CHARGE_RECOMMENDATION.value
     ):
         if not cr_map_id:

compliance-api/src/compliance_api/services/inspection_requirement.py

@@ -1323,30 +1323,29 @@ def _get_enforcement_number_by_type(result):
     return ""
 
 
-def _get_enforcement_status_by_type(result):
+def _get_enforcement_status_by_type(result):  # pylint: disable=too-many-return-statements
     """Get the correct enforcement status based on the enforcement action type."""
     enforcement_action_id = result.enforcement_action_id
 
     # Map enforcement action ID to the corresponding status field
     if enforcement_action_id == EnforcementActionOptionEnum.ORDER.value:
         return result.order_status
-    elif enforcement_action_id == EnforcementActionOptionEnum.WARNING_LETTER.value:
+    if enforcement_action_id == EnforcementActionOptionEnum.WARNING_LETTER.value:
         return result.warning_letter_status
-    elif enforcement_action_id == EnforcementActionOptionEnum.VIOLATION_TICKET.value:
+    if enforcement_action_id == EnforcementActionOptionEnum.VIOLATION_TICKET.value:
         return result.violation_ticket_status
-    elif (
+    if (
         enforcement_action_id
         == EnforcementActionOptionEnum.ADMINISTRATIVE_PENALTY_RECOMMENDATION.value
     ):
         return result.admin_penalty_status
-    elif (
+    if (
         enforcement_action_id == EnforcementActionOptionEnum.CHARGE_RECOMMENDATION.value
     ):
         return result.charge_rec_status
-    elif enforcement_action_id == EnforcementActionOptionEnum.RESTORATIVE_JUSTICE.value:
+    if enforcement_action_id == EnforcementActionOptionEnum.RESTORATIVE_JUSTICE.value:
         return result.restorative_justice_status
-    else:
-        return None
+    return None
 
 
 def _get_enforcement_progress_by_type(result):
@@ -1356,10 +1355,9 @@ def _get_enforcement_progress_by_type(result):
     # Only Order and Warning Letter have progress fields
     if enforcement_action_id == EnforcementActionOptionEnum.ORDER.value:
         return result.order_progress
-    elif enforcement_action_id == EnforcementActionOptionEnum.WARNING_LETTER.value:
+    if enforcement_action_id == EnforcementActionOptionEnum.WARNING_LETTER.value:
         return result.warning_letter_progress
-    else:
-        return None
+    return None
 
 
 def _apply_enforcement_number_sort(query, subq, sort_order):

compliance-api/src/compliance_api/services/restorative_justice.py

@@ -77,10 +77,18 @@ def update_restorative_justice(
     ) -> RestorativeJusticeModel:
         """Update an existing restorative justice."""
         # Check if restorative justice exists
-        cls.get_by_id(restorative_justice_id)
+        restorative_justice = cls.get_by_id(restorative_justice_id)
 
         # Extract inspection requirement IDs
         inspection_requirement_ids = update_data.pop("inspection_requirement_ids", None)
+        inspection = ServiceUtils.inspection_exist_check(
+            inspection_id=update_data.get("inspection_id")
+            or restorative_justice.inspection_id
+        )
+        if not restorative_justice.status == RestorativeJusticeStatusEnum.CLOSED:
+            ServiceUtils.access_check_update_for_inspection(inspection)
+        if restorative_justice.status == RestorativeJusticeStatusEnum.CLOSED:
+            ServiceUtils.access_check_for_super_user()
 
         with session_scope() as session:
             # Update the restorative justice

compliance-api/src/compliance_api/services/service_utils.py

@@ -52,6 +52,14 @@ def access_check_update_for_inspection(inspection: dict):
                 "Only the Superuser or the Primary officer on this inspection can perform this operation."
             )
 
+    @staticmethod
+    def access_check_for_super_user():
+        """Access check for super user."""
+        if not auth.has_permission([PermissionEnum.SUPERUSER]):
+            raise PermissionDeniedError(
+                "Only the Superuser can perform this operation."
+            )
+
     @staticmethod
     def inspection_exist_check(inspection_id: int):
         """Check if the inspection exist or not."""

compliance-api/src/compliance_api/services/violation_ticket.py

@@ -102,13 +102,14 @@ def update(
             raise ResourceNotFoundError(
                 f"Violation ticket with ID {violation_ticket_id} not found"
             )
-
-        # Perform access and status checks
-        ServiceUtils.access_check_update_for_inspection(violation_ticket.inspection)
-        # If violation ticket is closed, then check the inspection status
-        # No more modification possible once the VT and IR is closed
+        inspection = ServiceUtils.inspection_exist_check(
+            inspection_id=violation_ticket_data.get("inspection_id")
+            or violation_ticket.inspection_id
+        )
+        if not violation_ticket.is_closed:
+            ServiceUtils.access_check_update_for_inspection(inspection)
         if violation_ticket.is_closed:
-            ServiceUtils.inspection_status_check(violation_ticket.inspection)
+            ServiceUtils.access_check_for_super_user()
 
         # Check if violation ticket can be updated based on status
         if violation_ticket.status == ViolationTicketStatusEnum.PAID:

compliance-web/cypress/components/_components/_App/_Inspections/_Profile/InspectionEnforcements.cy.tsx

@@ -9,6 +9,8 @@ import { AdministrativePenalty } from "@/models/AdministrativePenalty";
 import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
 import { Box } from "@mui/material";
 import { EnforcementActionEnum } from "@/utils/constants";
+import { AuthProvider } from "react-oidc-context";
+import { OidcConfig } from "@/utils/config";
 
 describe("InspectionEnforcements", () => {
   let queryClient: QueryClient;
@@ -21,24 +23,26 @@ describe("InspectionEnforcements", () => {
   const mountComponent = (inspectionData = mockInspection) => {
     mount(
       <QueryClientProvider client={queryClient}>
-        <Box
-          sx={{
-            width: "800px",
-            height: "800px",
-            overflow: "visible",
-            position: "relative",
-          }}
-        >
-          <InspectionEnforcements inspectionData={inspectionData} />
-        </Box>
+        <AuthProvider {...OidcConfig}>
+          <Box
+            sx={{
+              width: "800px",
+              height: "800px",
+              overflow: "visible",
+              position: "relative",
+            }}
+          >
+            <InspectionEnforcements inspectionData={inspectionData} />
+          </Box>
+        </AuthProvider>
       </QueryClientProvider>
     );
   };
 
   beforeEach(() => {
     // Set viewport to ensure proper dimensions
     cy.viewport(1200, 800);
-
+    
     // Add CSS overrides to fix overflow issues in testing
     cy.document().then((doc) => {
       const style = doc.createElement("style");
@@ -62,6 +66,21 @@ describe("InspectionEnforcements", () => {
       },
     });
 
+    // Mock current user for useCurrentLoggedInUser hook
+    const mockCurrentUser = { preferred_username: "test.user@gov.bc.ca" };
+    queryClient.setQueryData(["current-user"], mockCurrentUser);
+
+    // Mock staff users data
+    const mockStaffUsers = [
+      {
+        id: 1,
+        name: "Test User",
+        auth_user_guid: "test.user@gov.bc.ca",
+        is_active: true,
+      },
+    ];
+    queryClient.setQueryData(["staff-users", true], mockStaffUsers);
+
     // Default mock inspection data
     mockInspection = {
       id: 1,

compliance-web/src/components/App/Inspections/Profile/Enforcements/AdministrativePenalty/AdministrativePenaltyUpdateModal.tsx

@@ -23,6 +23,7 @@ import { notify } from "@/store/snackbarStore";
 import { useModal } from "@/store/modalStore";
 import dayjs, { Dayjs } from "dayjs";
 import { APReferralStatus, APDecisionStatus } from "@/utils/constants";
+import { KC_USER_GROUPS, useIsRolesAllowed } from "@/hooks/useAuthorization";
 
 const administrativePenaltyUpdateSchema = yup.object().shape({
   referral_status: yup
@@ -76,20 +77,23 @@ const decisionOptions: DecisionOption[] = Object.values(APDecisionStatus).map(
 type AdministrativePenaltyUpdateModalProps = {
   administrativePenalty: AdministrativePenalty;
   inspectionData: Inspection;
+  isPrimaryOfficerOrSuperUser: boolean;
   onSuccess?: (data: AdministrativePenalty) => void;
 };
 
 const AdministrativePenaltyUpdateModal: FC<
   AdministrativePenaltyUpdateModalProps
-> = ({ administrativePenalty, inspectionData, onSuccess }) => {
+> = ({ administrativePenalty, inspectionData, isPrimaryOfficerOrSuperUser, onSuccess }) => {
   const queryClient = useQueryClient();
   const { setClose: setModalClose } = useModal();
-
+  const isSuperUser = useIsRolesAllowed([KC_USER_GROUPS.SUPERUSER]);
   // Calculate readonly mode based on inspection status and administrative penalty status
   const isReadonlyMode = useMemo(() => {
-    const isInspectionClosed = inspectionData.inspection_status.toLowerCase() === "closed";
-    return isInspectionClosed && administrativePenalty.is_closed;
-  }, [inspectionData.inspection_status, administrativePenalty.is_closed]);
+    if (administrativePenalty.is_closed) {
+      return !isSuperUser;
+    }
+    return !isPrimaryOfficerOrSuperUser;
+  }, [administrativePenalty.is_closed, isPrimaryOfficerOrSuperUser, isSuperUser]);
 
   const defaultValues = useMemo(() => {
     const currentReferralStatus =