1432: transfers grid

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

@@ -181,6 +181,7 @@ class TestEndpointPermissions(TestCase):
             {"method": "patch", "endpoint_name": "operation_bcghg_id", "kwargs": {"operation_id": mock_uuid}},
             {"method": "patch", "endpoint_name": "facility_bcghg_id", "kwargs": {'facility_id': mock_uuid}},
             {"method": "patch", "endpoint_name": "operation_bcghg_id", "kwargs": {'operation_id': mock_uuid}},
+            {"method": "get", "endpoint_name": "list_transfer_events"},
         ],
         "approved_authorized_roles": [
             {"method": "get", "endpoint_name": "list_operations"},

bc_obps/registration/api/v1/__init__.py

@@ -13,6 +13,7 @@
     users,
     facilities,
     contacts,
+    transfer_events,
 )
 from ._operations import operation_id
 from ._operations._operation_id import update_status, facilities, operation_representatives

bc_obps/registration/api/v1/transfer_events.py

@@ -0,0 +1,36 @@
+from typing import List, Literal, Optional
+from registration.models.event.transfer_event import TransferEvent
+from registration.schema.v1.transfer_event import TransferEventFilterSchema, TransferEventListOut
+from service.transfer_event_service import TransferEventService
+from common.permissions import authorize
+from django.http import HttpRequest
+from registration.utils import CustomPagination
+from registration.constants import TRANSFER_EVENT_TAGS
+from ninja.pagination import paginate
+from registration.decorators import handle_http_errors
+from ..router import router
+from service.error_service.custom_codes_4xx import custom_codes_4xx
+from ninja import Query
+from registration.schema.generic import Message
+from django.db.models import QuerySet
+
+
+@router.get(
+    "/transfer-events",
+    response={200: List[TransferEventListOut], custom_codes_4xx: Message},
+    tags=TRANSFER_EVENT_TAGS,
+    description="""Retrieves a paginated list of transfer events based on the provided filters.
+    The endpoint allows authorized users to view and sort transfer events filtered by various criteria such as operation, facility, and status.""",
+    auth=authorize("authorized_irc_user"),
+)
+@handle_http_errors()
+@paginate(CustomPagination)
+def list_transfer_events(
+    request: HttpRequest,
+    filters: TransferEventFilterSchema = Query(...),
+    sort_field: Optional[str] = "status",
+    sort_order: Optional[Literal["desc", "asc"]] = "desc",
+    paginate_result: bool = Query(True, description="Whether to paginate the results"),
+) -> QuerySet[TransferEvent]:
+    # NOTE: PageNumberPagination raises an error if we pass the response as a tuple (like 200, ...)
+    return TransferEventService.list_transfer_events(sort_field, sort_order, filters)

bc_obps/registration/constants.py

@@ -29,4 +29,5 @@
 USER_OPERATOR_TAGS_V2 = ["User Operator V2"]
 MISC_TAGS = ["Misc V1"]
 CONTACT_TAGS = ["Contact V1"]
+TRANSFER_EVENT_TAGS = ["Transfer Event V1"]
 V2 = ["V2"]

bc_obps/registration/fixtures/mock/transfer_event.json

@@ -11,8 +11,7 @@
       "description": "Sold my operation to the other operator so that I could move to Hawaii. Don't tell my employees they don't know.",
       "status": "Transferred",
       "other_operator": "4242ea9d-b917-4129-93c2-db00b7451051",
-      "other_operator_contact": 13,
-      "future_designated_operator": "Other Operator"
+      "other_operator_contact": 13
     }
   },
   {
@@ -24,8 +23,7 @@
       "effective_date": "2024-08-21T09:00:00Z",
       "description": "I was the designated operator but I'm being relieved of my duties because my other operators don't trust me anymore. Transfer this operation over to another one of the operators in our multiple-operators relationship. See if you can do better Mia!",
       "other_operator": "4242ea9d-b917-4129-93c2-db00b7451051",
-      "other_operator_contact": 15,
-      "future_designated_operator": "Other Operator"
+      "other_operator_contact": 15
     }
   },
   {
@@ -43,8 +41,7 @@
       "effective_date": "2024-12-25T09:00:00Z",
       "description": "I don't know what I'm doing but this seems like the right thing to do.",
       "other_operator": "4242ea9d-b917-4129-93c2-db00b7451051",
-      "other_operator_contact": 15,
-      "future_designated_operator": "Not sure"
+      "other_operator_contact": 15
     }
   }
 ]

bc_obps/registration/migrations/0056_remove_historicaltransferevent_future_designated_operator_and_more.py

@@ -0,0 +1,47 @@
+# Generated by Django 5.0.9 on 2024-11-15 23:20
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('registration', '0055_V1_14_0'),
+    ]
+
+    operations = [
+        migrations.RemoveField(
+            model_name='historicaltransferevent',
+            name='future_designated_operator',
+        ),
+        migrations.RemoveField(
+            model_name='transferevent',
+            name='future_designated_operator',
+        ),
+        migrations.AlterField(
+            model_name='historicaltransferevent',
+            name='status',
+            field=models.CharField(
+                choices=[
+                    ('Complete', 'Complete'),
+                    ('To be transferred', 'To Be Transferred'),
+                    ('Transferred', 'Transferred'),
+                ],
+                default='To be transferred',
+                max_length=100,
+            ),
+        ),
+        migrations.AlterField(
+            model_name='transferevent',
+            name='status',
+            field=models.CharField(
+                choices=[
+                    ('Complete', 'Complete'),
+                    ('To be transferred', 'To Be Transferred'),
+                    ('Transferred', 'Transferred'),
+                ],
+                default='To be transferred',
+                max_length=100,
+            ),
+        ),
+    ]

bc_obps/registration/models/event/transfer_event.py

@@ -5,22 +5,13 @@
 
 
 class TransferEvent(EventBaseModel):
+    # ok to display the db ones in the grid
     class Statuses(models.TextChoices):
         COMPLETE = "Complete"
-        PENDING = "Pending"
+        TO_BE_TRANSFERRED = "To be transferred"
         TRANSFERRED = "Transferred"
 
-    class FutureDesignatedOperatorChoices(models.TextChoices):
-        MY_OPERATOR = "My Operator"
-        OTHER_OPERATOR = "Other Operator"
-        NOT_SURE = "Not Sure"
-
     description = models.TextField(db_comment="Description of the transfer or change in designated operator.")
-    future_designated_operator = models.CharField(
-        max_length=1000,
-        choices=FutureDesignatedOperatorChoices.choices,
-        db_comment="The designated operator of the entit(y)/(ies) associated with the transfer, who will be responsible for matters related to GGERR.",
-    )
     other_operator = models.ForeignKey(
         Operator,
         on_delete=models.PROTECT,
@@ -36,7 +27,7 @@ class FutureDesignatedOperatorChoices(models.TextChoices):
     status = models.CharField(
         max_length=100,
         choices=Statuses.choices,
-        default=Statuses.PENDING,
+        default=Statuses.TO_BE_TRANSFERRED,
     )
     history = HistoricalRecords(
         table_name='erc_history"."transfer_event_history',

bc_obps/registration/schema/v1/transfer_event.py

@@ -0,0 +1,52 @@
+from typing import Optional
+from uuid import UUID
+
+from registration.models.event.transfer_event import TransferEvent
+from ninja import ModelSchema, Field, FilterSchema
+from django.db.models import Q
+import re
+from typing import Dict, Any
+
+
+class TransferEventListOut(ModelSchema):
+    operation__id: Optional[UUID] = None
+    operation__name: Optional[str] = Field(None, alias="operation__name")
+    facilities__name: Optional[str] = Field(None, alias="facilities__name")
+    facility__id: Optional[UUID] = Field(None, alias="facilities__id")
+    id: UUID
+
+    @staticmethod
+    def resolve_id(obj: Dict[str, Any]) -> UUID:
+        operation_id = obj.get('operation__id', None)
+        facility_id = obj.get('facilities__id', None)
+
+        record_id = operation_id if operation_id else facility_id
+        if not isinstance(record_id, UUID):
+            raise Exception('Missing valid UUID')
+        return record_id
+
+    class Meta:
+        model = TransferEvent
+        fields = ['effective_date', 'status', 'created_at']
+
+
+class TransferEventFilterSchema(FilterSchema):
+    # NOTE: we could simply use the `q` parameter to filter by related fields but,
+    # due to this issue: https://github.com/vitalik/django-ninja/issues/1037 mypy is unhappy so I'm using the `json_schema_extra` parameter
+    # If we want to achieve more by using the `q` parameter, we should use it and ignore the mypy error
+    effective_date: Optional[str] = Field(None, json_schema_extra={'q': 'effective_date__icontains'})
+    operation__name: Optional[str] = None
+    facilities__name: Optional[str] = Field(None, json_schema_extra={'q': 'facilities__name__icontains'})
+    status: Optional[str] = Field(None, json_schema_extra={'q': 'status__icontains'})
+
+    @staticmethod
+    def filtering_including_not_applicable(field: str, value: str) -> Q:
+        if value and re.search(value, 'n/a', re.IGNORECASE):
+            return Q(**{f"{field}__icontains": value}) | Q(**{f"{field}__isnull": True})
+        return Q(**{f"{field}__icontains": value}) if value else Q()
+
+    def filter_operation__name(self, value: str) -> Q:
+        return self.filtering_including_not_applicable('operation__name', value)
+
+    def filter_facilities__name(self, value: str) -> Q:
+        return self.filtering_including_not_applicable('facilities__name', value)

bc_obps/registration/tests/endpoints/v1/test_transfer_events.py

@@ -0,0 +1,138 @@
+from datetime import datetime, timedelta
+from django.utils import timezone
+from bc_obps.settings import NINJA_PAGINATION_PER_PAGE
+from model_bakery import baker
+from registration.tests.utils.helpers import CommonTestSetup, TestUtils
+
+from registration.utils import custom_reverse_lazy
+
+
+class TestTransferEventEndpoint(CommonTestSetup):
+    url = custom_reverse_lazy('list_transfer_events')
+    # GET
+    def test_list_transfer_events_unpaginated(self):
+        # transfer of an operation
+        baker.make_recipe('utils.transfer_event', operation=baker.make_recipe('utils.operation'))
+        # transfer of 50 facilities
+        baker.make_recipe('utils.transfer_event', facilities=baker.make_recipe('utils.facility', _quantity=50))
+        for role in ['cas_admin', 'cas_analyst']:
+            response = TestUtils.mock_get_with_auth_role(self, role, self.url + "?paginate_result=False")
+            assert response.status_code == 200
+            assert response.json().keys() == {'count', 'items'}
+            assert response.json()['count'] == 51
+
+            items = response.json().get('items', [])
+            for item in items:
+                assert set(item.keys()) == {
+                    'operation__name',
+                    'operation__id',
+                    'effective_date',
+                    'status',
+                    'id',
+                    'facilities__name',
+                    'facility__id',
+                    'created_at',
+                }
+
+    def test_list_transfer_events_paginated(self):
+        # transfer of an operation
+        baker.make_recipe('utils.transfer_event', operation=baker.make_recipe('utils.operation'))
+        # transfer of 50 facilities
+        baker.make_recipe('utils.transfer_event', facilities=baker.make_recipe('utils.facility', _quantity=50))
+        # Get the default page 1 response
+        response = TestUtils.mock_get_with_auth_role(self, "cas_admin", custom_reverse_lazy("list_transfer_events"))
+        assert response.status_code == 200
+
+        response_items_1 = response.json().get('items')
+        response_count_1 = response.json().get('count')
+        # save the id of the first paginated response item
+        page_1_response_id = response_items_1[0].get('id')
+        assert len(response_items_1) == NINJA_PAGINATION_PER_PAGE
+        assert response_count_1 == 51  # total count of transfers
+        # Get the page 2 response
+        response = TestUtils.mock_get_with_auth_role(
+            self,
+            "cas_admin",
+            self.url + "?page=2&sort_field=created_at&sort_order=desc",
+        )
+        assert response.status_code == 200
+        response_items_2 = response.json().get('items')
+        response_count_2 = response.json().get('count')
+        # save the id of the first paginated response item
+        page_2_response_id = response_items_2[0].get('id')
+        assert len(response_items_2) == NINJA_PAGINATION_PER_PAGE
+        # assert that the first item in the page 1 response is not the same as the first item in the page 2 response
+        assert page_1_response_id != page_2_response_id
+        assert response_count_2 == response_count_1  # total count of transfer_events should be the same
+        # Get the page 2 response but with a different sort order
+        response = TestUtils.mock_get_with_auth_role(
+            self,
+            "cas_admin",
+            self.url + "?page=2&sort_field=created_at&sort_order=asc",
+        )
+        assert response.status_code == 200
+        response_items_2_reverse = response.json().get('items')
+        # save the id of the first paginated response item
+        page_2_response_id_reverse = response_items_2_reverse[0].get('id')
+        assert len(response_items_2_reverse) == NINJA_PAGINATION_PER_PAGE
+        # assert that the first item in the page 2 response is not the same as the first item in the page 2 response with reversed order
+        assert page_2_response_id != page_2_response_id_reverse
+
+    def test_transfer_events_endpoint_list_transfer_events_with_sorting(self):
+        today = timezone.make_aware(datetime.now())
+        yesterday = today - timedelta(days=1)
+        # transfer of an operation
+        baker.make_recipe('utils.transfer_event', operation=baker.make_recipe('utils.operation'), effective_date=today)
+        # transfer of 50 facilities
+        baker.make_recipe(
+            'utils.transfer_event',
+            effective_date=yesterday,
+            facilities=baker.make_recipe('utils.facility', _quantity=50),
+        )
+
+        response_ascending = TestUtils.mock_get_with_auth_role(
+            self, "cas_admin", self.url + "?page=1&sort_field=effective_date&sort_order=asc"
+        )
+        # save the id of the first paginated response item
+        first_item_ascending = response_ascending.json()['items'][0]
+
+        # # Sort created at descending
+        response_descending = TestUtils.mock_get_with_auth_role(
+            self, "cas_admin", self.url + "?page=1&sort_field=effective_date&sort_order=desc"
+        )
+        first_item_descending = response_descending.json()['items'][0]
+        assert first_item_descending['effective_date'] > first_item_ascending['effective_date']
+
+    def test_transfer_events_endpoint_list_transfer_events_with_filter(self):
+        # transfer of an operation
+        baker.make_recipe('utils.transfer_event', operation=baker.make_recipe('utils.operation', name='Test Operation'))
+        # transfer of 50 facilities
+        baker.make_recipe('utils.transfer_event', facilities=baker.make_recipe('utils.facility', _quantity=50))
+
+        # Get the default page 1 response
+        response = TestUtils.mock_get_with_auth_role(
+            self, "cas_admin", self.url + "?facilities__name=010"
+        )  # filtering facilities__name
+        assert response.status_code == 200
+        response_items_1 = response.json().get('items')
+        for item in response_items_1:
+            assert item.get('facilities__name') == "Facility 010"
+
+        # Test with a status filter that doesn't exist
+        response = TestUtils.mock_get_with_auth_role(self, "cas_admin", self.url + "?status=unreal")
+        assert response.status_code == 200
+        assert response.json().get('count') == 0
+
+        # Test with two filters
+        facilities__name_to_filter, status_to_filter = response_items_1[0].get('facilities__name'), response_items_1[
+            0
+        ].get('status')
+        response = TestUtils.mock_get_with_auth_role(
+            self, "cas_admin", self.url + f"?facilities__name={facilities__name_to_filter}&status={status_to_filter}"
+        )
+        assert response.status_code == 200
+        response_items_2 = response.json().get('items')
+        assert len(response_items_2) == 1
+        assert response.json().get('count') == 1
+        assert response_items_2[0].get('facilities__name') == facilities__name_to_filter
+        assert response_items_2[0].get('status') == status_to_filter

bc_obps/registration/tests/models/event/test_transfer.py

@@ -42,22 +42,19 @@ def setUpTestData(cls):
             ("facilities", "facilities", None, None),
             ("other_operator", "other operator", None, None),
             ("other_operator_contact", "other operator contact", None, None),
-            ("future_designated_operator", "future designated operator", None, None),
         ]
         super().setUpTestData()
 
     def test_event_with_operation_only(self):
         self.create_event_with_operation_only(
             description="Why the transfer is happening",
-            future_designated_operator="My Operator",
             other_operator=operator_baker(),
             other_operator_contact=contact_baker(),
         )
 
     def test_event_with_facilities_only(self):
         self.create_event_with_facilities_only(
             description="Why the transfer is happening returns",
-            future_designated_operator="My Operator",
             other_operator=operator_baker(),
             other_operator_contact=contact_baker(),
         )

bc_obps/registration/tests/utils/baker_recipes.py

@@ -138,7 +138,6 @@
 
 transfer_event = Recipe(
     TransferEvent,
-    future_designated_operator=TransferEvent.FutureDesignatedOperatorChoices.MY_OPERATOR,
     other_operator=foreign_key(other_operator_for_transfer_event),
     other_operator_contact=foreign_key(contact_for_transfer_event),
 )