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

Merge pull request #1205 from cisagov/nl/981-test-domain-migration-script

Browse source 
Issue 981: test domain migration script
This commit is contained in:
CuriousX 2023-11-06 15:04:00 -06:00 • committed by GitHub
commit 173d65dfcd
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
11 changed files with 1298 additions and 178 deletions
Download patch file Download diff file Expand all files Collapse all files

249
docs/operations/data_migration.md
View file

@ -78,32 +78,151 @@ An example script using this technique is in
docker compose run app ./manage.py load_domain_invitations /app/escrow_domain_contacts.daily.dotgov.GOV.txt /app/escrow_contacts.daily.dotgov.GOV.txt
```
## Transition Domains
## Transition Domains (Part 1) - Setup Files for Import
We are provided with information about Transition Domains in 3 files:
FILE 1: **escrow_domain_contacts.daily.gov.GOV.txt** -> has the map of domain names to contact ID. Domains in this file will usually have 3 contacts each
FILE 2: **escrow_contacts.daily.gov.GOV.txt** -> has the mapping of contact id to contact email address (which is what we care about for sending domain invitations)
FILE 3: **escrow_domain_statuses.daily.gov.GOV.txt** -> has the map of domains and their statuses
Transferring this data from these files into our domain tables happens in two steps;
- FILE 1: **escrow_domain_contacts.daily.gov.GOV.txt** -> has the map of domain names to contact ID. Domains in this file will usually have 3 contacts each
- FILE 2: **escrow_contacts.daily.gov.GOV.txt** -> has the mapping of contact id to contact email address (which is what we care about for sending domain invitations)
- FILE 3: **escrow_domain_statuses.daily.gov.GOV.txt** -> has the map of domains and their statuses
***IMPORTANT: only run the following locally, to avoid publicizing PII in our public repo.***
We need to run a few scripts to parse these files into our domain tables.
We can do this both locally and in a sandbox.
### STEP 1: Load Transition Domain data into TransitionDomain table
### SECTION 1 - SANDBOX MIGRATION SETUP
Load migration data onto a production or sandbox environment
**SETUP**
In order to use the management command, we need to add the files to a folder under `src/`.
This will allow Docker to mount the files to a container (under `/app`) for our use.
**WARNING:** All files uploaded in this manner are temporary, i.e. they will be deleted when the app is restaged.
Do not use these environments to store data you want to keep around permanently. We don't want sensitive data to be accidentally present in our application environments.
- Create a folder called `tmp` underneath `src/`
- Add the above files to this folder
- Open a terminal and navigate to `src/`
#### STEP 1: Using cat to transfer data to sandboxes
Then run the following command (This will parse the three files in your `tmp` folder and load the information into the TransitionDomain table);
```shell
docker compose run -T app ./manage.py load_transition_domain /app/tmp/escrow_domain_contacts.daily.gov.GOV.txt /app/tmp/escrow_contacts.daily.gov.GOV.txt /app/tmp/escrow_domain_statuses.daily.gov.GOV.txt
```bash
cat {LOCAL_PATH_TO_FILE} | cf ssh {APP_NAME_IN_ENVIRONMENT} -c "cat > /home/vcap/tmp/{DESIRED_NAME_OF_FILE}"
```
**OPTIONAL COMMAND LINE ARGUMENTS**:
* APP_NAME_IN_ENVIRONMENT - Name of the app running in your environment, e.g. getgov-za or getgov-stable
* LOCAL_PATH_TO_FILE - Path to the file you want to copy, ex: src/tmp/escrow_contacts.daily.gov.GOV.txt
* DESIRED_NAME_OF_FILE - Use this to specify the filename and type, ex: test.txt or escrow_contacts.daily.gov.GOV.txt
**TROUBLESHOOTING:** Depending on your operating system (Windows for instance), this command may upload corrupt data. If you encounter the error `gzip: prfiles.tar.gz: not in gzip format` when trying to unzip a .tar.gz file, use the scp command instead.
#### STEP 1 (Alternative): Using scp to transfer data to sandboxes
**IMPORTANT:** Only follow these steps if cat does not work as expected. If it does, skip to step 2.
CloudFoundry supports scp as means of transferring data locally to our environment. If you are dealing with a batch of files, try sending across a tar.gz and unpacking that.
##### Login to Cloud.gov
```bash
cf login -a api.fr.cloud.gov --sso
```
##### Target your workspace
```bash
cf target -o cisa-dotgov -s {ENVIRONMENT_NAME}
```
*ENVIRONMENT_NAME* - Name of your sandbox, ex: za or ab
##### Run the scp command
Use the following command to transfer the desired file:
```shell
scp -P 2222 -o User=cf:$(cf curl /v3/apps/$(cf app {APP_NAME_IN_ENVIRONMENT} --guid)/processes | jq -r '.resources[]
| select(.type=="web") | .guid')/0 {LOCAL_PATH_TO_FILE} ssh.fr.cloud.gov:tmp/{DESIRED_NAME_OF_FILE}
```
The items in curly braces are the values that you will manually replace.
These are as follows:
* APP_NAME_IN_ENVIRONMENT - Name of the app running in your environment, e.g. getgov-za or getgov-stable
* LOCAL_PATH_TO_FILE - Path to the file you want to copy, ex: src/tmp/escrow_contacts.daily.gov.GOV.txt
* DESIRED_NAME_OF_FILE - Use this to specify the filename and type, ex: test.txt or escrow_contacts.daily.gov.GOV.txt
##### Get a temp auth code
The scp command requires a temporary authentication code. Open a new terminal instance (while keeping the current one open),
and enter the following command:
```shell
cf ssh-code
```
Copy this code into the password prompt from earlier.
NOTE: You can use different utilities to copy this onto the clipboard for you. If you are on Windows, try the command `cf ssh-code | clip`. On Mac, this will be `cf ssh-code | pbcopy`
#### STEP 2: Transfer uploaded files to the getgov directory
Due to the nature of how Cloud.gov operates, the getgov directory is dynamically generated whenever the app is built under the tmp/ folder. We can directly upload files to the tmp/ folder but cannot target the generated getgov folder directly, as we need to spin up a shell to access this. From here, we can move those uploaded files into the getgov directory using the `cat` command. Note that you will have to repeat this for each file you want to move, so it is better to use a tar.gz for multiple, and unpack it inside of the `datamigration` folder.
##### SSH into your sandbox
```shell
cf ssh {APP_NAME_IN_ENVIRONMENT}
```
##### Open a shell
```shell
/tmp/lifecycle/shell
```
##### From this directory, run the following command:
```shell
./manage.py cat_files_into_getgov --file_extension txt
```
NOTE: This will look for all files in /tmp with the .txt extension, but this can
be changed if you are dealing with different extensions. For instance, a .tar.gz could be expressed
as `--file_extension tar.gz`.
If you are using a tar.gz file, you will need to perform one additional step to extract it.
Run the following command from the same directory:
```shell
tar -xvf migrationdata/{FILE_NAME}.tar.gz -C migrationdata/ --strip-components=1
```
*FILE_NAME* - Name of the desired file, ex: exportdata
#### Manual method
If the `cat_files_into_getgov.py` script isn't working, follow these steps instead.
##### Move the desired file into the correct directory
```shell
cat ../tmp/{filename} > migrationdata/{filename}
```
*You are now ready to run migration scripts (see [Running the Migration Scripts](running-the-migration-scripts))*
### SECTION 2 - LOCAL MIGRATION SETUP (TESTING PURPOSES ONLY)
***IMPORTANT: only use test data, to avoid publicizing PII in our public repo.***
In order to run the scripts locally, we need to add the files to a folder under `src/`.
This will allow Docker to mount the files to a container (under `/app`) for our use.
- Add the above files to the `migrationdata/` folder
- Open a terminal and navigate to `src/`
*You are now ready to run migration scripts.*
## Transition Domains (Part 2) - Running the Migration Scripts
### STEP 1: Load Transition Domains
Run the following command, making sure the file paths point to the right location. This will parse the three given files and load the information into the TransitionDomain table.
(NOTE: If working in cloud.gov, change "/app/tmp" to point to the `migrationdata/` directory and and remove "docker compose run -T app" from the command)
```shell
docker compose run -T app ./manage.py load_transition_domain /app/tmp/escrow_domain_contacts.daily.gov.GOV.txt /app/tmp/escrow_contacts.daily.gov.GOV.txt /app/tmp/escrow_domain_statuses.daily.gov.GOV.txt --debug
```
##### COMMAND LINE ARGUMENTS:
`--debug`
This will print out additional, detailed logs.
@ -111,22 +230,108 @@ This will print out additional, detailed logs.
Directs the script to load only the first 100 entries into the table. You can adjust this number as needed for testing purposes.
`--resetTable`
This will delete all the data loaded into transtion_domain. It is helpful if you want to see the entries reload from scratch or for clearing test data.
This will delete all the data in transtion_domain. It is helpful if you want to see the entries reload from scratch or for clearing test data.
### STEP 2: Transfer Transition Domain data into main Domain tables
Now that we've loaded all the data into TransitionDomain, we need to update the main Domain and DomainInvitation tables with this information.
In the same terminal as used in STEP 1, run the command below;
(This will parse the data in TransitionDomain and either create a corresponding Domain object, OR, if a corresponding Domain already exists, it will update that Domain with the incoming status. It will also create DomainInvitation objects for each user associated with the domain):
(NOTE: If working in cloud.gov, and remove "docker compose run -T app" from the command)
```shell
docker compose run -T app ./manage.py transfer_transition_domains_to_domains
docker compose run -T app ./manage.py transfer_transition_domains_to_domains --debug
```
**OPTIONAL COMMAND LINE ARGUMENTS**:
##### COMMAND LINE ARGUMENTS:
`--debug`
This will print out additional, detailed logs.
`--limitParse 100`
Directs the script to load only the first 100 entries into the table. You can adjust this number as needed for testing purposes.
Directs the script to load only the first 100 entries into the table. You can adjust this number as needed for testing purposes.
### STEP 3: Send Domain invitations
To send invitation emails for every transition domain in the transition domain table, execute the following command:
(NOTE: If working in cloud.gov, and remove "docker compose run -T app" from the command)
```shell
docker compose run -T app ./manage.py send_domain_invitations -s
```
### STEP 4: Test the results (Run the analyzer script)
This script's main function is to scan the transition domain and domain tables for any anomalies. It produces a simple report of missing or duplicate data. NOTE: some missing data might be expected depending on the nature of our migrations so use best judgement when evaluating the results.
#### OPTION 1 - ANALYZE ONLY
To analyze our database without running migrations, execute the script without any optional arguments:
(NOTE: If working in cloud.gov, and remove "docker compose run -T app" from the command)
```shell
docker compose run -T app ./manage.py master_domain_migrations --debug
```
#### OPTION 2 - RUN MIGRATIONS FEATURE
To run the migrations again (all above migration steps) before analyzing, execute the following command (read the documentation on the terminal arguments below. Everything used by the migration scripts can also be passed into this script and will have the same effects). NOTE: --debug and --prompt allow you to step through the migration process and exit it after each step if you need to. It is recommended that you use these arguments when using the --runMigrations feature:
(NOTE: If working in cloud.gov, and remove "docker compose run -T app" from the command)
```shell
docker compose run -T app ./manage.py master_domain_migrations --runMigrations --debug --prompt
```
##### COMMAND LINE ARGUMENTS
`--runMigrations`
Runs all scripts (in sequence) for transition domain migrations
`--migrationDirectory`
The location of the files used for load_transition_domain migration script.
(default is "migrationdata" (This is the sandbox directory))
Example Usage:
*--migrationDirectory /app/tmp*
`--migrationFilenames`
The filenames used for load_transition_domain migration script.
Must appear *in oprder* and comma-delimited:
default is "escrow_domain_contacts.daily.gov.GOV.txt,escrow_contacts.daily.gov.GOV.txt,escrow_domain_statuses.daily.gov.GOV.txt"
where...
- domain_contacts_filename is the Data file with domain contact information
- contacts_filename is the Data file with contact information
- domain_statuses_filename is the Data file with domain status information
Example Usage:
*--migrationFilenames domain_contacts_filename.txt,contacts_filename.txt,domain_statuses_filename.txt*
`--sep`
Delimiter for the migration scripts to correctly parse the given text files.
(usually this can remain at default value of |)
`--debug`
Activates additional print statements
`--prompt`
Activates terminal prompts that allows the user to step through each portion of this script.
`--limitParse`
Used by the migration scripts (load_transition_domain) to set the limit for the
number of data entries to insert. Set to 0 (or just don't use this
argument) to parse every entry. This was provided primarily for testing
purposes
`--resetTable`
Used by the migration scripts to trigger a prompt for deleting all table entries.
Useful for testing purposes, but *use with caution*

8
src/migrationdata/README.md Normal file
View file

@ -0,0 +1,8 @@
## Purpose
Use this folder for storing files for the migration process. Should otherwise be empty on local dev environments unless necessary. This folder must exist due to the nature of how data is stored on cloud.gov and the nature of the data we want to send.
## How do I migrate registrar data?
This process is detailed in [data_migration.md](../../docs/operations/data_migration.md)
## What kind of files can I store here?
The intent is for PII data or otherwise, but this can exist in any format. Do note that the data contained in this file will be temporary, so after the app is restaged it will lose it. This is ideal for migration files as they write to our DB, but not for something you need to permanently hold onto.

56
src/registrar/management/commands/cat_files_into_getgov.py Normal file
View file

@ -0,0 +1,56 @@
"""Loads files from /tmp into our sandboxes"""
import glob
import logging
import os
import shutil
from django.core.management import BaseCommand
from registrar.management.commands.utility.terminal_helper import TerminalHelper
logger = logging.getLogger(__name__)
class Command(BaseCommand):
help = "Runs the cat command on files from /tmp into the getgov directory."
def add_arguments(self, parser):
"""Add our two filename arguments."""
parser.add_argument(
"--file_extension",
default="txt",
help="What file extensions to look for, like txt or gz",
)
parser.add_argument(
"--directory", default="migrationdata", help="Desired directory"
)
def handle(self, **options):
file_extension: str = options.get("file_extension").lstrip(".")
directory = options.get("directory")
helper = TerminalHelper()
# file_extension is always coerced as str, Truthy is OK to use here.
if not file_extension or not isinstance(file_extension, str):
raise ValueError(f"Invalid file extension '{file_extension}'")
matching_files = glob.glob(f"../tmp/*.{file_extension}")
if not matching_files:
logger.error(f"No files with the extension {file_extension} found")
return None
for src_file_path in matching_files:
filename = os.path.basename(src_file_path)
desired_file_path = os.path.join(directory, filename)
if os.path.exists(desired_file_path):
# For linter
prompt = "Do you want to replace it?"
replace = f"{desired_file_path} already exists. {prompt}"
if not helper.query_yes_no(replace):
continue
src_file_path = f"../tmp/{filename}"
shutil.copy(src_file_path, desired_file_path)

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

@ -9,56 +9,14 @@ from django.core.management import BaseCommand
from registrar.models import TransitionDomain
from registrar.management.commands.utility.terminal_helper import (
TerminalColors,
TerminalHelper,
)
logger = logging.getLogger(__name__)
class termColors:
"""Colors for terminal outputs
(makes reading the logs WAY easier)"""
HEADER = "\033[95m"
OKBLUE = "\033[94m"
OKCYAN = "\033[96m"
OKGREEN = "\033[92m"
YELLOW = "\033[93m"
FAIL = "\033[91m"
ENDC = "\033[0m"
BOLD = "\033[1m"
UNDERLINE = "\033[4m"
BackgroundLightYellow = "\033[103m"
def query_yes_no(question: str, default="yes") -> bool:
"""Ask a yes/no question via raw_input() and return their answer.
"question" is a string that is presented to the user.
"default" is the presumed answer if the user just hits <Enter>.
It must be "yes" (the default), "no" or None (meaning
an answer is required of the user).
The "answer" return value is True for "yes" or False for "no".
"""
valid = {"yes": True, "y": True, "ye": True, "no": False, "n": False}
if default is None:
prompt = " [y/n] "
elif default == "yes":
prompt = " [Y/n] "
elif default == "no":
prompt = " [y/N] "
else:
raise ValueError("invalid default answer: '%s'" % default)
while True:
logger.info(question + prompt)
choice = input().lower()
if default is not None and choice == "":
return valid[default]
elif choice in valid:
return valid[choice]
else:
logger.info("Please respond with 'yes' or 'no' " "(or 'y' or 'n').\n")
class Command(BaseCommand):
help = """Loads data for domains that are in transition
(populates transition_domain model objects)."""
@ -110,20 +68,20 @@ class Command(BaseCommand):
or --limitParse are in use"""
if debug_on:
logger.info(
f"""{termColors.OKCYAN}
f"""{TerminalColors.OKCYAN}
----------DEBUG MODE ON----------
Detailed print statements activated.
{termColors.ENDC}
{TerminalColors.ENDC}
"""
)
if debug_max_entries_to_parse > 0:
logger.info(
f"""{termColors.OKCYAN}
f"""{TerminalColors.OKCYAN}
----------LIMITER ON----------
Parsing of entries will be limited to
{debug_max_entries_to_parse} lines per file.")
Detailed print statements activated.
{termColors.ENDC}
{TerminalColors.ENDC}
"""
)
@ -146,7 +104,7 @@ class Command(BaseCommand):
) -> defaultdict[str, str]:
"""Creates mapping of userId -> emails"""
user_emails_dictionary = defaultdict(str)
logger.info("Reading domain-contacts data file %s", contacts_filename)
logger.info("Reading contacts data file %s", contacts_filename)
with open(contacts_filename, "r") as contacts_file:
for row in csv.reader(contacts_file, delimiter=sep):
user_id = row[0]
@ -195,7 +153,7 @@ class Command(BaseCommand):
", ".join(map(str, duplicate_domain_user_combos))
)
logger.warning(
f"{termColors.YELLOW} No e-mails found for users: {users_without_email_as_string}" # noqa
f"{TerminalColors.YELLOW} No e-mails found for users: {users_without_email_as_string}" # noqa
)
if total_duplicate_pairs > 0 or total_duplicate_domains > 0:
duplicate_pairs_as_string = "{}".format(
@ -205,7 +163,7 @@ class Command(BaseCommand):
", ".join(map(str, duplicate_domains))
)
logger.warning(
f"""{termColors.YELLOW}
f"""{TerminalColors.YELLOW}
----DUPLICATES FOUND-----
@ -218,7 +176,7 @@ class Command(BaseCommand):
the supplied data files;
{duplicate_domains_as_string}
{termColors.ENDC}"""
{TerminalColors.ENDC}"""
)
def print_summary_status_findings(
@ -237,7 +195,7 @@ class Command(BaseCommand):
", ".join(map(str, domains_without_status))
)
logger.warning(
f"""{termColors.YELLOW}
f"""{TerminalColors.YELLOW}
--------------------------------------------
Found {total_domains_without_status} domains
@ -245,7 +203,7 @@ class Command(BaseCommand):
---------------------------------------------
{domains_without_status_as_string}
{termColors.ENDC}"""
{TerminalColors.ENDC}"""
)
if total_outlier_statuses > 0:
@ -253,7 +211,7 @@ class Command(BaseCommand):
", ".join(map(str, outlier_statuses))
) # noqa
logger.warning(
f"""{termColors.YELLOW}
f"""{TerminalColors.YELLOW}
--------------------------------------------
Found {total_outlier_statuses} unaccounted
@ -264,36 +222,27 @@ class Command(BaseCommand):
(defaulted to Ready):
{domains_without_status_as_string}
{termColors.ENDC}"""
{TerminalColors.ENDC}"""
)
def print_debug(self, print_condition: bool, print_statement: str):
"""This function reduces complexity of debug statements
in other functions.
It uses the logger to write the given print_statement to the
terminal if print_condition is TRUE"""
# DEBUG:
if print_condition:
logger.info(print_statement)
def prompt_table_reset(self):
"""Brings up a prompt in the terminal asking
if the user wishes to delete data in the
TransitionDomain table. If the user confirms,
deletes all the data in the TransitionDomain table"""
confirm_reset = query_yes_no(
confirm_reset = TerminalHelper.query_yes_no(
f"""
{termColors.FAIL}
{TerminalColors.FAIL}
WARNING: Resetting the table will permanently delete all
the data!
Are you sure you want to continue?{termColors.ENDC}"""
Are you sure you want to continue?{TerminalColors.ENDC}"""
)
if confirm_reset:
logger.info(
f"""{termColors.YELLOW}
f"""{TerminalColors.YELLOW}
----------Clearing Table Data----------
(please wait)
{termColors.ENDC}"""
{TerminalColors.ENDC}"""
)
TransitionDomain.objects.all().delete()
@ -428,18 +377,18 @@ class Command(BaseCommand):
)
if existing_domain is not None:
# DEBUG:
self.print_debug(
TerminalHelper.print_conditional(
debug_on,
f"{termColors.YELLOW} DUPLICATE file entries found for domain: {new_entry_domain_name} {termColors.ENDC}", # noqa
f"{TerminalColors.YELLOW} DUPLICATE file entries found for domain: {new_entry_domain_name} {TerminalColors.ENDC}", # noqa
)
if new_entry_domain_name not in duplicate_domains:
duplicate_domains.append(new_entry_domain_name)
if existing_domain_user_pair is not None:
# DEBUG:
self.print_debug(
TerminalHelper.print_conditional(
debug_on,
f"""{termColors.YELLOW} DUPLICATE file entries found for domain - user {termColors.BackgroundLightYellow} PAIR {termColors.ENDC}{termColors.YELLOW}:
{new_entry_domain_name} - {new_entry_email} {termColors.ENDC}""", # noqa
f"""{TerminalColors.YELLOW} DUPLICATE file entries found for domain - user {TerminalColors.BackgroundLightYellow} PAIR {TerminalColors.ENDC}{TerminalColors.YELLOW}:
{new_entry_domain_name} - {new_entry_email} {TerminalColors.ENDC}""", # noqa
)
if existing_domain_user_pair not in duplicate_domain_user_combos:
duplicate_domain_user_combos.append(existing_domain_user_pair)
@ -456,20 +405,20 @@ class Command(BaseCommand):
if existing_entry.status != new_entry_status:
# DEBUG:
self.print_debug(
TerminalHelper.print_conditional(
debug_on,
f"{termColors.OKCYAN}"
f"{TerminalColors.OKCYAN}"
f"Updating entry: {existing_entry}"
f"Status: {existing_entry.status} > {new_entry_status}" # noqa
f"Email Sent: {existing_entry.email_sent} > {new_entry_emailSent}" # noqa
f"{termColors.ENDC}",
f"{TerminalColors.ENDC}",
)
existing_entry.status = new_entry_status
existing_entry.email_sent = new_entry_emailSent
existing_entry.save()
except TransitionDomain.MultipleObjectsReturned:
logger.info(
f"{termColors.FAIL}"
f"{TerminalColors.FAIL}"
f"!!! ERROR: duplicate entries exist in the"
f"transtion_domain table for domain:"
f"{new_entry_domain_name}"
@ -488,9 +437,9 @@ class Command(BaseCommand):
total_new_entries += 1
# DEBUG:
self.print_debug(
TerminalHelper.print_conditional(
debug_on,
f"{termColors.OKCYAN} Adding entry {total_new_entries}: {new_entry} {termColors.ENDC}", # noqa
f"{TerminalColors.OKCYAN} Adding entry {total_new_entries}: {new_entry} {TerminalColors.ENDC}", # noqa
)
# Check Parse limit and exit loop if needed
@ -499,20 +448,20 @@ class Command(BaseCommand):
and debug_max_entries_to_parse != 0
):
logger.info(
f"{termColors.YELLOW}"
f"{TerminalColors.YELLOW}"
f"----PARSE LIMIT REACHED. HALTING PARSER.----"
f"{termColors.ENDC}"
f"{TerminalColors.ENDC}"
)
break
TransitionDomain.objects.bulk_create(to_create)
logger.info(
f"""{termColors.OKGREEN}
f"""{TerminalColors.OKGREEN}
============= FINISHED ===============
Created {total_new_entries} transition domain entries,
updated {total_updated_domain_entries} transition domain entries
{termColors.ENDC}
{TerminalColors.ENDC}
"""
)

540
src/registrar/management/commands/master_domain_migrations.py Normal file
View file

@ -0,0 +1,540 @@
"""Data migration:
1 - generates a report of data integrity across all
transition domain related tables
2 - allows users to run all migration scripts for
transition domain data
"""
import logging
import argparse
import sys
from django.core.management import BaseCommand
from django.core.management import call_command
from registrar.models import (
Domain,
DomainInformation,
DomainInvitation,
TransitionDomain,
)
from registrar.management.commands.utility.terminal_helper import (
TerminalColors,
TerminalHelper,
)
logger = logging.getLogger(__name__)
class Command(BaseCommand):
help = """ """
def add_arguments(self, parser):
"""
OPTIONAL ARGUMENTS:
--runMigrations
Triggers running all scripts (in sequence)
for transition domain migrations
--migrationDirectory
The location of the files used for load_transition_domain migration script
EXAMPLE USAGE:
> --migrationDirectory /app/tmp
--migrationFilenames
The files used for load_transition_domain migration script.
Must appear IN ORDER and comma-delimiteds:
EXAMPLE USAGE:
> --migrationFilenames domain_contacts_filename.txt,contacts_filename.txt,domain_statuses_filename.txt
where...
- domain_contacts_filename is the Data file with domain contact information
- contacts_filename is the Data file with contact information
- domain_statuses_filename is the Data file with domain status information
--sep
Delimiter for the migration scripts to correctly parse the given text files.
(usually this can remain at default value of |)
--debug
Activates additional print statements
--prompt
Activates terminal prompts that allows
the user to step through each portion of this
script.
--limitParse
Used by the migration scripts (load_transition_domain) to set the limit for the
number of data entries to insert. Set to 0 (or just don't use this
argument) to parse every entry. This was provided primarily for testing
purposes
--resetTable
Used by the migration scripts to trigger a prompt for deleting all table entries.
Useful for testing purposes, but USE WITH CAUTION
""" # noqa - line length, impacts readability
parser.add_argument(
"--runMigrations",
help="Runs all scripts (in sequence) for transition domain migrations",
action=argparse.BooleanOptionalAction,
)
# --triggerLogins
# A boolean (default to true), which triggers running
# simulations of user logins for each user in domain invitation
parser.add_argument(
"--triggerLogins",
help="Simulates a user login for each user in domain invitation",
action=argparse.BooleanOptionalAction,
)
# The following file arguments have default values for running in the sandbox
parser.add_argument(
"--migrationDirectory",
default="migrationdata",
help=(
"The location of the files used for"
"load_transition_domain migration script"
),
)
parser.add_argument(
"--migrationFilenames",
default="escrow_domain_contacts.daily.gov.GOV.txt,"
"escrow_contacts.daily.gov.GOV.txt,"
"escrow_domain_statuses.daily.gov.GOV.txt",
help="""The files used for load_transition_domain migration script.
Must appear IN ORDER and separated by commas:
domain_contacts_filename.txt,contacts_filename.txt,domain_statuses_filename.txt
where...
- domain_contacts_filename is the Data file with domain contact
information
- contacts_filename is the Data file with contact information
- domain_statuses_filename is the Data file with domain status
information""",
)
parser.add_argument(
"--sep", default="|", help="Delimiter character for the migration files"
)
parser.add_argument("--debug", action=argparse.BooleanOptionalAction)
parser.add_argument("--prompt", action=argparse.BooleanOptionalAction)
parser.add_argument(
"--limitParse", default=0, help="Sets max number of entries to load"
)
parser.add_argument(
"--resetTable",
help="Deletes all data in the TransitionDomain table",
action=argparse.BooleanOptionalAction,
)
def compare_tables(self, debug_on: bool):
"""Does a diff between the transition_domain and the following tables:
domain, domain_information and the domain_invitation.
Produces the following report (printed to the terminal):
#1 - Print any domains that exist in the transition_domain table
but not in their corresponding domain, domain information or
domain invitation tables.
#2 - Print which table this domain is missing from
#3- Check for duplicate entries in domain or
domain_information tables and print which are
duplicates and in which tables
"""
logger.info(
f"""{TerminalColors.OKCYAN}
============= BEGINNING ANALYSIS ===============
{TerminalColors.ENDC}
"""
)
# TODO: would filteredRelation be faster?
missing_domains = []
duplicate_domains = []
missing_domain_informations = []
missing_domain_invites = []
for transition_domain in TransitionDomain.objects.all(): # DEBUG:
transition_domain_name = transition_domain.domain_name
transition_domain_email = transition_domain.username
TerminalHelper.print_conditional(
debug_on,
f"{TerminalColors.OKCYAN}Checking: {transition_domain_name} {TerminalColors.ENDC}", # noqa
)
# Check Domain table
matching_domains = Domain.objects.filter(name=transition_domain_name)
# Check Domain Information table
matching_domain_informations = DomainInformation.objects.filter(
domain__name=transition_domain_name
)
# Check Domain Invitation table
matching_domain_invitations = DomainInvitation.objects.filter(
email=transition_domain_email.lower(),
domain__name=transition_domain_name,
)
if len(matching_domains) == 0:
TerminalHelper.print_conditional(
debug_on,
f"""{TerminalColors.YELLOW}Missing Domain{TerminalColors.ENDC}""",
)
missing_domains.append(transition_domain_name)
elif len(matching_domains) > 1:
TerminalHelper.print_conditional(
debug_on,
f"""{TerminalColors.YELLOW}Duplicate Domain{TerminalColors.ENDC}""",
)
duplicate_domains.append(transition_domain_name)
if len(matching_domain_informations) == 0:
TerminalHelper.print_conditional(
debug_on,
f"""{TerminalColors.YELLOW}Missing Domain Information
{TerminalColors.ENDC}""",
)
missing_domain_informations.append(transition_domain_name)
if len(matching_domain_invitations) == 0:
TerminalHelper.print_conditional(
debug_on,
f"""{TerminalColors.YELLOW}Missing Domain Invitation
{TerminalColors.ENDC}""",
)
missing_domain_invites.append(transition_domain_name)
total_missing_domains = len(missing_domains)
total_duplicate_domains = len(duplicate_domains)
total_missing_domain_informations = len(missing_domain_informations)
total_missing_domain_invitations = len(missing_domain_invites)
missing_domains_as_string = "{}".format(", ".join(map(str, missing_domains)))
duplicate_domains_as_string = "{}".format(
", ".join(map(str, duplicate_domains))
)
missing_domain_informations_as_string = "{}".format(
", ".join(map(str, missing_domain_informations))
)
missing_domain_invites_as_string = "{}".format(
", ".join(map(str, missing_domain_invites))
)
logger.info(
f"""{TerminalColors.OKGREEN}
============= FINISHED ANALYSIS ===============
{total_missing_domains} Missing Domains:
(These are transition domains that are missing from the Domain Table)
{TerminalColors.YELLOW}{missing_domains_as_string}
{TerminalColors.OKGREEN}
{total_duplicate_domains} Duplicate Domains:
(These are transition domains which have duplicate
entries in the Domain Table)
{TerminalColors.YELLOW}{duplicate_domains_as_string}
{TerminalColors.OKGREEN}
{total_missing_domain_informations} Domain Information Entries missing:
(These are transition domains which have no entries
in the Domain Information Table)
{TerminalColors.YELLOW}{missing_domain_informations_as_string}
{TerminalColors.OKGREEN}
{total_missing_domain_invitations} Domain Invitations missing:
(These are transition domains which have no entires in
the Domain Invitation Table)
{TerminalColors.YELLOW}{missing_domain_invites_as_string}
{TerminalColors.OKGREEN}
{TerminalColors.ENDC}
"""
)
def run_load_transition_domain_script(
self,
file_location: str,
domain_contacts_filename: str,
contacts_filename: str,
domain_statuses_filename: str,
sep: str,
reset_table: bool,
debug_on: bool,
prompts_enabled: bool,
debug_max_entries_to_parse: int,
):
"""Runs the load_transition_domain script"""
# Create the command string
command_script = "load_transition_domain"
command_string = (
f"./manage.py {command_script} "
f"{file_location+domain_contacts_filename} "
f"{file_location+contacts_filename} "
f"{file_location+domain_statuses_filename} "
)
if sep is not None and sep != "|":
command_string += f"--sep {sep} "
if reset_table:
command_string += "--resetTable "
if debug_on:
command_string += "--debug "
if debug_max_entries_to_parse > 0:
command_string += f"--limitParse {debug_max_entries_to_parse} "
# Execute the command string
if prompts_enabled:
system_exit_on_terminate = True
TerminalHelper.prompt_for_execution(
system_exit_on_terminate,
command_string,
"Running load_transition_domain script",
)
# TODO: make this somehow run inside TerminalHelper prompt
call_command(
command_script,
f"{file_location+domain_contacts_filename}",
f"{file_location+contacts_filename}",
f"{file_location+domain_statuses_filename}",
sep=sep,
resetTable=reset_table,
debug=debug_on,
limitParse=debug_max_entries_to_parse,
)
def run_transfer_script(self, debug_on: bool, prompts_enabled: bool):
"""Runs the transfer_transition_domains_to_domains script"""
# Create the command string
command_script = "transfer_transition_domains_to_domains"
command_string = f"./manage.py {command_script}"
if debug_on:
command_string += "--debug "
# Execute the command string
if prompts_enabled:
system_exit_on_terminate = True
TerminalHelper.prompt_for_execution(
system_exit_on_terminate,
command_string,
"Running transfer_transition_domains_to_domains script",
)
# TODO: make this somehow run inside TerminalHelper prompt
call_command(command_script)
def run_send_invites_script(self, debug_on: bool, prompts_enabled: bool):
"""Runs the send_domain_invitations script"""
# Create the command string...
command_script = "send_domain_invitations"
command_string = f"./manage.py {command_script} -s"
# Execute the command string
if prompts_enabled:
system_exit_on_terminate = True
TerminalHelper.prompt_for_execution(
system_exit_on_terminate,
command_string,
"Running send_domain_invitations script",
)
# TODO: make this somehow run inside TerminalHelper prompt
call_command(command_script, send_emails=True)
def run_migration_scripts(
self,
file_location,
domain_contacts_filename,
contacts_filename,
domain_statuses_filename,
sep,
reset_table,
debug_on,
prompts_enabled,
debug_max_entries_to_parse,
):
"""Runs the following migration scripts (in order):
1 - imports for trans domains
2 - transfer to domain & domain invitation"""
if prompts_enabled:
# Allow the user to inspect the filepath
# data given in the arguments, and prompt
# the user to verify this info before proceeding
files_are_correct = TerminalHelper.query_yes_no(
f"""
{TerminalColors.OKCYAN}
*** IMPORTANT: VERIFY THE FOLLOWING ***
The migration scripts are looking in directory....
{file_location}
....for the following files:
- domain contacts: {domain_contacts_filename}
- contacts: {contacts_filename}
- domain statuses: {domain_statuses_filename}
{TerminalColors.FAIL}
Does this look correct?{TerminalColors.ENDC}"""
)
# If the user rejected the filepath information
# as incorrect, prompt the user to provide
# correct file inputs in their original command
# prompt and exit this subroutine
if not files_are_correct:
logger.info(
f"""
{TerminalColors.YELLOW}
PLEASE Re-Run the script with the correct
file location and filenames:
"""
)
return
# Proceed executing the migration scripts
self.run_load_transition_domain_script(
file_location,
domain_contacts_filename,
contacts_filename,
domain_statuses_filename,
sep,
reset_table,
debug_on,
prompts_enabled,
debug_max_entries_to_parse,
)
self.run_transfer_script(debug_on, prompts_enabled)
def handle(
self,
**options,
):
"""
Does the following;
1 - run migration scripts
2 - simulate logins
3 - send domain invitations (Emails should be sent to the appropriate users
note that all moved domains should now be accessible
on django admin for an analyst)
4 - analyze the data for transition domains
and generate a report
"""
# SETUP
# Grab all arguments relevant to
# orchestrating which parts of this script
# should execute. Print some indicators to
# the terminal so the user knows what is
# enabled.
# Get arguments
debug_on = options.get("debug")
prompts_enabled = options.get("prompt")
run_migrations_enabled = options.get("runMigrations")
TerminalHelper.print_conditional(
debug_on,
f"""{TerminalColors.OKCYAN}
----------DEBUG MODE ON----------
Detailed print statements activated.
{TerminalColors.ENDC}
""",
)
TerminalHelper.print_conditional(
run_migrations_enabled,
f"""{TerminalColors.OKCYAN}
----------RUNNING MIGRATIONS ON----------
All migration scripts will be run before
analyzing the data.
{TerminalColors.ENDC}
""",
)
TerminalHelper.print_conditional(
run_migrations_enabled,
f"""{TerminalColors.OKCYAN}
----------TRIGGER LOGINS ON----------
Will be simulating user logins
{TerminalColors.ENDC}
""",
)
# If a user decides to run all migration
# scripts, they may or may not wish to
# proceed with analysis of the data depending
# on the results of the migration.
# Provide a breakpoint for them to decide
# whether to continue or not.
# The same will happen if simulating user
# logins (to allow users to run only that
# portion of the script if desired)
prompt_continuation_of_analysis = False
# STEP 1 -- RUN MIGRATIONS
# Run migration scripts if specified by user
if run_migrations_enabled:
# grab arguments for running migrations
sep = options.get("sep")
reset_table = options.get("resetTable")
debug_max_entries_to_parse = int(options.get("limitParse"))
# Grab filepath information from the arguments
file_location = options.get("migrationDirectory") + "/"
filenames = options.get("migrationFilenames").split(",")
if len(filenames) < 3:
filenames_as_string = "{}".format(", ".join(map(str, filenames)))
logger.info(
f"""
{TerminalColors.FAIL}
--migrationFilenames expected 3 filenames to follow it,
but only {len(filenames)} were given:
{filenames_as_string}
PLEASE MODIFY THE SCRIPT AND TRY RUNNING IT AGAIN
============= TERMINATING =============
{TerminalColors.ENDC}
"""
)
sys.exit()
domain_contacts_filename = filenames[0]
contacts_filename = filenames[1]
domain_statuses_filename = filenames[2]
# Run migration scripts
self.run_migration_scripts(
file_location,
domain_contacts_filename,
contacts_filename,
domain_statuses_filename,
sep,
reset_table,
debug_on,
prompts_enabled,
debug_max_entries_to_parse,
)
prompt_continuation_of_analysis = True
# STEP 2 -- SEND INVITES
proceed_with_sending_invites = run_migrations_enabled
if prompts_enabled and run_migrations_enabled:
proceed_with_sending_invites = TerminalHelper.query_yes_no(
f"""{TerminalColors.FAIL}
Proceed with sending user invites for all transition domains?
(Y = proceed, N = skip)
{TerminalColors.ENDC}"""
)
if proceed_with_sending_invites:
self.run_send_invites_script(debug_on, prompts_enabled)
prompt_continuation_of_analysis = True
# STEP 3 -- ANALYZE TABLES & GENERATE REPORT
# Analyze tables for corrupt data...
if prompt_continuation_of_analysis and prompts_enabled:
# ^ (only prompt if we ran steps 1 and/or 2)
analyze_tables = TerminalHelper.query_yes_no(
f"""{TerminalColors.FAIL}
Proceed with table analysis?
(Y = proceed, N = exit)
{TerminalColors.ENDC}"""
)
if not analyze_tables:
return
self.compare_tables(debug_on)

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

@ -10,25 +10,14 @@ from registrar.models import TransitionDomain
from registrar.models import Domain
from registrar.models import DomainInvitation
from registrar.management.commands.utility.terminal_helper import (
TerminalColors,
TerminalHelper,
)
logger = logging.getLogger(__name__)
class termColors:
"""Colors for terminal outputs
(makes reading the logs WAY easier)"""
HEADER = "\033[95m"
OKBLUE = "\033[94m"
OKCYAN = "\033[96m"
OKGREEN = "\033[92m"
YELLOW = "\033[93m"
FAIL = "\033[91m"
ENDC = "\033[0m"
BOLD = "\033[1m"
UNDERLINE = "\033[4m"
BackgroundLightYellow = "\033[103m"
class Command(BaseCommand):
help = """Load data from transition domain tables
into main domain tables. Also create domain invitation
@ -49,34 +38,25 @@ class Command(BaseCommand):
):
"""Prints additional terminal statements to indicate if --debug
or --limitParse are in use"""
self.print_debug(
TerminalHelper.print_conditional(
debug_on,
f"""{termColors.OKCYAN}
f"""{TerminalColors.OKCYAN}
----------DEBUG MODE ON----------
Detailed print statements activated.
{termColors.ENDC}
{TerminalColors.ENDC}
""",
)
self.print_debug(
TerminalHelper.print_conditional(
debug_max_entries_to_parse > 0,
f"""{termColors.OKCYAN}
f"""{TerminalColors.OKCYAN}
----------LIMITER ON----------
Parsing of entries will be limited to
{debug_max_entries_to_parse} lines per file.")
Detailed print statements activated.
{termColors.ENDC}
{TerminalColors.ENDC}
""",
)
def print_debug(self, print_condition: bool, print_statement: str):
"""This function reduces complexity of debug statements
in other functions.
It uses the logger to write the given print_statement to the
terminal if print_condition is TRUE"""
# DEBUG:
if print_condition:
logger.info(print_statement)
def update_domain_status(
self, transition_domain: TransitionDomain, target_domain: Domain, debug_on: bool
) -> bool:
@ -96,13 +76,13 @@ class Command(BaseCommand):
target_domain.save()
# DEBUG:
self.print_debug(
TerminalHelper.print_conditional(
debug_on,
f"""{termColors.YELLOW}
f"""{TerminalColors.YELLOW}
>> Updated {target_domain.name} state from
'{existing_status}' to '{target_domain.state}'
(no domain invitation entry added)
{termColors.ENDC}""",
{TerminalColors.ENDC}""",
)
return True
return False
@ -123,22 +103,22 @@ class Command(BaseCommand):
total_domain_invitation_entries = len(domain_invitations_to_create)
logger.info(
f"""{termColors.OKGREEN}
f"""{TerminalColors.OKGREEN}
============= FINISHED ===============
Created {total_new_entries} transition domain entries,
Updated {total_updated_domain_entries} transition domain entries
Created {total_new_entries} domain entries,
Updated {total_updated_domain_entries} domain entries
Created {total_domain_invitation_entries} domain invitation entries
(NOTE: no invitations are SENT in this script)
{termColors.ENDC}
{TerminalColors.ENDC}
"""
)
if len(skipped_domain_entries) > 0:
logger.info(
f"""{termColors.FAIL}
f"""{TerminalColors.FAIL}
============= SKIPPED DOMAINS (ERRORS) ===============
{skipped_domain_entries}
{termColors.ENDC}
{TerminalColors.ENDC}
"""
)
@ -151,25 +131,25 @@ class Command(BaseCommand):
skipped_domain_invitations.remove(domain_invite.domain)
if len(skipped_domain_invitations) > 0:
logger.info(
f"""{termColors.FAIL}
f"""{TerminalColors.FAIL}
============= SKIPPED DOMAIN INVITATIONS (ERRORS) ===============
{skipped_domain_invitations}
{termColors.ENDC}
{TerminalColors.ENDC}
"""
)
# DEBUG:
self.print_debug(
TerminalHelper.print_conditional(
debug_on,
f"""{termColors.YELLOW}
f"""{TerminalColors.YELLOW}
======= DEBUG OUTPUT =======
Created Domains:
{domains_to_create}
Updated Domains:
{updated_domain_entries}
{termColors.ENDC}
{TerminalColors.ENDC}
""",
)
@ -184,7 +164,7 @@ class Command(BaseCommand):
if associated_domain is None:
logger.warning(
f"""
{termColors.FAIL}
{TerminalColors.FAIL}
!!! ERROR: Domain cannot be null for a
Domain Invitation object!
@ -241,11 +221,11 @@ class Command(BaseCommand):
total_rows_parsed = 0
logger.info(
f"""{termColors.OKGREEN}
f"""{TerminalColors.OKGREEN}
==========================
Beginning Data Transfer
==========================
{termColors.ENDC}"""
{TerminalColors.ENDC}"""
)
for transition_domain in TransitionDomain.objects.all():
@ -254,11 +234,11 @@ class Command(BaseCommand):
transition_domain_email = transition_domain.username
# DEBUG:
self.print_debug(
TerminalHelper.print_conditional(
debug_on,
f"""{termColors.OKCYAN}
f"""{TerminalColors.OKCYAN}
Processing Transition Domain: {transition_domain_name}, {transition_domain_status}, {transition_domain_email}
{termColors.ENDC}""", # noqa
{TerminalColors.ENDC}""", # noqa
)
new_domain_invitation = None
@ -269,11 +249,11 @@ class Command(BaseCommand):
# get the existing domain
domain_to_update = Domain.objects.get(name=transition_domain_name)
# DEBUG:
self.print_debug(
TerminalHelper.print_conditional(
debug_on,
f"""{termColors.YELLOW}
f"""{TerminalColors.YELLOW}
> Found existing entry in Domain table for: {transition_domain_name}, {domain_to_update.state}
{termColors.ENDC}""", # noqa
{TerminalColors.ENDC}""", # noqa
)
# for existing entry, update the status to
@ -300,7 +280,7 @@ class Command(BaseCommand):
# immediate attention.
logger.warning(
f"""
{termColors.FAIL}
{TerminalColors.FAIL}
!!! ERROR: duplicate entries already exist in the
Domain table for the following domain:
{transition_domain_name}
@ -316,7 +296,7 @@ class Command(BaseCommand):
except TransitionNotAllowed as err:
skipped_domain_entries.append(transition_domain_name)
logger.warning(
f"""{termColors.FAIL}
f"""{TerminalColors.FAIL}
Unable to change state for {transition_domain_name}
RECOMMENDATION:
@ -343,15 +323,15 @@ class Command(BaseCommand):
None,
)
if existing_domain_in_to_create is not None:
self.print_debug(
TerminalHelper.print_conditional(
debug_on,
f"""{termColors.YELLOW}
f"""{TerminalColors.YELLOW}
Duplicate Detected: {transition_domain_name}.
Cannot add duplicate entry for another username.
Violates Unique Key constraint.
Checking for unique user e-mail for Domain Invitations...
{termColors.ENDC}""",
{TerminalColors.ENDC}""",
)
new_domain_invitation = self.try_add_domain_invitation(
transition_domain_email, existing_domain_in_to_create
@ -363,9 +343,9 @@ class Command(BaseCommand):
)
domains_to_create.append(new_domain)
# DEBUG:
self.print_debug(
TerminalHelper.print_conditional(
debug_on,
f"{termColors.OKCYAN} Adding domain: {new_domain} {termColors.ENDC}", # noqa
f"{TerminalColors.OKCYAN} Adding domain: {new_domain} {TerminalColors.ENDC}", # noqa
)
new_domain_invitation = self.try_add_domain_invitation(
transition_domain_email, new_domain
@ -373,14 +353,14 @@ class Command(BaseCommand):
if new_domain_invitation is None:
logger.info(
f"{termColors.YELLOW} ! No new e-mail detected !" # noqa
f"(SKIPPED ADDING DOMAIN INVITATION){termColors.ENDC}"
f"{TerminalColors.YELLOW} ! No new e-mail detected !" # noqa
f"(SKIPPED ADDING DOMAIN INVITATION){TerminalColors.ENDC}"
)
else:
# DEBUG:
self.print_debug(
TerminalHelper.print_conditional(
debug_on,
f"{termColors.OKCYAN} Adding domain invitation: {new_domain_invitation} {termColors.ENDC}", # noqa
f"{TerminalColors.OKCYAN} Adding domain invitation: {new_domain_invitation} {TerminalColors.ENDC}", # noqa
)
domain_invitations_to_create.append(new_domain_invitation)
@ -390,9 +370,9 @@ class Command(BaseCommand):
and total_rows_parsed >= debug_max_entries_to_parse
):
logger.info(
f"""{termColors.YELLOW}
f"""{TerminalColors.YELLOW}
----PARSE LIMIT REACHED. HALTING PARSER.----
{termColors.ENDC}
{TerminalColors.ENDC}
"""
)
break

102
src/registrar/management/commands/utility/terminal_helper.py Normal file
View file

@ -0,0 +1,102 @@
import logging
import sys
logger = logging.getLogger(__name__)
class TerminalColors:
"""Colors for terminal outputs
(makes reading the logs WAY easier)"""
HEADER = "\033[95m"
OKBLUE = "\033[94m"
OKCYAN = "\033[96m"
OKGREEN = "\033[92m"
YELLOW = "\033[93m"
MAGENTA = "\033[35m"
FAIL = "\033[91m"
ENDC = "\033[0m"
BOLD = "\033[1m"
UNDERLINE = "\033[4m"
BackgroundLightYellow = "\033[103m"
class TerminalHelper:
@staticmethod
def query_yes_no(question: str, default="yes") -> bool:
"""Ask a yes/no question via raw_input() and return their answer.
"question" is a string that is presented to the user.
"default" is the presumed answer if the user just hits <Enter>.
It must be "yes" (the default), "no" or None (meaning
an answer is required of the user).
The "answer" return value is True for "yes" or False for "no".
"""
valid = {"yes": True, "y": True, "ye": True, "no": False, "n": False}
if default is None:
prompt = " [y/n] "
elif default == "yes":
prompt = " [Y/n] "
elif default == "no":
prompt = " [y/N] "
else:
raise ValueError("invalid default answer: '%s'" % default)
while True:
logger.info(question + prompt)
choice = input().lower()
if default is not None and choice == "":
return valid[default]
elif choice in valid:
return valid[choice]
else:
logger.info("Please respond with 'yes' or 'no' " "(or 'y' or 'n').\n")
@staticmethod
def print_conditional(print_condition: bool, print_statement: str):
"""This function reduces complexity of debug statements
in other functions.
It uses the logger to write the given print_statement to the
terminal if print_condition is TRUE"""
# DEBUG:
if print_condition:
logger.info(print_statement)
@staticmethod
def prompt_for_execution(
system_exit_on_terminate: bool, info_to_inspect: str, prompt_title: str
) -> bool:
"""Create to reduce code complexity.
Prompts the user to inspect the given string
and asks if they wish to proceed.
Returns true if the user responds (y),
Returns false if the user responds (n)"""
action_description_for_selecting_no = "skip"
if system_exit_on_terminate:
action_description_for_selecting_no = "exit"
# Allow the user to inspect the command string
# and ask if they wish to proceed
proceed_execution = TerminalHelper.query_yes_no(
f"""{TerminalColors.OKCYAN}
=====================================================
{prompt_title}
=====================================================
*** IMPORTANT: VERIFY THE FOLLOWING LOOKS CORRECT ***
{info_to_inspect}
{TerminalColors.FAIL}
Proceed? (Y = proceed, N = {action_description_for_selecting_no})
{TerminalColors.ENDC}"""
)
# If the user decided to proceed return true.
# Otherwise, either return false or exit this subroutine.
if not proceed_execution:
if system_exit_on_terminate:
sys.exit()
return False
return True

7
src/registrar/tests/data/test_contacts.txt Normal file
View file

@ -0,0 +1,7 @@
TESTUSER|12363_CONTACT|123-123-1234||918-000-0000||testuser@gmail.com|GSA|SOMECOMPANY|ctldbatch|2021-06-30T17:58:09Z|SOMECOMPANY|ctldbatch|2021-06-30T18:18:09Z|
USER1|12345_CONTACT|123-123-1234||918-000-0000||agustina.wyman7@test.com|GSA|SOMECOMPANY|ctldbatch|2021-06-29T18:53:09Z|SOMECOMPANY|ctldbatch|2021-06-29T18:58:08Z|
USER2|12355_CONTACT|123-123-1234||918-000-0000||susy.martin4@test.com|GSA|SOMECOMPANY|ctldbatch|2021-06-30T15:23:10Z|SOMECOMPANY|ctldbatch|2021-06-30T15:38:10Z|
USER3|12356_CONTACT|123-123-1234||918-000-0000||stephania.winters4@test.com|GSA|SOMECOMPANY|ctldbatch|2021-06-30T15:23:10Z|SOMECOMPANY|ctldbatch|2021-06-30T18:28:09Z|
USER4|12357_CONTACT|123-123-1234||918-000-0000||alexandra.bobbitt5@test.com|GSA|SOMECOMPANY|ctldbatch|2021-06-30T15:23:10Z|SOMECOMPANY|ctldbatch|2021-08-02T22:13:09Z|
USER5|12362_CONTACT|123-123-1234||918-000-0000||jospeh.mcdowell3@test.com|GSA|SOMECOMPANY|ctldbatch|2021-06-30T17:58:09Z|SOMECOMPANY|ctldbatch|2021-06-30T18:33:09Z|
USER6|12363_CONTACT|123-123-1234||918-000-0000||reginald.ratcliff4@test.com|GSA|SOMECOMPANY|ctldbatch|2021-06-30T17:58:09Z|SOMECOMPANY|ctldbatch|2021-06-30T18:18:09Z|

8
src/registrar/tests/data/test_domain_contacts.txt Normal file
View file

@ -0,0 +1,8 @@
Anomaly.gov|ANOMALY|tech
TestDomain.gov|TESTUSER|admin
FakeWebsite1|USER1|admin
FakeWebsite1|USER2|tech
FakeWebsite1|USER3|billing
FakeWebsite2.GOV|USER4|admin
FakeWebsite2.GOV|USER5|billing
FakeWebsite2.GOV|USER6|tech

4
src/registrar/tests/data/test_domain_statuses.txt Normal file
View file

@ -0,0 +1,4 @@
Anomaly.gov|muahaha|
TestDomain.gov|ok|
FakeWebsite1.GOV|serverHold|
FakeWebsite2.GOV|Hold|

261
src/registrar/tests/test_transition_domain_migrations.py Normal file
View file

@ -0,0 +1,261 @@
from django.test import TestCase
from registrar.models import (
User,
Domain,
DomainInvitation,
TransitionDomain,
DomainInformation,
UserDomainRole,
)
from django.core.management import call_command
class TestLogins(TestCase):
""" """
def setUp(self):
""" """
# self.load_transition_domain_script = "load_transition_domain",
# self.transfer_script = "transfer_transition_domains_to_domains",
# self.master_script = "load_transition_domain",
self.test_data_file_location = "/app/registrar/tests/data"
self.test_domain_contact_filename = "test_domain_contacts.txt"
self.test_contact_filename = "test_contacts.txt"
self.test_domain_status_filename = "test_domain_statuses.txt"
def tearDown(self):
super().tearDown()
TransitionDomain.objects.all().delete()
Domain.objects.all().delete()
DomainInvitation.objects.all().delete()
DomainInformation.objects.all().delete()
User.objects.all().delete()
UserDomainRole.objects.all().delete()
def run_load_domains(self):
call_command(
"load_transition_domain",
f"{self.test_data_file_location}/{self.test_domain_contact_filename}",
f"{self.test_data_file_location}/{self.test_contact_filename}",
f"{self.test_data_file_location}/{self.test_domain_status_filename}",
)
def run_transfer_domains(self):
call_command("transfer_transition_domains_to_domains")
def run_master_script(self):
call_command(
"master_domain_migrations",
runMigrations=True,
migrationDirectory=f"{self.test_data_file_location}",
migrationFilenames=(
f"{self.test_domain_contact_filename},"
f"{self.test_contact_filename},"
f"{self.test_domain_status_filename}"
),
)
def compare_tables(
self,
expected_total_transition_domains,
expected_total_domains,
expected_total_domain_informations,
expected_total_domain_invitations,
expected_missing_domains,
expected_duplicate_domains,
expected_missing_domain_informations,
expected_missing_domain_invitations,
):
"""Does a diff between the transition_domain and the following tables:
domain, domain_information and the domain_invitation.
Verifies that the data loaded correctly."""
missing_domains = []
duplicate_domains = []
missing_domain_informations = []
missing_domain_invites = []
for transition_domain in TransitionDomain.objects.all(): # DEBUG:
transition_domain_name = transition_domain.domain_name
transition_domain_email = transition_domain.username
# Check Domain table
matching_domains = Domain.objects.filter(name=transition_domain_name)
# Check Domain Information table
matching_domain_informations = DomainInformation.objects.filter(
domain__name=transition_domain_name
)
# Check Domain Invitation table
matching_domain_invitations = DomainInvitation.objects.filter(
email=transition_domain_email.lower(),
domain__name=transition_domain_name,
)
if len(matching_domains) == 0:
missing_domains.append(transition_domain_name)
elif len(matching_domains) > 1:
duplicate_domains.append(transition_domain_name)
if len(matching_domain_informations) == 0:
missing_domain_informations.append(transition_domain_name)
if len(matching_domain_invitations) == 0:
missing_domain_invites.append(transition_domain_name)
total_missing_domains = len(missing_domains)
total_duplicate_domains = len(duplicate_domains)
total_missing_domain_informations = len(missing_domain_informations)
total_missing_domain_invitations = len(missing_domain_invites)
total_transition_domains = len(TransitionDomain.objects.all())
total_domains = len(Domain.objects.all())
total_domain_informations = len(DomainInformation.objects.all())
total_domain_invitations = len(DomainInvitation.objects.all())
print(
f"""
total_missing_domains = {len(missing_domains)}
total_duplicate_domains = {len(duplicate_domains)}
total_missing_domain_informations = {len(missing_domain_informations)}
total_missing_domain_invitations = {len(missing_domain_invites)}
total_transition_domains = {len(TransitionDomain.objects.all())}
total_domains = {len(Domain.objects.all())}
total_domain_informations = {len(DomainInformation.objects.all())}
total_domain_invitations = {len(DomainInvitation.objects.all())}
"""
)
self.assertTrue(total_missing_domains == expected_missing_domains)
self.assertTrue(total_duplicate_domains == expected_duplicate_domains)
self.assertTrue(
total_missing_domain_informations == expected_missing_domain_informations
)
self.assertTrue(
total_missing_domain_invitations == expected_missing_domain_invitations
)
self.assertTrue(total_transition_domains == expected_total_transition_domains)
self.assertTrue(total_domains == expected_total_domains)
self.assertTrue(total_domain_informations == expected_total_domain_informations)
self.assertTrue(total_domain_invitations == expected_total_domain_invitations)
def test_master_migration_functions(self):
"""Run the full master migration script using local test data.
NOTE: This is more of an integration test and so far does not
follow best practice of limiting the number of assertions per test.
But for now, this will double-check that the script
works as intended."""
self.run_master_script()
# STEP 2: (analyze the tables just like the
# migration script does, but add assert statements)
expected_total_transition_domains = 8
expected_total_domains = 4
expected_total_domain_informations = 0
expected_total_domain_invitations = 7
expected_missing_domains = 0
expected_duplicate_domains = 0
# we expect 8 missing domain invites since the
# migration does not auto-login new users
expected_missing_domain_informations = 8
# we expect 1 missing invite from anomaly.gov (an injected error)
expected_missing_domain_invitations = 1
self.compare_tables(
expected_total_transition_domains,
expected_total_domains,
expected_total_domain_informations,
expected_total_domain_invitations,
expected_missing_domains,
expected_duplicate_domains,
expected_missing_domain_informations,
expected_missing_domain_invitations,
)
def test_load_transition_domain(self):
self.run_load_domains()
# STEP 2: (analyze the tables just like the migration
# script does, but add assert statements)
expected_total_transition_domains = 8
expected_total_domains = 0
expected_total_domain_informations = 0
expected_total_domain_invitations = 0
expected_missing_domains = 8
expected_duplicate_domains = 0
expected_missing_domain_informations = 8
expected_missing_domain_invitations = 8
self.compare_tables(
expected_total_transition_domains,
expected_total_domains,
expected_total_domain_informations,
expected_total_domain_invitations,
expected_missing_domains,
expected_duplicate_domains,
expected_missing_domain_informations,
expected_missing_domain_invitations,
)
def test_transfer_transition_domains_to_domains(self):
# TODO: setup manually instead of calling other script
self.run_load_domains()
self.run_transfer_domains()
# Analyze the tables
expected_total_transition_domains = 8
expected_total_domains = 4
expected_total_domain_informations = 0
expected_total_domain_invitations = 7
expected_missing_domains = 0
expected_duplicate_domains = 0
expected_missing_domain_informations = 8
expected_missing_domain_invitations = 1
self.compare_tables(
expected_total_transition_domains,
expected_total_domains,
expected_total_domain_informations,
expected_total_domain_invitations,
expected_missing_domains,
expected_duplicate_domains,
expected_missing_domain_informations,
expected_missing_domain_invitations,
)
def test_logins(self):
# TODO: setup manually instead of calling other scripts
self.run_load_domains()
self.run_transfer_domains()
# Simluate Logins
for invite in DomainInvitation.objects.all():
# get a user with this email address
user, user_created = User.objects.get_or_create(
email=invite.email, username=invite.email
)
user.first_login()
# Analyze the tables
expected_total_transition_domains = 8
expected_total_domains = 4
expected_total_domain_informations = 3
expected_total_domain_invitations = 7
expected_missing_domains = 0
expected_duplicate_domains = 0
expected_missing_domain_informations = 1
expected_missing_domain_invitations = 1
self.compare_tables(
expected_total_transition_domains,
expected_total_domains,
expected_total_domain_informations,
expected_total_domain_invitations,
expected_missing_domains,
expected_duplicate_domains,
expected_missing_domain_informations,
expected_missing_domain_invitations,
)