FECFILE-2421: delete load test mirror data

django-backend/fecfiler/devops/management/commands/delete_locust_load_test_data.py

@@ -0,0 +1,24 @@
+from django.core.management.base import CommandError
+from .fecfile_base import FECCommand
+from fecfiler.devops.utils.load_test import LoadTestUtils
+import structlog
+
+logger = structlog.get_logger(__name__)
+
+
+class Command(FECCommand):
+    help = "Delete locust test data from load mirror"
+    command_name = "delete_locust_load_test_data"
+
+    def command(self, *args, **options):
+        load_test_utils = LoadTestUtils()
+        try:
+            load_test_utils.validate_load_mirror_runtime()
+        except ValueError as error:
+            raise CommandError(str(error)) from error
+
+        try:
+            load_test_utils.delete_load_test_committees_and_data()
+        except Exception as e:
+            logger.error(f"Error occurred while deleting locust load test data: {e}")
+            raise

django-backend/fecfiler/devops/tests/test_delete_locust_load_test_data.py

@@ -0,0 +1,35 @@
+from django.core.management import call_command
+from django.core.management.base import CommandError
+from django.test import TestCase
+from unittest.mock import patch
+
+
+class DeleteLocustLoadTestDataCommandTest(TestCase):
+    @patch(
+        "fecfiler.devops.management.commands"
+        ".delete_locust_load_test_data.LoadTestUtils"
+    )
+    def test_command_runs_when_load_mirror_validation_passes(self, mock_load_test_utils):
+        call_command("delete_locust_load_test_data")
+
+        mock_load_test_utils.return_value.validate_load_mirror_runtime \
+            .assert_called_once()
+        mock_load_test_utils.return_value.delete_load_test_committees_and_data \
+            .assert_called_once()
+
+    @patch(
+        "fecfiler.devops.management.commands"
+        ".delete_locust_load_test_data.LoadTestUtils"
+    )
+    def test_command_errors_when_load_mirror_validation_fails(self, mock_load_test_utils):
+        mock_load_test_utils.return_value.validate_load_mirror_runtime.side_effect = (
+            ValueError(
+                "delete_locust_load_test_data can only be run on a load testing mirror"
+            )
+        )
+
+        with self.assertRaisesRegex(CommandError, "load testing mirror"):
+            call_command("delete_locust_load_test_data")
+
+        mock_load_test_utils.return_value \
+            .delete_load_test_committees_and_data.assert_not_called()

django-backend/fecfiler/devops/tests/test_load_test_utils.py

@@ -1,6 +1,9 @@
 from django.test import TestCase
 from unittest.mock import patch, MagicMock
 from fecfiler.devops.utils.load_test import LoadTestUtils
+from fecfiler.committee_accounts.models import CommitteeAccount, Membership
+from fecfiler.user.models import User
+import json
 
 
 class LoadTestUtilsTestCase(TestCase):
@@ -69,3 +72,176 @@ def test_create_new_committee(
             committee_account_id=mock_committee.id,
             user=mock_user,
         )
+
+    @patch.dict(
+        "os.environ",
+        {
+            "VCAP_APPLICATION": json.dumps(
+                {
+                    "application_name": "load-fecfile-web-api",
+                }
+            ),
+            "VCAP_SERVICES": json.dumps(
+                {
+                    "aws-rds": [
+                        {"name": "load-fecfile-api-rds"},
+                    ],
+                    "s3": [
+                        {"name": "load-fecfile-api-s3"},
+                    ],
+                }
+            ),
+        },
+        clear=False,
+    )
+    def test_validate_load_mirror_runtime_success(self):
+        self.utils.validate_load_mirror_runtime()
+
+    @patch.dict(
+        "os.environ",
+        {
+            "VCAP_APPLICATION": json.dumps(
+                {
+                    "application_name": "fecfile-web-api",
+                }
+            ),
+            "VCAP_SERVICES": json.dumps(
+                {
+                    "aws-rds": [
+                        {"name": "load-fecfile-api-rds"},
+                    ],
+                    "s3": [
+                        {"name": "load-fecfile-api-s3"},
+                    ],
+                }
+            ),
+        },
+        clear=False,
+    )
+    def test_validate_load_mirror_runtime_fails_for_non_load_app(self):
+        with self.assertRaisesRegex(ValueError, "load testing mirror"):
+            self.utils.validate_load_mirror_runtime()
+
+    @patch.dict(
+        "os.environ",
+        {
+            "VCAP_APPLICATION": json.dumps(
+                {
+                    "application_name": "load-fecfile-web-api",
+                }
+            ),
+            "VCAP_SERVICES": json.dumps(
+                {
+                    "aws-rds": [
+                        {"name": "fecfile-api-rds"},
+                    ],
+                }
+            ),
+        },
+        clear=False,
+    )
+    def test_validate_load_mirror_runtime_fails_without_required_rds(self):
+        with self.assertRaisesRegex(ValueError, "load testing mirror"):
+            self.utils.validate_load_mirror_runtime()
+
+    @patch.dict(
+        "os.environ",
+        {
+            "VCAP_APPLICATION": json.dumps(
+                {
+                    "application_name": "load-fecfile-web-api",
+                }
+            ),
+            "VCAP_SERVICES": json.dumps(
+                {
+                    "aws-rds": [
+                        {"name": "load-fecfile-api-rds"},
+                    ],
+                }
+            ),
+        },
+        clear=False,
+    )
+    def test_validate_load_mirror_runtime_succeeds_with_only_required_rds(self):
+        self.utils.validate_load_mirror_runtime()
+
+    @patch.dict(
+        "os.environ",
+        {
+            "VCAP_APPLICATION": json.dumps(
+                {
+                    "application_name": "load-fecfile-web-api",
+                }
+            ),
+            "VCAP_SERVICES": "not-json",
+        },
+        clear=False,
+    )
+    def test_validate_load_mirror_runtime_fails_for_invalid_service_json(self):
+        with self.assertRaisesRegex(ValueError, "Could not parse VCAP_SERVICES"):
+            self.utils.validate_load_mirror_runtime()
+
+    def test_delete_load_test_committees_and_data_deletes_only_marked_data(self):
+        load_user = User.objects.create(
+            email="test@test.com",
+            username="test@test.com",
+        )
+        keep_user = User.objects.create(
+            email="keep@example.com",
+            username="keep@example.com",
+        )
+
+        load_committee = CommitteeAccount.objects.create(committee_id="C33333333")
+        keep_committee = CommitteeAccount.objects.create(committee_id="C12345678")
+
+        Membership.objects.create(
+            role=Membership.CommitteeRole.COMMITTEE_ADMINISTRATOR,
+            committee_account=load_committee,
+            user=load_user,
+        )
+        Membership.objects.create(
+            role=Membership.CommitteeRole.COMMITTEE_ADMINISTRATOR,
+            committee_account=keep_committee,
+            user=keep_user,
+        )
+
+        self.utils.delete_load_test_committees_and_data()
+
+        self.assertFalse(
+            CommitteeAccount.all_objects.filter(committee_id="C33333333").exists()
+        )
+        self.assertTrue(
+            CommitteeAccount.all_objects.filter(committee_id="C12345678").exists()
+        )
+        self.assertFalse(User.objects.filter(email__iexact="test@test.com").exists())
+        self.assertTrue(User.objects.filter(email__iexact="keep@example.com").exists())
+
+    def test_delete_load_test_committees_and_data_keeps_test_user_with_memberships(self):
+        load_user = User.objects.create(
+            email="test@test.com",
+            username="test@test.com",
+        )
+        committee = CommitteeAccount.objects.create(committee_id="C33333333")
+        other_committee = CommitteeAccount.objects.create(committee_id="C87654321")
+
+        Membership.objects.create(
+            role=Membership.CommitteeRole.COMMITTEE_ADMINISTRATOR,
+            committee_account=committee,
+            user=load_user,
+        )
+        Membership.objects.create(
+            role=Membership.CommitteeRole.COMMITTEE_ADMINISTRATOR,
+            committee_account=other_committee,
+            user=load_user,
+        )
+
+        with patch.object(
+            LoadTestUtils,
+            "delete_load_test_committees_and_data",
+            wraps=self.utils.delete_load_test_committees_and_data,
+        ):
+            CommitteeAccount.objects.filter(committee_id="C33333333") \
+                .first().hard_delete()
+            self.utils.delete_orphaned_test_user()
+
+        self.assertTrue(User.objects.filter(email__iexact="test@test.com").exists())

django-backend/fecfiler/devops/utils/load_test.py

@@ -1,5 +1,9 @@
+import json
+import os
 import structlog
 import math
+from django.conf import settings
+from django.db.models import Q
 from fecfiler.committee_accounts.models import CommitteeAccount, Membership
 from fecfiler.user.models import User
 
@@ -8,9 +12,120 @@
 logger = structlog.get_logger(__name__)
 
 TEST_USER_EMAIL = "test@test.com"
+LOAD_PREFIX = "load-"
+REQUIRED_LOAD_RDS_SERVICE = "load-fecfile-api-rds"
+LOAD_TEST_CONTACT_CITY = "Testville"
+LOAD_TEST_CONTACT_EMPLOYER = "Business Inc."
+LOAD_TEST_CONTACT_OCCUPATION = "Job"
+LOAD_TEST_REPORT_FORM_TYPE = "F3XN"
 
 
 class LoadTestUtils:
+    def validate_load_mirror_runtime(self):
+        app_name, service_names = self.get_cloud_foundry_runtime_identity()
+
+        has_load_app_name = app_name.startswith(LOAD_PREFIX)
+        has_required_rds = REQUIRED_LOAD_RDS_SERVICE in service_names
+
+        if has_load_app_name and has_required_rds:
+            logger.info(
+                "Validated load test runtime context",
+                app_name=app_name,
+                required_service=REQUIRED_LOAD_RDS_SERVICE,
+            )
+            return
+
+        raise ValueError(
+            "delete_locust_load_test_data can only be run on a load testing mirror"
+        )
+
+    def get_cloud_foundry_runtime_identity(self):
+        app_name = self.get_cloud_foundry_app_name()
+        service_names = self.get_cloud_foundry_service_names()
+
+        if not app_name or not service_names:
+            raise ValueError(
+                "delete_locust_load_test_data can only be run on a load testing mirror"
+            )
+
+        return app_name, service_names
+
+    def get_cloud_foundry_app_name(self):
+        vcap_application = os.environ.get("VCAP_APPLICATION")
+        if vcap_application:
+            try:
+                vcap_application_json = json.loads(vcap_application)
+            except json.JSONDecodeError as error:
+                raise ValueError(
+                    "Could not parse VCAP_APPLICATION to determine app name"
+                ) from error
+
+            app_name = vcap_application_json.get("application_name")
+            if app_name:
+                return app_name
+
+        return settings.APPLICATION_NAME
+
+    def get_cloud_foundry_service_names(self):
+        vcap_services = os.environ.get("VCAP_SERVICES")
+        if not vcap_services:
+            return set()
+
+        try:
+            vcap_services_json = json.loads(vcap_services)
+        except json.JSONDecodeError as error:
+            raise ValueError(
+                "Could not parse VCAP_SERVICES to determine service names"
+            ) from error
+
+        service_names = set()
+        for service_instances in vcap_services_json.values():
+            for service_instance in service_instances:
+                service_name = service_instance.get("name")
+                if service_name:
+                    service_names.add(service_name)
+
+        return service_names
+
+    def delete_load_test_committees_and_data(self):
+        committees_to_delete = CommitteeAccount.all_objects.filter(
+            Q(membership__user__email__iexact=TEST_USER_EMAIL)
+            | Q(
+                contact__city=LOAD_TEST_CONTACT_CITY,
+                contact__employer=LOAD_TEST_CONTACT_EMPLOYER,
+                contact__occupation=LOAD_TEST_CONTACT_OCCUPATION,
+            )
+            | Q(report__form_type=LOAD_TEST_REPORT_FORM_TYPE)
+        ).distinct()
+
+        committee_ids = list(committees_to_delete.values_list("committee_id", flat=True))
+        logger.info(
+            "Deleting load test committees",
+            committee_count=len(committee_ids),
+            committee_ids=committee_ids,
+        )
+
+        for committee in committees_to_delete:
+            committee.hard_delete()
+
+        self.delete_orphaned_test_user()
+
+    def delete_orphaned_test_user(self):
+        test_user = User.objects.filter(email__iexact=TEST_USER_EMAIL).first()
+        if not test_user:
+            return
+
+        has_memberships = Membership.objects.filter(user=test_user).exists()
+        if has_memberships:
+            logger.info(
+                "Skipping test user delete due to remaining memberships",
+                user_email=TEST_USER_EMAIL,
+            )
+            return
+
+        test_user.delete()
+        logger.info("Deleted load test user", user_email=TEST_USER_EMAIL)
+
     def create_load_test_committees_and_data(
         self,
         base_committee_number,

django-backend/manage.py

@@ -38,6 +38,7 @@
     # COMMITTEE ACCOUNT COMMANDS #
     "load_mocked_committee_data",
     "gen_locust_load_test_data",
+    "delete_locust_load_test_data",
     "delete_committee_account",
     "load_committee_data",
     # REPORT COMMANDS #