Merge pull request #2198 from uktrade/uat

PROD Release

api/applications/management/commands/update_good_name.py

@@ -1,10 +1,9 @@
 import csv
+
 from django.core.management.base import BaseCommand
+
 from api.goods.models import Good
-from api.users.models import BaseUser
-from api.audit_trail import service as audit_trail_service
-from api.audit_trail.enums import AuditType
-from django.db import transaction
+from api.support.helpers import developer_intervention
 
 
 class Command(BaseCommand):
@@ -17,24 +16,24 @@ def handle(self, *args, **kwargs):
         csv_file = kwargs["csv_file"]
 
         reader = csv.DictReader(csv_file)
-        with transaction.atomic():
+        with developer_intervention(dry_run=False) as audit_log:
             for row in reader:
                 good_id = row["good_id"]
                 name = row["name"]
                 new_name = row["new_name"]
                 additional_text = row["additional_text"]
 
-                self.update_good_name(good_id, name, new_name, additional_text)
+                self.update_good_name(good_id, name, new_name, additional_text, audit_log)
 
-    def update_good_name(self, good_id, name, new_name, additional_text):
+    def update_good_name(self, good_id, name, new_name, additional_text, audit_log):
         good = Good.objects.get(id=good_id, name=name)
-        system_user = BaseUser.objects.get(id="00000000-0000-0000-0000-000000000001")
 
-        audit_trail_service.create(
-            actor=system_user,
-            verb=AuditType.DEVELOPER_INTERVENTION,
-            target=good,
-            payload={"name": {"new": new_name, "old": name}, "additional_text": additional_text},
+        audit_log(
+            good,
+            additional_text,
+            {
+                "name": {"new": new_name, "old": name},
+            },
         )
 
         good.name = new_name

api/applications/management/commands/update_party_address.py

@@ -1,10 +1,9 @@
 import csv
+
 from django.core.management.base import BaseCommand
+
 from api.parties.models import Party
-from api.users.models import BaseUser
-from api.audit_trail import service as audit_trail_service
-from api.audit_trail.enums import AuditType
-from django.db import transaction
+from api.support.helpers import developer_intervention
 
 
 class Command(BaseCommand):
@@ -17,24 +16,24 @@ def handle(self, *args, **kwargs):
         csv_file = kwargs["csv_file"]
 
         reader = csv.DictReader(csv_file)
-        with transaction.atomic():
+        with developer_intervention(dry_run=False) as audit_log:
             for row in reader:
                 party_id = row["party_id"]
                 address = row["address"].replace("\\r\\n", "\r\n")
                 new_address = row["new_address"]
                 additional_text = row["additional_text"]
 
-                self.update_field_on_party(party_id, address, new_address, additional_text)
+                self.update_field_on_party(party_id, address, new_address, additional_text, audit_log)
 
-    def update_field_on_party(self, party_id, address, new_address, additional_text):
+    def update_field_on_party(self, party_id, address, new_address, additional_text, audit_log):
         party = Party.objects.get(id=party_id, address=address)
-        system_user = BaseUser.objects.get(id="00000000-0000-0000-0000-000000000001")
 
-        audit_trail_service.create(
-            actor=system_user,
-            verb=AuditType.DEVELOPER_INTERVENTION,
-            target=party,
-            payload={"address": {"new": new_address, "old": address}, "additional_text": additional_text},
+        audit_log(
+            party,
+            additional_text,
+            {
+                "address": {"new": new_address, "old": address},
+            },
         )
 
         party.address = new_address

api/applications/management/commands/update_party_name.py

@@ -0,0 +1,42 @@
+import csv
+from django.core.management.base import BaseCommand
+from api.parties.models import Party
+from api.users.models import BaseUser
+from api.audit_trail import service as audit_trail_service
+from api.audit_trail.enums import AuditType
+from django.db import transaction
+
+
+class Command(BaseCommand):
+    help = "Update Name for multiple parties from a CSV file"
+
+    def add_arguments(self, parser):
+        parser.add_argument("csv_file", type=open, help="The path to the CSV file containing updates")
+
+    def handle(self, *args, **kwargs):
+        csv_file = kwargs["csv_file"]
+
+        reader = csv.DictReader(csv_file)
+        with transaction.atomic():
+            for row in reader:
+                party_id = row["party_id"]
+                name = row["name"].replace("\\r\\n", "\r\n")
+                new_name = row["new_name"]
+                additional_text = row["additional_text"]
+
+                self.update_field_on_party(party_id, name, new_name, additional_text)
+
+    def update_field_on_party(self, party_id, name, new_name, additional_text):
+        party = Party.objects.get(id=party_id, name=name)
+        system_user = BaseUser.objects.get(id="00000000-0000-0000-0000-000000000001")
+
+        audit_trail_service.create(
+            actor=system_user,
+            verb=AuditType.DEVELOPER_INTERVENTION,
+            target=party,
+            payload={"name": {"new": new_name, "old": name}, "additional_text": additional_text},
+        )
+
+        party.name = new_name
+        party.save()
+        self.stdout.write(f"Updated name for Party {party_id} from {name} to {new_name}.")

api/applications/tests/test_update_party_name.py

@@ -0,0 +1,61 @@
+from api.audit_trail.models import Audit
+from api.audit_trail.enums import AuditType
+from django.core.management import call_command
+from tempfile import NamedTemporaryFile
+import pytest
+from api.parties.models import Party
+from test_helpers.clients import DataTestClient
+
+
+class UpdatePartyFromCSVTests(DataTestClient):
+    def setUp(self):
+        super().setUp()
+        self.standard_application = self.create_draft_standard_application(self.organisation)
+
+    def test_update_field_on_party_from_csv(self):
+
+        new_name = "Bangarang 3000"
+        old_name = self.standard_application.end_user.party.name
+        party_id = self.standard_application.end_user.party.id
+
+        with NamedTemporaryFile(suffix=".csv", delete=True) as tmp_file:
+            rows = [
+                "party_id,name,new_name,additional_text",
+                f"""{party_id},"{old_name}",{new_name},added by John Smith as per LTD-XXX""",
+            ]
+            tmp_file.write("\n".join(rows).encode("utf-8"))
+            tmp_file.flush()
+
+            call_command("update_party_name", tmp_file.name)
+            self.standard_application.refresh_from_db()
+            self.assertEqual(self.standard_application.end_user.party.name, new_name)
+
+            audit = Audit.objects.get()
+
+            self.assertEqual(audit.actor, self.system_user)
+            self.assertEqual(audit.target.id, party_id)
+            self.assertEqual(audit.verb, AuditType.DEVELOPER_INTERVENTION)
+            self.assertEqual(
+                audit.payload,
+                {
+                    "name": {"new": new_name, "old": old_name},
+                    "additional_text": "added by John Smith as per LTD-XXX",
+                },
+            )
+
+    def test_update_field_on_party_from_csv_invalid(self):
+
+        new_name = "Bangarang 3000"
+        old_name = "This is not an name"
+        party_id = self.standard_application.end_user.party.id
+
+        with NamedTemporaryFile(suffix=".csv", delete=True) as tmp_file:
+            rows = [
+                "party_id,name,new_name,additional_text",
+                f"""{party_id},"{old_name}",{new_name},added by John Smith as per LTD-XXX""",
+            ]
+            tmp_file.write("\n".join(rows).encode("utf-8"))
+            tmp_file.flush()
+
+            with pytest.raises(Party.DoesNotExist):
+                call_command("update_party_name", tmp_file.name)

api/audit_trail/enums.py

@@ -145,6 +145,7 @@ class AuditType(LiteEnum):
     EXPORTER_SUBMITTED_AMENDMENT = autostr()
     AMENDMENT_CREATED = autostr()
     DEVELOPER_INTERVENTION = autostr()
+    ADD_EXPORTER_USER_TO_ORGANISATION = autostr()
 
     def human_readable(self):
         """