Merge pull request #3303 from cisagov/ag/3234-test-existing-script

#3234: Modify create_federal_portfolio script to add requested suborg fields - [AG]

src/registrar/management/commands/create_federal_portfolio.py

@@ -5,6 +5,7 @@
 from django.core.management import BaseCommand, CommandError
 from registrar.management.commands.utility.terminal_helper import TerminalColors, TerminalHelper
 from registrar.models import DomainInformation, DomainRequest, FederalAgency, Suborganization, Portfolio, User
+from registrar.models.utility.generic_helper import normalize_string
 
 
 logger = logging.getLogger(__name__)
@@ -21,10 +22,21 @@ def __init__(self, *args, **kwargs):
         self.failed_portfolios = set()
 
     def add_arguments(self, parser):
-        """Add three arguments:
-        1. agency_name => the value of FederalAgency.agency
-        2. --parse_requests => if true, adds the given portfolio to each related DomainRequest
-        3. --parse_domains => if true, adds the given portfolio to each related DomainInformation
+        """Add command line arguments to create federal portfolios.
+
+        Required (mutually exclusive) arguments:
+            --agency_name: Name of a specific FederalAgency to create a portfolio for
+            --branch: Federal branch to process ("executive", "legislative", or "judicial").
+                    Creates portfolios for all FederalAgencies in that branch.
+
+        Required (at least one):
+            --parse_requests: Add the created portfolio(s) to related DomainRequest records
+            --parse_domains: Add the created portfolio(s) to related DomainInformation records
+            Note: You can use both --parse_requests and --parse_domains together
+
+        Optional (mutually exclusive with parse options):
+            --both: Shorthand for using both --parse_requests and --parse_domains
+                Cannot be used with --parse_requests or --parse_domains
         """
         group = parser.add_mutually_exclusive_group(required=True)
         group.add_argument(
@@ -78,12 +90,14 @@ def handle(self, **options):
             else:
                 raise CommandError(f"Cannot find '{branch}' federal agencies in our database.")
 
+        portfolios = []
         for federal_agency in agencies:
             message = f"Processing federal agency '{federal_agency.agency}'..."
             TerminalHelper.colorful_logger(logger.info, TerminalColors.MAGENTA, message)
             try:
                 # C901 'Command.handle' is too complex (12)
-                self.handle_populate_portfolio(federal_agency, parse_domains, parse_requests, both)
+                portfolio = self.handle_populate_portfolio(federal_agency, parse_domains, parse_requests, both)
+                portfolios.append(portfolio)
             except Exception as exec:
                 self.failed_portfolios.add(federal_agency)
                 logger.error(exec)
@@ -99,9 +113,65 @@ def handle(self, **options):
             display_as_str=True,
         )
 
+        # POST PROCESSING STEP: Remove the federal agency if it matches the portfolio name.
+        # We only do this for started domain requests.
+        if parse_requests or both:
+            TerminalHelper.prompt_for_execution(
+                system_exit_on_terminate=True,
+                prompt_message="This action will update domain requests even if they aren't on a portfolio.",
+                prompt_title=(
+                    "POST PROCESS STEP: Do you want to clear federal agency on (related) started domain requests?"
+                ),
+                verify_message=None,
+            )
+            self.post_process_started_domain_requests(agencies, portfolios)
+
+    def post_process_started_domain_requests(self, agencies, portfolios):
+        """
+        Removes duplicate organization data by clearing federal_agency when it matches the portfolio name.
+        Only processes domain requests in STARTED status.
+        """
+        message = "Removing duplicate portfolio and federal_agency values from domain requests..."
+        TerminalHelper.colorful_logger(logger.info, TerminalColors.MAGENTA, message)
+
+        # For each request, clear the federal agency under these conditions:
+        # 1. A portfolio *already exists* with the same name as the federal agency.
+        # 2. Said portfolio (or portfolios) are only the ones specified at the start of the script.
+        # 3. The domain request is in status "started".
+        # Note: Both names are normalized so excess spaces are stripped and the string is lowercased.
+        domain_requests_to_update = DomainRequest.objects.filter(
+            federal_agency__in=agencies,
+            federal_agency__agency__isnull=False,
+            status=DomainRequest.DomainRequestStatus.STARTED,
+            organization_name__isnull=False,
+        )
+        portfolio_set = {normalize_string(portfolio.organization_name) for portfolio in portfolios if portfolio}
+
+        # Update the request, assuming the given agency name matches the portfolio name
+        updated_requests = []
+        for req in domain_requests_to_update:
+            agency_name = normalize_string(req.federal_agency.agency)
+            if agency_name in portfolio_set:
+                req.federal_agency = None
+                updated_requests.append(req)
+
+        # Execute the update and Log the results
+        if TerminalHelper.prompt_for_execution(
+            system_exit_on_terminate=False,
+            prompt_message=(
+                f"{len(domain_requests_to_update)} domain requests will be updated. "
+                f"These records will be changed: {[str(req) for req in updated_requests]}"
+            ),
+            prompt_title="Do you wish to commit this update to the database?",
+        ):
+            DomainRequest.objects.bulk_update(updated_requests, ["federal_agency"])
+            TerminalHelper.colorful_logger(logger.info, TerminalColors.OKBLUE, "Action completed successfully.")
+
     def handle_populate_portfolio(self, federal_agency, parse_domains, parse_requests, both):
         """Attempts to create a portfolio. If successful, this function will
-        also create new suborganizations"""
+        also create new suborganizations.
+        Returns the portfolio for the given federal_agency.
+        """
         portfolio, created = self.create_portfolio(federal_agency)
         if created:
             self.create_suborganizations(portfolio, federal_agency)
@@ -111,6 +181,8 @@ def handle_populate_portfolio(self, federal_agency, parse_domains, parse_request
         if parse_requests or both:
             self.handle_portfolio_requests(portfolio, federal_agency)
 
+        return portfolio
+
     def create_portfolio(self, federal_agency):
         """Creates a portfolio if it doesn't presently exist.
         Returns portfolio, created."""
@@ -172,17 +244,15 @@ def create_suborganizations(self, portfolio: Portfolio, federal_agency: FederalA
             return
 
         # Check for existing suborgs on the current portfolio
-        existing_suborgs = Suborganization.objects.filter(name__in=org_names)
+        existing_suborgs = Suborganization.objects.filter(name__in=org_names, name__isnull=False)
         if existing_suborgs.exists():
             message = f"Some suborganizations already exist for portfolio '{portfolio}'."
             TerminalHelper.colorful_logger(logger.info, TerminalColors.OKBLUE, message)
 
         # Create new suborgs, as long as they don't exist in the db already
         new_suborgs = []
         for name in org_names - set(existing_suborgs.values_list("name", flat=True)):
-            # Stored in variables due to linter wanting type information here.
-            portfolio_name: str = portfolio.organization_name if portfolio.organization_name is not None else ""
-            if name is not None and name.lower() == portfolio_name.lower():
+            if normalize_string(name) == normalize_string(portfolio.organization_name):
                 # You can use this to populate location information, when this occurs.
                 # However, this isn't needed for now so we can skip it.
                 message = (
@@ -229,19 +299,39 @@ def handle_portfolio_requests(self, portfolio: Portfolio, federal_agency: Federa
         # Get all suborg information and store it in a dict to avoid doing a db call
         suborgs = Suborganization.objects.filter(portfolio=portfolio).in_bulk(field_name="name")
         for domain_request in domain_requests:
+            # Set the portfolio
             domain_request.portfolio = portfolio
-            if domain_request.organization_name in suborgs:
-                domain_request.sub_organization = suborgs.get(domain_request.organization_name)
+
+            # Set suborg info
+            domain_request.sub_organization = suborgs.get(domain_request.organization_name, None)
+            if domain_request.sub_organization is None:
+                domain_request.requested_suborganization = normalize_string(
+                    domain_request.organization_name, lowercase=False
+                )
+                domain_request.suborganization_city = normalize_string(domain_request.city, lowercase=False)
+                domain_request.suborganization_state_territory = domain_request.state_territory
+
             self.updated_portfolios.add(portfolio)
 
-        DomainRequest.objects.bulk_update(domain_requests, ["portfolio", "sub_organization"])
+        DomainRequest.objects.bulk_update(
+            domain_requests,
+            [
+                "portfolio",
+                "sub_organization",
+                "requested_suborganization",
+                "suborganization_city",
+                "suborganization_state_territory",
+            ],
+        )
         message = f"Added portfolio '{portfolio}' to {len(domain_requests)} domain requests."
         TerminalHelper.colorful_logger(logger.info, TerminalColors.OKGREEN, message)
 
     def handle_portfolio_domains(self, portfolio: Portfolio, federal_agency: FederalAgency):
         """
         Associate portfolio with domains for a federal agency.
         Updates all relevant domain information records.
+
+        Returns a queryset of DomainInformation objects, or None if nothing changed.
         """
         domain_infos = DomainInformation.objects.filter(federal_agency=federal_agency, portfolio__isnull=True)
         if not domain_infos.exists():
@@ -257,8 +347,7 @@ def handle_portfolio_domains(self, portfolio: Portfolio, federal_agency: Federal
         suborgs = Suborganization.objects.filter(portfolio=portfolio).in_bulk(field_name="name")
         for domain_info in domain_infos:
             domain_info.portfolio = portfolio
-            if domain_info.organization_name in suborgs:
-                domain_info.sub_organization = suborgs.get(domain_info.organization_name)
+            domain_info.sub_organization = suborgs.get(domain_info.organization_name, None)
 
         DomainInformation.objects.bulk_update(domain_infos, ["portfolio", "sub_organization"])
         message = f"Added portfolio '{portfolio}' to {len(domain_infos)} domains."

src/registrar/models/utility/generic_helper.py

@@ -343,3 +343,13 @@ def value_of_attribute(obj, attribute_name: str):
     if callable(value):
         value = value()
     return value
+
+
+def normalize_string(string_to_normalize, lowercase=True):
+    """Normalizes a given string. Returns a string without extra spaces, in all lowercase."""
+    if not isinstance(string_to_normalize, str):
+        logger.error(f"normalize_string => {string_to_normalize} is not type str.")
+        return string_to_normalize
+
+    new_string = " ".join(string_to_normalize.split())
+    return new_string.lower() if lowercase else new_string

src/registrar/tests/common.py

@@ -1039,6 +1039,8 @@ def completed_domain_request(  # noqa
     federal_agency=None,
     federal_type=None,
     action_needed_reason=None,
+    city=None,
+    state_territory=None,
     portfolio=None,
     organization_name=None,
     sub_organization=None,
@@ -1081,7 +1083,7 @@ def completed_domain_request(  # noqa
         organization_name=organization_name if organization_name else "Testorg",
         address_line1="address 1",
         address_line2="address 2",
-        state_territory="NY",
+        state_territory="NY" if not state_territory else state_territory,
         zipcode="10002",
         senior_official=so,
         requested_domain=domain,
@@ -1090,6 +1092,10 @@ def completed_domain_request(  # noqa
         investigator=investigator,
         federal_agency=federal_agency,
     )
+
+    if city:
+        domain_request_kwargs["city"] = city
+
     if has_about_your_organization:
         domain_request_kwargs["about_your_organization"] = "e-Government"
     if has_anything_else:

src/registrar/tests/test_management_scripts.py

@@ -1516,6 +1516,91 @@ def run_create_federal_portfolio(self, **kwargs):
         ):
             call_command("create_federal_portfolio", **kwargs)
 
+    @less_console_noise_decorator
+    def test_post_process_started_domain_requests_existing_portfolio(self):
+        """Ensures that federal agency is cleared when agency name matches portfolio name.
+        As the name implies, this implicitly tests the "post_process_started_domain_requests" function.
+        """
+        federal_agency_2 = FederalAgency.objects.create(agency="Sugarcane", federal_type=BranchChoices.EXECUTIVE)
+
+        # Test records with portfolios and no org names
+        # Create a portfolio. This script skips over "started"
+        portfolio = Portfolio.objects.create(organization_name="Sugarcane", creator=self.user)
+        # Create a domain request with matching org name
+        matching_request = completed_domain_request(
+            name="matching.gov",
+            status=DomainRequest.DomainRequestStatus.STARTED,
+            generic_org_type=DomainRequest.OrganizationChoices.FEDERAL,
+            federal_agency=federal_agency_2,
+            user=self.user,
+            portfolio=portfolio,
+        )
+
+        # Create a request not in started (no change should occur)
+        matching_request_in_wrong_status = completed_domain_request(
+            name="kinda-matching.gov",
+            status=DomainRequest.DomainRequestStatus.IN_REVIEW,
+            generic_org_type=DomainRequest.OrganizationChoices.FEDERAL,
+            federal_agency=self.federal_agency,
+            user=self.user,
+        )
+
+        self.run_create_federal_portfolio(agency_name="Sugarcane", parse_requests=True)
+        self.run_create_federal_portfolio(agency_name="Test Federal Agency", parse_requests=True)
+
+        # Refresh from db
+        matching_request.refresh_from_db()
+        matching_request_in_wrong_status.refresh_from_db()
+
+        # Request with matching name should have federal_agency cleared
+        self.assertIsNone(matching_request.federal_agency)
+        self.assertIsNotNone(matching_request.portfolio)
+        self.assertEqual(matching_request.portfolio.organization_name, "Sugarcane")
+
+        # Request with matching name but wrong state should keep its federal agency
+        self.assertEqual(matching_request_in_wrong_status.federal_agency, self.federal_agency)
+        self.assertIsNotNone(matching_request_in_wrong_status.portfolio)
+        self.assertEqual(matching_request_in_wrong_status.portfolio.organization_name, "Test Federal Agency")
+
+    @less_console_noise_decorator
+    def test_post_process_started_domain_requests(self):
+        """Tests that federal agency is cleared when agency name
+        matches an existing portfolio's name, even if the domain request isn't
+        directly on that portfolio."""
+
+        federal_agency_2 = FederalAgency.objects.create(agency="Sugarcane", federal_type=BranchChoices.EXECUTIVE)
+
+        # Create a request with matching federal_agency name but no direct portfolio association
+        matching_agency_request = completed_domain_request(
+            name="agency-match.gov",
+            status=DomainRequest.DomainRequestStatus.STARTED,
+            generic_org_type=DomainRequest.OrganizationChoices.FEDERAL,
+            federal_agency=federal_agency_2,
+            user=self.user,
+        )
+
+        # Create a control request that shouldn't match
+        non_matching_request = completed_domain_request(
+            name="no-match.gov",
+            status=DomainRequest.DomainRequestStatus.STARTED,
+            generic_org_type=DomainRequest.OrganizationChoices.FEDERAL,
+            federal_agency=self.federal_agency,
+            user=self.user,
+        )
+
+        # We expect the matching agency to have its fed agency cleared.
+        self.run_create_federal_portfolio(agency_name="Sugarcane", parse_requests=True)
+        matching_agency_request.refresh_from_db()
+        non_matching_request.refresh_from_db()
+
+        # Request with matching agency name should have federal_agency cleared
+        self.assertIsNone(matching_agency_request.federal_agency)
+
+        # Non-matching request should keep its federal_agency
+        self.assertIsNotNone(non_matching_request.federal_agency)
+        self.assertEqual(non_matching_request.federal_agency, self.federal_agency)
+
+    @less_console_noise_decorator
     def test_create_single_portfolio(self):
         """Test portfolio creation with suborg and senior official."""
         self.run_create_federal_portfolio(agency_name="Test Federal Agency", parse_requests=True)
@@ -1588,6 +1673,34 @@ def test_create_multiple_portfolios_for_branch_legislative(self):
         self.assertTrue(all([creator == User.get_default_user() for creator in creators]))
         self.assertTrue(all([note == "Auto-generated record" for note in notes]))
 
+    def test_script_adds_requested_suborganization_information(self):
+        """Tests that the script adds the requested suborg fields for domain requests"""
+        # Create a new domain request with some errant spacing
+        custom_suborg_request = completed_domain_request(
+            name="custom_org.gov",
+            status=DomainRequest.DomainRequestStatus.IN_REVIEW,
+            generic_org_type=DomainRequest.OrganizationChoices.FEDERAL,
+            federal_agency=self.executive_agency_2,
+            user=self.user,
+            organization_name=" requested org name ",
+            city="Austin   ",
+            state_territory=DomainRequest.StateTerritoryChoices.TEXAS,
+        )
+
+        self.assertIsNone(custom_suborg_request.requested_suborganization)
+        self.assertIsNone(custom_suborg_request.suborganization_city)
+        self.assertIsNone(custom_suborg_request.suborganization_state_territory)
+
+        # Run the script and test it
+        self.run_create_federal_portfolio(branch="executive", parse_requests=True)
+        custom_suborg_request.refresh_from_db()
+
+        self.assertEqual(custom_suborg_request.requested_suborganization, "requested org name")
+        self.assertEqual(custom_suborg_request.suborganization_city, "Austin")
+        self.assertEqual(
+            custom_suborg_request.suborganization_state_territory, DomainRequest.StateTerritoryChoices.TEXAS
+        )
+
     def test_create_multiple_portfolios_for_branch_executive(self):
         """Tests creating all portfolios under a given branch"""
         federal_choice = DomainRequest.OrganizationChoices.FEDERAL