resolve dja-collapse conflict

src/registrar/management/commands/populate_domain_updated_federal_agency.py

@@ -0,0 +1,121 @@
+""""
+Data migration: Renaming deprecated Federal Agencies to
+their new updated names ie (U.S. Peace Corps to Peace Corps)
+within Domain Information and Domain Requests
+"""
+
+import logging
+
+from django.core.management import BaseCommand
+from registrar.models import DomainInformation, DomainRequest, FederalAgency
+from registrar.management.commands.utility.terminal_helper import ScriptDataHelper
+
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    help = "Transfers Domain Request and Domain Information federal agency field from string to FederalAgency object"
+
+    # Deprecated federal agency names mapped to designated replacements {old_value, new value}
+    rename_deprecated_federal_agency = {
+        "Appraisal Subcommittee": "Appraisal Subcommittee of the Federal Financial Institutions Examination Council",
+        "Barry Goldwater Scholarship and Excellence in Education Program": "Barry Goldwater Scholarship and Excellence in Education Foundation",  # noqa
+        "Federal Reserve System": "Federal Reserve Board of Governors",
+        "Harry S Truman Scholarship Foundation": "Harry S. Truman Scholarship Foundation",
+        "Japan-US Friendship Commission": "Japan-U.S. Friendship Commission",
+        "Japan-United States Friendship Commission": "Japan-U.S. Friendship Commission",
+        "John F. Kennedy Center for Performing Arts": "John F. Kennedy Center for the Performing Arts",
+        "Occupational Safety & Health Review Commission": "Occupational Safety and Health Review Commission",
+        "Corporation for National & Community Service": "Corporation for National and Community Service",
+        "Export/Import Bank of the U.S.": "Export-Import Bank of the United States",
+        "Medical Payment Advisory Commission": "Medicare Payment Advisory Commission",
+        "U.S. Peace Corps": "Peace Corps",
+        "Chemical Safety Board": "U.S. Chemical Safety Board",
+        "Nuclear Waste Technical Review Board": "U.S. Nuclear Waste Technical Review Board",
+        "State, Local, and Tribal Government": "Non-Federal Agency",
+        # "U.S. China Economic and Security Review Commission": "U.S.-China Economic and Security Review Commission",
+    }
+
+    def find_federal_agency_row(self, domain_object):
+        federal_agency = domain_object.federal_agency
+        # Domain Information objects without a federal agency default to Non-Federal Agency
+        if (federal_agency is None) or (federal_agency == ""):
+            federal_agency = "Non-Federal Agency"
+        if federal_agency in self.rename_deprecated_federal_agency.keys():
+            federal_agency = self.rename_deprecated_federal_agency[federal_agency]
+        return FederalAgency.objects.filter(agency=federal_agency).get()
+
+    def handle(self, **options):
+        """
+        Renames the Federal Agency to the correct new naming
+        for both Domain Information and Domain Requests objects.
+
+        NOTE: If it's None for a domain request, we skip it as
+        a user most likely hasn't gotten to it yet.
+        """
+        logger.info("Transferring federal agencies to FederalAgency object")
+        # DomainInformation object we populate with updated_federal_agency which are then bulk updated
+        domain_infos_to_update = []
+        domain_requests_to_update = []
+        # Domain Requests with null federal_agency that are not populated with updated_federal_agency
+        domain_requests_skipped = []
+        domain_infos_with_errors = []
+        domain_requests_with_errors = []
+
+        domain_infos = DomainInformation.objects.all()
+        domain_requests = DomainRequest.objects.all()
+
+        logger.info(f"Found {len(domain_infos)} DomainInfo objects with federal agency.")
+        logger.info(f"Found {len(domain_requests)} Domain Request objects with federal agency.")
+
+        for domain_info in domain_infos:
+            try:
+                federal_agency_row = self.find_federal_agency_row(domain_info)
+                domain_info.updated_federal_agency = federal_agency_row
+                domain_infos_to_update.append(domain_info)
+                logger.info(
+                    f"DomainInformation {domain_info} => updated_federal_agency set to: \
+                    {domain_info.updated_federal_agency}"
+                )
+            except Exception as err:
+                domain_infos_with_errors.append(domain_info)
+                logger.info(
+                    f"DomainInformation {domain_info} failed to update updated_federal_agency \
+                from federal_agency {domain_info.federal_agency}. Error: {err}"
+                )
+
+        ScriptDataHelper.bulk_update_fields(DomainInformation, domain_infos_to_update, ["updated_federal_agency"])
+
+        for domain_request in domain_requests:
+            try:
+                if (domain_request.federal_agency is None) or (domain_request.federal_agency == ""):
+                    domain_requests_skipped.append(domain_request)
+                else:
+                    federal_agency_row = self.find_federal_agency_row(domain_request)
+                    domain_request.updated_federal_agency = federal_agency_row
+                    domain_requests_to_update.append(domain_request)
+                    logger.info(
+                        f"DomainRequest {domain_request} => updated_federal_agency set to: \
+                        {domain_request.updated_federal_agency}"
+                    )
+            except Exception as err:
+                domain_requests_with_errors.append(domain_request)
+                logger.info(
+                    f"DomainRequest {domain_request} failed to update updated_federal_agency \
+                from federal_agency {domain_request.federal_agency}. Error: {err}"
+                )
+
+        ScriptDataHelper.bulk_update_fields(DomainRequest, domain_requests_to_update, ["updated_federal_agency"])
+
+        logger.info(f"{len(domain_infos_to_update)} DomainInformation rows updated update_federal_agency.")
+        logger.info(
+            f"{len(domain_infos_with_errors)} DomainInformation rows errored when updating update_federal_agency. \
+            {domain_infos_with_errors}"
+        )
+        logger.info(f"{len(domain_requests_to_update)} DomainRequest rows updated update_federal_agency.")
+        logger.info(f"{len(domain_requests_skipped)} DomainRequest rows with null federal_agency skipped.")
+        logger.info(
+            f"{len(domain_requests_with_errors)} DomainRequest rows errored when updating update_federal_agency. \
+            {domain_requests_with_errors}"
+        )

src/registrar/tests/common.py

@@ -789,6 +789,7 @@ def create_ready_domain():
     return domain
 
 
+# TODO in 1793: Remove the federal agency/updated federal agency fields
 def completed_domain_request(
     has_other_contacts=True,
     has_current_website=True,
@@ -803,6 +804,8 @@ def completed_domain_request(
     generic_org_type="federal",
     is_election_board=False,
     organization_type=None,
+    federal_agency=None,
+    updated_federal_agency=None,
 ):
     """A completed domain request."""
     if not user:
@@ -839,6 +842,7 @@ def completed_domain_request(
             last_name="Bob",
             is_staff=True,
         )
+
     domain_request_kwargs = dict(
         generic_org_type=generic_org_type,
         is_election_board=is_election_board,
@@ -856,6 +860,8 @@ def completed_domain_request(
         creator=user,
         status=status,
         investigator=investigator,
+        federal_agency=federal_agency,
+        updated_federal_agency=updated_federal_agency,
     )
     if has_about_your_organization:
         domain_request_kwargs["about_your_organization"] = "e-Government"
@@ -864,7 +870,6 @@ def completed_domain_request(
 
     if organization_type:
         domain_request_kwargs["organization_type"] = organization_type
-
     domain_request, _ = DomainRequest.objects.get_or_create(**domain_request_kwargs)
 
     if has_other_contacts:

src/registrar/tests/test_management_scripts.py

@@ -841,3 +841,120 @@ class TestDiscloseEmails(MockEppLib):
                     )
                 ]
             )
+
+
+# TODO in #1793: Remove this whole test class
+class TestPopulateDomainUpdatedFederalAgency(TestCase):
+    def setUp(self):
+        super().setUp()
+
+        # Get the domain requests
+        self.domain_request_1 = completed_domain_request(
+            name="stitches.gov",
+            generic_org_type=DomainRequest.OrganizationChoices.FEDERAL,
+            is_election_board=True,
+            status=DomainRequest.DomainRequestStatus.IN_REVIEW,
+            federal_agency="U.S. Peace Corps",
+        )
+        self.domain_request_2 = completed_domain_request(
+            name="fadoesntexist.gov",
+            generic_org_type=DomainRequest.OrganizationChoices.FEDERAL,
+            is_election_board=True,
+            status=DomainRequest.DomainRequestStatus.IN_REVIEW,
+            federal_agency="MEOWARDRULES",
+        )
+        self.domain_request_3 = completed_domain_request(
+            name="nullfederalagency.gov",
+            generic_org_type=DomainRequest.OrganizationChoices.FEDERAL,
+            is_election_board=True,
+            status=DomainRequest.DomainRequestStatus.IN_REVIEW,
+            federal_agency=None,
+        )
+
+        # Approve all three requests
+        self.domain_request_1.approve()
+        self.domain_request_2.approve()
+        self.domain_request_3.approve()
+
+        # Get the domains
+        self.domain_1 = Domain.objects.get(name="stitches.gov")
+        self.domain_2 = Domain.objects.get(name="fadoesntexist.gov")
+        self.domain_3 = Domain.objects.get(name="nullfederalagency.gov")
+
+        # Get the domain infos
+        self.domain_info_1 = DomainInformation.objects.get(domain=self.domain_1)
+        self.domain_info_2 = DomainInformation.objects.get(domain=self.domain_2)
+        self.domain_info_3 = DomainInformation.objects.get(domain=self.domain_3)
+
+    def tearDown(self):
+        super().tearDown()
+        DomainInformation.objects.all().delete()
+        DomainRequest.objects.all().delete()
+        Domain.objects.all().delete()
+
+    def run_populate_domain_updated_federal_agency(self):
+        """
+        This method executes the populate_domain_updated_federal_agency command.
+
+        The 'call_command' function from Django's management framework is then used to
+        execute the populate_domain_updated_federal_agency command.
+        """
+        with less_console_noise():
+            call_command("populate_domain_updated_federal_agency")
+
+    def test_domain_information_renaming_federal_agency_success(self):
+        """
+        Domain Information updates successfully for an "outdated" Federal Agency
+        """
+
+        self.run_populate_domain_updated_federal_agency()
+
+        self.domain_info_1.refresh_from_db()
+
+        previous_federal_agency_name = self.domain_info_1.federal_agency
+
+        updated_federal_agency_name = self.domain_info_1.updated_federal_agency.agency
+
+        self.assertEqual(previous_federal_agency_name, "U.S. Peace Corps")
+        self.assertEqual(updated_federal_agency_name, "Peace Corps")
+
+    def test_domain_information_does_not_exist(self):
+        """
+        Update a Federal Agency that doesn't exist
+        (should return None bc the Federal Agency didn't exist before)
+        """
+
+        self.run_populate_domain_updated_federal_agency()
+
+        self.domain_info_2.refresh_from_db()
+
+        self.assertEqual(self.domain_info_2.updated_federal_agency, None)
+
+    def test_domain_request_is_skipped(self):
+        """
+        Update a Domain Request that doesn't exist
+        (should return None bc the Federal Agency didn't exist before)
+        """
+
+        # Test case #2
+        self.run_populate_domain_updated_federal_agency()
+
+        self.domain_request_2.refresh_from_db()
+
+        self.assertEqual(self.domain_request_2.updated_federal_agency, None)
+
+    def test_domain_information_updating_null_federal_agency_to_non_federal_agency(self):
+        """
+        Updating a Domain Information that was previously None
+        to Non-Federal Agency
+        """
+
+        self.run_populate_domain_updated_federal_agency()
+
+        self.domain_info_3.refresh_from_db()
+
+        previous_federal_agency_name = self.domain_info_3.federal_agency
+        updated_federal_agency_name = self.domain_info_3.updated_federal_agency.agency
+
+        self.assertEqual(previous_federal_agency_name, None)
+        self.assertEqual(updated_federal_agency_name, "Non-Federal Agency")