diff --git a/src/registrar/management/commands/transfer_transition_domains_to_domains.py b/src/registrar/management/commands/transfer_transition_domains_to_domains.py index 81e5ed041..82e94ef2d 100644 --- a/src/registrar/management/commands/transfer_transition_domains_to_domains.py +++ b/src/registrar/management/commands/transfer_transition_domains_to_domains.py @@ -254,7 +254,7 @@ class Command(BaseCommand): # ----------------------- CREATE DOMAIN ----------------------- # no matching entry, make one target_domain = Domain( - name=transition_domain_name, + name=str(transition_domain_name), state=transition_domain_status, expiration_date=transition_domain_expiration_date, ) @@ -377,11 +377,12 @@ class Command(BaseCommand): org_choices, debug_on, ) -> DomainInformation: - org_type = transition_domain.organization_type + org_type_current = transition_domain.organization_type fed_type = transition_domain.federal_type fed_agency = transition_domain.federal_agency - match org_type: + org_type = ("", "") + match org_type_current: case "Federal": org_type = ("federal", "Federal") case "Interstate": @@ -414,7 +415,7 @@ class Command(BaseCommand): elif debug_on: logger.debug(f"No org type found on {domain.name}") - if valid_fed_type: + if valid_fed_type and isinstance(fed_type, str): new_domain_info_data["federal_type"] = fed_type.lower() elif debug_on: logger.debug(f"No federal type found on {domain.name}") diff --git a/src/registrar/management/commands/utility/extra_transition_domain_helper.py b/src/registrar/management/commands/utility/extra_transition_domain_helper.py index e7e1ae745..b68b9dc27 100644 --- a/src/registrar/management/commands/utility/extra_transition_domain_helper.py +++ b/src/registrar/management/commands/utility/extra_transition_domain_helper.py @@ -9,7 +9,7 @@ import logging import os import sys -from typing import List, Tuple +from typing import Dict, List, Tuple from registrar.models.transition_domain import TransitionDomain @@ -321,7 +321,10 @@ class LoadExtraTransitionDomain: and transition_domain.federal_agency.strip() != "" ) - if not info.active.lower() == "y": + if ( + not isinstance(info.active, str) or + not info.active.lower() == "y" + ): self.parse_logs.create_log_item( EnumFilenames.DOMAIN_ADHOC, LogCode.ERROR, @@ -331,7 +334,10 @@ class LoadExtraTransitionDomain: ) return transition_domain - if not info.isfederal.lower() == "y": + if ( + not isinstance(info.isfederal, str) or + not info.isfederal.lower() == "y" + ): self.parse_logs.create_log_item( EnumFilenames.DOMAIN_ADHOC, LogCode.ERROR, @@ -377,7 +383,9 @@ class LoadExtraTransitionDomain: # This data is stored as follows: FEDERAL - Judicial # For all other records, it is stored as so: Interstate # We can infer if it is federal or not based on this fact. - domain_type = info.domaintype.split("-") + domain_type = [] + if isinstance(info.domaintype, str): + domain_type = info.domaintype.split("-") domain_type_length = len(domain_type) if domain_type_length < 1 or domain_type_length > 2: raise ValueError("Found invalid data on DOMAIN_ADHOC") @@ -387,7 +395,10 @@ class LoadExtraTransitionDomain: # Check if this domain_type is active or not. # If not, we don't want to add this. - if not info.active.lower() == "y": + if ( + not isinstance(info.active, str) or + not info.active.lower() == "y" + ): self.parse_logs.create_log_item( EnumFilenames.DOMAIN_ADHOC, LogCode.ERROR, @@ -681,7 +692,7 @@ class FileDataHolder: self.id_field = id_field # Object data # - self.data = {} + self.data: Dict[str, type] = {} def try_infer_filename(self, current_file_name, default_file_name): """Tries to match a given filename to a regex, @@ -798,7 +809,7 @@ class ExtraTransitionDomain: # TODO - revise comment def populate_file_data( - self, pattern_map_params: List[Tuple[EnumFilenames, str, type, str]] + self, pattern_map_params ): """Populates the self.file_data field given a set of tuple params. @@ -935,6 +946,21 @@ class ExtraTransitionDomain: ) return dict_data + def _grab_row_id(self, row, id_field, file, dataclass_type): + try: + row_id = row[id_field] + except KeyError as err: + logger.error( + f"{TerminalColors.FAIL}" + "\n Key mismatch! Did you upload the wrong file?" + f"\n File: {file}" + f"\n Expected type: {dataclass_type}" + f"{TerminalColors.ENDC}" + ) + raise err + else: + return row_id + def _read_csv_file(self, file, seperator, dataclass_type, id_field): dict_data = {} # Used when we encounter bad data @@ -955,8 +981,8 @@ class ExtraTransitionDomain: ) dict_data = {} break - - row_id = row[id_field] + + row_id = self._grab_row_id(row, id_field, file, dataclass_type) # To maintain pairity with the load_transition_domain # script, we store this data in lowercase. diff --git a/src/registrar/management/commands/utility/terminal_helper.py b/src/registrar/management/commands/utility/terminal_helper.py index 1d6fb23a5..24e7147ce 100644 --- a/src/registrar/management/commands/utility/terminal_helper.py +++ b/src/registrar/management/commands/utility/terminal_helper.py @@ -1,6 +1,7 @@ from enum import Enum import logging import sys +from typing import List logger = logging.getLogger(__name__) @@ -42,7 +43,7 @@ class TerminalColors: class TerminalHelper: @staticmethod - def query_yes_no(question: str, default="yes") -> bool: + def query_yes_no(question: str, default="yes"): """Ask a yes/no question via raw_input() and return their answer. "question" is a string that is presented to the user. @@ -73,7 +74,7 @@ class TerminalHelper: logger.info("Please respond with 'yes' or 'no' " "(or 'y' or 'n').\n") @staticmethod - def query_yes_no_exit(question: str, default="yes") -> bool: + def query_yes_no_exit(question: str, default="yes"): """Ask a yes/no question via raw_input() and return their answer. "question" is a string that is presented to the user. @@ -113,7 +114,7 @@ class TerminalHelper: logger.info("Please respond with a valid selection.\n") # @staticmethod - def array_as_string(array_to_convert: []) -> str: + def array_as_string(array_to_convert: List[str]) -> str: array_as_string = "{}".format("\n".join(map(str, array_to_convert))) return array_as_string