zydronium/manage.get.gov
1
0
Fork 0
mirror of https://github.com/cisagov/manage.get.gov.git synced 2025-05-19 02:49:21 +02:00
Code Issues Projects Releases Packages Wiki Activity

Black linting

Browse source
This commit is contained in:
zandercymatics 2023-11-13 08:31:20 -07:00
parent cfaafb8ef3
commit 8aa840f427
No known key found for this signature in database
GPG key ID: FF4636ABEC9682B7
10 changed files with 73 additions and 234 deletions
Download patch file Download diff file Expand all files Collapse all files

40
src/registrar/management/commands/load_transition_domain.py
View file

@ -45,10 +45,7 @@ class Command(BaseCommand):
""" """
parser.add_argument( parser.add_argument(
"migration_json_filename", "migration_json_filename",
help=( help=("A JSON file that holds the location and filenames" "of all the data files used for migrations"),
"A JSON file that holds the location and filenames"
"of all the data files used for migrations"
),
) )
parser.add_argument("--sep", default="|", help="Delimiter character") parser.add_argument("--sep", default="|", help="Delimiter character")
@ -73,9 +70,7 @@ class Command(BaseCommand):
"Recommended to be enabled only in a development or testing setting.", "Recommended to be enabled only in a development or testing setting.",
) )
parser.add_argument( parser.add_argument("--directory", default="migrationdata", help="Desired directory")
"--directory", default="migrationdata", help="Desired directory"
)
parser.add_argument( parser.add_argument(
"--domain_contacts_filename", "--domain_contacts_filename",
help="Data file with domain contact information", help="Data file with domain contact information",
@ -119,9 +114,7 @@ class Command(BaseCommand):
help="Defines the filename for domain type adhocs", help="Defines the filename for domain type adhocs",
) )
def print_debug_mode_statements( def print_debug_mode_statements(self, debug_on: bool, debug_max_entries_to_parse: int):
self, debug_on: bool, debug_max_entries_to_parse: int
):
"""Prints additional terminal statements to indicate if --debug """Prints additional terminal statements to indicate if --debug
or --limitParse are in use""" or --limitParse are in use"""
if debug_on: if debug_on:
@ -356,42 +349,31 @@ class Command(BaseCommand):
debug_on = args.debug debug_on = args.debug
# Get --LimitParse argument # Get --LimitParse argument
debug_max_entries_to_parse = int( debug_max_entries_to_parse = int(args.limitParse) # set to 0 to parse all entries
args.limitParse
) # set to 0 to parse all entries
# Variables for Additional TransitionDomain Information # # Variables for Additional TransitionDomain Information #
# Main script filenames - these do not have defaults # Main script filenames - these do not have defaults
domain_contacts_filename = None domain_contacts_filename = None
try: try:
domain_contacts_filename = directory + options.get( domain_contacts_filename = directory + options.get("domain_contacts_filename")
"domain_contacts_filename"
)
except TypeError: except TypeError:
logger.error( logger.error(
f"Invalid filename of '{args.domain_contacts_filename}'" f"Invalid filename of '{args.domain_contacts_filename}'" " was provided for domain_contacts_filename"
" was provided for domain_contacts_filename"
) )
contacts_filename = None contacts_filename = None
try: try:
contacts_filename = directory + options.get("contacts_filename") contacts_filename = directory + options.get("contacts_filename")
except TypeError: except TypeError:
logger.error( logger.error(f"Invalid filename of '{args.contacts_filename}'" " was provided for contacts_filename")
f"Invalid filename of '{args.contacts_filename}'"
" was provided for contacts_filename"
)
domain_statuses_filename = None domain_statuses_filename = None
try: try:
domain_statuses_filename = directory + options.get( domain_statuses_filename = directory + options.get("domain_statuses_filename")
"domain_statuses_filename"
)
except TypeError: except TypeError:
logger.error( logger.error(
f"Invalid filename of '{args.domain_statuses_filename}'" f"Invalid filename of '{args.domain_statuses_filename}'" " was provided for domain_statuses_filename"
" was provided for domain_statuses_filename"
) )
# Agency information # Agency information
@ -630,9 +612,7 @@ class Command(BaseCommand):
# Print a summary of findings (duplicate entries, # Print a summary of findings (duplicate entries,
# missing data..etc.) # missing data..etc.)
self.print_summary_duplications( self.print_summary_duplications(duplicate_domain_user_combos, duplicate_domains, users_without_email)
duplicate_domain_user_combos, duplicate_domains, users_without_email
)
self.print_summary_status_findings(domains_without_status, outlier_statuses) self.print_summary_status_findings(domains_without_status, outlier_statuses)
logger.info( logger.info(

5
src/registrar/management/commands/master_domain_migrations.py
View file

@ -94,10 +94,7 @@ class Command(BaseCommand):
parser.add_argument( parser.add_argument(
"--migrationJSON", "--migrationJSON",
default="migrationFilepaths.json", default="migrationFilepaths.json",
help=( help=("A JSON file that holds the location and filenames" "of all the data files used for migrations"),
"A JSON file that holds the location and filenames"
"of all the data files used for migrations"
),
) )
# TODO: deprecate this once JSON module is done? (or keep as an override) # TODO: deprecate this once JSON module is done? (or keep as an override)

98
src/registrar/management/commands/transfer_transition_domains_to_domains.py
View file

@ -42,9 +42,7 @@ class Command(BaseCommand):
# ====================================================== # ======================================================
# ===================== PRINTING ====================== # ===================== PRINTING ======================
# ====================================================== # ======================================================
def print_debug_mode_statements( def print_debug_mode_statements(self, debug_on: bool, debug_max_entries_to_parse: int):
self, debug_on: bool, debug_max_entries_to_parse: int
):
"""Prints additional terminal statements to indicate if --debug """Prints additional terminal statements to indicate if --debug
or --limitParse are in use""" or --limitParse are in use"""
TerminalHelper.print_conditional( TerminalHelper.print_conditional(
@ -66,13 +64,8 @@ class Command(BaseCommand):
""", """,
) )
def parse_limit_reached( def parse_limit_reached(self, debug_max_entries_to_parse: bool, total_rows_parsed: int) -> bool:
self, debug_max_entries_to_parse: bool, total_rows_parsed: int if debug_max_entries_to_parse > 0 and total_rows_parsed >= debug_max_entries_to_parse:
) -> bool:
if (
debug_max_entries_to_parse > 0
and total_rows_parsed >= debug_max_entries_to_parse
):
logger.info( logger.info(
f"""{TerminalColors.YELLOW} f"""{TerminalColors.YELLOW}
----PARSE LIMIT REACHED. HALTING PARSER.---- ----PARSE LIMIT REACHED. HALTING PARSER.----
@ -159,9 +152,7 @@ class Command(BaseCommand):
# ====================================================== # ======================================================
# =================== DOMAIN ===================== # =================== DOMAIN =====================
# ====================================================== # ======================================================
def update_or_create_domain( def update_or_create_domain(self, transition_domain: TransitionDomain, debug_on: bool):
self, transition_domain: TransitionDomain, debug_on: bool
):
"""Given a transition domain, either finds & updates an existing """Given a transition domain, either finds & updates an existing
corresponding domain, or creates a new corresponding domain in corresponding domain, or creates a new corresponding domain in
the Domain table. the Domain table.
@ -260,9 +251,7 @@ class Command(BaseCommand):
) )
return (target_domain, True) return (target_domain, True)
def update_domain_status( def update_domain_status(self, transition_domain: TransitionDomain, target_domain: Domain, debug_on: bool) -> bool:
self, transition_domain: TransitionDomain, target_domain: Domain, debug_on: bool
) -> bool:
"""Given a transition domain that matches an existing domain, """Given a transition domain that matches an existing domain,
updates the existing domain object with that status of updates the existing domain object with that status of
the transition domain. the transition domain.
@ -293,9 +282,7 @@ class Command(BaseCommand):
# ====================================================== # ======================================================
# ================ DOMAIN INVITATION ================== # ================ DOMAIN INVITATION ==================
# ====================================================== # ======================================================
def try_add_domain_invitation( def try_add_domain_invitation(self, domain_email: str, associated_domain: Domain) -> DomainInvitation | None:
self, domain_email: str, associated_domain: Domain
) -> DomainInvitation | None:
"""If no domain invitation exists for the given domain and """If no domain invitation exists for the given domain and
e-mail, create and return a new domain invitation object. e-mail, create and return a new domain invitation object.
If one already exists, or if the email is invalid, return NONE""" If one already exists, or if the email is invalid, return NONE"""
@ -334,17 +321,11 @@ class Command(BaseCommand):
# ====================================================== # ======================================================
# ================ DOMAIN INFORMATION ================= # ================ DOMAIN INFORMATION =================
# ====================================================== # ======================================================
def update_domain_information( def update_domain_information(self, current: DomainInformation, target: DomainInformation, debug_on: bool) -> bool:
self, current: DomainInformation, target: DomainInformation, debug_on: bool
) -> bool:
# DEBUG: # DEBUG:
TerminalHelper.print_conditional( TerminalHelper.print_conditional(
debug_on, debug_on,
( (f"{TerminalColors.OKCYAN}" f"Updating: {current}" f"{TerminalColors.ENDC}"), # noqa
f"{TerminalColors.OKCYAN}"
f"Updating: {current}"
f"{TerminalColors.ENDC}"
), # noqa
) )
updated = False updated = False
@ -466,15 +447,11 @@ class Command(BaseCommand):
debug_on, debug_on,
) )
target_domain_information = None target_domain_information = None
domain_information_exists = DomainInformation.objects.filter( domain_information_exists = DomainInformation.objects.filter(domain__name=transition_domain_name).exists()
domain__name=transition_domain_name
).exists()
if domain_information_exists: if domain_information_exists:
try: try:
# get the existing domain information object # get the existing domain information object
target_domain_information = DomainInformation.objects.get( target_domain_information = DomainInformation.objects.get(domain__name=transition_domain_name)
domain__name=transition_domain_name
)
# DEBUG: # DEBUG:
TerminalHelper.print_conditional( TerminalHelper.print_conditional(
debug_on, debug_on,
@ -488,9 +465,7 @@ class Command(BaseCommand):
# for existing entry, update the status to # for existing entry, update the status to
# the transition domain status # the transition domain status
self.update_domain_information( self.update_domain_information(target_domain_information, template_domain_information, debug_on)
target_domain_information, template_domain_information, debug_on
)
# TODO: not all domains need to be updated # TODO: not all domains need to be updated
# (the information is the same). # (the information is the same).
# Need to bubble this up to the final report. # Need to bubble this up to the final report.
@ -560,9 +535,7 @@ class Command(BaseCommand):
if target_domain_information is None: if target_domain_information is None:
# ---------------- SKIPPED ---------------- # ---------------- SKIPPED ----------------
skipped_domain_information_entries.append(target_domain_information) skipped_domain_information_entries.append(target_domain_information)
debug_string = ( debug_string = f"skipped domain information: {target_domain_information}"
f"skipped domain information: {target_domain_information}"
)
elif was_created: elif was_created:
# DEBUG: # DEBUG:
TerminalHelper.print_conditional( TerminalHelper.print_conditional(
@ -577,11 +550,7 @@ class Command(BaseCommand):
# The unique key constraint does not allow multiple domain # The unique key constraint does not allow multiple domain
# information objects to share the same domain # information objects to share the same domain
existing_domain_information_in_to_create = next( existing_domain_information_in_to_create = next(
( (x for x in domain_information_to_create if x.domain.name == target_domain_information.domain.name),
x
for x in domain_information_to_create
if x.domain.name == target_domain_information.domain.name
),
None, None,
) )
# TODO: this is redundant. # TODO: this is redundant.
@ -590,10 +559,7 @@ class Command(BaseCommand):
existing_domain_info = DomainInformation.objects.filter( existing_domain_info = DomainInformation.objects.filter(
domain__name=target_domain_information.domain.name domain__name=target_domain_information.domain.name
).exists() ).exists()
if ( if existing_domain_information_in_to_create is not None or existing_domain_info:
existing_domain_information_in_to_create is not None
or existing_domain_info
):
debug_string = f"""{TerminalColors.YELLOW} debug_string = f"""{TerminalColors.YELLOW}
Duplicate Detected: {existing_domain_information_in_to_create}. Duplicate Detected: {existing_domain_information_in_to_create}.
Cannot add duplicate Domain Information object Cannot add duplicate Domain Information object
@ -601,15 +567,11 @@ class Command(BaseCommand):
else: else:
# ---------------- CREATED ---------------- # ---------------- CREATED ----------------
domain_information_to_create.append(target_domain_information) domain_information_to_create.append(target_domain_information)
debug_string = ( debug_string = f"created domain information: {target_domain_information}"
f"created domain information: {target_domain_information}"
)
elif not was_created: elif not was_created:
# ---------------- UPDATED ---------------- # ---------------- UPDATED ----------------
updated_domain_information.append(target_domain_information) updated_domain_information.append(target_domain_information)
debug_string = ( debug_string = f"updated domain information: {target_domain_information}"
f"updated domain information: {target_domain_information}"
)
else: else:
debug_string = "domain information already exists and " debug_string = "domain information already exists and "
f"matches incoming data (NO CHANGES MADE): {target_domain_information}" f"matches incoming data (NO CHANGES MADE): {target_domain_information}"
@ -664,9 +626,7 @@ class Command(BaseCommand):
# ====================================================== # ======================================================
# ====================== DOMAIN ======================= # ====================== DOMAIN =======================
target_domain, was_created = self.update_or_create_domain( target_domain, was_created = self.update_or_create_domain(transition_domain, debug_on)
transition_domain, debug_on
)
debug_string = "" debug_string = ""
if target_domain is None: if target_domain is None:
@ -704,9 +664,7 @@ class Command(BaseCommand):
# ====================================================== # ======================================================
# ================ DOMAIN INVITATIONS ================== # ================ DOMAIN INVITATIONS ==================
new_domain_invitation = self.try_add_domain_invitation( new_domain_invitation = self.try_add_domain_invitation(transition_domain_email, target_domain)
transition_domain_email, target_domain
)
if new_domain_invitation is None: if new_domain_invitation is None:
logger.info( logger.info(
f"{TerminalColors.YELLOW} ! No new e-mail detected !" # noqa f"{TerminalColors.YELLOW} ! No new e-mail detected !" # noqa
@ -812,19 +770,11 @@ class Command(BaseCommand):
invitation.domain = existing_domain.get() invitation.domain = existing_domain.get()
else: else:
# Raise an err for now # Raise an err for now
raise Exception( raise Exception(f"Domain {existing_domain} wants to be added" "but doesn't exist in the DB")
f"Domain {existing_domain} wants to be added"
"but doesn't exist in the DB"
)
invitation.save() invitation.save()
valid_org_choices = [ valid_org_choices = [(name, value) for name, value in DomainApplication.OrganizationChoices.choices]
(name, value) valid_fed_choices = [value for name, value in DomainApplication.BranchChoices.choices]
for name, value in DomainApplication.OrganizationChoices.choices
]
valid_fed_choices = [
value for name, value in DomainApplication.BranchChoices.choices
]
valid_agency_choices = DomainApplication.AGENCIES valid_agency_choices = DomainApplication.AGENCIES
# ====================================================== # ======================================================
# ================= DOMAIN INFORMATION ================= # ================= DOMAIN INFORMATION =================
@ -852,11 +802,7 @@ class Command(BaseCommand):
TerminalHelper.print_conditional( TerminalHelper.print_conditional(
debug_on, debug_on,
( (f"{TerminalColors.YELLOW}" f"Trying to add: {domain_information_to_create}" f"{TerminalColors.ENDC}"),
f"{TerminalColors.YELLOW}"
f"Trying to add: {domain_information_to_create}"
f"{TerminalColors.ENDC}"
),
) )
DomainInformation.objects.bulk_create(domain_information_to_create) DomainInformation.objects.bulk_create(domain_information_to_create)

93
src/registrar/management/commands/utility/extra_transition_domain_helper.py
View file

@ -111,9 +111,7 @@ class FileTransitionLog:
"""Logs every LogItem contained in this object""" """Logs every LogItem contained in this object"""
for parent_log in self.logs: for parent_log in self.logs:
for child_log in parent_log: for child_log in parent_log:
TerminalHelper.print_conditional( TerminalHelper.print_conditional(True, child_log.message, child_log.severity)
True, child_log.message, child_log.severity
)
def display_logs_by_domain_name(self, domain_name, restrict_type=LogCode.DEFAULT): def display_logs_by_domain_name(self, domain_name, restrict_type=LogCode.DEFAULT):
"""Displays all logs of a given domain_name. """Displays all logs of a given domain_name.
@ -130,9 +128,7 @@ class FileTransitionLog:
return None return None
for log in domain_logs: for log in domain_logs:
TerminalHelper.print_conditional( TerminalHelper.print_conditional(restrict_type != log.code, log.message, log.code)
restrict_type != log.code, log.message, log.code
)
def get_logs(self, file_type, domain_name): def get_logs(self, file_type, domain_name):
"""Grabs the logs associated with """Grabs the logs associated with
@ -166,33 +162,21 @@ class LoadExtraTransitionDomain:
updated_transition_domain = transition_domain updated_transition_domain = transition_domain
try: try:
# STEP 1: Parse organization data # STEP 1: Parse organization data
updated_transition_domain = self.parse_org_data( updated_transition_domain = self.parse_org_data(domain_name, transition_domain)
domain_name, transition_domain
)
# STEP 2: Parse domain type data # STEP 2: Parse domain type data
updated_transition_domain = self.parse_domain_type_data( updated_transition_domain = self.parse_domain_type_data(domain_name, transition_domain)
domain_name, transition_domain
)
# STEP 3: Parse agency data # STEP 3: Parse agency data
updated_transition_domain = self.parse_agency_data( updated_transition_domain = self.parse_agency_data(domain_name, transition_domain)
domain_name, transition_domain
)
# STEP 4: Parse creation and expiration data # STEP 4: Parse creation and expiration data
updated_transition_domain = self.parse_creation_expiration_data( updated_transition_domain = self.parse_creation_expiration_data(domain_name, transition_domain)
domain_name, transition_domain
)
# Check if the instance has changed before saving # Check if the instance has changed before saving
updated_transition_domain.save() updated_transition_domain.save()
updated_transition_domains.append(updated_transition_domain) updated_transition_domains.append(updated_transition_domain)
logger.info( logger.info(f"{TerminalColors.OKCYAN}" f"Successfully updated {domain_name}" f"{TerminalColors.ENDC}")
f"{TerminalColors.OKCYAN}"
f"Successfully updated {domain_name}"
f"{TerminalColors.ENDC}"
)
# If we run into an exception on this domain, # If we run into an exception on this domain,
# Just skip over it and log that it happened. # Just skip over it and log that it happened.
@ -267,8 +251,7 @@ class LoadExtraTransitionDomain:
self.parse_logs.create_log_item( self.parse_logs.create_log_item(
EnumFilenames.DOMAIN_ESCROW, EnumFilenames.DOMAIN_ESCROW,
LogCode.ERROR, LogCode.ERROR,
"Could not add epp_creation_date and epp_expiration_date " "Could not add epp_creation_date and epp_expiration_date " f"on {domain_name}, no data exists.",
f"on {domain_name}, no data exists.",
domain_name, domain_name,
not self.debug, not self.debug,
) )
@ -316,10 +299,7 @@ class LoadExtraTransitionDomain:
) )
return transition_domain return transition_domain
agency_exists = ( agency_exists = transition_domain.federal_agency is not None and transition_domain.federal_agency.strip() != ""
transition_domain.federal_agency is not None
and transition_domain.federal_agency.strip() != ""
)
if not isinstance(info.active, str) or not info.active.lower() == "y": if not isinstance(info.active, str) or not info.active.lower() == "y":
self.parse_logs.create_log_item( self.parse_logs.create_log_item(
@ -354,9 +334,7 @@ class LoadExtraTransitionDomain:
return transition_domain return transition_domain
def parse_domain_type_data( def parse_domain_type_data(self, domain_name, transition_domain: TransitionDomain) -> TransitionDomain:
self, domain_name, transition_domain: TransitionDomain
) -> TransitionDomain:
"""Grabs organization_type and federal_type from the parsed files """Grabs organization_type and federal_type from the parsed files
and associates it with a transition_domain object, then returns that object.""" and associates it with a transition_domain object, then returns that object."""
if not isinstance(transition_domain, TransitionDomain): if not isinstance(transition_domain, TransitionDomain):
@ -401,12 +379,10 @@ class LoadExtraTransitionDomain:
# Are we updating data that already exists, # Are we updating data that already exists,
# or are we adding new data in its place? # or are we adding new data in its place?
organization_type_exists = ( organization_type_exists = (
transition_domain.organization_type is not None transition_domain.organization_type is not None and transition_domain.organization_type.strip() != ""
and transition_domain.organization_type.strip() != ""
) )
federal_type_exists = ( federal_type_exists = (
transition_domain.federal_type is not None transition_domain.federal_type is not None and transition_domain.federal_type.strip() != ""
and transition_domain.federal_type.strip() != ""
) )
# If we get two records, then we know it is federal. # If we get two records, then we know it is federal.
@ -440,9 +416,7 @@ class LoadExtraTransitionDomain:
return transition_domain return transition_domain
def parse_org_data( def parse_org_data(self, domain_name, transition_domain: TransitionDomain) -> TransitionDomain:
self, domain_name, transition_domain: TransitionDomain
) -> TransitionDomain:
"""Grabs organization_name from the parsed files and associates it """Grabs organization_name from the parsed files and associates it
with a transition_domain object, then returns that object.""" with a transition_domain object, then returns that object."""
if not isinstance(transition_domain, TransitionDomain): if not isinstance(transition_domain, TransitionDomain):
@ -460,8 +434,7 @@ class LoadExtraTransitionDomain:
return transition_domain return transition_domain
desired_property_exists = ( desired_property_exists = (
transition_domain.organization_name is not None transition_domain.organization_name is not None and transition_domain.organization_name.strip() != ""
and transition_domain.organization_name.strip() != ""
) )
transition_domain.organization_name = org_info.orgname transition_domain.organization_name = org_info.orgname
@ -478,9 +451,7 @@ class LoadExtraTransitionDomain:
return transition_domain return transition_domain
def _add_or_change_message( def _add_or_change_message(self, file_type, var_name, changed_value, domain_name, is_update=False):
self, file_type, var_name, changed_value, domain_name, is_update=False
):
"""Creates a log instance when a property """Creates a log instance when a property
is successfully changed on a given TransitionDomain.""" is successfully changed on a given TransitionDomain."""
if not is_update: if not is_update:
@ -868,9 +839,7 @@ class ExtraTransitionDomain:
else: else:
if not infer_filenames: if not infer_filenames:
raise FileNotFoundError( raise FileNotFoundError(
f"{TerminalColors.FAIL}" f"{TerminalColors.FAIL}" f"Could not find file {filename} for {name}" f"{TerminalColors.ENDC}"
f"Could not find file {filename} for {name}"
f"{TerminalColors.ENDC}"
) )
# Infer filename logic # # Infer filename logic #
@ -903,9 +872,7 @@ class ExtraTransitionDomain:
) )
continue continue
raise FileNotFoundError( raise FileNotFoundError(
f"{TerminalColors.FAIL}" f"{TerminalColors.FAIL}" f"Could not find file {filename} for {name}" f"{TerminalColors.ENDC}"
f"Could not find file {filename} for {name}"
f"{TerminalColors.ENDC}"
) )
def clear_file_data(self): def clear_file_data(self):
@ -913,17 +880,13 @@ class ExtraTransitionDomain:
file_type: FileDataHolder = item file_type: FileDataHolder = item
file_type.data = {} file_type.data = {}
def parse_csv_file( def parse_csv_file(self, file, seperator, dataclass_type, id_field, is_domain_escrow=False):
self, file, seperator, dataclass_type, id_field, is_domain_escrow=False
):
# Domain escrow is an edge case # Domain escrow is an edge case
if is_domain_escrow: if is_domain_escrow:
item_to_return = self._read_domain_escrow(file, seperator) item_to_return = self._read_domain_escrow(file, seperator)
return item_to_return return item_to_return
else: else:
item_to_return = self._read_csv_file( item_to_return = self._read_csv_file(file, seperator, dataclass_type, id_field)
file, seperator, dataclass_type, id_field
)
return item_to_return return item_to_return
# Domain escrow is an edgecase given that its structured differently data-wise. # Domain escrow is an edgecase given that its structured differently data-wise.
@ -938,9 +901,7 @@ class ExtraTransitionDomain:
creation_date = datetime.strptime(row[7], date_format) creation_date = datetime.strptime(row[7], date_format)
expiration_date = datetime.strptime(row[11], date_format) expiration_date = datetime.strptime(row[11], date_format)
dict_data[domain_name] = DomainEscrow( dict_data[domain_name] = DomainEscrow(domain_name, creation_date, expiration_date)
domain_name, creation_date, expiration_date
)
return dict_data return dict_data
def _grab_row_id(self, row, id_field, file, dataclass_type): def _grab_row_id(self, row, id_field, file, dataclass_type):
@ -973,9 +934,7 @@ class ExtraTransitionDomain:
f"Found bad data in {file}. Attempting to clean." f"Found bad data in {file}. Attempting to clean."
f"{TerminalColors.ENDC}" f"{TerminalColors.ENDC}"
) )
updated_file_content = self.replace_bad_seperators( updated_file_content = self.replace_bad_seperators(file, f"{seperator}", ";badseperator;")
file, f"{seperator}", ";badseperator;"
)
dict_data = {} dict_data = {}
break break
@ -989,11 +948,7 @@ class ExtraTransitionDomain:
# After we clean the data, try to parse it again # After we clean the data, try to parse it again
if updated_file_content: if updated_file_content:
logger.info( logger.info(f"{TerminalColors.MAGENTA}" f"Retrying load for {file}" f"{TerminalColors.ENDC}")
f"{TerminalColors.MAGENTA}"
f"Retrying load for {file}"
f"{TerminalColors.ENDC}"
)
# Store the file locally rather than writing to the file. # Store the file locally rather than writing to the file.
# This is to avoid potential data corruption. # This is to avoid potential data corruption.
updated_file = io.StringIO(updated_file_content) updated_file = io.StringIO(updated_file_content)
@ -1004,9 +959,7 @@ class ExtraTransitionDomain:
# is wrong with the file. # is wrong with the file.
if None in row: if None in row:
logger.error( logger.error(
f"{TerminalColors.FAIL}" f"{TerminalColors.FAIL}" f"Corrupt data found for {row_id}. Skipping." f"{TerminalColors.ENDC}"
f"Corrupt data found for {row_id}. Skipping."
f"{TerminalColors.ENDC}"
) )
continue continue

10
src/registrar/management/commands/utility/terminal_helper.py
View file

@ -193,9 +193,7 @@ class TerminalHelper:
return total_line return total_line
@staticmethod @staticmethod
def print_to_file_conditional( def print_to_file_conditional(print_condition: bool, filename: str, file_directory: str, file_contents: str):
print_condition: bool, filename: str, file_directory: str, file_contents: str
):
"""Sometimes logger outputs get insanely huge.""" """Sometimes logger outputs get insanely huge."""
if print_condition: if print_condition:
# Add a slash if the last character isn't one # Add a slash if the last character isn't one
@ -204,10 +202,6 @@ class TerminalHelper:
# Assemble filepath # Assemble filepath
filepath = f"{file_directory}{filename}.txt" filepath = f"{file_directory}{filename}.txt"
# Write to file # Write to file
logger.info( logger.info(f"{TerminalColors.MAGENTA}Writing to file " f" {filepath}..." f"{TerminalColors.ENDC}")
f"{TerminalColors.MAGENTA}Writing to file "
f" {filepath}..."
f"{TerminalColors.ENDC}"
)
with open(f"{filepath}", "w+") as f: with open(f"{filepath}", "w+") as f:
f.write(file_contents) f.write(file_contents)

24
src/registrar/management/commands/utility/transition_domain_arguments.py
View file

@ -37,26 +37,14 @@ class TransitionDomainArguments:
# Filenames # # Filenames #
# = Adhocs =# # = Adhocs =#
agency_adhoc_filename: Optional[str] = field( agency_adhoc_filename: Optional[str] = field(default=EnumFilenames.AGENCY_ADHOC.value[1], repr=True)
default=EnumFilenames.AGENCY_ADHOC.value[1], repr=True domain_adhoc_filename: Optional[str] = field(default=EnumFilenames.DOMAIN_ADHOC.value[1], repr=True)
) organization_adhoc_filename: Optional[str] = field(default=EnumFilenames.ORGANIZATION_ADHOC.value[1], repr=True)
domain_adhoc_filename: Optional[str] = field( authority_adhoc_filename: Optional[str] = field(default=EnumFilenames.AUTHORITY_ADHOC.value[1], repr=True)
default=EnumFilenames.DOMAIN_ADHOC.value[1], repr=True
)
organization_adhoc_filename: Optional[str] = field(
default=EnumFilenames.ORGANIZATION_ADHOC.value[1], repr=True
)
authority_adhoc_filename: Optional[str] = field(
default=EnumFilenames.AUTHORITY_ADHOC.value[1], repr=True
)
# = Data files =# # = Data files =#
domain_escrow_filename: Optional[str] = field( domain_escrow_filename: Optional[str] = field(default=EnumFilenames.DOMAIN_ESCROW.value[1], repr=True)
default=EnumFilenames.DOMAIN_ESCROW.value[1], repr=True domain_additional_filename: Optional[str] = field(default=EnumFilenames.DOMAIN_ADDITIONAL.value[1], repr=True)
)
domain_additional_filename: Optional[str] = field(
default=EnumFilenames.DOMAIN_ADDITIONAL.value[1], repr=True
)
domain_contacts_filename: Optional[str] = field(default=None, repr=True) domain_contacts_filename: Optional[str] = field(default=None, repr=True)
domain_statuses_filename: Optional[str] = field(default=None, repr=True) domain_statuses_filename: Optional[str] = field(default=None, repr=True)
contacts_filename: Optional[str] = field(default=None, repr=True) contacts_filename: Optional[str] = field(default=None, repr=True)

8
src/registrar/migrations/0045_transitiondomain_federal_agency_and_more.py
View file

@ -27,16 +27,12 @@ class Migration(migrations.Migration):
migrations.AddField( migrations.AddField(
model_name="transitiondomain", model_name="transitiondomain",
name="organization_type", name="organization_type",
field=models.TextField( field=models.TextField(blank=True, help_text="Type of organization", max_length=255, null=True),
blank=True, help_text="Type of organization", max_length=255, null=True
),
), ),
migrations.AddField( migrations.AddField(
model_name="transitiondomain", model_name="transitiondomain",
name="organization_name", name="organization_name",
field=models.TextField( field=models.TextField(blank=True, db_index=True, help_text="Organization name", null=True),
blank=True, db_index=True, help_text="Organization name", null=True
),
), ),
migrations.AddField( migrations.AddField(
model_name="transitiondomain", model_name="transitiondomain",

5
src/registrar/models/domain_application.py
View file

@ -166,10 +166,7 @@ class DomainApplication(TimeStampedModel):
"American Battle Monuments Commission", "American Battle Monuments Commission",
"AMTRAK", "AMTRAK",
"Appalachian Regional Commission", "Appalachian Regional Commission",
( ("Appraisal Subcommittee of the Federal Financial " "Institutions Examination Council"),
"Appraisal Subcommittee of the Federal Financial "
"Institutions Examination Council"
),
"Appraisal Subcommittee", "Appraisal Subcommittee",
"Architect of the Capitol", "Architect of the Capitol",
"Armed Forces Retirement Home", "Armed Forces Retirement Home",

8
src/registrar/models/transition_domain.py
View file

@ -67,15 +67,11 @@ class TransitionDomain(TimeStampedModel):
) )
epp_creation_date = models.DateField( epp_creation_date = models.DateField(
null=True, null=True,
help_text=( help_text=("Duplication of registry's creation " "date saved for ease of reporting"),
"Duplication of registry's creation " "date saved for ease of reporting"
),
) )
epp_expiration_date = models.DateField( epp_expiration_date = models.DateField(
null=True, null=True,
help_text=( help_text=("Duplication of registry's expiration " "date saved for ease of reporting"),
"Duplication of registry's expiration " "date saved for ease of reporting"
),
) )
def __str__(self): def __str__(self):

16
src/registrar/tests/test_transition_domain_migrations.py
View file

@ -103,9 +103,7 @@ class TestMigrations(TestCase):
# Check Domain table # Check Domain table
matching_domains = Domain.objects.filter(name=transition_domain_name) matching_domains = Domain.objects.filter(name=transition_domain_name)
# Check Domain Information table # Check Domain Information table
matching_domain_informations = DomainInformation.objects.filter( matching_domain_informations = DomainInformation.objects.filter(domain__name=transition_domain_name)
domain__name=transition_domain_name
)
# Check Domain Invitation table # Check Domain Invitation table
matching_domain_invitations = DomainInvitation.objects.filter( matching_domain_invitations = DomainInvitation.objects.filter(
email=transition_domain_email.lower(), email=transition_domain_email.lower(),
@ -146,12 +144,8 @@ class TestMigrations(TestCase):
) )
self.assertEqual(total_missing_domains, expected_missing_domains) self.assertEqual(total_missing_domains, expected_missing_domains)
self.assertEqual(total_duplicate_domains, expected_duplicate_domains) self.assertEqual(total_duplicate_domains, expected_duplicate_domains)
self.assertEqual( self.assertEqual(total_missing_domain_informations, expected_missing_domain_informations)
total_missing_domain_informations, expected_missing_domain_informations self.assertEqual(total_missing_domain_invitations, expected_missing_domain_invitations)
)
self.assertEqual(
total_missing_domain_invitations, expected_missing_domain_invitations
)
self.assertEqual(total_transition_domains, expected_total_transition_domains) self.assertEqual(total_transition_domains, expected_total_transition_domains)
self.assertEqual(total_domains, expected_total_domains) self.assertEqual(total_domains, expected_total_domains)
@ -352,9 +346,7 @@ class TestMigrations(TestCase):
# Simluate Logins # Simluate Logins
for invite in DomainInvitation.objects.all(): for invite in DomainInvitation.objects.all():
# get a user with this email address # get a user with this email address
user, user_created = User.objects.get_or_create( user, user_created = User.objects.get_or_create(email=invite.email, username=invite.email)
email=invite.email, username=invite.email
)
user.on_each_login() user.on_each_login()
# Analyze the tables # Analyze the tables