diff --git a/src/registrar/admin.py b/src/registrar/admin.py index 36d6b4fd9..1ded3de82 100644 --- a/src/registrar/admin.py +++ b/src/registrar/admin.py @@ -1,17 +1,18 @@ +from datetime import date import logging from django import forms from django.db.models.functions import Concat, Coalesce from django.db.models import Value, CharField -from django.http import HttpResponse +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 import Contact, Domain, DomainApplication, DraftDomain, User, Website from registrar.utility import csv_export @@ -24,6 +25,7 @@ from django_fsm import TransitionNotAllowed # type: ignore from django.utils.safestring import mark_safe from django.utils.html import escape + logger = logging.getLogger(__name__) @@ -158,10 +160,8 @@ class AdminSortFields: match db_field.name: case "investigator": # We should only return users who are staff. - # Currently a fallback. Consider removing this if it is not needed. return model.objects.filter(is_staff=True).order_by(*order_by) case _: - # If no case is defined, return the default if isinstance(order_by, list) or isinstance(order_by, tuple): return model.objects.order_by(*order_by) else: @@ -201,8 +201,8 @@ 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. @@ -285,7 +285,7 @@ class UserContactInline(admin.StackedInline): model = models.Contact -class UserAdmin(BaseUserAdmin): +class MyUserAdmin(BaseUserAdmin): """Custom user admin class to use our inlines.""" class Meta: @@ -1111,8 +1111,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 @@ -1166,6 +1166,26 @@ 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) + curr_exp_date = domain.registry_expiration_date + 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") @@ -1224,6 +1244,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 @@ -1234,6 +1255,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, @@ -1422,7 +1518,7 @@ class VerifiedByStaffAdmin(ListHeaderAdmin): admin.site.unregister(LogEntry) # Unregister the default registration admin.site.register(LogEntry, CustomLogEntryAdmin) -admin.site.register(models.User, UserAdmin) +admin.site.register(models.User, MyUserAdmin) # Unregister the built-in Group model admin.site.unregister(Group) # Register UserGroup diff --git a/src/registrar/assets/js/get-gov-admin.js b/src/registrar/assets/js/get-gov-admin.js index 866c7bd7d..29aa9ce03 100644 --- a/src/registrar/assets/js/get-gov-admin.js +++ b/src/registrar/assets/js/get-gov-admin.js @@ -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)); } } diff --git a/src/registrar/assets/sass/_theme/_admin.scss b/src/registrar/assets/sass/_theme/_admin.scss index 93a0d7338..b57c6a015 100644 --- a/src/registrar/assets/sass/_theme/_admin.scss +++ b/src/registrar/assets/sass/_theme/_admin.scss @@ -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); } @@ -279,4 +283,22 @@ h1, h2, h3, // Fixes a display issue where the list was entirely white, or had too much whitespace .select2-dropdown { display: inline-grid !important; -} \ No newline at end of file +} + +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; + } +} diff --git a/src/registrar/models/domain.py b/src/registrar/models/domain.py index 7ebe3dc34..020ba43c0 100644 --- a/src/registrar/models/domain.py +++ b/src/registrar/models/domain.py @@ -265,15 +265,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 diff --git a/src/registrar/models/domain_information.py b/src/registrar/models/domain_information.py index acaa330bb..1a50efe2c 100644 --- a/src/registrar/models/domain_information.py +++ b/src/registrar/models/domain_information.py @@ -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) diff --git a/src/registrar/models/utility/domain_helper.py b/src/registrar/models/utility/domain_helper.py index 230b23e16..9e3559676 100644 --- a/src/registrar/models/utility/domain_helper.py +++ b/src/registrar/models/utility/domain_helper.py @@ -180,8 +180,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 diff --git a/src/registrar/templates/admin/base_site.html b/src/registrar/templates/admin/base_site.html index f9ff23455..73e9ba1f0 100644 --- a/src/registrar/templates/admin/base_site.html +++ b/src/registrar/templates/admin/base_site.html @@ -18,11 +18,12 @@ + + {% endblock %} {% block title %}{% if subtitle %}{{ subtitle }} | {% endif %}{{ title }} | {{ site_title|default:_('Django site admin') }}{% endblock %} - {% block extrastyle %}{{ block.super }} {% endblock %} diff --git a/src/registrar/templates/django/admin/domain_change_form.html b/src/registrar/templates/django/admin/domain_change_form.html index c4461d07f..67c5ac291 100644 --- a/src/registrar/templates/django/admin/domain_change_form.html +++ b/src/registrar/templates/django/admin/domain_change_form.html @@ -2,21 +2,117 @@ {% load i18n static %} {% block field_sets %} -
- - -
- {% if original.state == original.State.READY %} - - {% elif original.state == original.State.ON_HOLD %} - - {% endif %} - {% if original.state == original.State.READY or original.state == original.State.ON_HOLD %} - | - {% endif %} - {% if original.state != original.State.DELETED %} - +
+
+ + {# Dja has margin styles defined on inputs as is. Lets work with it, rather than fight it. #} + + +
+
+ {% if original.state != original.State.DELETED %} + + Extend expiration date + + | + {% endif %} + {% if original.state == original.State.READY %} + + {% elif original.state == original.State.ON_HOLD %} + + {% endif %} + {% if original.state == original.State.READY or original.state == original.State.ON_HOLD %} + | + {% endif %} + {% if original.state != original.State.DELETED %} + {% endif %} +
{{ 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 %} +
+
+
+

+ Are you sure you want to extend the expiration date? +

+
+

+ This will extend the expiration date by one year. + {# Acts as a
#} +

+ This action cannot be undone. +

+

+ Domain: {{ original.name }} + {# Acts as a
#} +

+ New expiration date: {{ extended_expiration_date }} + {{test}} +

+
+ +
+
    +
  • + +
    • +
  • + +
    • +
+
+
+ +
+
+{{ block.super }} +{% endblock %} \ No newline at end of file diff --git a/src/registrar/templates/home.html b/src/registrar/templates/home.html index a79065f50..93f1243ea 100644 --- a/src/registrar/templates/home.html +++ b/src/registrar/templates/home.html @@ -15,6 +15,7 @@ {% endblock %}

Manage your domains

+

diff --git a/src/registrar/tests/common.py b/src/registrar/tests/common.py index 8d2385c23..17833d689 100644 --- a/src/registrar/tests/common.py +++ b/src/registrar/tests/common.py @@ -922,6 +922,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), @@ -1050,11 +1055,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) diff --git a/src/registrar/tests/test_admin.py b/src/registrar/tests/test_admin.py index ce2b40122..949636f3e 100644 --- a/src/registrar/tests/test_admin.py +++ b/src/registrar/tests/test_admin.py @@ -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 ( @@ -9,7 +11,7 @@ from registrar.admin import ( DomainApplicationAdminForm, DomainInvitationAdmin, ListHeaderAdmin, - UserAdmin, + MyUserAdmin, AuditedAdmin, ContactAdmin, DomainInformationAdmin, @@ -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 @@ -949,7 +1124,7 @@ class TestDomainApplicationAdmin(MockEppLib): user_request = self.factory.post( "/admin/autocomplete/?app_label=registrar&model_name=domainapplication&field_name=investigator" ) - user_admin = UserAdmin(User, self.site) + user_admin = MyUserAdmin(User, self.site) user_queryset = user_admin.get_search_results(user_request, application_queryset, None)[0] current_dropdown = list(user_queryset) @@ -1350,10 +1525,10 @@ class ListHeaderAdminTest(TestCase): User.objects.all().delete() -class UserAdminTest(TestCase): +class MyUserAdminTest(TestCase): def setUp(self): admin_site = AdminSite() - self.admin = UserAdmin(model=get_user_model(), admin_site=admin_site) + self.admin = MyUserAdmin(model=get_user_model(), admin_site=admin_site) def test_list_display_without_username(self): request = self.client.request().wsgi_request @@ -1375,7 +1550,7 @@ class UserAdminTest(TestCase): request = self.client.request().wsgi_request request.user = create_superuser() fieldsets = self.admin.get_fieldsets(request) - expected_fieldsets = super(UserAdmin, self.admin).get_fieldsets(request) + expected_fieldsets = super(MyUserAdmin, self.admin).get_fieldsets(request) self.assertEqual(fieldsets, expected_fieldsets) def test_get_fieldsets_cisa_analyst(self): @@ -1435,7 +1610,7 @@ class AuditedAdminTest(TestCase): desired_sort_order = list(User.objects.filter(is_staff=True).order_by(*sorted_fields)) # Grab the data returned from get search results - admin = UserAdmin(User, self.site) + admin = MyUserAdmin(User, self.site) search_queryset = admin.get_search_results(request, application_queryset, None)[0] current_sort_order = list(search_queryset) diff --git a/src/registrar/tests/test_reports.py b/src/registrar/tests/test_reports.py index 630904218..011c60b93 100644 --- a/src/registrar/tests/test_reports.py +++ b/src/registrar/tests/test_reports.py @@ -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.""" diff --git a/src/registrar/utility/csv_export.py b/src/registrar/utility/csv_export.py index a188fb91c..90e80f551 100644 --- a/src/registrar/utility/csv_export.py +++ b/src/registrar/utility/csv_export.py @@ -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, + )