Merge pull request #2151 from uktrade/LTD-5342-special-character-manu…

…al-fix

LTD-5342 special character management command

api/applications/management/commands/update_good_name.py

@@ -0,0 +1,42 @@
+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
+
+
+class Command(BaseCommand):
+    help = "Update name for multiple goods 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:
+                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)
+
+    def update_good_name(self, good_id, name, new_name, additional_text):
+        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},
+        )
+
+        good.name = new_name
+        good.save()
+        self.stdout.write(f"Updated name for Good {good_id} from {name} to {new_name}.")

api/applications/management/commands/update_party_address.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 address 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"]
+                address = row["address"]
+                new_address = row["new_address"]
+                additional_text = row["additional_text"]
+
+                self.update_field_on_party(party_id, address, new_address, additional_text)
+
+    def update_field_on_party(self, party_id, address, new_address, additional_text):
+        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},
+        )
+
+        party.address = new_address
+        party.save()
+        self.stdout.write(f"Updated address for Party {party_id} from {address} to {new_address}.")

api/applications/tests/test_update_good_name.py

@@ -0,0 +1,65 @@
+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.goods.models import Good
+from test_helpers.clients import DataTestClient
+
+
+class UpdateGoodFromCSVTests(DataTestClient):
+    def setUp(self):
+        super().setUp()
+        self.standard_application = self.create_draft_standard_application(self.organisation)
+
+    def test_update_good_name_from_csv(self):
+
+        new_name = "Bangarang 3000"
+        goodonapplication = self.standard_application.goods.get()
+        good = goodonapplication.good
+        old_name = good.name
+        good_id = good.id
+
+        with NamedTemporaryFile(suffix=".csv", delete=True) as tmp_file:
+            rows = [
+                "good_id,name,new_name,additional_text",
+                f"""{good_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_good_name", tmp_file.name)
+            good.refresh_from_db()
+            self.assertEqual(good.name, new_name)
+
+            audit = Audit.objects.get()
+
+            self.assertEqual(audit.actor, self.system_user)
+            self.assertEqual(audit.target.id, good_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_good_name_from_csv_invalid(self):
+
+        new_name = "Bangarang 3000"
+        goodonapplication = self.standard_application.goods.get()
+        good = goodonapplication.good
+        old_name = "Definitely not this"
+        good_id = good.id
+
+        with NamedTemporaryFile(suffix=".csv", delete=True) as tmp_file:
+            rows = [
+                "good_id,name,new_name,additional_text",
+                f"""{good_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(Good.DoesNotExist):
+                call_command("update_good_name", tmp_file.name)

api/applications/tests/test_update_parties_address.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_address = "56 Heathwood Road Broadstairs Kent"
+        old_address = self.standard_application.end_user.party.address
+        party_id = self.standard_application.end_user.party.id
+
+        with NamedTemporaryFile(suffix=".csv", delete=True) as tmp_file:
+            rows = [
+                "party_id,address,new_address,additional_text",
+                f"""{party_id},"{old_address}",{new_address},added by John Smith as per LTD-XXX""",
+            ]
+            tmp_file.write("\n".join(rows).encode("utf-8"))
+            tmp_file.flush()
+
+            call_command("update_party_address", tmp_file.name)
+            self.standard_application.refresh_from_db()
+            self.assertEqual(self.standard_application.end_user.party.address, new_address)
+
+            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,
+                {
+                    "address": {"new": new_address, "old": old_address},
+                    "additional_text": "added by John Smith as per LTD-XXX",
+                },
+            )
+
+    def test_update_field_on_party_from_csv_invalid(self):
+
+        new_address = "56 Heathwood Road Broadstairs Kent"
+        old_address = "This is not an address"
+        party_id = self.standard_application.end_user.party.id
+
+        with NamedTemporaryFile(suffix=".csv", delete=True) as tmp_file:
+            rows = [
+                "party_id,address,new_address,additional_text",
+                f"""{party_id},"{old_address}",{new_address},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_address", tmp_file.name)

api/audit_trail/enums.py

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