chore: all tests pass - is mypy happy now?

bc_obps/registration/schema/v2/operation.py

@@ -82,7 +82,7 @@ class OperationInformationIn(ModelSchema):
     activities: List[int]
     boundary_map: Optional[str] = None
     process_flow_diagram: Optional[str] = None
-    naics_code_id: int
+    naics_code_id: Optional[int] = None
     opt_in: Optional[bool] = False
     secondary_naics_code_id: Optional[int] = None
     tertiary_naics_code_id: Optional[int] = None

bc_obps/registration/tests/integration/test_changing_registration_purpose.py

@@ -83,16 +83,28 @@ def _set_facilities(self):
 
     def _set_new_registration_purpose(self, new_purpose):
         operation_payload = {
-            "registration_purpose": new_purpose,
-            "boundary_map": MOCK_DATA_URL,
-            "process_flow_diagram": MOCK_DATA_URL,
-            "activities": [2, 3],
-            "naics_code_id": self.operation.naics_code_id,
-            "secondary_naics_code_id": self.operation.secondary_naics_code_id,
             "name": self.operation.name,
             "type": self.operation.type,
-            "regulated_products": [1, 2] if new_purpose != Operation.Purposes.ELECTRICITY_IMPORT_OPERATION else [],
+            "registration_purpose": new_purpose,
         }
+        if new_purpose == Operation.Purposes.ELECTRICITY_IMPORT_OPERATION:
+            operation_payload.update(
+                {
+                    "activities": [],
+                    "regulated_products": [],
+                }
+            )
+        else:
+            operation_payload.update(
+                {
+                    "boundary_map": MOCK_DATA_URL,
+                    "process_flow_diagram": MOCK_DATA_URL,
+                    "activities": [2, 3],
+                    "naics_code_id": self.operation.naics_code_id,
+                    "secondary_naics_code_id": self.operation.secondary_naics_code_id,
+                    "regulated_products": [1, 2],
+                }
+            )
         response = TestUtils.mock_put_with_auth_role(
             self,
             "industry_user",

bc_obps/registration/tests/integration/test_operation_registration.py

@@ -234,12 +234,14 @@ def test_operation_registration_workflow(self, operation_type, purpose):
             assert self.operation.opted_in_operation_id is None
 
         # make sure we have the two required documents
-        assert (
-            self.operation.documents.filter(Q(type__name='process_flow_diagram') | Q(type__name='boundary_map'))
-            .distinct()
-            .count()
-            == 2
-        )
+        # (unless the registration_purpose is EIO, in which case no documents are required)
+        if purpose != Operation.Purposes.ELECTRICITY_IMPORT_OPERATION:
+            assert (
+                self.operation.documents.filter(Q(type__name='process_flow_diagram') | Q(type__name='boundary_map'))
+                .distinct()
+                .count()
+                == 2
+            )
 
         # make sure we have the facility
         if operation_type == OperationTypes.LFO:

bc_obps/service/data_access_service/opted_in_operation_detail_service.py

@@ -1,6 +1,8 @@
 from uuid import UUID
+from service.operation_service import OperationService
 from registration.schema.v2.operation import OptedInOperationDetailIn
 from registration.models.opted_in_operation_detail import OptedInOperationDetail
+from registration.models.operation import Operation
 from registration.utils import update_model_instance
 
 
@@ -21,3 +23,14 @@ def update_opted_in_operation_detail(
         updated_opted_in_operation_detail_instance.save()
         updated_opted_in_operation_detail_instance.set_create_or_update(user_guid)
         return updated_opted_in_operation_detail_instance
+
+    @classmethod
+    def archive_or_delete_opted_in_operation_detail(cls, user_guid: UUID, operation_id: UUID) -> None:
+        operation = OperationService.get_if_authorized(user_guid, operation_id)
+        if operation.opted_in_operation:
+            opted_in_operation_detail_id = operation.opted_in_operation.id
+            opted_in_operation_detail = OptedInOperationDetail.objects.get(pk=opted_in_operation_detail_id)
+            if operation.status == Operation.Statuses.REGISTERED:
+                opted_in_operation_detail.set_archive(user_guid)
+            else:
+                opted_in_operation_detail.delete()

bc_obps/service/operation_service_v2.py

@@ -177,9 +177,7 @@ def remove_operation_representative(
     def remove_opted_in_operation_detail(cls, user_guid: UUID, operation_id: UUID) -> Operation:
         operation: Operation = OperationService.get_if_authorized(user_guid, operation_id)
         operation.opt_in = False
-        opted_in_detail: OptedInOperationDetail = operation.opted_in_operation
-        # TODO - determine whether this should be archived or deleted??
-        opted_in_detail.set_archive(user_guid)
+        OptedInOperationDataAccessService.archive_or_delete_opted_in_operation_detail(user_guid, operation_id)
         operation.opted_in_operation = None
         operation.save(update_fields=['opt_in', 'opted_in_operation'])
 
@@ -200,7 +198,6 @@ def create_or_update_operation_v2(
             original_operation = OperationService.get_if_authorized(user_guid, operation_id)
             if payload.registration_purpose != original_operation.registration_purpose:
                 payload = cls.handle_change_of_registration_purpose(user_guid, original_operation, payload)
-                print(payload)
 
         operation: Operation
         operation_data = payload.dict(
@@ -217,6 +214,7 @@ def create_or_update_operation_v2(
         )
         operation_data['pk'] = operation_id if operation_id else None
         operation_data['operator_id'] = user_operator.operator_id
+
         operation, _ = Operation.custom_update_or_create(Operation, user_guid, **operation_data)
 
         # set m2m relationships
@@ -444,12 +442,12 @@ def check_conditions() -> Generator[Tuple[Callable[[], bool], str], None, None]:
 
             # Check if the operation has both a process flow diagram and a boundary map (unless it is an EIO)
             yield (
-                lambda: (
+                lambda: not (
                     operation.registration_purpose != Operation.Purposes.ELECTRICITY_IMPORT_OPERATION
                     and operation.documents.filter(Q(type__name='process_flow_diagram') | Q(type__name='boundary_map'))
                     .distinct()
                     .count()
-                    == 2
+                    < 2
                 ),
                 "Operation must have a process flow diagram and a boundary map.",
             )
@@ -526,10 +524,10 @@ def handle_change_of_registration_purpose(
     ) -> OperationInformationIn:
         if original_operation.registration_purpose == Operation.Purposes.OPTED_IN_OPERATION:
             payload.opt_in = False
-            opted_in_detail: OptedInOperationDetail = original_operation.opted_in_operation
-            if original_operation.status == Operation.Statuses.REGISTERED:
+            opted_in_detail = original_operation.opted_in_operation
+            if opted_in_detail and original_operation.status == Operation.Statuses.REGISTERED:
                 opted_in_detail.set_archive(user_guid)
-            else:
+            elif opted_in_detail:
                 opted_in_detail.delete()
         elif original_operation.registration_purpose == Operation.Purposes.NEW_ENTRANT_OPERATION:
             payload.date_of_first_shipment = None
@@ -545,18 +543,10 @@ def handle_change_of_registration_purpose(
             payload.tertiary_naics_code_id = None
             payload.boundary_map = None
             payload.process_flow_diagram = None
-            if original_operation.status == Operation.Statuses.REGISTERED:
-                DocumentService.archive_or_delete_operation_document(
-                    user_guid, original_operation.id, 'process_flow_diagram'
-                )
-                DocumentService.archive_or_delete_operation_document(user_guid, original_operation.id, 'boundary_map')
-            else:
-                boundary_map = original_operation.get_boundary_map()
-                if boundary_map:
-                    boundary_map.delete()
-                process_flow_diagram = original_operation.get_process_flow_diagram()
-                if process_flow_diagram:
-                    process_flow_diagram.delete()
+            DocumentService.archive_or_delete_operation_document(
+                user_guid, original_operation.id, 'process_flow_diagram'
+            )
+            DocumentService.archive_or_delete_operation_document(user_guid, original_operation.id, 'boundary_map')
         elif payload.registration_purpose in [
             Operation.Purposes.REPORTING_OPERATION,
             Operation.Purposes.POTENTIAL_REPORTING_OPERATION,

bc_obps/service/tests/test_operation_service_v2.py

@@ -146,8 +146,6 @@ def test_remove_opted_in_operation_detail():
         assert operation.opt_in is True
         assert operation.opted_in_operation is not None
 
-        opted_in_operation_detail_id = operation.opted_in_operation.id
-
         opted_in_payload = OptedInOperationDetailIn(
             meets_section_3_emissions_requirements=False,
             meets_electricity_import_operation_criteria=False,
@@ -163,15 +161,11 @@ def test_remove_opted_in_operation_detail():
         )
 
         OperationServiceV2.remove_opted_in_operation_detail(approved_user_operator.user.user_guid, operation.id)
-        opted_in_operation_detail.refresh_from_db()
         operation.refresh_from_db()
 
         assert operation.opt_in is False
         assert operation.opted_in_operation is None
-        # assert OptedInOperationDetail.objects.get(pk=opted_in_operation_detail_id).exists()
         assert opted_in_operation_detail is not None
-        assert opted_in_operation_detail.archived_by == approved_user_operator.user
-        assert opted_in_operation_detail.archived_at is not None
 
     @staticmethod
     def test_assigning_opted_in_operation_will_create_and_opted_in_operation_detail():
@@ -509,7 +503,7 @@ def test_create_operation_with_multiple_operators():
     @staticmethod
     def test_update_operation_with_multiple_operators():
         approved_user_operator = baker.make_recipe('utils.approved_user_operator')
-        existing_operation = baker.make_recipe('utils.operation')
+        existing_operation = baker.make_recipe('utils.operation', operator=approved_user_operator.operator)
         multiple_operators = baker.make_recipe('utils.multiple_operator', operation=existing_operation, _quantity=3)
         existing_operation.multiple_operators.set(multiple_operators)
 
@@ -568,7 +562,7 @@ def test_update_operation_with_multiple_operators():
     @staticmethod
     def test_update_operation_archive_multiple_operators():
         approved_user_operator = baker.make_recipe('utils.approved_user_operator')
-        existing_operation = baker.make_recipe('utils.operation')
+        existing_operation = baker.make_recipe('utils.operation', operator=approved_user_operator.operator)
         multiple_operators = baker.make_recipe('utils.multiple_operator', operation=existing_operation, _quantity=3)
         existing_operation.multiple_operators.set(multiple_operators)