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

merged main

Browse source
This commit is contained in:
David Kennedy 2024-02-26 15:10:34 -05:00
commit 38db00bd79
No known key found for this signature in database
GPG key ID: 6528A5386E66B96B
16 changed files with 1285 additions and 294 deletions
Download patch file Download diff file Expand all files Collapse all files

321
src/registrar/admin.py
View file

@ -1,19 +1,20 @@
from datetime import date
import logging
from django import forms
from django.db.models import Q
from django.db.models.functions import Concat
from django.http import HttpResponse
from django.db.models.functions import Concat, Coalesce
from django.db.models import Value, CharField, Q
from django.http import HttpResponse, HttpResponseRedirect
from django.shortcuts import redirect
from django_fsm import get_available_FIELD_transitions
from django.contrib import admin, messages
from django.contrib.auth.admin import UserAdmin as BaseUserAdmin
from django.contrib.auth.models import Group
from django.contrib.contenttypes.models import ContentType
from django.http.response import HttpResponseRedirect
from django.urls import reverse
from dateutil.relativedelta import relativedelta # type: ignore
from epplibwrapper.errors import ErrorCode, RegistryError
from registrar.models.domain import Domain
from registrar.models.user import User
from registrar.models import Contact, Domain, DomainApplication, DraftDomain, User, Website
from registrar.utility import csv_export
from registrar.views.utility.mixins import OrderableFieldsMixin
from django.contrib.admin.views.main import ORDER_VAR
@ -23,7 +24,6 @@ from auditlog.admin import LogEntryAdmin # type: ignore
from django_fsm import TransitionNotAllowed # type: ignore
from django.utils.safestring import mark_safe
from django.utils.html import escape
from django.utils.translation import gettext_lazy as _
logger = logging.getLogger(__name__)
@ -120,41 +120,52 @@ class CustomLogEntryAdmin(LogEntryAdmin):
class AdminSortFields:
def get_queryset(db_field):
_name_sort = ["first_name", "last_name", "email"]
# Define a mapping of field names to model querysets and sort expressions.
# A dictionary is used for specificity, but the downside is some degree of repetition.
# To eliminate this, this list can be generated dynamically but the readability of that
# is impacted.
sort_mapping = {
# == Contact == #
"other_contacts": (Contact, _name_sort),
"authorizing_official": (Contact, _name_sort),
"submitter": (Contact, _name_sort),
# == User == #
"creator": (User, _name_sort),
"user": (User, _name_sort),
"investigator": (User, _name_sort),
# == Website == #
"current_websites": (Website, "website"),
"alternative_domains": (Website, "website"),
# == DraftDomain == #
"requested_domain": (DraftDomain, "name"),
# == DomainApplication == #
"domain_application": (DomainApplication, "requested_domain__name"),
# == Domain == #
"domain": (Domain, "name"),
"approved_domain": (Domain, "name"),
}
@classmethod
def get_queryset(cls, db_field):
"""This is a helper function for formfield_for_manytomany and formfield_for_foreignkey"""
# customize sorting
if db_field.name in (
"other_contacts",
"authorizing_official",
"submitter",
):
# Sort contacts by first_name, then last_name, then email
return models.Contact.objects.all().order_by(Concat("first_name", "last_name", "email"))
elif db_field.name in ("current_websites", "alternative_domains"):
# sort web sites
return models.Website.objects.all().order_by("website")
elif db_field.name in (
"creator",
"user",
"investigator",
):
# Sort users by first_name, then last_name, then email
return models.User.objects.all().order_by(Concat("first_name", "last_name", "email"))
elif db_field.name in (
"domain",
"approved_domain",
):
# Sort domains by name
return models.Domain.objects.all().order_by("name")
elif db_field.name in ("requested_domain",):
# Sort draft domains by name
return models.DraftDomain.objects.all().order_by("name")
elif db_field.name in ("domain_application",):
# Sort domain applications by name
return models.DomainApplication.objects.all().order_by("requested_domain__name")
else:
queryset_info = cls.sort_mapping.get(db_field.name, None)
if queryset_info is None:
return None
# Grab the model we want to order, and grab how we want to order it
model, order_by = queryset_info
match db_field.name:
case "investigator":
# We should only return users who are staff.
return model.objects.filter(is_staff=True).order_by(*order_by)
case _:
if isinstance(order_by, list) or isinstance(order_by, tuple):
return model.objects.order_by(*order_by)
else:
return model.objects.order_by(order_by)
class AuditedAdmin(admin.ModelAdmin):
"""Custom admin to make auditing easier."""
@ -172,9 +183,14 @@ class AuditedAdmin(admin.ModelAdmin):
def formfield_for_manytomany(self, db_field, request, **kwargs):
"""customize the behavior of formfields with manytomany relationships. the customized
behavior includes sorting of objects in lists as well as customizing helper text"""
# Define a queryset. Note that in the super of this,
# a new queryset will only be generated if one does not exist.
# Thus, the order in which we define queryset matters.
queryset = AdminSortFields.get_queryset(db_field)
if queryset:
kwargs["queryset"] = queryset
formfield = super().formfield_for_manytomany(db_field, request, **kwargs)
# customize the help text for all formfields for manytomany
formfield.help_text = (
@ -184,11 +200,16 @@ class AuditedAdmin(admin.ModelAdmin):
return formfield
def formfield_for_foreignkey(self, db_field, request, **kwargs):
"""customize the behavior of formfields with foreign key relationships. this will customize
the behavior of selects. customized behavior includes sorting of objects in list"""
"""Customize the behavior of formfields with foreign key relationships. This will customize
the behavior of selects. Customized behavior includes sorting of objects in list."""
# Define a queryset. Note that in the super of this,
# a new queryset will only be generated if one does not exist.
# Thus, the order in which we define queryset matters.
queryset = AdminSortFields.get_queryset(db_field)
if queryset:
kwargs["queryset"] = queryset
return super().formfield_for_foreignkey(db_field, request, **kwargs)
@ -223,7 +244,6 @@ class ListHeaderAdmin(AuditedAdmin, OrderableFieldsMixin):
parameter_value: string}
TODO: convert investigator id to investigator username
"""
filters = []
# Retrieve the filter parameters
for param in request.GET.keys():
@ -267,6 +287,14 @@ class UserContactInline(admin.StackedInline):
class MyUserAdmin(BaseUserAdmin):
"""Custom user admin class to use our inlines."""
class Meta:
"""Contains meta information about this class"""
model = models.User
fields = "__all__"
_meta = Meta()
inlines = [UserContactInline]
list_display = (
@ -355,6 +383,42 @@ class MyUserAdmin(BaseUserAdmin):
# in autocomplete_fields for user
ordering = ["first_name", "last_name", "email"]
def get_search_results(self, request, queryset, search_term):
"""
Override for get_search_results. This affects any upstream model using autocomplete_fields,
such as DomainApplication. This is because autocomplete_fields uses an API call to fetch data,
and this fetch comes from this method.
"""
# Custom filtering logic
queryset, use_distinct = super().get_search_results(request, queryset, search_term)
# If we aren't given a request to modify, we shouldn't try to
if request is None or not hasattr(request, "GET"):
return queryset, use_distinct
# Otherwise, lets modify it!
request_get = request.GET
# The request defines model name and field name.
# For instance, model_name could be "DomainApplication"
# and field_name could be "investigator".
model_name = request_get.get("model_name", None)
field_name = request_get.get("field_name", None)
# Make sure we're only modifying requests from these models.
models_to_target = {"domainapplication"}
if model_name in models_to_target:
# Define rules per field
match field_name:
case "investigator":
# We should not display investigators who don't have a staff role
queryset = queryset.filter(is_staff=True)
case _:
# In the default case, do nothing
pass
return queryset, use_distinct
# Let's define First group
# (which should in theory be the ONLY group)
def group(self, obj):
@ -419,6 +483,9 @@ class ContactAdmin(ListHeaderAdmin):
"contact",
"email",
]
# this ordering effects the ordering of results
# in autocomplete_fields for user
ordering = ["first_name", "last_name", "email"]
# We name the custom prop 'contact' because linter
# is not allowing a short_description attr on it
@ -690,6 +757,14 @@ class DomainInformationAdmin(ListHeaderAdmin):
# to activate the edit/delete/view buttons
filter_horizontal = ("other_contacts",)
autocomplete_fields = [
"creator",
"domain_application",
"authorizing_official",
"domain",
"submitter",
]
# Table ordering
ordering = ["domain__name"]
@ -754,8 +829,23 @@ class DomainApplicationAdmin(ListHeaderAdmin):
"""Lookup reimplementation, gets users of is_staff.
Returns a list of tuples consisting of (user.id, user)
"""
privileged_users = User.objects.filter(is_staff=True).order_by("first_name", "last_name", "email")
return [(user.id, user) for user in privileged_users]
# Select all investigators that are staff, then order by name and email
privileged_users = (
DomainApplication.objects.select_related("investigator")
.filter(investigator__is_staff=True)
.order_by("investigator__first_name", "investigator__last_name", "investigator__email")
)
# Annotate the full name and return a values list that lookups can use
privileged_users_annotated = privileged_users.annotate(
full_name=Coalesce(
Concat("investigator__first_name", Value(" "), "investigator__last_name", output_field=CharField()),
"investigator__email",
output_field=CharField(),
)
).values_list("investigator__id", "full_name")
return privileged_users_annotated
def queryset(self, request, queryset):
"""Custom queryset implementation, filters by investigator"""
@ -885,26 +975,19 @@ class DomainApplicationAdmin(ListHeaderAdmin):
"anything_else",
"is_policy_acknowledged",
]
autocomplete_fields = [
"approved_domain",
"requested_domain",
"submitter",
"creator",
"authorizing_official",
"investigator",
]
filter_horizontal = ("current_websites", "alternative_domains", "other_contacts")
# Table ordering
ordering = ["requested_domain__name"]
# lists in filter_horizontal are not sorted properly, sort them
# by website
def formfield_for_manytomany(self, db_field, request, **kwargs):
if db_field.name in ("current_websites", "alternative_domains"):
kwargs["queryset"] = models.Website.objects.all().order_by("website") # Sort websites
return super().formfield_for_manytomany(db_field, request, **kwargs)
def formfield_for_foreignkey(self, db_field, request, **kwargs):
# Removes invalid investigator options from the investigator dropdown
if db_field.name == "investigator":
kwargs["queryset"] = User.objects.filter(is_staff=True)
return db_field.formfield(**kwargs)
return super().formfield_for_foreignkey(db_field, request, **kwargs)
# Trigger action when a fieldset is changed
def save_model(self, request, obj, form, change):
if obj and obj.creator.status != models.User.RESTRICTED:
@ -972,7 +1055,6 @@ class DomainApplicationAdmin(ListHeaderAdmin):
admin user permissions and the application creator's status, so
we'll use the baseline readonly_fields and extend it as needed.
"""
readonly_fields = list(self.readonly_fields)
# Check if the creator is restricted
@ -1035,6 +1117,14 @@ class DomainInformationInline(admin.StackedInline):
# to activate the edit/delete/view buttons
filter_horizontal = ("other_contacts",)
autocomplete_fields = [
"creator",
"domain_application",
"authorizing_official",
"domain",
"submitter",
]
def formfield_for_manytomany(self, db_field, request, **kwargs):
"""customize the behavior of formfields with manytomany relationships. the customized
behavior includes sorting of objects in lists as well as customizing helper text"""
@ -1050,8 +1140,8 @@ class DomainInformationInline(admin.StackedInline):
return formfield
def formfield_for_foreignkey(self, db_field, request, **kwargs):
"""customize the behavior of formfields with foreign key relationships. this will customize
the behavior of selects. customized behavior includes sorting of objects in list"""
"""Customize the behavior of formfields with foreign key relationships. This will customize
the behavior of selects. Customized behavior includes sorting of objects in list."""
queryset = AdminSortFields.get_queryset(db_field)
if queryset:
kwargs["queryset"] = queryset
@ -1164,6 +1254,33 @@ class DomainAdmin(ListHeaderAdmin):
# Table ordering
ordering = ["name"]
def changeform_view(self, request, object_id=None, form_url="", extra_context=None):
"""Custom changeform implementation to pass in context information"""
if extra_context is None:
extra_context = {}
# Pass in what the an extended expiration date would be for the expiration date modal
if object_id is not None:
domain = Domain.objects.get(pk=object_id)
years_to_extend_by = self._get_calculated_years_for_exp_date(domain)
try:
curr_exp_date = domain.registry_expiration_date
except KeyError:
# No expiration date was found. Return none.
extra_context["extended_expiration_date"] = None
return super().changeform_view(request, object_id, form_url, extra_context)
if curr_exp_date < date.today():
extra_context["extended_expiration_date"] = date.today() + relativedelta(years=years_to_extend_by)
else:
new_date = domain.registry_expiration_date + relativedelta(years=years_to_extend_by)
extra_context["extended_expiration_date"] = new_date
else:
extra_context["extended_expiration_date"] = None
return super().changeform_view(request, object_id, form_url, extra_context)
def export_data_type(self, request):
# match the CSV example with all the fields
response = HttpResponse(content_type="text/csv")
@ -1222,6 +1339,7 @@ class DomainAdmin(ListHeaderAdmin):
"_edit_domain": self.do_edit_domain,
"_delete_domain": self.do_delete_domain,
"_get_status": self.do_get_status,
"_extend_expiration_date": self.do_extend_expiration_date,
}
# Check which action button was pressed and call the corresponding function
@ -1232,6 +1350,81 @@ class DomainAdmin(ListHeaderAdmin):
# If no matching action button is found, return the super method
return super().response_change(request, obj)
def do_extend_expiration_date(self, request, obj):
"""Extends a domains expiration date by one year from the current date"""
# Make sure we're dealing with a Domain
if not isinstance(obj, Domain):
self.message_user(request, "Object is not of type Domain.", messages.ERROR)
return None
years = self._get_calculated_years_for_exp_date(obj)
# Renew the domain.
try:
obj.renew_domain(length=years)
self.message_user(
request,
"Successfully extended the expiration date.",
)
except RegistryError as err:
if err.is_connection_error():
error_message = "Error connecting to the registry."
else:
error_message = f"Error extending this domain: {err}."
self.message_user(request, error_message, messages.ERROR)
except KeyError:
# In normal code flow, a keyerror can only occur when
# fresh data can't be pulled from the registry, and thus there is no cache.
self.message_user(
request,
"Error connecting to the registry. No expiration date was found.",
messages.ERROR,
)
except Exception as err:
logger.error(err, stack_info=True)
self.message_user(request, "Could not delete: An unspecified error occured", messages.ERROR)
return HttpResponseRedirect(".")
def _get_calculated_years_for_exp_date(self, obj, extension_period: int = 1):
"""Given the current date, an extension period, and a registry_expiration_date
on the domain object, calculate the number of years needed to extend the
current expiration date by the extension period.
"""
# Get the date we want to update to
desired_date = self._get_current_date() + relativedelta(years=extension_period)
# Grab the current expiration date
try:
exp_date = obj.registry_expiration_date
except KeyError:
# if no expiration date from registry, set it to today
logger.warning("current expiration date not set; setting to today")
exp_date = self._get_current_date()
# If the expiration date is super old (2020, for example), we need to
# "catch up" to the current year, so we add the difference.
# If both years match, then lets just proceed as normal.
calculated_exp_date = exp_date + relativedelta(years=extension_period)
year_difference = desired_date.year - exp_date.year
years = extension_period
if desired_date > calculated_exp_date:
# Max probably isn't needed here (no code flow), but it guards against negative and 0.
# In both of those cases, we just want to extend by the extension_period.
years = max(extension_period, year_difference)
return years
# Workaround for unit tests, as we cannot mock date directly.
# it is immutable. Rather than dealing with a convoluted workaround,
# lets wrap this in a function.
def _get_current_date(self):
"""Gets the current date"""
return date.today()
def do_delete_domain(self, request, obj):
if not isinstance(obj, Domain):
# Could be problematic if the type is similar,
@ -1390,6 +1583,10 @@ class DraftDomainAdmin(ListHeaderAdmin):
search_fields = ["name"]
search_help_text = "Search by draft domain name."
# this ordering effects the ordering of results
# in autocomplete_fields for user
ordering = ["name"]
class VerifiedByStaffAdmin(ListHeaderAdmin):
list_display = ("email", "requestor", "truncated_notes", "created_at")

31
src/registrar/assets/js/get-gov-admin.js
View file

@ -23,6 +23,33 @@ function openInNewTab(el, removeAttribute = false){
// <<>><<>><<>><<>><<>><<>><<>><<>><<>><<>><<>><<>><<>><<>><<>>
// Initialization code.
/** An IIFE for pages in DjangoAdmin that use modals.
* Dja strips out form elements, and modals generate their content outside
* of the current form scope, so we need to "inject" these inputs.
*/
(function (){
function createPhantomModalFormButtons(){
let submitButtons = document.querySelectorAll('.usa-modal button[type="submit"]');
form = document.querySelector("form")
submitButtons.forEach((button) => {
let input = document.createElement("input");
input.type = "submit";
input.name = button.name;
input.value = button.value;
input.style.display = "none"
// Add the hidden input to the form
form.appendChild(input);
button.addEventListener("click", () => {
console.log("clicking")
input.click();
})
})
}
createPhantomModalFormButtons();
})();
/** An IIFE for pages in DjangoAdmin which may need custom JS implementation.
* Currently only appends target="_blank" to the domain_form object,
* but this can be expanded.
@ -41,8 +68,8 @@ function openInNewTab(el, removeAttribute = false){
let domainFormElement = document.getElementById("domain_form");
let domainSubmitButton = document.getElementById("manageDomainSubmitButton");
if(domainSubmitButton && domainFormElement){
domainSubmitButton.addEventListener("mouseover", () => openInNewTab(domainFormElement, true));
domainSubmitButton.addEventListener("mouseout", () => openInNewTab(domainFormElement, false));
domainSubmitButton.addEventListener("mouseover", () => openInNewTab(domainFormElement, true));
domainSubmitButton.addEventListener("mouseout", () => openInNewTab(domainFormElement, false));
}
}

32
src/registrar/assets/sass/_theme/_admin.scss
View file

@ -183,6 +183,10 @@ h1, h2, h3,
.submit-row div.spacer {
flex-grow: 1;
}
.submit-row .mini-spacer{
margin-left: 2px;
margin-right: 2px;
}
.submit-row span {
margin-top: units(1);
}
@ -270,3 +274,31 @@ h1, h2, h3,
margin: 0!important;
}
}
// Hides the "clear" button on autocomplete, as we already have one to use
.select2-selection__clear {
display: none;
}
// Fixes a display issue where the list was entirely white, or had too much whitespace
.select2-dropdown {
display: inline-grid !important;
}
input.admin-confirm-button {
text-transform: none;
}
// Button groups in /admin incorrectly have bullets.
// Remove that!
.usa-modal__footer .usa-button-group__item {
list-style-type: none;
}
// USWDS media checks are overzealous in this situation,
// we should manually define this behaviour.
@media (max-width: 768px) {
.button-list-mobile {
display: contents !important;
}
}

28
src/registrar/models/domain.py
View file

@ -13,6 +13,7 @@ from typing import Any
from registrar.models.host import Host
from registrar.models.host_ip import HostIP
from registrar.utility.enums import DefaultEmail
from registrar.utility import errors
from registrar.utility.errors import (
ActionNotAllowed,
@ -192,9 +193,17 @@ class Domain(TimeStampedModel, DomainHelper):
@classmethod
def available(cls, domain: str) -> bool:
"""Check if a domain is available."""
"""Check if a domain is available.
This is called by the availablility api and
is called in the validate function on the request/domain page
throws- RegistryError or InvalidDomainError"""
if not cls.string_could_be_domain(domain):
raise ValueError("Not a valid domain: %s" % str(domain))
logger.warning("Not a valid domain: %s" % str(domain))
# throw invalid domain error so that it can be caught in
# validate_and_handle_errors in domain_helper
raise errors.InvalidDomainError()
domain_name = domain.lower()
req = commands.CheckDomain([domain_name])
return registry.send(req, cleaned=True).res_data[0].avail
@ -265,15 +274,17 @@ class Domain(TimeStampedModel, DomainHelper):
Default length and unit of time are 1 year.
"""
# if no expiration date from registry, set to today
# If no date is specified, grab the registry_expiration_date
try:
cur_exp_date = self.registry_expiration_date
exp_date = self.registry_expiration_date
except KeyError:
# if no expiration date from registry, set it to today
logger.warning("current expiration date not set; setting to today")
cur_exp_date = date.today()
exp_date = date.today()
# create RenewDomain request
request = commands.RenewDomain(name=self.name, cur_exp_date=cur_exp_date, period=epp.Period(length, unit))
request = commands.RenewDomain(name=self.name, cur_exp_date=exp_date, period=epp.Period(length, unit))
try:
# update expiration date in registry, and set the updated
@ -427,7 +438,6 @@ class Domain(TimeStampedModel, DomainHelper):
raise NameserverError(code=nsErrorCodes.INVALID_HOST, nameserver=nameserver)
elif cls.isSubdomain(name, nameserver) and (ip is None or ip == []):
raise NameserverError(code=nsErrorCodes.MISSING_IP, nameserver=nameserver)
elif not cls.isSubdomain(name, nameserver) and (ip is not None and ip != []):
raise NameserverError(code=nsErrorCodes.GLUE_RECORD_NOT_ALLOWED, nameserver=nameserver, ip=ip)
elif ip is not None and ip != []:
@ -1778,6 +1788,10 @@ class Domain(TimeStampedModel, DomainHelper):
for cleaned_host in cleaned_hosts:
# Check if the cleaned_host already exists
host_in_db, host_created = Host.objects.get_or_create(domain=self, name=cleaned_host["name"])
# Check if the nameserver is a subdomain of the current domain
# If it is NOT a subdomain, we remove the IP address
if not Domain.isSubdomain(self.name, cleaned_host["name"]):
cleaned_host["addrs"] = []
# Get cleaned list of ips for update
cleaned_ips = cleaned_host["addrs"]
if not host_created:

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

@ -255,6 +255,14 @@ class DomainInformation(TimeStampedModel):
else:
da_many_to_many_dict[field] = getattr(domain_application, field).all()
# This will not happen in normal code flow, but having some redundancy doesn't hurt.
# da_dict should not have "id" under any circumstances.
# If it does have it, then this indicates that common_fields is overzealous in the data
# that it is returning. Try looking in DomainHelper.get_common_fields.
if "id" in da_dict:
logger.warning("create_from_da() -> Found attribute 'id' when trying to create")
da_dict.pop("id", None)
# Create a placeholder DomainInformation object
domain_info = DomainInformation(**da_dict)

15
src/registrar/models/utility/domain_helper.py
View file

@ -33,11 +33,12 @@ class DomainHelper:
# Split into pieces for the linter
domain = cls._validate_domain_string(domain, blank_ok)
try:
if not check_domain_available(domain):
raise errors.DomainUnavailableError()
except RegistryError as err:
raise errors.RegistrySystemError() from err
if domain != "":
try:
if not check_domain_available(domain):
raise errors.DomainUnavailableError()
except RegistryError as err:
raise errors.RegistrySystemError() from err
return domain
@staticmethod
@ -180,8 +181,8 @@ class DomainHelper:
"""
# Get a list of the existing fields on model_1 and model_2
model_1_fields = set(field.name for field in model_1._meta.get_fields() if field != "id")
model_2_fields = set(field.name for field in model_2._meta.get_fields() if field != "id")
model_1_fields = set(field.name for field in model_1._meta.get_fields() if field.name != "id")
model_2_fields = set(field.name for field in model_2._meta.get_fields() if field.name != "id")
# Get the fields that exist on both DomainApplication and DomainInformation
common_fields = model_1_fields & model_2_fields

37
src/registrar/models/utility/generic_helper.py Normal file
View file

@ -0,0 +1,37 @@
"""This file contains general purpose helpers that don't belong in any specific location"""
import time
import logging
logger = logging.getLogger(__name__)
class Timer:
"""
This class is used to measure execution time for performance profiling.
__enter__ and __exit__ is used such that you can wrap any code you want
around a with statement. After this exits, logger.info will print
the execution time in seconds.
Note that this class does not account for general randomness as more
robust libraries do, so there is some tiny amount of latency involved
in using this, but it is minimal enough that for most applications it is not
noticable.
Usage:
with Timer():
...some code
"""
def __enter__(self):
"""Starts the timer"""
self.start = time.time()
# This allows usage of the instance within the with block
return self
def __exit__(self, *args):
"""Ends the timer and logs what happened"""
self.end = time.time()
self.duration = self.end - self.start
logger.info(f"Execution time: {self.duration} seconds")

3
src/registrar/templates/admin/base_site.html
View file

@ -18,11 +18,12 @@
<link rel="apple-touch-icon" size="180x180"
href="{% static 'img/registrar/favicons/favicon-180.png' %}"
>
<script src="{% static 'js/uswds-init.min.js' %}" defer></script>
<script src="{% static 'js/uswds.min.js' %}" defer></script>
<script type="application/javascript" src="{% static 'js/get-gov-admin.js' %}" defer></script>
{% endblock %}
{% block title %}{% if subtitle %}{{ subtitle }} | {% endif %}{{ title }} | {{ site_title|default:_('Django site admin') }}{% endblock %}
{% block extrastyle %}{{ block.super }}
<link rel="stylesheet" type="text/css" href="{% static 'css/styles.css' %}" />
{% endblock %}

124
src/registrar/templates/django/admin/domain_change_form.html
View file

@ -2,21 +2,117 @@
{% load i18n static %}
{% block field_sets %}
<div class="submit-row">
<input id="manageDomainSubmitButton" type="submit" value="Manage domain" name="_edit_domain">
<input type="submit" value="Get registry status" name="_get_status">
<div class="spacer"></div>
{% if original.state == original.State.READY %}
<input type="submit" value="Place hold" name="_place_client_hold" class="custom-link-button">
{% elif original.state == original.State.ON_HOLD %}
<input type="submit" value="Remove hold" name="_remove_client_hold" class="custom-link-button">
{% endif %}
{% if original.state == original.State.READY or original.state == original.State.ON_HOLD %}
<span> | </span>
{% endif %}
{% if original.state != original.State.DELETED %}
<input type="submit" value="Remove from registry" name="_delete_domain" class="custom-link-button">
<div class="display-flex flex-row flex-justify submit-row">
<div class="flex-align-self-start button-list-mobile">
<input id="manageDomainSubmitButton" type="submit" value="Manage domain" name="_edit_domain">
{# Dja has margin styles defined on inputs as is. Lets work with it, rather than fight it. #}
<span class="mini-spacer"></span>
<input type="submit" value="Get registry status" name="_get_status">
</div>
<div class="desktop:flex-align-self-end">
{% if original.state != original.State.DELETED %}
<a
class="text-middle"
href="#toggle-extend-expiration-alert"
aria-controls="toggle-extend-expiration-alert"
data-open-modal
>
Extend expiration date
</a>
<span class="margin-left-05 margin-right-05 text-middle"> | </span>
{% endif %}
{% if original.state == original.State.READY %}
<input type="submit" value="Place hold" name="_place_client_hold" class="custom-link-button">
{% elif original.state == original.State.ON_HOLD %}
<input type="submit" value="Remove hold" name="_remove_client_hold" class="custom-link-button">
{% endif %}
{% if original.state == original.State.READY or original.state == original.State.ON_HOLD %}
<span class="margin-left-05 margin-right-05 text-middle"> | </span>
{% endif %}
{% if original.state != original.State.DELETED %}
<input type="submit" value="Remove from registry" name="_delete_domain" class="custom-link-button">
{% endif %}
</div>
</div>
{{ block.super }}
{% endblock %}
{% block submit_buttons_bottom %}
{% comment %}
Modals behave very weirdly in django admin.
They tend to "strip out" any injected form elements, leaving only the main form.
In addition, USWDS handles modals by first destroying the element, then repopulating it toward the end of the page.
In effect, this means that the modal is not, and cannot, be surrounded by any form element at compile time.
The current workaround for this is to use javascript to inject a hidden input, and bind submit of that
element to the click of the confirmation button within this modal.
This is controlled by the class `dja-form-placeholder` on the button.
In addition, the modal element MUST be placed low in the DOM. The script loads slower on DJA than on other portions
of the application, so this means that it will briefly "populate", causing unintended visual effects.
{% endcomment %}
<div
class="usa-modal"
id="toggle-extend-expiration-alert"
aria-labelledby="Are you sure you want to extend the expiration date?"
aria-describedby="This expiration date will be extended."
>
<div class="usa-modal__content">
<div class="usa-modal__main">
<h2 class="usa-modal__heading" id="modal-1-heading">
Are you sure you want to extend the expiration date?
</h2>
<div class="usa-prose">
<p>
This will extend the expiration date by one year.
{# Acts as a <br> #}
<div class="display-inline"></div>
This action cannot be undone.
</p>
<p>
Domain: <b>{{ original.name }}</b>
{# Acts as a <br> #}
<div class="display-inline"></div>
New expiration date: <b>{{ extended_expiration_date }}</b>
{{test}}
</p>
</div>
<div class="usa-modal__footer">
<ul class="usa-button-group">
<li class="usa-button-group__item">
<button
type="submit"
class="usa-button dja-form-placeholder"
name="_extend_expiration_date"
>
Yes, extend date
</button>
</li>
<li class="usa-button-group__item">
<button
type="button"
class="usa-button usa-button--unstyled padding-105 text-center"
data-close-modal
>
Cancel
</button>
</li>
</ul>
</div>
</div>
<button
type="button"
class="usa-button usa-modal__close"
aria-label="Close this window"
data-close-modal
>
<svg class="usa-icon" aria-hidden="true" focusable="false" role="img">
<use xlink:href="{%static 'img/sprite.svg'%}#close"></use>
</svg>
</button>
</div>
</div>
{{ block.super }}
{% endblock %}

1
src/registrar/templates/home.html
View file

@ -15,6 +15,7 @@
{% endblock %}
<h1>Manage your domains</h2>
<p class="margin-top-4">
<a href="{% url 'application:' %}" class="usa-button"
>

117
src/registrar/tests/common.py
View file

@ -231,6 +231,7 @@ class AuditedAdminMockData:
first_name="{} first_name:{}".format(item_name, short_hand),
last_name="{} last_name:{}".format(item_name, short_hand),
username="{} username:{}".format(item_name + str(uuid.uuid4())[:8], short_hand),
is_staff=True,
)[0]
return user
@ -691,6 +692,56 @@ class MockEppLib(TestCase):
],
ex_date=datetime.date(2023, 5, 25),
)
mockDataInfoDomainSubdomain = fakedEppObject(
"fakePw",
cr_date=make_aware(datetime.datetime(2023, 5, 25, 19, 45, 35)),
contacts=[common.DomainContact(contact="123", type=PublicContact.ContactTypeChoices.SECURITY)],
hosts=["fake.meoward.gov"],
statuses=[
common.Status(state="serverTransferProhibited", description="", lang="en"),
common.Status(state="inactive", description="", lang="en"),
],
ex_date=datetime.date(2023, 5, 25),
)
mockDataInfoDomainSubdomainAndIPAddress = fakedEppObject(
"fakePw",
cr_date=make_aware(datetime.datetime(2023, 5, 25, 19, 45, 35)),
contacts=[common.DomainContact(contact="123", type=PublicContact.ContactTypeChoices.SECURITY)],
hosts=["fake.meow.gov"],
statuses=[
common.Status(state="serverTransferProhibited", description="", lang="en"),
common.Status(state="inactive", description="", lang="en"),
],
ex_date=datetime.date(2023, 5, 25),
addrs=[common.Ip(addr="2.0.0.8")],
)
mockDataInfoDomainNotSubdomainNoIP = fakedEppObject(
"fakePw",
cr_date=make_aware(datetime.datetime(2023, 5, 25, 19, 45, 35)),
contacts=[common.DomainContact(contact="123", type=PublicContact.ContactTypeChoices.SECURITY)],
hosts=["fake.meow.com"],
statuses=[
common.Status(state="serverTransferProhibited", description="", lang="en"),
common.Status(state="inactive", description="", lang="en"),
],
ex_date=datetime.date(2023, 5, 25),
)
mockDataInfoDomainSubdomainNoIP = fakedEppObject(
"fakePw",
cr_date=make_aware(datetime.datetime(2023, 5, 25, 19, 45, 35)),
contacts=[common.DomainContact(contact="123", type=PublicContact.ContactTypeChoices.SECURITY)],
hosts=["fake.subdomainwoip.gov"],
statuses=[
common.Status(state="serverTransferProhibited", description="", lang="en"),
common.Status(state="inactive", description="", lang="en"),
],
ex_date=datetime.date(2023, 5, 25),
)
mockDataExtensionDomain = fakedEppObject(
"fakePw",
cr_date=make_aware(datetime.datetime(2023, 5, 25, 19, 45, 35)),
@ -828,6 +879,24 @@ class MockEppLib(TestCase):
addrs=[common.Ip(addr="1.2.3.4"), common.Ip(addr="2.3.4.5")],
)
mockDataInfoHosts1IP = fakedEppObject(
"lastPw",
cr_date=make_aware(datetime.datetime(2023, 8, 25, 19, 45, 35)),
addrs=[common.Ip(addr="2.0.0.8")],
)
mockDataInfoHostsNotSubdomainNoIP = fakedEppObject(
"lastPw",
cr_date=make_aware(datetime.datetime(2023, 8, 26, 19, 45, 35)),
addrs=[],
)
mockDataInfoHostsSubdomainNoIP = fakedEppObject(
"lastPw",
cr_date=make_aware(datetime.datetime(2023, 8, 27, 19, 45, 35)),
addrs=[],
)
mockDataHostChange = fakedEppObject("lastPw", cr_date=make_aware(datetime.datetime(2023, 8, 25, 19, 45, 35)))
addDsData1 = {
"keyTag": 1234,
@ -921,6 +990,11 @@ class MockEppLib(TestCase):
ex_date=datetime.date(2023, 5, 25),
)
mockButtonRenewedDomainExpDate = fakedEppObject(
"fake.gov",
ex_date=datetime.date(2025, 5, 25),
)
mockDnsNeededRenewedDomainExpDate = fakedEppObject(
"fakeneeded.gov",
ex_date=datetime.date(2023, 2, 15),
@ -989,6 +1063,8 @@ class MockEppLib(TestCase):
return self.mockDeleteDomainCommands(_request, cleaned)
case commands.RenewDomain:
return self.mockRenewDomainCommand(_request, cleaned)
case commands.InfoHost:
return self.mockInfoHostCommmands(_request, cleaned)
case _:
return MagicMock(res_data=[self.mockDataInfoHosts])
@ -1003,6 +1079,25 @@ class MockEppLib(TestCase):
code=ErrorCode.COMMAND_COMPLETED_SUCCESSFULLY,
)
def mockInfoHostCommmands(self, _request, cleaned):
request_name = getattr(_request, "name", None)
# Define a dictionary to map request names to data and extension values
request_mappings = {
"fake.meow.gov": (self.mockDataInfoHosts1IP, None), # is subdomain and has ip
"fake.meow.com": (self.mockDataInfoHostsNotSubdomainNoIP, None), # not subdomain w no ip
"fake.subdomainwoip.gov": (self.mockDataInfoHostsSubdomainNoIP, None), # subdomain w no ip
}
# Retrieve the corresponding values from the dictionary
default_mapping = (self.mockDataInfoHosts, None)
res_data, extensions = request_mappings.get(request_name, default_mapping)
return MagicMock(
res_data=[res_data],
extensions=[extensions] if extensions is not None else [],
)
def mockUpdateHostCommands(self, _request, cleaned):
test_ws_ip = common.Ip(addr="1.1. 1.1")
addrs_submitted = getattr(_request, "addrs", [])
@ -1049,11 +1144,19 @@ class MockEppLib(TestCase):
res_data=[self.mockMaximumRenewedDomainExpDate],
code=ErrorCode.COMMAND_COMPLETED_SUCCESSFULLY,
)
else:
return MagicMock(
res_data=[self.mockRenewedDomainExpDate],
code=ErrorCode.COMMAND_COMPLETED_SUCCESSFULLY,
)
elif getattr(_request, "name", None) == "fake.gov":
period = getattr(_request, "period", None)
extension_period = getattr(period, "length", None)
if extension_period == 2:
return MagicMock(
res_data=[self.mockButtonRenewedDomainExpDate],
code=ErrorCode.COMMAND_COMPLETED_SUCCESSFULLY,
)
else:
return MagicMock(
res_data=[self.mockRenewedDomainExpDate],
code=ErrorCode.COMMAND_COMPLETED_SUCCESSFULLY,
)
def mockInfoDomainCommands(self, _request, cleaned):
request_name = getattr(_request, "name", None)
@ -1083,6 +1186,10 @@ class MockEppLib(TestCase):
"adomain2.gov": (self.InfoDomainWithVerisignSecurityContact, None),
"defaulttechnical.gov": (self.InfoDomainWithDefaultTechnicalContact, None),
"justnameserver.com": (self.justNameserver, None),
"meoward.gov": (self.mockDataInfoDomainSubdomain, None),
"meow.gov": (self.mockDataInfoDomainSubdomainAndIPAddress, None),
"fakemeow.gov": (self.mockDataInfoDomainNotSubdomainNoIP, None),
"subdomainwoip.gov": (self.mockDataInfoDomainSubdomainNoIP, None),
}
# Retrieve the corresponding values from the dictionary

462
src/registrar/tests/test_admin.py
View file

@ -1,6 +1,8 @@
from datetime import date
from django.test import TestCase, RequestFactory, Client
from django.contrib.admin.sites import AdminSite
from contextlib import ExitStack
from django_webtest import WebTest # type: ignore
from django.contrib import messages
from django.urls import reverse
from registrar.admin import (
@ -35,7 +37,7 @@ from .common import (
)
from django.contrib.sessions.backends.db import SessionStore
from django.contrib.auth import get_user_model
from unittest.mock import patch
from unittest.mock import ANY, call, patch
from unittest import skip
from django.conf import settings
@ -45,7 +47,11 @@ import logging
logger = logging.getLogger(__name__)
class TestDomainAdmin(MockEppLib):
class TestDomainAdmin(MockEppLib, WebTest):
# csrf checks do not work with WebTest.
# We disable them here. TODO for another ticket.
csrf_checks = False
def setUp(self):
self.site = AdminSite()
self.admin = DomainAdmin(model=Domain, admin_site=self.site)
@ -53,8 +59,177 @@ class TestDomainAdmin(MockEppLib):
self.superuser = create_superuser()
self.staffuser = create_user()
self.factory = RequestFactory()
self.app.set_user(self.superuser.username)
self.client.force_login(self.superuser)
super().setUp()
@skip("TODO for another ticket. This test case is grabbing old db data.")
@patch("registrar.admin.DomainAdmin._get_current_date", return_value=date(2024, 1, 1))
def test_extend_expiration_date_button(self, mock_date_today):
"""
Tests if extend_expiration_date button extends correctly
"""
# Create a ready domain with a preset expiration date
domain, _ = Domain.objects.get_or_create(name="fake.gov", state=Domain.State.READY)
response = self.app.get(reverse("admin:registrar_domain_change", args=[domain.pk]))
# Make sure the ex date is what we expect it to be
domain_ex_date = Domain.objects.get(id=domain.id).expiration_date
self.assertEqual(domain_ex_date, date(2023, 5, 25))
# Make sure that the page is loading as expected
self.assertEqual(response.status_code, 200)
self.assertContains(response, domain.name)
self.assertContains(response, "Extend expiration date")
# Grab the form to submit
form = response.forms["domain_form"]
with patch("django.contrib.messages.add_message") as mock_add_message:
# Submit the form
response = form.submit("_extend_expiration_date")
# Follow the response
response = response.follow()
# refresh_from_db() does not work for objects with protected=True.
# https://github.com/viewflow/django-fsm/issues/89
new_domain = Domain.objects.get(id=domain.id)
# Check that the current expiration date is what we expect
self.assertEqual(new_domain.expiration_date, date(2025, 5, 25))
# Assert that everything on the page looks correct
self.assertEqual(response.status_code, 200)
self.assertContains(response, domain.name)
self.assertContains(response, "Extend expiration date")
# Ensure the message we recieve is in line with what we expect
expected_message = "Successfully extended the expiration date."
expected_call = call(
# The WGSI request doesn't need to be tested
ANY,
messages.INFO,
expected_message,
extra_tags="",
fail_silently=False,
)
mock_add_message.assert_has_calls([expected_call], 1)
@patch("registrar.admin.DomainAdmin._get_current_date", return_value=date(2024, 1, 1))
def test_extend_expiration_date_button_epp(self, mock_date_today):
"""
Tests if extend_expiration_date button sends the right epp command
"""
# Create a ready domain with a preset expiration date
domain, _ = Domain.objects.get_or_create(name="fake.gov", state=Domain.State.READY)
response = self.app.get(reverse("admin:registrar_domain_change", args=[domain.pk]))
# Make sure that the page is loading as expected
self.assertEqual(response.status_code, 200)
self.assertContains(response, domain.name)
self.assertContains(response, "Extend expiration date")
# Grab the form to submit
form = response.forms["domain_form"]
with patch("django.contrib.messages.add_message") as mock_add_message:
with patch("registrar.models.Domain.renew_domain") as renew_mock:
# Submit the form
response = form.submit("_extend_expiration_date")
# Follow the response
response = response.follow()
# This value is based off of the current year - the expiration date.
# We "freeze" time to 2024, so 2024 - 2023 will always result in an
# "extension" of 2, as that will be one year of extension from that date.
extension_length = 2
# Assert that it is calling the function with the right extension length.
# We only need to test the value that EPP sends, as we can assume the other
# test cases cover the "renew" function.
renew_mock.assert_has_calls([call(length=extension_length)], any_order=False)
# We should not make duplicate calls
self.assertEqual(renew_mock.call_count, 1)
# Assert that everything on the page looks correct
self.assertEqual(response.status_code, 200)
self.assertContains(response, domain.name)
self.assertContains(response, "Extend expiration date")
# Ensure the message we recieve is in line with what we expect
expected_message = "Successfully extended the expiration date."
expected_call = call(
# The WGSI request doesn't need to be tested
ANY,
messages.INFO,
expected_message,
extra_tags="",
fail_silently=False,
)
mock_add_message.assert_has_calls([expected_call], 1)
@patch("registrar.admin.DomainAdmin._get_current_date", return_value=date(2023, 1, 1))
def test_extend_expiration_date_button_date_matches_epp(self, mock_date_today):
"""
Tests if extend_expiration_date button sends the right epp command
when the current year matches the expiration date
"""
# Create a ready domain with a preset expiration date
domain, _ = Domain.objects.get_or_create(name="fake.gov", state=Domain.State.READY)
response = self.app.get(reverse("admin:registrar_domain_change", args=[domain.pk]))
# Make sure that the page is loading as expected
self.assertEqual(response.status_code, 200)
self.assertContains(response, domain.name)
self.assertContains(response, "Extend expiration date")
# Grab the form to submit
form = response.forms["domain_form"]
with patch("django.contrib.messages.add_message") as mock_add_message:
with patch("registrar.models.Domain.renew_domain") as renew_mock:
# Submit the form
response = form.submit("_extend_expiration_date")
# Follow the response
response = response.follow()
extension_length = 1
# Assert that it is calling the function with the right extension length.
# We only need to test the value that EPP sends, as we can assume the other
# test cases cover the "renew" function.
renew_mock.assert_has_calls([call(length=extension_length)], any_order=False)
# We should not make duplicate calls
self.assertEqual(renew_mock.call_count, 1)
# Assert that everything on the page looks correct
self.assertEqual(response.status_code, 200)
self.assertContains(response, domain.name)
self.assertContains(response, "Extend expiration date")
# Ensure the message we recieve is in line with what we expect
expected_message = "Successfully extended the expiration date."
expected_call = call(
# The WGSI request doesn't need to be tested
ANY,
messages.INFO,
expected_message,
extra_tags="",
fail_silently=False,
)
mock_add_message.assert_has_calls([expected_call], 1)
def test_short_org_name_in_domains_list(self):
"""
Make sure the short name is displaying in admin on the list page
@ -966,8 +1141,17 @@ class TestDomainApplicationAdmin(MockEppLib):
investigator_field = DomainApplication._meta.get_field("investigator")
# We should only be displaying staff users, in alphabetical order
expected_dropdown = list(User.objects.filter(is_staff=True))
current_dropdown = list(self.admin.formfield_for_foreignkey(investigator_field, request).queryset)
sorted_fields = ["first_name", "last_name", "email"]
expected_dropdown = list(User.objects.filter(is_staff=True).order_by(*sorted_fields))
# Grab the current dropdown. We do an API call to autocomplete to get this info.
application_queryset = self.admin.formfield_for_foreignkey(investigator_field, request).queryset
user_request = self.factory.post(
"/admin/autocomplete/?app_label=registrar&model_name=domainapplication&field_name=investigator"
)
user_admin = MyUserAdmin(User, self.site)
user_queryset = user_admin.get_search_results(user_request, application_queryset, None)[0]
current_dropdown = list(user_queryset)
self.assertEqual(expected_dropdown, current_dropdown)
@ -1002,7 +1186,13 @@ class TestDomainApplicationAdmin(MockEppLib):
self.client.login(username="staffuser", password=p)
request = RequestFactory().get("/")
expected_list = list(User.objects.filter(is_staff=True).order_by("first_name", "last_name", "email"))
# These names have metadata embedded in them. :investigator implicitly tests if
# these are actually from the attribute "investigator".
expected_list = [
"AGuy AGuy last_name:investigator",
"FinalGuy FinalGuy last_name:investigator",
"SomeGuy first_name:investigator SomeGuy last_name:investigator",
]
# Get the actual sorted list of investigators from the lookups method
actual_list = [item for _, item in self.admin.InvestigatorFilter.lookups(self, request, self.admin)]
@ -1434,12 +1624,47 @@ class AuditedAdminTest(TestCase):
return ordered_list
def test_alphabetically_sorted_domain_application_investigator(self):
"""Tests if the investigator field is alphabetically sorted by mimicking
the call event flow"""
# Creates multiple domain applications - review status does not matter
applications = multiple_unalphabetical_domain_objects("application")
# Create a mock request
application_request = self.factory.post(
"/admin/registrar/domainapplication/{}/change/".format(applications[0].pk)
)
# Get the formfield data from the application page
application_admin = AuditedAdmin(DomainApplication, self.site)
field = DomainApplication.investigator.field
application_queryset = application_admin.formfield_for_foreignkey(field, application_request).queryset
request = self.factory.post(
"/admin/autocomplete/?app_label=registrar&model_name=domainapplication&field_name=investigator"
)
sorted_fields = ["first_name", "last_name", "email"]
desired_sort_order = list(User.objects.filter(is_staff=True).order_by(*sorted_fields))
# Grab the data returned from get search results
admin = MyUserAdmin(User, self.site)
search_queryset = admin.get_search_results(request, application_queryset, None)[0]
current_sort_order = list(search_queryset)
self.assertEqual(
desired_sort_order,
current_sort_order,
"Investigator is not ordered alphabetically",
)
# This test case should be refactored in general, as it is too overly specific and engineered
def test_alphabetically_sorted_fk_fields_domain_application(self):
with less_console_noise():
tested_fields = [
DomainApplication.authorizing_official.field,
DomainApplication.submitter.field,
DomainApplication.investigator.field,
# DomainApplication.investigator.field,
DomainApplication.creator.field,
DomainApplication.requested_domain.field,
]
@ -1457,142 +1682,141 @@ class AuditedAdminTest(TestCase):
# but both fields are of a fixed length.
# For test case purposes, this should be performant.
for field in tested_fields:
isNamefield: bool = field == DomainApplication.requested_domain.field
if isNamefield:
sorted_fields = ["name"]
else:
sorted_fields = ["first_name", "last_name"]
# We want both of these to be lists, as it is richer test wise.
desired_order = self.order_by_desired_field_helper(model_admin, request, field.name, *sorted_fields)
current_sort_order = list(model_admin.formfield_for_foreignkey(field, request).queryset)
# Conforms to the same object structure as desired_order
current_sort_order_coerced_type = []
# This is necessary as .queryset and get_queryset
# return lists of different types/structures.
# We need to parse this data and coerce them into the same type.
for contact in current_sort_order:
if not isNamefield:
first = contact.first_name
last = contact.last_name
with self.subTest(field=field):
isNamefield: bool = field == DomainApplication.requested_domain.field
if isNamefield:
sorted_fields = ["name"]
else:
first = contact.name
last = None
sorted_fields = ["first_name", "last_name"]
# We want both of these to be lists, as it is richer test wise.
name_tuple = self.coerced_fk_field_helper(first, last, field.name, ":")
if name_tuple is not None:
current_sort_order_coerced_type.append(name_tuple)
desired_order = self.order_by_desired_field_helper(model_admin, request, field.name, *sorted_fields)
current_sort_order = list(model_admin.formfield_for_foreignkey(field, request).queryset)
self.assertEqual(
desired_order,
current_sort_order_coerced_type,
"{} is not ordered alphabetically".format(field.name),
)
# Conforms to the same object structure as desired_order
current_sort_order_coerced_type = []
# This is necessary as .queryset and get_queryset
# return lists of different types/structures.
# We need to parse this data and coerce them into the same type.
for contact in current_sort_order:
if not isNamefield:
first = contact.first_name
last = contact.last_name
else:
first = contact.name
last = None
name_tuple = self.coerced_fk_field_helper(first, last, field.name, ":")
if name_tuple is not None:
current_sort_order_coerced_type.append(name_tuple)
self.assertEqual(
desired_order,
current_sort_order_coerced_type,
"{} is not ordered alphabetically".format(field.name),
)
def test_alphabetically_sorted_fk_fields_domain_information(self):
with less_console_noise():
tested_fields = [
DomainInformation.authorizing_official.field,
DomainInformation.submitter.field,
# DomainInformation.creator.field,
(DomainInformation.domain.field, ["name"]),
(DomainInformation.domain_application.field, ["requested_domain__name"]),
]
# Creates multiple domain applications - review status does not matter
applications = multiple_unalphabetical_domain_objects("information")
tested_fields = [
DomainInformation.authorizing_official.field,
DomainInformation.submitter.field,
# DomainInformation.creator.field,
(DomainInformation.domain.field, ["name"]),
(DomainInformation.domain_application.field, ["requested_domain__name"]),
]
# Creates multiple domain applications - review status does not matter
applications = multiple_unalphabetical_domain_objects("information")
# Create a mock request
request = self.factory.post("/admin/registrar/domaininformation/{}/change/".format(applications[0].pk))
# Create a mock request
request = self.factory.post("/admin/registrar/domaininformation/{}/change/".format(applications[0].pk))
model_admin = AuditedAdmin(DomainInformation, self.site)
model_admin = AuditedAdmin(DomainInformation, self.site)
sorted_fields = []
# Typically we wouldn't want two nested for fields,
# but both fields are of a fixed length.
# For test case purposes, this should be performant.
for field in tested_fields:
isOtherOrderfield: bool = isinstance(field, tuple)
field_obj = None
if isOtherOrderfield:
sorted_fields = field[1]
field_obj = field[0]
else:
sorted_fields = ["first_name", "last_name"]
field_obj = field
# We want both of these to be lists, as it is richer test wise.
desired_order = self.order_by_desired_field_helper(model_admin, request, field_obj.name, *sorted_fields)
current_sort_order = list(model_admin.formfield_for_foreignkey(field_obj, request).queryset)
sorted_fields = []
# Typically we wouldn't want two nested for fields,
# but both fields are of a fixed length.
# For test case purposes, this should be performant.
for field in tested_fields:
isOtherOrderfield: bool = isinstance(field, tuple)
field_obj = None
if isOtherOrderfield:
sorted_fields = field[1]
field_obj = field[0]
else:
sorted_fields = ["first_name", "last_name"]
field_obj = field
# We want both of these to be lists, as it is richer test wise.
desired_order = self.order_by_desired_field_helper(model_admin, request, field_obj.name, *sorted_fields)
current_sort_order = list(model_admin.formfield_for_foreignkey(field_obj, request).queryset)
# Conforms to the same object structure as desired_order
current_sort_order_coerced_type = []
# Conforms to the same object structure as desired_order
current_sort_order_coerced_type = []
# This is necessary as .queryset and get_queryset
# return lists of different types/structures.
# We need to parse this data and coerce them into the same type.
for obj in current_sort_order:
last = None
if not isOtherOrderfield:
first = obj.first_name
last = obj.last_name
elif field_obj == DomainInformation.domain.field:
first = obj.name
elif field_obj == DomainInformation.domain_application.field:
first = obj.requested_domain.name
# This is necessary as .queryset and get_queryset
# return lists of different types/structures.
# We need to parse this data and coerce them into the same type.
for obj in current_sort_order:
last = None
if not isOtherOrderfield:
first = obj.first_name
last = obj.last_name
elif field_obj == DomainInformation.domain.field:
first = obj.name
elif field_obj == DomainInformation.domain_application.field:
first = obj.requested_domain.name
name_tuple = self.coerced_fk_field_helper(first, last, field_obj.name, ":")
if name_tuple is not None:
current_sort_order_coerced_type.append(name_tuple)
name_tuple = self.coerced_fk_field_helper(first, last, field_obj.name, ":")
if name_tuple is not None:
current_sort_order_coerced_type.append(name_tuple)
self.assertEqual(
desired_order,
current_sort_order_coerced_type,
"{} is not ordered alphabetically".format(field_obj.name),
)
self.assertEqual(
desired_order,
current_sort_order_coerced_type,
"{} is not ordered alphabetically".format(field_obj.name),
)
def test_alphabetically_sorted_fk_fields_domain_invitation(self):
with less_console_noise():
tested_fields = [DomainInvitation.domain.field]
tested_fields = [DomainInvitation.domain.field]
# Creates multiple domain applications - review status does not matter
applications = multiple_unalphabetical_domain_objects("invitation")
# Creates multiple domain applications - review status does not matter
applications = multiple_unalphabetical_domain_objects("invitation")
# Create a mock request
request = self.factory.post("/admin/registrar/domaininvitation/{}/change/".format(applications[0].pk))
# Create a mock request
request = self.factory.post("/admin/registrar/domaininvitation/{}/change/".format(applications[0].pk))
model_admin = AuditedAdmin(DomainInvitation, self.site)
model_admin = AuditedAdmin(DomainInvitation, self.site)
sorted_fields = []
# Typically we wouldn't want two nested for fields,
# but both fields are of a fixed length.
# For test case purposes, this should be performant.
for field in tested_fields:
sorted_fields = ["name"]
# We want both of these to be lists, as it is richer test wise.
sorted_fields = []
# Typically we wouldn't want two nested for fields,
# but both fields are of a fixed length.
# For test case purposes, this should be performant.
for field in tested_fields:
sorted_fields = ["name"]
# We want both of these to be lists, as it is richer test wise.
desired_order = self.order_by_desired_field_helper(model_admin, request, field.name, *sorted_fields)
current_sort_order = list(model_admin.formfield_for_foreignkey(field, request).queryset)
desired_order = self.order_by_desired_field_helper(model_admin, request, field.name, *sorted_fields)
current_sort_order = list(model_admin.formfield_for_foreignkey(field, request).queryset)
# Conforms to the same object structure as desired_order
current_sort_order_coerced_type = []
# Conforms to the same object structure as desired_order
current_sort_order_coerced_type = []
# This is necessary as .queryset and get_queryset
# return lists of different types/structures.
# We need to parse this data and coerce them into the same type.
for contact in current_sort_order:
first = contact.name
last = None
# This is necessary as .queryset and get_queryset
# return lists of different types/structures.
# We need to parse this data and coerce them into the same type.
for contact in current_sort_order:
first = contact.name
last = None
name_tuple = self.coerced_fk_field_helper(first, last, field.name, ":")
if name_tuple is not None:
current_sort_order_coerced_type.append(name_tuple)
name_tuple = self.coerced_fk_field_helper(first, last, field.name, ":")
if name_tuple is not None:
current_sort_order_coerced_type.append(name_tuple)
self.assertEqual(
desired_order,
current_sort_order_coerced_type,
"{} is not ordered alphabetically".format(field.name),
)
self.assertEqual(
desired_order,
current_sort_order_coerced_type,
"{} is not ordered alphabetically".format(field.name),
)
def coerced_fk_field_helper(self, first_name, last_name, field_name, queryset_shorthand):
"""Handles edge cases for test cases"""

164
src/registrar/tests/test_models_domain.py
View file

@ -20,7 +20,7 @@ from registrar.models.user import User
from registrar.utility.errors import ActionNotAllowed, NameserverError
from registrar.models.utility.contact_error import ContactError, ContactErrorCodes
from registrar.utility import errors
from django_fsm import TransitionNotAllowed # type: ignore
from epplibwrapper import (
@ -96,7 +96,7 @@ class TestDomainCache(MockEppLib):
self.mockedSendFunction.assert_has_calls(expectedCalls)
def test_cache_nested_elements(self):
def test_cache_nested_elements_not_subdomain(self):
"""Cache works correctly with the nested objects cache and hosts"""
with less_console_noise():
domain, _ = Domain.objects.get_or_create(name="igorville.gov")
@ -113,7 +113,7 @@ class TestDomainCache(MockEppLib):
}
expectedHostsDict = {
"name": self.mockDataInfoDomain.hosts[0],
"addrs": [item.addr for item in self.mockDataInfoHosts.addrs],
"addrs": [], # should return empty bc fake.host.com is not a subdomain of igorville.gov
"cr_date": self.mockDataInfoHosts.cr_date,
}
@ -138,6 +138,59 @@ class TestDomainCache(MockEppLib):
# invalidate cache
domain._cache = {}
# get host
domain._get_property("hosts")
# Should return empty bc fake.host.com is not a subdomain of igorville.gov
self.assertEqual(domain._cache["hosts"], [expectedHostsDict])
# get contacts
domain._get_property("contacts")
self.assertEqual(domain._cache["hosts"], [expectedHostsDict])
self.assertEqual(domain._cache["contacts"], expectedContactsDict)
def test_cache_nested_elements_is_subdomain(self):
"""Cache works correctly with the nested objects cache and hosts"""
with less_console_noise():
domain, _ = Domain.objects.get_or_create(name="meoward.gov")
# The contact list will initially contain objects of type 'DomainContact'
# this is then transformed into PublicContact, and cache should NOT
# hold onto the DomainContact object
expectedUnfurledContactsList = [
common.DomainContact(contact="123", type="security"),
]
expectedContactsDict = {
PublicContact.ContactTypeChoices.ADMINISTRATIVE: None,
PublicContact.ContactTypeChoices.SECURITY: "123",
PublicContact.ContactTypeChoices.TECHNICAL: None,
}
expectedHostsDict = {
"name": self.mockDataInfoDomainSubdomain.hosts[0],
"addrs": [item.addr for item in self.mockDataInfoHosts.addrs],
"cr_date": self.mockDataInfoHosts.cr_date,
}
# this can be changed when the getter for contacts is implemented
domain._get_property("contacts")
# check domain info is still correct and not overridden
self.assertEqual(domain._cache["auth_info"], self.mockDataInfoDomainSubdomain.auth_info)
self.assertEqual(domain._cache["cr_date"], self.mockDataInfoDomainSubdomain.cr_date)
# check contacts
self.assertEqual(domain._cache["_contacts"], self.mockDataInfoDomainSubdomain.contacts)
# The contact list should not contain what is sent by the registry by default,
# as _fetch_cache will transform the type to PublicContact
self.assertNotEqual(domain._cache["contacts"], expectedUnfurledContactsList)
self.assertEqual(domain._cache["contacts"], expectedContactsDict)
# get and check hosts is set correctly
domain._get_property("hosts")
self.assertEqual(domain._cache["hosts"], [expectedHostsDict])
self.assertEqual(domain._cache["contacts"], expectedContactsDict)
# invalidate cache
domain._cache = {}
# get host
domain._get_property("hosts")
self.assertEqual(domain._cache["hosts"], [expectedHostsDict])
@ -502,16 +555,26 @@ class TestDomainAvailable(MockEppLib):
self.assertFalse(available)
patcher.stop()
def test_domain_available_with_value_error(self):
def test_domain_available_with_invalid_error(self):
"""
Scenario: Testing whether an invalid domain is available
Should throw ValueError
Should throw InvalidDomainError
Validate ValueError is raised
Validate InvalidDomainError is raised
"""
with self.assertRaises(ValueError):
with self.assertRaises(errors.InvalidDomainError):
Domain.available("invalid-string")
def test_domain_available_with_empty_string(self):
"""
Scenario: Testing whether an empty string domain name is available
Should throw InvalidDomainError
Validate InvalidDomainError is raised
"""
with self.assertRaises(errors.InvalidDomainError):
Domain.available("")
def test_domain_available_unsuccessful(self):
"""
Scenario: Testing behavior when registry raises a RegistryError
@ -1572,31 +1635,100 @@ class TestRegistrantNameservers(MockEppLib):
self.assertEqual(nameservers[0][1], ["1.1.1.1"])
patcher.stop()
def test_nameservers_stored_on_fetch_cache(self):
def test_nameservers_stored_on_fetch_cache_a_subdomain_with_ip(self):
"""
#1: Nameserver is a subdomain, and has an IP address
referenced by mockDataInfoDomainSubdomainAndIPAddress
"""
with less_console_noise():
# make the domain
domain, _ = Domain.objects.get_or_create(name="meow.gov", state=Domain.State.READY)
# mock the get_or_create methods for Host and HostIP
with patch.object(Host.objects, "get_or_create") as mock_host_get_or_create, patch.object(
HostIP.objects, "get_or_create"
) as mock_host_ip_get_or_create:
mock_host_get_or_create.return_value = (Host(domain=domain), True)
mock_host_ip_get_or_create.return_value = (HostIP(), True)
# force fetch_cache to be called, which will return above documented mocked hosts
domain.nameservers
mock_host_get_or_create.assert_called_once_with(domain=domain, name="fake.meow.gov")
# Retrieve the mocked_host from the return value of the mock
actual_mocked_host, _ = mock_host_get_or_create.return_value
mock_host_ip_get_or_create.assert_called_with(address="2.0.0.8", host=actual_mocked_host)
self.assertEqual(mock_host_ip_get_or_create.call_count, 1)
def test_nameservers_stored_on_fetch_cache_a_subdomain_without_ip(self):
"""
#2: Nameserver is a subdomain, but doesn't have an IP address associated
referenced by mockDataInfoDomainSubdomainNoIP
"""
with less_console_noise():
# make the domain
domain, _ = Domain.objects.get_or_create(name="subdomainwoip.gov", state=Domain.State.READY)
# mock the get_or_create methods for Host and HostIP
with patch.object(Host.objects, "get_or_create") as mock_host_get_or_create, patch.object(
HostIP.objects, "get_or_create"
) as mock_host_ip_get_or_create:
mock_host_get_or_create.return_value = (Host(domain=domain), True)
mock_host_ip_get_or_create.return_value = (HostIP(), True)
# force fetch_cache to be called, which will return above documented mocked hosts
domain.nameservers
mock_host_get_or_create.assert_called_once_with(domain=domain, name="fake.subdomainwoip.gov")
mock_host_ip_get_or_create.assert_not_called()
self.assertEqual(mock_host_ip_get_or_create.call_count, 0)
def test_nameservers_stored_on_fetch_cache_not_subdomain_with_ip(self):
"""
Scenario: Nameservers are stored in db when they are retrieved from fetch_cache.
Verify the success of this by asserting get_or_create calls to db.
The mocked data for the EPP calls returns a host name
of 'fake.host.com' from InfoDomain and an array of 2 IPs: 1.2.3.4 and 2.3.4.5
from InfoHost
#3: Nameserver is not a subdomain, but it does have an IP address returned
due to how we set up our defaults
"""
with less_console_noise():
domain, _ = Domain.objects.get_or_create(name="fake.gov", state=Domain.State.READY)
# mock the get_or_create methods for Host and HostIP
with patch.object(Host.objects, "get_or_create") as mock_host_get_or_create, patch.object(
HostIP.objects, "get_or_create"
) as mock_host_ip_get_or_create:
# Set the return value for the mocks
mock_host_get_or_create.return_value = (Host(), True)
mock_host_get_or_create.return_value = (Host(domain=domain), True)
mock_host_ip_get_or_create.return_value = (HostIP(), True)
# force fetch_cache to be called, which will return above documented mocked hosts
domain.nameservers
# assert that the mocks are called
mock_host_get_or_create.assert_called_once_with(domain=domain, name="fake.host.com")
# Retrieve the mocked_host from the return value of the mock
actual_mocked_host, _ = mock_host_get_or_create.return_value
mock_host_ip_get_or_create.assert_called_with(address="2.3.4.5", host=actual_mocked_host)
self.assertEqual(mock_host_ip_get_or_create.call_count, 2)
mock_host_ip_get_or_create.assert_not_called()
self.assertEqual(mock_host_ip_get_or_create.call_count, 0)
def test_nameservers_stored_on_fetch_cache_not_subdomain_without_ip(self):
"""
#4: Nameserver is not a subdomain and doesn't have an associated IP address
referenced by self.mockDataInfoDomainNotSubdomainNoIP
"""
with less_console_noise():
domain, _ = Domain.objects.get_or_create(name="fakemeow.gov", state=Domain.State.READY)
with patch.object(Host.objects, "get_or_create") as mock_host_get_or_create, patch.object(
HostIP.objects, "get_or_create"
) as mock_host_ip_get_or_create:
mock_host_get_or_create.return_value = (Host(domain=domain), True)
mock_host_ip_get_or_create.return_value = (HostIP(), True)
# force fetch_cache to be called, which will return above documented mocked hosts
domain.nameservers
mock_host_get_or_create.assert_called_once_with(domain=domain, name="fake.meow.com")
mock_host_ip_get_or_create.assert_not_called()
self.assertEqual(mock_host_ip_get_or_create.call_count, 0)
@skip("not implemented yet")
def test_update_is_unsuccessful(self):

110
src/registrar/tests/test_reports.py
View file

@ -7,13 +7,14 @@ from registrar.models.domain import Domain
from registrar.models.public_contact import PublicContact
from registrar.models.user import User
from django.contrib.auth import get_user_model
from registrar.models.user_domain_role import UserDomainRole
from registrar.tests.common import MockEppLib
from registrar.utility.csv_export import (
write_header,
write_body,
write_csv,
get_default_start_date,
get_default_end_date,
)
from django.core.management import call_command
from unittest.mock import MagicMock, call, mock_open, patch
from api.views import get_current_federal, get_current_full
@ -336,11 +337,30 @@ class ExportDataTest(MockEppLib):
federal_agency="Armed Forces Retirement Home",
)
meoward_user = get_user_model().objects.create(
username="meoward_username", first_name="first_meoward", last_name="last_meoward", email="meoward@rocks.com"
)
# Test for more than 1 domain manager
_, created = UserDomainRole.objects.get_or_create(
user=meoward_user, domain=self.domain_1, role=UserDomainRole.Roles.MANAGER
)
_, created = UserDomainRole.objects.get_or_create(
user=self.user, domain=self.domain_1, role=UserDomainRole.Roles.MANAGER
)
# Test for just 1 domain manager
_, created = UserDomainRole.objects.get_or_create(
user=meoward_user, domain=self.domain_2, role=UserDomainRole.Roles.MANAGER
)
def tearDown(self):
PublicContact.objects.all().delete()
Domain.objects.all().delete()
DomainInformation.objects.all().delete()
User.objects.all().delete()
UserDomainRole.objects.all().delete()
super().tearDown()
def test_export_domains_to_writer_security_emails(self):
@ -383,8 +403,10 @@ class ExportDataTest(MockEppLib):
}
self.maxDiff = None
# Call the export functions
write_header(writer, columns)
write_body(writer, columns, sort_fields, filter_condition)
write_csv(
writer, columns, sort_fields, filter_condition, get_domain_managers=False, should_write_header=True
)
# Reset the CSV file's position to the beginning
csv_file.seek(0)
# Read the content into a variable
@ -405,7 +427,7 @@ class ExportDataTest(MockEppLib):
expected_content = expected_content.replace(",,", "").replace(",", "").replace(" ", "").strip()
self.assertEqual(csv_content, expected_content)
def test_write_body(self):
def test_write_csv(self):
"""Test that write_body returns the
existing domain, test that sort by domain name works,
test that filter works"""
@ -440,8 +462,9 @@ class ExportDataTest(MockEppLib):
],
}
# Call the export functions
write_header(writer, columns)
write_body(writer, columns, sort_fields, filter_condition)
write_csv(
writer, columns, sort_fields, filter_condition, get_domain_managers=False, should_write_header=True
)
# Reset the CSV file's position to the beginning
csv_file.seek(0)
# Read the content into a variable
@ -489,8 +512,9 @@ class ExportDataTest(MockEppLib):
],
}
# Call the export functions
write_header(writer, columns)
write_body(writer, columns, sort_fields, filter_condition)
write_csv(
writer, columns, sort_fields, filter_condition, get_domain_managers=False, should_write_header=True
)
# Reset the CSV file's position to the beginning
csv_file.seek(0)
# Read the content into a variable
@ -567,20 +591,22 @@ class ExportDataTest(MockEppLib):
}
# Call the export functions
write_header(writer, columns)
write_body(
write_csv(
writer,
columns,
sort_fields,
filter_condition,
get_domain_managers=False,
should_write_header=True,
)
write_body(
write_csv(
writer,
columns,
sort_fields_for_deleted_domains,
filter_conditions_for_deleted_domains,
get_domain_managers=False,
should_write_header=False,
)
# Reset the CSV file's position to the beginning
csv_file.seek(0)
@ -606,6 +632,64 @@ class ExportDataTest(MockEppLib):
self.assertEqual(csv_content, expected_content)
def test_export_domains_to_writer_domain_managers(self):
"""Test that export_domains_to_writer returns the
expected domain managers"""
with less_console_noise():
# Create a CSV file in memory
csv_file = StringIO()
writer = csv.writer(csv_file)
# Define columns, sort fields, and filter condition
columns = [
"Domain name",
"Status",
"Expiration date",
"Domain type",
"Agency",
"Organization name",
"City",
"State",
"AO",
"AO email",
"Security contact email",
]
sort_fields = ["domain__name"]
filter_condition = {
"domain__state__in": [
Domain.State.READY,
Domain.State.DNS_NEEDED,
Domain.State.ON_HOLD,
],
}
self.maxDiff = None
# Call the export functions
write_csv(
writer, columns, sort_fields, filter_condition, get_domain_managers=True, should_write_header=True
)
# Reset the CSV file's position to the beginning
csv_file.seek(0)
# Read the content into a variable
csv_content = csv_file.read()
# We expect READY domains,
# sorted alphabetially by domain name
expected_content = (
"Domain name,Status,Expiration date,Domain type,Agency,"
"Organization name,City,State,AO,AO email,"
"Security contact email,Domain manager email 1,Domain manager email 2,\n"
"adomain10.gov,Ready,,Federal,Armed Forces Retirement Home,,,, , ,\n"
"adomain2.gov,Dns needed,,Interstate,,,,, , , ,meoward@rocks.com\n"
"cdomain1.gov,Ready,,Federal - Executive,World War I Centennial Commission,,,"
", , , ,meoward@rocks.com,info@example.com\n"
"ddomain3.gov,On hold,,Federal,Armed Forces Retirement Home,,,, , , ,,\n"
)
# Normalize line endings and remove commas,
# spaces and leading/trailing whitespace
csv_content = csv_content.replace(",,", "").replace(",", "").replace(" ", "").replace("\r\n", "\n").strip()
expected_content = expected_content.replace(",,", "").replace(",", "").replace(" ", "").strip()
self.assertEqual(csv_content, expected_content)
class HelperFunctions(TestCase):
"""This asserts that 1=1. Its limited usefulness lies in making sure the helper methods stay healthy."""

12
src/registrar/tests/test_views_domain.py
View file

@ -821,14 +821,15 @@ class TestDomainNameservers(TestDomainOverview):
nameserver1 = "ns1.igorville.gov"
nameserver2 = "ns2.igorville.gov"
valid_ip = "1.1. 1.1"
# initial nameservers page has one server with two ips
valid_ip_2 = "2.2. 2.2"
# have to throw an error in order to test that the whitespace has been stripped from ip
nameservers_page = self.app.get(reverse("domain-dns-nameservers", kwargs={"pk": self.domain.id}))
session_id = self.app.cookies[settings.SESSION_COOKIE_NAME]
self.app.set_cookie(settings.SESSION_COOKIE_NAME, session_id)
# attempt to submit the form without one host and an ip with whitespace
nameservers_page.form["form-0-server"] = nameserver1
nameservers_page.form["form-1-ip"] = valid_ip
nameservers_page.form["form-0-ip"] = valid_ip
nameservers_page.form["form-1-ip"] = valid_ip_2
nameservers_page.form["form-1-server"] = nameserver2
with less_console_noise(): # swallow log warning message
result = nameservers_page.form.submit()
@ -937,15 +938,14 @@ class TestDomainNameservers(TestDomainOverview):
nameserver1 = "ns1.igorville.gov"
nameserver2 = "ns2.igorville.gov"
valid_ip = "127.0.0.1"
# initial nameservers page has one server with two ips
valid_ip_2 = "128.0.0.2"
nameservers_page = self.app.get(reverse("domain-dns-nameservers", kwargs={"pk": self.domain.id}))
session_id = self.app.cookies[settings.SESSION_COOKIE_NAME]
self.app.set_cookie(settings.SESSION_COOKIE_NAME, session_id)
# attempt to submit the form without two hosts, both subdomains,
# only one has ips
nameservers_page.form["form-0-server"] = nameserver1
nameservers_page.form["form-0-ip"] = valid_ip
nameservers_page.form["form-1-server"] = nameserver2
nameservers_page.form["form-1-ip"] = valid_ip
nameservers_page.form["form-1-ip"] = valid_ip_2
with less_console_noise(): # swallow log warning message
result = nameservers_page.form.submit()
# form submission was a successful post, response should be a 302

114
src/registrar/utility/csv_export.py
View file

@ -17,8 +17,9 @@ logger = logging.getLogger(__name__)
def write_header(writer, columns):
"""
Receives params from the parent methods and outputs a CSV with a header row.
Works with write_header as longas the same writer object is passed.
Works with write_header as long as the same writer object is passed.
"""
writer.writerow(columns)
@ -43,7 +44,7 @@ def get_domain_infos(filter_condition, sort_fields):
return domain_infos_cleaned
def parse_row(columns, domain_info: DomainInformation, security_emails_dict=None):
def parse_row(columns, domain_info: DomainInformation, security_emails_dict=None, get_domain_managers=False):
"""Given a set of columns, generate a new row from cleaned column data"""
# Domain should never be none when parsing this information
@ -77,6 +78,8 @@ def parse_row(columns, domain_info: DomainInformation, security_emails_dict=None
# create a dictionary of fields which can be included in output
FIELDS = {
"Domain name": domain.name,
"Status": domain.get_state_display(),
"Expiration date": domain.expiration_date,
"Domain type": domain_type,
"Agency": domain_info.federal_agency,
"Organization name": domain_info.organization_name,
@ -85,39 +88,27 @@ def parse_row(columns, domain_info: DomainInformation, security_emails_dict=None
"AO": domain_info.ao, # type: ignore
"AO email": domain_info.authorizing_official.email if domain_info.authorizing_official else " ",
"Security contact email": security_email,
"Status": domain.get_state_display(),
"Expiration date": domain.expiration_date,
"Created at": domain.created_at,
"First ready": domain.first_ready,
"Deleted": domain.deleted,
}
# user_emails = [user.email for user in domain.permissions]
if get_domain_managers:
# Get each domain managers email and add to list
dm_emails = [dm.user.email for dm in domain.permissions.all()]
# Dynamically add user emails to the FIELDS dictionary
# for i, user_email in enumerate(user_emails, start=1):
# FIELDS[f"User{i} email"] = user_email
# Set up the "matching header" + row field data
for i, dm_email in enumerate(dm_emails, start=1):
FIELDS[f"Domain manager email {i}"] = dm_email
row = [FIELDS.get(column, "") for column in columns]
return row
def write_body(
writer,
columns,
sort_fields,
filter_condition,
):
def _get_security_emails(sec_contact_ids):
"""
Receives params from the parent methods and outputs a CSV with fltered and sorted domains.
Works with write_header as longas the same writer object is passed.
Retrieve security contact emails for the given security contact IDs.
"""
# Get the domainInfos
all_domain_infos = get_domain_infos(filter_condition, sort_fields)
# Store all security emails to avoid epp calls or excessive filters
sec_contact_ids = all_domain_infos.values_list("domain__security_contact_registry_id", flat=True)
security_emails_dict = {}
public_contacts = (
PublicContact.objects.only("email", "domain__name")
@ -133,24 +124,55 @@ def write_body(
else:
logger.warning("csv_export -> Domain was none for PublicContact")
# all_user_nums = 0
# for domain_info in all_domain_infos:
# user_num = len(domain_info.domain.permissions)
# all_user_nums.append(user_num)
return security_emails_dict
# if user_num > highest_user_nums:
# highest_user_nums = user_num
# Build the header here passing to it highest_user_nums
def update_columns_with_domain_managers(columns, max_dm_count):
"""
Update the columns list to include "Domain manager email {#}" headers
based on the maximum domain manager count.
"""
for i in range(1, max_dm_count + 1):
columns.append(f"Domain manager email {i}")
def write_csv(
writer,
columns,
sort_fields,
filter_condition,
get_domain_managers=False,
should_write_header=True,
):
"""
Receives params from the parent methods and outputs a CSV with fltered and sorted domains.
Works with write_header as longas the same writer object is passed.
get_domain_managers: Conditional bc we only use domain manager info for export_data_full_to_csv
should_write_header: Conditional bc export_data_growth_to_csv calls write_body twice
"""
all_domain_infos = get_domain_infos(filter_condition, sort_fields)
# Store all security emails to avoid epp calls or excessive filters
sec_contact_ids = all_domain_infos.values_list("domain__security_contact_registry_id", flat=True)
security_emails_dict = _get_security_emails(sec_contact_ids)
# Reduce the memory overhead when performing the write operation
paginator = Paginator(all_domain_infos, 1000)
if get_domain_managers and len(all_domain_infos) > 0:
# We want to get the max amont of domain managers an
# account has to set the column header dynamically
max_dm_count = max(len(domain_info.domain.permissions.all()) for domain_info in all_domain_infos)
update_columns_with_domain_managers(columns, max_dm_count)
for page_num in paginator.page_range:
page = paginator.page(page_num)
rows = []
for domain_info in page.object_list:
try:
row = parse_row(columns, domain_info, security_emails_dict)
row = parse_row(columns, domain_info, security_emails_dict, get_domain_managers)
rows.append(row)
except ValueError:
# This should not happen. If it does, just skip this row.
@ -158,7 +180,10 @@ def write_body(
logger.error("csv_export -> Error when parsing row, domain was None")
continue
writer.writerows(rows)
if should_write_header:
write_header(writer, columns)
writer.writerows(rows)
def export_data_type_to_csv(csv_file):
@ -168,6 +193,8 @@ def export_data_type_to_csv(csv_file):
# define columns to include in export
columns = [
"Domain name",
"Status",
"Expiration date",
"Domain type",
"Agency",
"Organization name",
@ -176,9 +203,9 @@ def export_data_type_to_csv(csv_file):
"AO",
"AO email",
"Security contact email",
"Status",
"Expiration date",
# For domain manager we are pass it in as a parameter below in write_body
]
# Coalesce is used to replace federal_type of None with ZZZZZ
sort_fields = [
"organization_type",
@ -193,8 +220,7 @@ def export_data_type_to_csv(csv_file):
Domain.State.ON_HOLD,
],
}
write_header(writer, columns)
write_body(writer, columns, sort_fields, filter_condition)
write_csv(writer, columns, sort_fields, filter_condition, get_domain_managers=True, should_write_header=True)
def export_data_full_to_csv(csv_file):
@ -225,8 +251,7 @@ def export_data_full_to_csv(csv_file):
Domain.State.ON_HOLD,
],
}
write_header(writer, columns)
write_body(writer, columns, sort_fields, filter_condition)
write_csv(writer, columns, sort_fields, filter_condition, get_domain_managers=False, should_write_header=True)
def export_data_federal_to_csv(csv_file):
@ -258,8 +283,7 @@ def export_data_federal_to_csv(csv_file):
Domain.State.ON_HOLD,
],
}
write_header(writer, columns)
write_body(writer, columns, sort_fields, filter_condition)
write_csv(writer, columns, sort_fields, filter_condition, get_domain_managers=False, should_write_header=True)
def get_default_start_date():
@ -326,6 +350,12 @@ def export_data_growth_to_csv(csv_file, start_date, end_date):
"domain__deleted__gte": start_date_formatted,
}
write_header(writer, columns)
write_body(writer, columns, sort_fields, filter_condition)
write_body(writer, columns, sort_fields_for_deleted_domains, filter_condition_for_deleted_domains)
write_csv(writer, columns, sort_fields, filter_condition, get_domain_managers=False, should_write_header=True)
write_csv(
writer,
columns,
sort_fields_for_deleted_domains,
filter_condition_for_deleted_domains,
get_domain_managers=False,
should_write_header=False,
)