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

added logic to bypass prompts if debug is enabled

Browse source
This commit is contained in:
CocoByte 2023-10-24 18:37:58 -06:00
parent a7bd6709fa
commit d5e9e45a2f
No known key found for this signature in database
GPG key ID: BBFAA2526384C97F
1 changed files with 112 additions and 117 deletions
Download patch file Download diff file Expand all files Collapse all files

133
src/registrar/management/commands/test_domain_migration.py
View file

@ -237,13 +237,18 @@ class Command(BaseCommand):
if not proceed_execution: if not proceed_execution:
sys.exit() sys.exit()
self.execute_command(command_string)
return True
def execute_command(self, command_string:str):
"""Executes the given command string"""
logger.info(f"""{TerminalColors.OKCYAN} logger.info(f"""{TerminalColors.OKCYAN}
==== EXECUTING... ==== ==== EXECUTING... ====
{TerminalColors.ENDC}""") {TerminalColors.ENDC}""")
os.system(f"{command_string}") os.system(f"{command_string}")
return True
def run_load_transition_domain_script(self, def run_load_transition_domain_script(self,
file_location: str, file_location: str,
domain_contacts_filename: str, domain_contacts_filename: str,
@ -252,6 +257,7 @@ class Command(BaseCommand):
sep: str, sep: str,
reset_table: bool, reset_table: bool,
debug_on: bool, debug_on: bool,
prompts_enabled: bool,
debug_max_entries_to_parse: int): debug_max_entries_to_parse: int):
"""Runs the load_transition_domain script""" """Runs the load_transition_domain script"""
# Create the command string # Create the command string
@ -269,33 +275,51 @@ class Command(BaseCommand):
command_string += f"--limitParse {debug_max_entries_to_parse} " command_string += f"--limitParse {debug_max_entries_to_parse} "
# Execute the command string # Execute the command string
if prompts_enabled:
self.prompt_for_execution(command_string, "Running load_transition_domain script") self.prompt_for_execution(command_string, "Running load_transition_domain script")
return
self.execute_command(command_string)
def run_transfer_script(self, debug_on:bool): def run_transfer_script(self, debug_on:bool, prompts_enabled: bool):
"""Runs the transfer_transition_domains_to_domains script""" """Runs the transfer_transition_domains_to_domains script"""
# Create the command string # Create the command string
command_string = "./manage.py transfer_transition_domains_to_domains " command_string = "./manage.py transfer_transition_domains_to_domains "
if debug_on: if debug_on:
command_string += "--debug " command_string += "--debug "
# Execute the command string # Execute the command string
if prompts_enabled:
self.prompt_for_execution(command_string, "Running transfer_transition_domains_to_domains script") self.prompt_for_execution(command_string, "Running transfer_transition_domains_to_domains script")
return
self.execute_command(command_string)
def run_send_invites_script(self): def run_send_invites_script(self, debug_on: bool, prompts_enabled: bool):
"""Runs the send_domain_invitations script""" """Runs the send_domain_invitations script"""
# Create the command string... # Create the command string...
command_string = "./manage.py send_domain_invitations -s" command_string = "./manage.py send_domain_invitations -s"
# Execute the command string # Execute the command string
if prompts_enabled:
self.prompt_for_execution(command_string, "Running send_domain_invitations script") self.prompt_for_execution(command_string, "Running send_domain_invitations script")
return
self.execute_command(command_string)
def run_migration_scripts(self, def run_migration_scripts(self,
prompts_enabled: bool,
options): options):
"""Runs the following migration scripts (in order): """Runs the following migration scripts (in order):
1 - imports for trans domains 1 - imports for trans domains
2 - transfer to domain & domain invitation""" 2 - transfer to domain & domain invitation"""
# Get arguments
sep = options.get("sep")
reset_table = options.get("resetTable")
debug_on = options.get("debug")
debug_max_entries_to_parse = int(
options.get("limitParse")
)
# Grab filepath information from the arguments # Grab filepath information from the arguments
file_location = options.get("loaderDirectory")+"/" file_location = options.get("loaderDirectory")+"/"
filenames = options.get("loaderFilenames").split() filenames = options.get("loaderFilenames").split()
@ -311,11 +335,12 @@ class Command(BaseCommand):
============= TERMINATING ============= ============= TERMINATING =============
{TerminalColors.ENDC} {TerminalColors.ENDC}
""") """)
return sys.exit()
domain_contacts_filename = filenames[0] domain_contacts_filename = filenames[0]
contacts_filename = filenames[1] contacts_filename = filenames[1]
domain_statuses_filename = filenames[2] domain_statuses_filename = filenames[2]
if prompts_enabled:
# Allow the user to inspect the filepath # Allow the user to inspect the filepath
# data given in the arguments, and prompt # data given in the arguments, and prompt
# the user to verify this info before proceeding # the user to verify this info before proceeding
@ -351,16 +376,7 @@ class Command(BaseCommand):
""") """)
return return
# Proceed executing the migration scripts... # Proceed executing the migration scripts
# ...First, Get all the arguments
sep = options.get("sep")
reset_table = options.get("resetTable")
debug_on = options.get("debug")
debug_max_entries_to_parse = int(
options.get("limitParse")
)
#...Second, Run the migration scripts in order
self.run_load_transition_domain_script(file_location, self.run_load_transition_domain_script(file_location,
domain_contacts_filename, domain_contacts_filename,
contacts_filename, contacts_filename,
@ -368,8 +384,9 @@ class Command(BaseCommand):
sep, sep,
reset_table, reset_table,
debug_on, debug_on,
prompts_enabled,
debug_max_entries_to_parse) debug_max_entries_to_parse)
self.run_transfer_script(debug_on) self.run_transfer_script(debug_on, prompts_enabled)
def simulate_user_logins(self, debug_on): def simulate_user_logins(self, debug_on):
@ -380,52 +397,27 @@ class Command(BaseCommand):
f"{TerminalColors.OKCYAN}" f"{TerminalColors.OKCYAN}"
f"================== SIMULATING LOGINS ==================" f"================== SIMULATING LOGINS =================="
f"{TerminalColors.ENDC}") f"{TerminalColors.ENDC}")
for invite in DomainInvitation.objects.all(): #TODO: limit to our stuff
#DEBUG:
TerminalHelper.print_conditional(debug_on,
f"{TerminalColors.OKCYAN}"
f"Processing invite: {invite}"
f"{TerminalColors.ENDC}")
# get a user with this email address
user, user_created = User.objects.get_or_create(email=invite.email, username=invite.email)
#DEBUG:
TerminalHelper.print_conditional(user_created,
f"""{TerminalColors.OKCYAN}No user found (creating temporary user object){TerminalColors.ENDC}""")
TerminalHelper.print_conditional(debug_on,
f"""{TerminalColors.OKCYAN}Executing first-time login for user: {user}{TerminalColors.ENDC}""")
user.first_login()
if user_created:
logger.info(f"""{TerminalColors.YELLOW}(Deleting temporary user object){TerminalColors.ENDC}""")
user.delete()
# get a user with this email address
# user_exists = User.objects.filter(email=invite.email).exists()
# if user_exists: # for invite in DomainInvitation.objects.all(): #TODO: limit to our stuff
# user = User.objects.get(email=invite.email) # #DEBUG:
# TerminalHelper.print_conditional(debug_on, # TerminalHelper.print_conditional(debug_on,
# f"""{TerminalColors.OKCYAN}Logging in user: {user}{TerminalColors.ENDC}""") # f"{TerminalColors.OKCYAN}"
# user.first_login() # f"Processing invite: {invite}"
# else: # f"{TerminalColors.ENDC}")
# logger.info(f"""{TerminalColors.YELLOW}No user found -- creating temp user object...{TerminalColors.ENDC}""") # # get a user with this email address
# user = User(email=invite.email) # user, user_created = User.objects.get_or_create(email=invite.email, username=invite.email)
# user.save() # #DEBUG:
# TerminalHelper.print_conditional(user_created,
# f"""{TerminalColors.OKCYAN}No user found (creating temporary user object){TerminalColors.ENDC}""")
# TerminalHelper.print_conditional(debug_on,
# f"""{TerminalColors.OKCYAN}Executing first-time login for user: {user}{TerminalColors.ENDC}""")
# user.first_login() # user.first_login()
# if user_created:
# logger.info(f"""{TerminalColors.YELLOW}(Deleting temporary user object){TerminalColors.ENDC}""") # logger.info(f"""{TerminalColors.YELLOW}(Deleting temporary user object){TerminalColors.ENDC}""")
# user.delete() # user.delete()
# for invite in DomainInvitation.objects.all():
# #DEBUG:
# TerminalHelper.print_debug(debug_on,f"""{TerminalColors.OKCYAN}Processing invite: {invite}{TerminalColors.ENDC}""")
# # get a user with this email address
# User = get_user_model()
# try:
# user = User.objects.get(email=invite.email)
# #DEBUG:
# TerminalHelper.print_debug(debug_on,f"""{TerminalColors.OKCYAN}Logging in user: {user}{TerminalColors.ENDC}""")
# Client.force_login(user)
# except User.DoesNotExist:
# #TODO: how should we handle this?
# logger.warn(f"""{TerminalColors.FAIL}No user found {invite.email}{TerminalColors.ENDC}""")
def handle( def handle(
self, self,
@ -449,7 +441,8 @@ class Command(BaseCommand):
# the terminal so the user knows what is # the terminal so the user knows what is
# enabled. # enabled.
debug_on = options.get("debug") debug_on = options.get("debug")
run_loaders_on = options.get("runLoaders") prompts_enabled = debug_on #TODO: add as argument?
run_loaders_enabled = options.get("runLoaders")
simulate_user_login_enabled = options.get("triggerLogins") simulate_user_login_enabled = options.get("triggerLogins")
TerminalHelper.print_conditional( TerminalHelper.print_conditional(
debug_on, debug_on,
@ -460,7 +453,7 @@ class Command(BaseCommand):
""" """
) )
TerminalHelper.print_conditional( TerminalHelper.print_conditional(
run_loaders_on, run_loaders_enabled,
f"""{TerminalColors.OKCYAN} f"""{TerminalColors.OKCYAN}
----------RUNNING LOADERS ON---------- ----------RUNNING LOADERS ON----------
All migration scripts will be run before All migration scripts will be run before
@ -469,7 +462,7 @@ class Command(BaseCommand):
""" """
) )
TerminalHelper.print_conditional( TerminalHelper.print_conditional(
run_loaders_on, run_loaders_enabled,
f"""{TerminalColors.OKCYAN} f"""{TerminalColors.OKCYAN}
----------TRIGGER LOGINS ON---------- ----------TRIGGER LOGINS ON----------
Will be simulating user logins Will be simulating user logins
@ -489,8 +482,8 @@ class Command(BaseCommand):
# STEP 1 -- RUN LOADERS # STEP 1 -- RUN LOADERS
# Run migration scripts if specified by user # Run migration scripts if specified by user
if run_loaders_on: if run_loaders_enabled:
self.run_migration_scripts(options) self.run_migration_scripts(options, prompts_enabled)
prompt_continuation_of_analysis = True prompt_continuation_of_analysis = True
# STEP 2 -- SIMULATE LOGINS # STEP 2 -- SIMULATE LOGINS
@ -503,32 +496,34 @@ class Command(BaseCommand):
# automatically execute this as the final step # automatically execute this as the final step
# to ensure Domain Information objects get added # to ensure Domain Information objects get added
# to the database.) # to the database.)
if run_loaders_on: if run_loaders_enabled:
if prompts_enabled:
simulate_user_login_enabled = TerminalHelper.query_yes_no( simulate_user_login_enabled = TerminalHelper.query_yes_no(
f"""{TerminalColors.FAIL} f"""{TerminalColors.FAIL}
Proceed with simulating user logins? Proceed with simulating user logins?
{TerminalColors.ENDC}""" {TerminalColors.ENDC}"""
) )
if simulate_user_login_enabled: if not simulate_user_login_enabled:
return
self.simulate_user_logins(debug_on) self.simulate_user_logins(debug_on)
prompt_continuation_of_analysis = True prompt_continuation_of_analysis = True
# STEP 3 -- SEND INVITES # STEP 3 -- SEND INVITES
if prompts_enabled:
proceed_with_sending_invites = TerminalHelper.query_yes_no( proceed_with_sending_invites = TerminalHelper.query_yes_no(
f"""{TerminalColors.FAIL} f"""{TerminalColors.FAIL}
Proceed with sending user invites? Proceed with sending user invites?
{TerminalColors.ENDC}""" {TerminalColors.ENDC}"""
) )
if proceed_with_sending_invites: if not proceed_with_sending_invites:
self.run_send_invites_script() return
self.run_send_invites_script(debug_on)
prompt_continuation_of_analysis = True prompt_continuation_of_analysis = True
# STEP 4 -- ANALYZE TABLES & GENERATE REPORT # STEP 4 -- ANALYZE TABLES & GENERATE REPORT
# Analyze tables for corrupt data... # Analyze tables for corrupt data...
if prompt_continuation_of_analysis & prompts_enabled:
# only prompt if we ran steps 1 and/or 2 # ^ (only prompt if we ran steps 1 and/or 2)
if prompt_continuation_of_analysis:
analyze_tables = TerminalHelper.query_yes_no( analyze_tables = TerminalHelper.query_yes_no(
f"""{TerminalColors.FAIL} f"""{TerminalColors.FAIL}
Proceed with table analysis? Proceed with table analysis?