Merge pull request #1957 from uktrade/uat

PROD Release

api/applications/tests/test_matching_denials.py

@@ -1,40 +1,25 @@
-import csv
-import io
 import pytest
+import os
 
+from django.conf import settings
 from django.urls import reverse
-from faker import Faker
 from rest_framework import status
 
 from api.applications.tests.factories import DenialMatchFactory
-from api.external_data import models, serializers
+from api.external_data import models
 from test_helpers.clients import DataTestClient
 
 
 class ApplicationDenialMatchesOnApplicationTests(DataTestClient):
     def setUp(self):
         super().setUp()
-        self.faker = Faker()
         self.application = self.create_standard_application_case(self.organisation)
-        denials = [
-            {name: self.faker.word() for name in serializers.DenialFromCSVFileSerializer.required_headers}
-            for _ in range(5)
-        ]
-
-        content = io.StringIO()
-        writer = csv.DictWriter(
-            content,
-            fieldnames=[*serializers.DenialFromCSVFileSerializer.required_headers, "field_n"],
-            delimiter=",",
-            quoting=csv.QUOTE_MINIMAL,
-        )
-        writer.writeheader()
-        writer.writerows(denials)
-        response = self.client.post(
-            reverse("external_data:denial-list"), {"csv_file": content.getvalue()}, **self.gov_headers
-        )
+        file_path = os.path.join(settings.BASE_DIR, "external_data/tests/denial_valid.csv")
+        with open(file_path, "rb") as f:
+            content = f.read()
+        response = self.client.post(reverse("external_data:denial-list"), {"csv_file": content}, **self.gov_headers)
         self.assertEqual(response.status_code, status.HTTP_201_CREATED)
-        self.assertEqual(models.DenialEntity.objects.count(), 5)
+        self.assertEqual(models.DenialEntity.objects.count(), 4)
 
     @pytest.mark.xfail(reason="This test is flaky and should be rewritten")
     # Occasionally causes this error:
@@ -60,7 +45,7 @@ def test_adding_denials_to_application(self):
     def test_revoke_denial_without_comment_failure(self):
         response = self.client.get(reverse("external_data:denial-list"), **self.gov_headers)
         self.assertEqual(response.status_code, status.HTTP_200_OK)
-        self.assertEqual(response.json()["count"], 5)
+        self.assertEqual(response.json()["count"], 4)
 
         denials = response.json()["results"]
 
@@ -77,7 +62,7 @@ def test_revoke_denial_without_comment_failure(self):
     def test_revoke_denial_success(self):
         response = self.client.get(reverse("external_data:denial-list"), **self.gov_headers)
         self.assertEqual(response.status_code, status.HTTP_200_OK)
-        self.assertEqual(response.json()["count"], 5)
+        self.assertEqual(response.json()["count"], 4)
 
         denials = response.json()["results"]
 

api/external_data/migrations/0023_set_denial_entity_type.py

@@ -0,0 +1,48 @@
+# Generated by Django 4.2.11 on 2024-04-18 12:00
+
+from django.db import migrations
+from api.external_data.enums import DenialEntityType
+
+
+def get_denial_entity_type(data):
+
+    if isinstance(data, dict):
+        entity_type = ""
+        normalised_entity_type_dict = {keys.lower(): values.lower() for keys, values in data.items()}
+
+        is_end_user_flag = normalised_entity_type_dict.get("end_user_flag", "false") == "true"
+        is_consignee_flag = normalised_entity_type_dict.get("consignee_flag", "false") == "true"
+        is_other_role = len(normalised_entity_type_dict.get("other_role", "")) > 0
+
+        if is_end_user_flag and is_consignee_flag:
+            entity_type = DenialEntityType.END_USER
+        elif not is_end_user_flag and is_consignee_flag:
+            entity_type = DenialEntityType.CONSIGNEE
+        elif is_end_user_flag and not is_consignee_flag:
+            entity_type = DenialEntityType.END_USER
+        elif not is_end_user_flag and not is_consignee_flag and is_other_role:
+            entity_type = DenialEntityType.THIRD_PARTY
+
+        return entity_type
+
+
+def set_denial_entity_type(apps, schema_editor):
+
+    DenialEntity = apps.get_model("external_data", "DenialEntity")
+
+    for denial_entity in DenialEntity.objects.filter(entity_type__isnull=True):
+
+        denial_entity_type = get_denial_entity_type(denial_entity.data)
+
+        if denial_entity_type in ["end_user", "consignee", "third_party"]:
+            denial_entity.entity_type = denial_entity_type
+            denial_entity.save()
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("external_data", "0022_denialentity_entity_type"),
+    ]
+
+    operations = [migrations.RunPython(set_denial_entity_type, migrations.RunPython.noop)]

api/external_data/migrations/0024_denials_data_migration.py

@@ -0,0 +1,64 @@
+# Generated by Django 4.2.10 on 2024-04-02 14:59
+
+import logging
+from django.db import migrations
+from django.forms.models import model_to_dict
+from django.db.models import Q
+from django.db import IntegrityError
+from django.db import transaction
+
+log = logging.getLogger(__name__)
+
+required_fields = [
+        "reference",
+        "regime_reg_ref",
+        "notifying_government",
+        "item_list_codes",
+        "item_description",
+        "end_use",
+        "is_revoked",
+        "is_revoked_comment",
+        "reason_for_refusal",
+]
+
+def denials_data_migration(apps, schema_editor):
+    DenialEntity = apps.get_model("external_data", "DenialEntity")
+    Denial = apps.get_model("external_data", "Denial")
+
+    # There are a handfull (10) of regime_reg_refs that are null which was in the initial load
+    # Assumption here is that they can be deleted since it's erroneous data as we now know 
+    # regime_reg_ref is considered a unique DN record. 
+
+    total_null_regime_reg_ref =  DenialEntity.objects.filter(Q(regime_reg_ref__isnull=True) | Q(regime_reg_ref=''))
+    log.info(
+            "Delete null regime_reg_ref total -> %s",
+            total_null_regime_reg_ref.count(),
+    )
+    total_null_regime_reg_ref.delete()
+    duplicate_denial_errors = []
+    with transaction.atomic():
+        sid = transaction.savepoint()
+        for denial_entity in DenialEntity.objects.all():
+            try:
+                denial_entity_dict = {key:value for (key,value) in model_to_dict(denial_entity).items() if key in required_fields}
+                denial , _ = Denial.objects.get_or_create(**denial_entity_dict)
+                denial_entity.denial = denial
+                denial_entity.save()
+            except IntegrityError as e:
+                    duplicate_denial_errors.append(denial_entity.regime_reg_ref)
+
+        if duplicate_denial_errors:
+            log.info(
+                "There are the following duplicate denials in the database rolling back this migration: -> %s",
+                    duplicate_denial_errors,
+            )
+            transaction.savepoint_rollback(sid)
+        else:
+            transaction.savepoint_commit(sid)   
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("external_data", "0023_set_denial_entity_type"),
+    ]
+
+    operations = [migrations.RunPython(denials_data_migration, migrations.RunPython.noop, atomic=False)]

api/external_data/migrations/tests/test_0023_set_denial_entity_type.py

@@ -0,0 +1,90 @@
+import pytest
+
+from django_test_migrations.contrib.unittest_case import MigratorTestCase
+from api.external_data.enums import DenialEntityType
+
+test_data = [
+    {
+        "reference": "DN2000/0000",
+        "regime_reg_ref": "AB-CD-EF-000",
+        "name": "Organisation Name",
+        "address": "1000 Street Name, City Name",
+        "notifying_government": "Country Name",
+        "country": "Country Name",
+        "item_list_codes": "0A00100",
+        "item_description": "Medium Size Widget",
+        "consignee_name": "Example Name",
+        "end_use": "Used in industry",
+        "reason_for_refusal": "Risk of outcome",
+        "spire_entity_id": 123,
+        "data": {"END_USER_FLAG": "true", "CONSIGNEE_FLAG": "false", "OTHER_ROLE": ""},
+        "entity_type": None,
+    },
+    {
+        "reference": "DN2000/0010",
+        "regime_reg_ref": "AB-CD-EF-300",
+        "name": "Organisation Name 3",
+        "address": "2001 Street Name, City Name 3",
+        "notifying_government": "Country Name 3",
+        "country": "Country Name 3",
+        "item_list_codes": "0A00201",
+        "item_description": "Unspecified Size Widget",
+        "consignee_name": "Example Name 3",
+        "end_use": "Used in other industry",
+        "reason_for_refusal": "Risk of outcome 3",
+        "spire_entity_id": 125,
+        "data": {"END_USER_FLAG": "false", "CONSIGNEE_FLAG": "true", "OTHER_ROLE": ""},
+        "entity_type": None,
+    },
+    {
+        "reference": "DN2000/0000",
+        "regime_reg_ref": "AB-CD-EF-000",
+        "name": "Organisation Name",
+        "address": "1000 Street Name, City Name",
+        "notifying_government": "Country Name",
+        "country": "Country Name",
+        "item_list_codes": "0A00100",
+        "item_description": "Medium Size Widget",
+        "consignee_name": "Example Name",
+        "end_use": "Used in industry",
+        "reason_for_refusal": "Risk of outcome",
+        "spire_entity_id": 123,
+        "data": {"END_USER_FLAG": "true", "CONSIGNEE_FLAG": "true", "OTHER_ROLE": ""},
+        "entity_type": None,
+    },
+    {
+        "reference": "DN3000/0000",
+        "regime_reg_ref": "AB-CD-EF-100",
+        "name": "Organisation Name XYZ",
+        "address": "2000 Street Name, City Name 2",
+        "notifying_government": "Country Name 2",
+        "country": "Country Name 2",
+        "item_list_codes": "0A00200",
+        "item_description": "Large Size Widget",
+        "consignee_name": "Example Name 2",
+        "end_use": "Used in other industry",
+        "reason_for_refusal": "Risk of outcome 2",
+        "spire_entity_id": 124,
+        "data": {"END_USER_FLAG": "false", "CONSIGNEE_FLAG": "false", "OTHER_ROLE": "other"},
+        "entity_type": None,
+    },
+]
+
+
+@pytest.mark.django_db()
+class TestDenialEntityTypeSet(MigratorTestCase):
+    migrate_from = ("external_data", "0022_denialentity_entity_type")
+    migrate_to = ("external_data", "0023_set_denial_entity_type")
+
+    def prepare(self):
+        DenialEntity = self.old_state.apps.get_model("external_data", "DenialEntity")
+        for row in test_data:
+            DenialEntity.objects.create(**row)
+
+    def test_0023_set_denial_entity_type(self):
+        DenialEntity = self.new_state.apps.get_model("external_data", "DenialEntity")
+
+        self.assertEqual(DenialEntity.objects.all().count(),4)
+        self.assertEqual(DenialEntity.objects.filter(entity_type=DenialEntityType.END_USER).count(),2)
+        self.assertEqual(DenialEntity.objects.filter(entity_type=DenialEntityType.CONSIGNEE).count(),1)
+        self.assertEqual(DenialEntity.objects.filter(entity_type=DenialEntityType.THIRD_PARTY).count(),1)

api/external_data/migrations/tests/test_0024_denials_data_migration.py

@@ -0,0 +1,59 @@
+import pytest
+
+from django_test_migrations.contrib.unittest_case import MigratorTestCase
+
+
+test_data = [
+{"reference":"DN2010\/0057","regime_reg_ref":"reg.123.123","name":"name 1","address":"address 1","notifying_government":"UK","country":"UK","item_list_codes":"all","item_description":"desc a","end_use":"use 1","reason_for_refusal":"a"},
+{"reference":"DN2010\/0057","regime_reg_ref":"reg.123.1234","name":"name 2","address":"address 2","notifying_government":"UK","country":"UK","item_list_codes":"all","item_description":"desc a","end_use":"use 1","reason_for_refusal":"a"},
+{"reference":"DN2010\/0057","regime_reg_ref":"reg.123.1234","name":"name 3","address":"address 3","notifying_government":"UK","country":"UK","item_list_codes":"all","item_description":"desc a","end_use":"use 1","reason_for_refusal":"a"},
+{"reference":"DN2010\/0057","regime_reg_ref":"reg.123.1234","name":"name 4","address":"address 4","notifying_government":"UK","country":"UK","item_list_codes":"all","item_description":"desc a","end_use":"use 1","reason_for_refusal":"a"},
+{"reference":"DN2010\/0057","name":"bad record","address":"bad record","notifying_government":"UK","country":"bad","item_list_codes":"all","item_description":"bad","end_use":"bad","reason_for_refusal":"bad "}
+]
+
+
+@pytest.mark.django_db()
+class TestDenialDataMigration(MigratorTestCase):
+
+    migrate_from = ("external_data", "0023_set_denial_entity_type")
+    migrate_to = ("external_data", "0024_denials_data_migration")
+
+
+    def prepare(self):
+        DenialEntity = self.old_state.apps.get_model("external_data", "DenialEntity")
+        for row in test_data:
+            DenialEntity.objects.create(**row)  
+
+
+
+
+    def test_0023_denials_data_migration(self):
+        DenialEntity = self.new_state.apps.get_model("external_data", "DenialEntity")
+        Denial = self.new_state.apps.get_model("external_data", "Denial")
+
+        self.assertEqual(DenialEntity.objects.all().count(), 4)
+        self.assertEqual(Denial.objects.all().count(), 2)
+        self.assertEqual(Denial.objects.get(regime_reg_ref='reg.123.1234').denial_entity.count(), 3)
+
+
+@pytest.mark.django_db()
+class TestDenialDataDuplicatesMigration(MigratorTestCase):
+
+    migrate_from = ("external_data", "0023_set_denial_entity_type")
+    migrate_to = ("external_data", "0024_denials_data_migration")
+
+
+    def prepare(self):
+        DenialEntity = self.old_state.apps.get_model("external_data", "DenialEntity")
+        for row in test_data:
+            DenialEntity.objects.create(**row)
+        test_data[0]["end_use"] = "end_use b"
+        DenialEntity.objects.create(**test_data[0])
+
+
+
+    def test_0024_denials_data_migration_duplicates(self):
+        DenialEntity = self.new_state.apps.get_model("external_data", "DenialEntity")
+        Denial = self.new_state.apps.get_model("external_data", "Denial")
+        self.assertEqual(DenialEntity.objects.all().count(), 5)
+        self.assertEqual(Denial.objects.all().count(), 0)