Merge pull request #4777 from bcgov/chore/2123-refactor-registration-file-uploads

chore: registration file upload

Author: pbastia

bc_obps/common/lib/dataclasses/__init__.py

@@ -0,0 +1 @@
+from .asdict import asdict  # noqa: F401

bc_obps/common/lib/dataclasses/asdict.py

@@ -0,0 +1,11 @@
+import dataclasses
+from typing import Any, Optional
+from .dict_factory import dict_factory
+
+
+def asdict(obj: Any, *, include: Optional[set] = None, exclude_none: bool = False) -> dict:
+    """
+    Providing an alternative to dataclasses.asdict to give an API similar to pydantic's dict method.
+    """
+
+    return dataclasses.asdict(obj, dict_factory=dict_factory(include=include, exclude_none=exclude_none))

bc_obps/common/lib/dataclasses/dict_factory.py

@@ -0,0 +1,30 @@
+from typing import Any, Callable, Iterable, List, Optional, Tuple
+
+
+def _field_filter(
+    obj: List[Tuple[str, Any]], include: Optional[set] = None, exclude_none: bool = False
+) -> Iterable[Tuple[str, Any]]:
+    """
+    Helper function to include only specified fields from a dataclass in the output dict.
+    The field information passed to the factory by dataclasses.asdict is a list of tuples in the form (field_name, field_value).
+    """
+    if include is not None and not isinstance(include, set):
+        raise ValueError("The 'include' parameter must be a set of field names.")
+
+    for field in obj:
+        if include is not None and field[0] not in include:
+            continue
+        if exclude_none and field[1] is None:
+            continue
+        yield field
+
+
+def dict_factory(include: Optional[set] = None, exclude_none: bool = False) -> Callable[[List[Tuple[str, Any]]], dict]:
+    """
+    Factory function that returns a function to include only specified fields in the output dict.
+    """
+
+    def factory_func(obj: Any) -> dict:
+        return dict(_field_filter(obj, include=include, exclude_none=exclude_none))
+
+    return factory_func

bc_obps/common/tests/endpoints/auth/constants.py

@@ -236,7 +236,7 @@
             "kwargs": {"version_id": MOCK_INT},
         },
         {
-            "method": "put",
+            "method": "post",
             "endpoint_name": "register_edit_operation_information",
             "kwargs": {"operation_id": MOCK_UUID},
         },
@@ -266,7 +266,7 @@
             "kwargs": {"version_id": MOCK_INT},
         },
         {
-            "method": "put",
+            "method": "post",
             "endpoint_name": "update_operation",
             "kwargs": {"operation_id": MOCK_UUID},
         },
@@ -290,7 +290,7 @@
             "kwargs": {"operation_id": MOCK_UUID},
         },
         {
-            "method": "put",
+            "method": "post",
             "endpoint_name": "create_or_replace_new_entrant_application",
             "kwargs": {"operation_id": MOCK_UUID},
         },

bc_obps/common/tests/lib/dataclasses/test_asdict.py

@@ -0,0 +1,50 @@
+from dataclasses import dataclass
+
+from common.lib.dataclasses.asdict import asdict
+
+
+@dataclass
+class Child:
+    value: int
+    optional_value: str | None = None
+
+
+@dataclass
+class Parent:
+    name: str
+    child: Child
+    optional_field: str | None = None
+
+
+class TestAsdict:
+    def test_returns_nested_dataclass_as_dict(self):
+        subject = Parent(name="parent", child=Child(value=1, optional_value="child"))
+
+        assert asdict(subject) == {
+            "name": "parent",
+            "child": {"value": 1, "optional_value": "child"},
+            "optional_field": None,
+        }
+
+    def test_include_limits_top_level_fields(self):
+        subject = Parent(name="parent", child=Child(value=1), optional_field="kept")
+
+        assert asdict(subject, include={"name", "optional_field"}) == {
+            "name": "parent",
+            "optional_field": "kept",
+        }
+
+    def test_exclude_none_removes_none_values(self):
+        subject = Parent(name="parent", child=Child(value=1), optional_field=None)
+
+        assert asdict(subject, exclude_none=True) == {
+            "name": "parent",
+            "child": {"value": 1},
+        }
+
+    def test_include_and_exclude_none_together(self):
+        subject = Parent(name="parent", child=Child(value=1), optional_field=None)
+
+        assert asdict(subject, include={"name", "optional_field"}, exclude_none=True) == {
+            "name": "parent",
+        }

bc_obps/common/tests/utils/test_files.py

@@ -0,0 +1,6 @@
+from django.core.files.base import ContentFile
+from django.core.files.uploadedfile import InMemoryUploadedFile, UploadedFile
+
+
+def create_test_file(name: str) -> UploadedFile:
+    return InMemoryUploadedFile(ContentFile(b"file_content", name=name), None, name, None, 12, "utf-8")

bc_obps/registration/api/_operations/_operation_id/_registration/new_entrant_application.py

@@ -1,11 +1,11 @@
 from typing import Literal, Tuple
 from uuid import UUID
 from django.http import HttpRequest
+from ninja import File, UploadedFile
 from registration.constants import OPERATION_TAGS
 from service.error_service.custom_codes_4xx import custom_codes_4xx
 from registration.schema import (
     OperationUpdateOut,
-    OperationNewEntrantApplicationIn,
     OperationNewEntrantApplicationOut,
     Message,
 )
@@ -15,7 +15,6 @@
 from registration.models import Operation
 from registration.api.router import router
 
-
 ##### GET #####
 
 
@@ -32,16 +31,18 @@ def get_operation_new_entrant_application(request: HttpRequest, operation_id: UU
     return 200, OperationService.get_if_authorized(get_current_user_guid(request), operation_id, ['id', 'operator_id'])
 
 
-@router.put(
+@router.post(
     "/operations/{uuid:operation_id}/registration/new-entrant-application",
     response={200: OperationUpdateOut, custom_codes_4xx: Message},
     tags=OPERATION_TAGS,
     description="Creates or replaces a new entrant application document for an Operation",
     auth=authorize("approved_industry_user"),
 )
 def create_or_replace_new_entrant_application(
-    request: HttpRequest, operation_id: UUID, payload: OperationNewEntrantApplicationIn
+    request: HttpRequest,
+    operation_id: UUID,
+    new_entrant_application: File[UploadedFile],
 ) -> Tuple[Literal[200], Operation]:
     return 200, OperationService.create_or_replace_new_entrant_application(
-        get_current_user_guid(request), operation_id, payload
+        get_current_user_guid(request), operation_id, new_entrant_application
     )

bc_obps/registration/api/_operations/_operation_id/_registration/operation.py

@@ -1,8 +1,9 @@
 from typing import Literal, Tuple
 from uuid import UUID
 from django.http import HttpRequest
+from ninja import File, UploadedFile
 from registration.schema import OperationInformationIn, OperationUpdateOut, OperationRegistrationOut, Message
-from service.operation_service import OperationService
+from service.operation_service import OperationData, OperationService, MultipleOperatorData
 from registration.constants import OPERATION_TAGS
 from common.permissions import authorize
 from common.api.utils import get_current_user_guid
@@ -25,10 +26,7 @@ def register_get_operation_information(request: HttpRequest, operation_id: UUID)
     return 200, OperationService.get_if_authorized(get_current_user_guid(request), operation_id)
 
 
-##### PUT #####
-
-
-@router.put(
+@router.post(
     "/operations/{uuid:operation_id}/registration/operation",
     response={200: OperationUpdateOut, custom_codes_4xx: Message},
     tags=OPERATION_TAGS,
@@ -37,6 +35,29 @@ def register_get_operation_information(request: HttpRequest, operation_id: UUID)
     auth=authorize('approved_industry_user'),
 )
 def register_edit_operation_information(
-    request: HttpRequest, operation_id: UUID, payload: OperationInformationIn
+    request: HttpRequest,
+    operation_id: UUID,
+    payload: OperationInformationIn,
+    # django-ninja doesn't parse multipart requests properly if the type is marked Optional
+    boundary_map: File[UploadedFile] = None,  # type: ignore
+    process_flow_diagram: File[UploadedFile] = None,  # type: ignore
+    new_entrant_application: File[UploadedFile] = None,  # type: ignore
 ) -> Tuple[Literal[200], Operation]:
-    return 200, OperationService.register_operation_information(get_current_user_guid(request), operation_id, payload)
+
+    data = OperationData(
+        boundary_map=boundary_map,
+        process_flow_diagram=process_flow_diagram,
+        new_entrant_application=new_entrant_application,
+        **payload.model_dump(exclude={'multiple_operators_array'}),
+        multiple_operators_array=(
+            [
+                MultipleOperatorData(
+                    **op.model_dump(exclude={'business_structure'}), business_structure_id=op.business_structure.name  # type: ignore
+                )
+                for op in payload.multiple_operators_array or []
+            ]
+        ),
+    )
+    operation = OperationService.register_operation_information(get_current_user_guid(request), operation_id, data)
+
+    return 200, operation

bc_obps/registration/api/_operations/_operation_id/_registration/operation_representative.py

@@ -41,4 +41,10 @@ def create_operation_representative(
 def remove_operation_representative(
     request: HttpRequest, operation_id: UUID, payload: OperationRepresentativeRemove
 ) -> Tuple[Literal[200], OperationRepresentativeRemove]:
-    return 200, OperationService.remove_operation_representative(get_current_user_guid(request), operation_id, payload)
+    removed_id = OperationService.remove_operation_representative(
+        get_current_user_guid(request),
+        operation_id,
+        payload.id,  # type: ignore
+    )
+
+    return 200, OperationRepresentativeRemove(id=removed_id)

bc_obps/registration/api/_operations/_operation_id/_registration/opted_in_operation.py

@@ -3,6 +3,7 @@
 from django.http import HttpRequest
 from registration.models.opted_in_operation_detail import OptedInOperationDetail
 from registration.schema import OptedInOperationDetailIn, OptedInOperationDetailOut, OptedOutOperationDetailIn, Message
+from service.data_types.operation_service import OptedInOperationDetailData
 from service.operation_service import OperationService
 from registration.constants import OPERATION_TAGS
 from common.permissions import authorize
@@ -36,7 +37,13 @@ def operation_registration_get_opted_in_operation_detail(
 def operation_registration_update_opted_in_operation_detail(
     request: HttpRequest, operation_id: UUID, payload: OptedInOperationDetailIn
 ) -> Tuple[Literal[200, 400], OptedInOperationDetail]:
-    return 200, OperationService.update_opted_in_operation_detail(get_current_user_guid(request), operation_id, payload)
+
+    opted_in_operation_detail_data = OptedInOperationDetailData(**payload.dict())
+    operation = OperationService.update_opted_in_operation_detail(
+        get_current_user_guid(request), operation_id, opted_in_operation_detail_data
+    )
+
+    return 200, operation
 
 
 @router.put(