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

Merge branch 'main' into ms/3125-fix-sandbox-log-output

Browse source
This commit is contained in:
Matt-Spence 2025-02-06 14:05:58 -05:00 • committed by GitHub
commit 37d160bd1e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
21 changed files with 1980 additions and 232 deletions
Download patch file Download diff file Expand all files Collapse all files

61
src/registrar/admin.py
View file

@ -28,7 +28,11 @@ from django.shortcuts import redirect
from django_fsm import get_available_FIELD_transitions, FSMField
from registrar.models import DomainInformation, Portfolio, UserPortfolioPermission, DomainInvitation
from registrar.models.utility.portfolio_helper import UserPortfolioPermissionChoices, UserPortfolioRoleChoices
from registrar.utility.email_invitations import send_domain_invitation_email, send_portfolio_invitation_email
from registrar.utility.email_invitations import (
send_domain_invitation_email,
send_portfolio_admin_addition_emails,
send_portfolio_invitation_email,
)
from registrar.views.utility.invitation_helper import (
get_org_membership,
get_requested_user,
@ -1551,7 +1555,9 @@ class DomainInvitationAdmin(BaseInvitationAdmin):
and not member_of_this_org
and not member_of_a_different_org
):
send_portfolio_invitation_email(email=requested_email, requestor=requestor, portfolio=domain_org)
send_portfolio_invitation_email(
email=requested_email, requestor=requestor, portfolio=domain_org, is_admin_invitation=False
)
portfolio_invitation, _ = PortfolioInvitation.objects.get_or_create(
email=requested_email,
portfolio=domain_org,
@ -1642,30 +1648,57 @@ class PortfolioInvitationAdmin(BaseInvitationAdmin):
Emails sent to requested user / email.
When exceptions are raised, return without saving model.
"""
if not change: # Only send email if this is a new PortfolioInvitation (creation)
try:
portfolio = obj.portfolio
requested_email = obj.email
requestor = request.user
# Look up a user with that email
requested_user = get_requested_user(requested_email)
is_admin_invitation = UserPortfolioRoleChoices.ORGANIZATION_ADMIN in obj.roles
if not change: # Only send email if this is a new PortfolioInvitation (creation)
# Look up a user with that email
requested_user = get_requested_user(requested_email)
permission_exists = UserPortfolioPermission.objects.filter(
user__email=requested_email, portfolio=portfolio, user__email__isnull=False
).exists()
try:
permission_exists = UserPortfolioPermission.objects.filter(
user__email=requested_email, portfolio=portfolio, user__email__isnull=False
).exists()
if not permission_exists:
# if permission does not exist for a user with requested_email, send email
send_portfolio_invitation_email(email=requested_email, requestor=requestor, portfolio=portfolio)
if not send_portfolio_invitation_email(
email=requested_email,
requestor=requestor,
portfolio=portfolio,
is_admin_invitation=is_admin_invitation,
):
messages.warning(
self.request, "Could not send email notification to existing organization admins."
)
# if user exists for email, immediately retrieve portfolio invitation upon creation
if requested_user is not None:
obj.retrieve()
messages.success(request, f"{requested_email} has been invited.")
else:
messages.warning(request, "User is already a member of this portfolio.")
except Exception as e:
# when exception is raised, handle and do not save the model
handle_invitation_exceptions(request, e, requested_email)
return
else: # Handle the case when updating an existing PortfolioInvitation
# Retrieve the existing object from the database
existing_obj = PortfolioInvitation.objects.get(pk=obj.pk)
# Check if the previous roles did NOT include ORGANIZATION_ADMIN
# and the new roles DO include ORGANIZATION_ADMIN
was_not_admin = UserPortfolioRoleChoices.ORGANIZATION_ADMIN not in existing_obj.roles
# Check also if status is INVITED, ignore role changes for other statuses
is_invited = obj.status == PortfolioInvitation.PortfolioInvitationStatus.INVITED
if was_not_admin and is_admin_invitation and is_invited:
# send email to existing portfolio admins if new admin
if not send_portfolio_admin_addition_emails(
email=requested_email,
requestor=requestor,
portfolio=portfolio,
):
messages.warning(request, "Could not send email notification to existing organization admins.")
except Exception as e:
# when exception is raised, handle and do not save the model
handle_invitation_exceptions(request, e, requested_email)
return
# Call the parent save method to save the object
super().save_model(request, obj, form, change)

22
src/registrar/fixtures/fixtures_domains.py
View file

@ -3,7 +3,6 @@ from django.utils import timezone
import logging
import random
from faker import Faker
from django.db import transaction
from registrar.fixtures.fixtures_requests import DomainRequestFixture
from registrar.fixtures.fixtures_users import UserFixture
@ -29,19 +28,18 @@ class DomainFixture(DomainRequestFixture):
def load(cls):
# Lumped under .atomic to ensure we don't make redundant DB calls.
# This bundles them all together, and then saves it in a single call.
with transaction.atomic():
try:
# Get the usernames of users created in the UserFixture
created_usernames = [user_data["username"] for user_data in UserFixture.ADMINS + UserFixture.STAFF]
try:
# Get the usernames of users created in the UserFixture
created_usernames = [user_data["username"] for user_data in UserFixture.ADMINS + UserFixture.STAFF]
# Filter users to only include those created by the fixture
users = list(User.objects.filter(username__in=created_usernames))
except Exception as e:
logger.warning(e)
return
# Filter users to only include those created by the fixture
users = list(User.objects.filter(username__in=created_usernames))
except Exception as e:
logger.warning(e)
return
# Approve each user associated with `in review` status domains
cls._approve_domain_requests(users)
# Approve each user associated with `in review` status domains
cls._approve_domain_requests(users)
@staticmethod
def _generate_fake_expiration_date(days_in_future=365):

69
src/registrar/fixtures/fixtures_portfolios.py
View file

@ -1,7 +1,6 @@
import logging
import random
from faker import Faker
from django.db import transaction
from registrar.models import User, DomainRequest, FederalAgency
from registrar.models.portfolio import Portfolio
@ -84,42 +83,38 @@ class PortfolioFixture:
def load(cls):
"""Creates portfolios."""
logger.info("Going to load %s portfolios" % len(cls.PORTFOLIOS))
try:
user = User.objects.all().last()
except Exception as e:
logger.warning(e)
return
# Lumped under .atomic to ensure we don't make redundant DB calls.
# This bundles them all together, and then saves it in a single call.
with transaction.atomic():
portfolios_to_create = []
for portfolio_data in cls.PORTFOLIOS:
organization_name = portfolio_data["organization_name"]
# Check if portfolio with the organization name already exists
if Portfolio.objects.filter(organization_name=organization_name).exists():
logger.info(
f"Portfolio with organization name '{organization_name}' already exists, skipping creation."
)
continue
try:
portfolio = Portfolio(
creator=user,
organization_name=portfolio_data["organization_name"],
)
cls._set_non_foreign_key_fields(portfolio, portfolio_data)
cls._set_foreign_key_fields(portfolio, portfolio_data, user)
portfolios_to_create.append(portfolio)
except Exception as e:
logger.warning(e)
# Bulk create portfolios
if len(portfolios_to_create) > 0:
try:
user = User.objects.all().last()
Portfolio.objects.bulk_create(portfolios_to_create)
logger.info(f"Successfully created {len(portfolios_to_create)} portfolios")
except Exception as e:
logger.warning(e)
return
portfolios_to_create = []
for portfolio_data in cls.PORTFOLIOS:
organization_name = portfolio_data["organization_name"]
# Check if portfolio with the organization name already exists
if Portfolio.objects.filter(organization_name=organization_name).exists():
logger.info(
f"Portfolio with organization name '{organization_name}' already exists, skipping creation."
)
continue
try:
portfolio = Portfolio(
creator=user,
organization_name=portfolio_data["organization_name"],
)
cls._set_non_foreign_key_fields(portfolio, portfolio_data)
cls._set_foreign_key_fields(portfolio, portfolio_data, user)
portfolios_to_create.append(portfolio)
except Exception as e:
logger.warning(e)
# Bulk create domain requests
if len(portfolios_to_create) > 0:
try:
Portfolio.objects.bulk_create(portfolios_to_create)
logger.info(f"Successfully created {len(portfolios_to_create)} portfolios")
except Exception as e:
logger.warning(f"Error bulk creating portfolios: {e}")
logger.warning(f"Error bulk creating portfolios: {e}")

26
src/registrar/fixtures/fixtures_requests.py
View file

@ -3,7 +3,6 @@ from django.utils import timezone
import logging
import random
from faker import Faker
from django.db import transaction
from registrar.fixtures.fixtures_portfolios import PortfolioFixture
from registrar.fixtures.fixtures_suborganizations import SuborganizationFixture
@ -303,24 +302,17 @@ class DomainRequestFixture:
def load(cls):
"""Creates domain requests for each user in the database."""
logger.info("Going to load %s domain requests" % len(cls.DOMAINREQUESTS))
try:
# Get the usernames of users created in the UserFixture
created_usernames = [user_data["username"] for user_data in UserFixture.ADMINS + UserFixture.STAFF]
# Lumped under .atomic to ensure we don't make redundant DB calls.
# This bundles them all together, and then saves it in a single call.
# The atomic block will cause the code to stop executing if one instance in the
# nested iteration fails, which will cause an early exit and make it hard to debug.
# Comment out with transaction.atomic() when debugging.
with transaction.atomic():
try:
# Get the usernames of users created in the UserFixture
created_usernames = [user_data["username"] for user_data in UserFixture.ADMINS + UserFixture.STAFF]
# Filter users to only include those created by the fixture
users = list(User.objects.filter(username__in=created_usernames))
except Exception as e:
logger.warning(e)
return
# Filter users to only include those created by the fixture
users = list(User.objects.filter(username__in=created_usernames))
except Exception as e:
logger.warning(e)
return
cls._create_domain_requests(users)
cls._create_domain_requests(users)
@classmethod
def _create_domain_requests(cls, users): # noqa: C901

13
src/registrar/fixtures/fixtures_suborganizations.py
View file

@ -1,6 +1,5 @@
import logging
from faker import Faker
from django.db import transaction
from registrar.models.portfolio import Portfolio
from registrar.models.suborganization import Suborganization
@ -34,14 +33,12 @@ class SuborganizationFixture:
def load(cls):
"""Creates suborganizations."""
logger.info(f"Going to load {len(cls.SUBORGS)} suborgs")
portfolios = cls._get_portfolios()
if not portfolios:
return
with transaction.atomic():
portfolios = cls._get_portfolios()
if not portfolios:
return
suborgs_to_create = cls._prepare_suborgs_to_create(portfolios)
cls._bulk_create_suborgs(suborgs_to_create)
suborgs_to_create = cls._prepare_suborgs_to_create(portfolios)
cls._bulk_create_suborgs(suborgs_to_create)
@classmethod
def _get_portfolios(cls):

86
src/registrar/fixtures/fixtures_user_portfolio_permissions.py
View file

@ -1,7 +1,6 @@
import logging
import random
from faker import Faker
from django.db import transaction
from registrar.fixtures.fixtures_portfolios import PortfolioFixture
from registrar.fixtures.fixtures_users import UserFixture
@ -26,56 +25,55 @@ class UserPortfolioPermissionFixture:
# Lumped under .atomic to ensure we don't make redundant DB calls.
# This bundles them all together, and then saves it in a single call.
with transaction.atomic():
try:
# Get the usernames of users created in the UserFixture
created_usernames = [user_data["username"] for user_data in UserFixture.ADMINS + UserFixture.STAFF]
try:
# Get the usernames of users created in the UserFixture
created_usernames = [user_data["username"] for user_data in UserFixture.ADMINS + UserFixture.STAFF]
# Filter users to only include those created by the fixture
users = list(User.objects.filter(username__in=created_usernames))
# Filter users to only include those created by the fixture
users = list(User.objects.filter(username__in=created_usernames))
organization_names = [portfolio["organization_name"] for portfolio in PortfolioFixture.PORTFOLIOS]
organization_names = [portfolio["organization_name"] for portfolio in PortfolioFixture.PORTFOLIOS]
portfolios = list(Portfolio.objects.filter(organization_name__in=organization_names))
portfolios = list(Portfolio.objects.filter(organization_name__in=organization_names))
if not users:
logger.warning("User fixtures missing.")
return
if not portfolios:
logger.warning("Portfolio fixtures missing.")
return
except Exception as e:
logger.warning(e)
if not users:
logger.warning("User fixtures missing.")
return
user_portfolio_permissions_to_create = []
for user in users:
# Assign a random portfolio to a user
portfolio = random.choice(portfolios) # nosec
try:
if not UserPortfolioPermission.objects.filter(user=user, portfolio=portfolio).exists():
user_portfolio_permission = UserPortfolioPermission(
user=user,
portfolio=portfolio,
roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN],
additional_permissions=[
UserPortfolioPermissionChoices.EDIT_MEMBERS,
UserPortfolioPermissionChoices.EDIT_REQUESTS,
],
)
user_portfolio_permissions_to_create.append(user_portfolio_permission)
else:
logger.info(
f"Permission exists for user '{user.username}' "
f"on portfolio '{portfolio.organization_name}'."
)
except Exception as e:
logger.warning(e)
if not portfolios:
logger.warning("Portfolio fixtures missing.")
return
# Bulk create permissions
cls._bulk_create_permissions(user_portfolio_permissions_to_create)
except Exception as e:
logger.warning(e)
return
user_portfolio_permissions_to_create = []
for user in users:
# Assign a random portfolio to a user
portfolio = random.choice(portfolios) # nosec
try:
if not UserPortfolioPermission.objects.filter(user=user, portfolio=portfolio).exists():
user_portfolio_permission = UserPortfolioPermission(
user=user,
portfolio=portfolio,
roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN],
additional_permissions=[
UserPortfolioPermissionChoices.EDIT_MEMBERS,
UserPortfolioPermissionChoices.EDIT_REQUESTS,
],
)
user_portfolio_permissions_to_create.append(user_portfolio_permission)
else:
logger.info(
f"Permission exists for user '{user.username}' "
f"on portfolio '{portfolio.organization_name}'."
)
except Exception as e:
logger.warning(e)
# Bulk create permissions
cls._bulk_create_permissions(user_portfolio_permissions_to_create)
@classmethod
def _bulk_create_permissions(cls, user_portfolio_permissions_to_create):

12
src/registrar/fixtures/fixtures_users.py
View file

@ -1,6 +1,5 @@
import logging
from faker import Faker
from django.db import transaction
from registrar.models import (
User,
@ -455,10 +454,9 @@ class UserFixture:
@classmethod
def load(cls):
with transaction.atomic():
cls.load_users(cls.ADMINS, "full_access_group", are_superusers=True)
cls.load_users(cls.STAFF, "cisa_analysts_group")
cls.load_users(cls.ADMINS, "full_access_group", are_superusers=True)
cls.load_users(cls.STAFF, "cisa_analysts_group")
# Combine ADMINS and STAFF lists
all_users = cls.ADMINS + cls.STAFF
cls.load_allowed_emails(cls, all_users, additional_emails=cls.ADDITIONAL_ALLOWED_EMAILS)
# Combine ADMINS and STAFF lists
all_users = cls.ADMINS + cls.STAFF
cls.load_allowed_emails(cls, all_users, additional_emails=cls.ADDITIONAL_ALLOWED_EMAILS)

26
src/registrar/forms/portfolio.py
View file

@ -312,6 +312,32 @@ class BasePortfolioMemberForm(forms.ModelForm):
self.initial["domain_permissions"] = selected_domain_permission
self.initial["member_permissions"] = selected_member_permission
def is_change_from_member_to_admin(self) -> bool:
"""
Checks if the roles have changed from not containing ORGANIZATION_ADMIN
to containing ORGANIZATION_ADMIN.
"""
previous_roles = set(self.initial.get("roles", [])) # Initial roles before change
new_roles = set(self.cleaned_data.get("roles", [])) # New roles after change
return (
UserPortfolioRoleChoices.ORGANIZATION_ADMIN not in previous_roles
and UserPortfolioRoleChoices.ORGANIZATION_ADMIN in new_roles
)
def is_change_from_admin_to_member(self) -> bool:
"""
Checks if the roles have changed from containing ORGANIZATION_ADMIN
to not containing ORGANIZATION_ADMIN.
"""
previous_roles = set(self.initial.get("roles", [])) # Initial roles before change
new_roles = set(self.cleaned_data.get("roles", [])) # New roles after change
return (
UserPortfolioRoleChoices.ORGANIZATION_ADMIN in previous_roles
and UserPortfolioRoleChoices.ORGANIZATION_ADMIN not in new_roles
)
class PortfolioMemberForm(BasePortfolioMemberForm):
"""

22
src/registrar/management/commands/remove_unused_portfolios.py
View file

@ -149,9 +149,9 @@ class Command(BaseCommand):
)
return
with transaction.atomic():
# Try to delete the portfolios
try:
# Try to delete the portfolios
try:
with transaction.atomic():
summary = []
for portfolio in portfolios_to_delete:
portfolio_summary = [f"---- CASCADE SUMMARY for {portfolio.organization_name} -----"]
@ -222,14 +222,14 @@ class Command(BaseCommand):
"""
)
except IntegrityError as e:
logger.info(
f"""{TerminalColors.FAIL}
Could not delete some portfolios due to integrity constraints:
{e}
{TerminalColors.ENDC}
"""
)
except IntegrityError as e:
logger.info(
f"""{TerminalColors.FAIL}
Could not delete some portfolios due to integrity constraints:
{e}
{TerminalColors.ENDC}
"""
)
def handle(self, *args, **options):
# Get all Portfolio entries not in the allowed portfolios list

12
src/registrar/models/utility/portfolio_helper.py
View file

@ -1,5 +1,6 @@
from registrar.utility import StrEnum
from django.db import models
from django.db.models import Q
from django.apps import apps
from django.forms import ValidationError
from registrar.utility.waffle import flag_is_active_for_user
@ -132,9 +133,10 @@ def validate_user_portfolio_permission(user_portfolio_permission):
"Based on current waffle flag settings, users cannot be assigned to multiple portfolios."
)
existing_invitations = PortfolioInvitation.objects.exclude(
portfolio=user_portfolio_permission.portfolio
).filter(email=user_portfolio_permission.user.email)
existing_invitations = PortfolioInvitation.objects.filter(email=user_portfolio_permission.user.email).exclude(
Q(portfolio=user_portfolio_permission.portfolio)
| Q(status=PortfolioInvitation.PortfolioInvitationStatus.RETRIEVED)
)
if existing_invitations.exists():
raise ValidationError(
"This user is already assigned to a portfolio invitation. "
@ -191,8 +193,8 @@ def validate_portfolio_invitation(portfolio_invitation):
if not flag_is_active_for_user(user, "multiple_portfolios"):
existing_permissions = UserPortfolioPermission.objects.filter(user=user)
existing_invitations = PortfolioInvitation.objects.exclude(id=portfolio_invitation.id).filter(
email=portfolio_invitation.email
existing_invitations = PortfolioInvitation.objects.filter(email=portfolio_invitation.email).exclude(
Q(id=portfolio_invitation.id) | Q(status=PortfolioInvitation.PortfolioInvitationStatus.RETRIEVED)
)
if existing_permissions.exists():

40
src/registrar/templates/emails/portfolio_admin_addition_notification.txt Normal file
View file

@ -0,0 +1,40 @@
{% autoescape off %}{# In a text file, we don't want to have HTML entities escaped #}
Hi,{% if portfolio_admin and portfolio_admin.first_name %} {{ portfolio_admin.first_name }}.{% endif %}
An admin was invited to your .gov organization.
ORGANIZATION: {{ portfolio.organization_name }}
INVITED BY: {{ requestor_email }}
INVITED ON: {{date}}
ADMIN INVITED: {{ invited_email_address }}
----------------------------------------------------------------
NEXT STEPS
The person who received the invitation will become an admin once they log in to the
.gov registrar. They'll need to access the registrar using a Login.gov account that's
associated with the invited email address.
If you need to cancel this invitation or remove the admin, you can do that by going to
the Members section for your organization <https://manage.get.gov/>.
WHY DID YOU RECEIVE THIS EMAIL?
You’re listed as an admin for {{ portfolio.organization_name }}. That means you'll receive a notification
whenever a new admin is invited to that organization.
If you have questions or concerns, reach out to the person who sent the invitation or reply to this email.
THANK YOU
.Gov helps the public identify official, trusted information. Thank you for using a .gov domain.
----------------------------------------------------------------
The .gov team
Contact us: <https://get.gov/contact/>
Learn about .gov <https://get.gov>
The .gov registry is a part of the Cybersecurity and Infrastructure Security Agency
(CISA) <https://cisa.gov/>
{% endautoescape %}

1
src/registrar/templates/emails/portfolio_admin_addition_notification_subject.txt Normal file
View file

@ -0,0 +1 @@
An admin was invited to your .gov organization

33
src/registrar/templates/emails/portfolio_admin_removal_notification.txt Normal file
View file

@ -0,0 +1,33 @@
{% autoescape off %}{# In a text file, we don't want to have HTML entities escaped #}
Hi,{% if portfolio_admin and portfolio_admin.first_name %} {{ portfolio_admin.first_name }}.{% endif %}
An admin was removed from your .gov organization.
ORGANIZATION: {{ portfolio.organization_name }}
REMOVED BY: {{ requestor_email }}
REMOVED ON: {{date}}
ADMIN REMOVED: {{ removed_email_address }}
You can view this update by going to the Members section for your .gov organization <https://manage.get.gov/>.
----------------------------------------------------------------
WHY DID YOU RECEIVE THIS EMAIL?
You’re listed as an admin for {{ portfolio.organization_name }}. That means you'll receive a notification
whenever an admin is removed from that organization.
If you have questions or concerns, reach out to the person who removed the admin or reply to this email.
THANK YOU
.Gov helps the public identify official, trusted information. Thank you for using a .gov domain.
----------------------------------------------------------------
The .gov team
Contact us: <https://get.gov/contact/>
Learn about .gov <https://get.gov>
The .gov registry is a part of the Cybersecurity and Infrastructure Security Agency
(CISA) <https://cisa.gov/>
{% endautoescape %}

1
src/registrar/templates/emails/portfolio_admin_removal_notification_subject.txt Normal file
View file

@ -0,0 +1 @@
An admin was removed from your .gov organization

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

@ -254,6 +254,7 @@ class TestDomainInvitationAdmin(TestCase):
email="test@example.com",
requestor=self.superuser,
portfolio=self.portfolio,
is_admin_invitation=False,
)
# Assert success message
@ -504,6 +505,7 @@ class TestDomainInvitationAdmin(TestCase):
email="test@example.com",
requestor=self.superuser,
portfolio=self.portfolio,
is_admin_invitation=False,
)
# Assert retrieve on domain invite only was called
@ -567,6 +569,7 @@ class TestDomainInvitationAdmin(TestCase):
email="test@example.com",
requestor=self.superuser,
portfolio=self.portfolio,
is_admin_invitation=False,
)
# Assert retrieve on domain invite only was called
@ -693,6 +696,7 @@ class TestDomainInvitationAdmin(TestCase):
email="nonexistent@example.com",
requestor=self.superuser,
portfolio=self.portfolio,
is_admin_invitation=False,
)
# Assert retrieve was not called
@ -918,6 +922,7 @@ class TestDomainInvitationAdmin(TestCase):
email="nonexistent@example.com",
requestor=self.superuser,
portfolio=self.portfolio,
is_admin_invitation=False,
)
# Assert retrieve on domain invite only was called
@ -979,6 +984,7 @@ class TestDomainInvitationAdmin(TestCase):
email="nonexistent@example.com",
requestor=self.superuser,
portfolio=self.portfolio,
is_admin_invitation=False,
)
# Assert retrieve on domain invite only was called
@ -1204,7 +1210,7 @@ class TestPortfolioInvitationAdmin(TestCase):
@less_console_noise_decorator
@patch("registrar.admin.send_portfolio_invitation_email")
@patch("django.contrib.messages.success") # Mock the `messages.warning` call
@patch("django.contrib.messages.success") # Mock the `messages.success` call
def test_save_sends_email(self, mock_messages_success, mock_send_email):
"""On save_model, an email is sent if an invitation already exists."""
@ -1455,6 +1461,94 @@ class TestPortfolioInvitationAdmin(TestCase):
# Assert that messages.error was called with the correct message
mock_messages_error.assert_called_once_with(request, "Could not send email invitation.")
@less_console_noise_decorator
@patch("registrar.admin.send_portfolio_admin_addition_emails")
def test_save_existing_sends_email_notification(self, mock_send_email):
"""On save_model to an existing invitation, an email is set to notify existing
admins, if the invitation changes from member to admin."""
# Create an instance of the admin class
admin_instance = PortfolioInvitationAdmin(PortfolioInvitation, admin_site=None)
# Mock the response value of the email send
mock_send_email.return_value = True
# Create and save a PortfolioInvitation instance
portfolio_invitation = PortfolioInvitation.objects.create(
email="james.gordon@gotham.gov",
portfolio=self.portfolio,
roles=[UserPortfolioRoleChoices.ORGANIZATION_MEMBER], # Initially NOT an admin
status=PortfolioInvitation.PortfolioInvitationStatus.INVITED, # Must be "INVITED"
)
# Create a request object
request = self.factory.post(f"/admin/registrar/PortfolioInvitation/{portfolio_invitation.pk}/change/")
request.user = self.superuser
# Change roles from MEMBER to ADMIN
portfolio_invitation.roles = [UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
# Call the save_model method
admin_instance.save_model(request, portfolio_invitation, None, True)
# Assert that send_portfolio_admin_addition_emails is called
mock_send_email.assert_called_once()
# Get the arguments passed to send_portfolio_admin_addition_emails
_, called_kwargs = mock_send_email.call_args
# Assert the email content
self.assertEqual(called_kwargs["email"], "james.gordon@gotham.gov")
self.assertEqual(called_kwargs["requestor"], self.superuser)
self.assertEqual(called_kwargs["portfolio"], self.portfolio)
@less_console_noise_decorator
@patch("registrar.admin.send_portfolio_admin_addition_emails")
@patch("django.contrib.messages.warning") # Mock the `messages.warning` call
def test_save_existing_email_notification_warning(self, mock_messages_warning, mock_send_email):
"""On save_model for an existing invitation, a warning is displayed if method to
send email to notify admins returns False."""
# Create an instance of the admin class
admin_instance = PortfolioInvitationAdmin(PortfolioInvitation, admin_site=None)
# Mock the response value of the email send
mock_send_email.return_value = False
# Create and save a PortfolioInvitation instance
portfolio_invitation = PortfolioInvitation.objects.create(
email="james.gordon@gotham.gov",
portfolio=self.portfolio,
roles=[UserPortfolioRoleChoices.ORGANIZATION_MEMBER], # Initially NOT an admin
status=PortfolioInvitation.PortfolioInvitationStatus.INVITED, # Must be "INVITED"
)
# Create a request object
request = self.factory.post(f"/admin/registrar/PortfolioInvitation/{portfolio_invitation.pk}/change/")
request.user = self.superuser
# Change roles from MEMBER to ADMIN
portfolio_invitation.roles = [UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
# Call the save_model method
admin_instance.save_model(request, portfolio_invitation, None, True)
# Assert that send_portfolio_admin_addition_emails is called
mock_send_email.assert_called_once()
# Get the arguments passed to send_portfolio_admin_addition_emails
_, called_kwargs = mock_send_email.call_args
# Assert the email content
self.assertEqual(called_kwargs["email"], "james.gordon@gotham.gov")
self.assertEqual(called_kwargs["requestor"], self.superuser)
self.assertEqual(called_kwargs["portfolio"], self.portfolio)
# Assert that messages.error was called with the correct message
mock_messages_warning.assert_called_once_with(
request, "Could not send email notification to existing organization admins."
)
class TestHostAdmin(TestCase):
"""Tests for the HostAdmin class as super user

447
src/registrar/tests/test_email_invitations.py
View file

@ -2,12 +2,24 @@ import unittest
from unittest.mock import patch, MagicMock
from datetime import date
from registrar.models.domain import Domain
from registrar.models.portfolio import Portfolio
from registrar.models.user import User
from registrar.models.user_domain_role import UserDomainRole
from registrar.models.user_portfolio_permission import UserPortfolioPermission
from registrar.models.utility.portfolio_helper import UserPortfolioRoleChoices
from registrar.utility.email import EmailSendingError
from registrar.utility.email_invitations import send_domain_invitation_email, send_emails_to_domain_managers
from registrar.utility.email_invitations import (
_send_portfolio_admin_addition_emails_to_portfolio_admins,
_send_portfolio_admin_removal_emails_to_portfolio_admins,
send_domain_invitation_email,
send_emails_to_domain_managers,
send_portfolio_admin_addition_emails,
send_portfolio_admin_removal_emails,
send_portfolio_invitation_email,
)
from api.tests.common import less_console_noise_decorator
from registrar.utility.errors import MissingEmailError
class DomainInvitationEmail(unittest.TestCase):
@ -16,9 +28,9 @@ class DomainInvitationEmail(unittest.TestCase):
@patch("registrar.utility.email_invitations.send_templated_email")
@patch("registrar.utility.email_invitations.UserDomainRole.objects.filter")
@patch("registrar.utility.email_invitations._validate_invitation")
@patch("registrar.utility.email_invitations.get_requestor_email")
@patch("registrar.utility.email_invitations._get_requestor_email")
@patch("registrar.utility.email_invitations.send_invitation_email")
@patch("registrar.utility.email_invitations.normalize_domains")
@patch("registrar.utility.email_invitations._normalize_domains")
def test_send_domain_invitation_email(
self,
mock_normalize_domains,
@ -58,7 +70,7 @@ class DomainInvitationEmail(unittest.TestCase):
# Assertions
mock_normalize_domains.assert_called_once_with(mock_domain)
mock_get_requestor_email.assert_called_once_with(mock_requestor, [mock_domain])
mock_get_requestor_email.assert_called_once_with(mock_requestor, domains=[mock_domain])
mock_validate_invitation.assert_called_once_with(
email, None, [mock_domain], mock_requestor, is_member_of_different_org
)
@ -81,9 +93,9 @@ class DomainInvitationEmail(unittest.TestCase):
@patch("registrar.utility.email_invitations.send_templated_email")
@patch("registrar.utility.email_invitations.UserDomainRole.objects.filter")
@patch("registrar.utility.email_invitations._validate_invitation")
@patch("registrar.utility.email_invitations.get_requestor_email")
@patch("registrar.utility.email_invitations._get_requestor_email")
@patch("registrar.utility.email_invitations.send_invitation_email")
@patch("registrar.utility.email_invitations.normalize_domains")
@patch("registrar.utility.email_invitations._normalize_domains")
def test_send_domain_invitation_email_multiple_domains(
self,
mock_normalize_domains,
@ -137,7 +149,7 @@ class DomainInvitationEmail(unittest.TestCase):
# Assertions
mock_normalize_domains.assert_called_once_with([mock_domain1, mock_domain2])
mock_get_requestor_email.assert_called_once_with(mock_requestor, [mock_domain1, mock_domain2])
mock_get_requestor_email.assert_called_once_with(mock_requestor, domains=[mock_domain1, mock_domain2])
mock_validate_invitation.assert_called_once_with(
email, None, [mock_domain1, mock_domain2], mock_requestor, is_member_of_different_org
)
@ -197,7 +209,7 @@ class DomainInvitationEmail(unittest.TestCase):
mock_validate_invitation.assert_called_once()
@less_console_noise_decorator
@patch("registrar.utility.email_invitations.get_requestor_email")
@patch("registrar.utility.email_invitations._get_requestor_email")
def test_send_domain_invitation_email_raises_get_requestor_email_exception(self, mock_get_requestor_email):
"""Test sending domain invitation email for one domain and assert exception
when get_requestor_email fails.
@ -217,9 +229,9 @@ class DomainInvitationEmail(unittest.TestCase):
@less_console_noise_decorator
@patch("registrar.utility.email_invitations._validate_invitation")
@patch("registrar.utility.email_invitations.get_requestor_email")
@patch("registrar.utility.email_invitations._get_requestor_email")
@patch("registrar.utility.email_invitations.send_invitation_email")
@patch("registrar.utility.email_invitations.normalize_domains")
@patch("registrar.utility.email_invitations._normalize_domains")
def test_send_domain_invitation_email_raises_sending_email_exception(
self,
mock_normalize_domains,
@ -258,7 +270,7 @@ class DomainInvitationEmail(unittest.TestCase):
# Assertions
mock_normalize_domains.assert_called_once_with(mock_domain)
mock_get_requestor_email.assert_called_once_with(mock_requestor, [mock_domain])
mock_get_requestor_email.assert_called_once_with(mock_requestor, domains=[mock_domain])
mock_validate_invitation.assert_called_once_with(
email, None, [mock_domain], mock_requestor, is_member_of_different_org
)
@ -267,9 +279,9 @@ class DomainInvitationEmail(unittest.TestCase):
@less_console_noise_decorator
@patch("registrar.utility.email_invitations.send_emails_to_domain_managers")
@patch("registrar.utility.email_invitations._validate_invitation")
@patch("registrar.utility.email_invitations.get_requestor_email")
@patch("registrar.utility.email_invitations._get_requestor_email")
@patch("registrar.utility.email_invitations.send_invitation_email")
@patch("registrar.utility.email_invitations.normalize_domains")
@patch("registrar.utility.email_invitations._normalize_domains")
def test_send_domain_invitation_email_manager_emails_send_mail_exception(
self,
mock_normalize_domains,
@ -306,7 +318,7 @@ class DomainInvitationEmail(unittest.TestCase):
# Assertions
mock_normalize_domains.assert_called_once_with(mock_domain)
mock_get_requestor_email.assert_called_once_with(mock_requestor, [mock_domain])
mock_get_requestor_email.assert_called_once_with(mock_requestor, domains=[mock_domain])
mock_validate_invitation.assert_called_once_with(
email, None, [mock_domain], mock_requestor, is_member_of_different_org
)
@ -469,3 +481,410 @@ class DomainInvitationEmail(unittest.TestCase):
"date": date.today(),
},
)
class PortfolioInvitationEmailTests(unittest.TestCase):
def setUp(self):
"""Setup common test data for all test cases"""
self.email = "invitee@example.com"
self.requestor = MagicMock(name="User")
self.requestor.email = "requestor@example.com"
self.portfolio = MagicMock(name="Portfolio")
@less_console_noise_decorator
@patch("registrar.utility.email_invitations.send_templated_email")
def test_send_portfolio_invitation_email_success(self, mock_send_templated_email):
"""Test successful email sending"""
is_admin_invitation = False
result = send_portfolio_invitation_email(self.email, self.requestor, self.portfolio, is_admin_invitation)
self.assertTrue(result)
mock_send_templated_email.assert_called_once()
@less_console_noise_decorator
@patch(
"registrar.utility.email_invitations.send_templated_email",
side_effect=EmailSendingError("Failed to send email"),
)
def test_send_portfolio_invitation_email_failure(self, mock_send_templated_email):
"""Test failure when sending email"""
is_admin_invitation = False
with self.assertRaises(EmailSendingError) as context:
send_portfolio_invitation_email(self.email, self.requestor, self.portfolio, is_admin_invitation)
self.assertIn("Could not sent email invitation to", str(context.exception))
@less_console_noise_decorator
@patch(
"registrar.utility.email_invitations._get_requestor_email",
side_effect=MissingEmailError("Requestor has no email"),
)
@less_console_noise_decorator
def test_send_portfolio_invitation_email_missing_requestor_email(self, mock_get_email):
"""Test when requestor has no email"""
is_admin_invitation = False
with self.assertRaises(MissingEmailError) as context:
send_portfolio_invitation_email(self.email, self.requestor, self.portfolio, is_admin_invitation)
self.assertIn(
"Can't send invitation email. No email is associated with your user account.", str(context.exception)
)
@less_console_noise_decorator
@patch(
"registrar.utility.email_invitations._send_portfolio_admin_addition_emails_to_portfolio_admins",
return_value=False,
)
@patch("registrar.utility.email_invitations.send_templated_email")
def test_send_portfolio_invitation_email_admin_invitation(self, mock_send_templated_email, mock_admin_email):
"""Test admin invitation email logic"""
is_admin_invitation = True
result = send_portfolio_invitation_email(self.email, self.requestor, self.portfolio, is_admin_invitation)
self.assertFalse(result) # Admin email sending failed
mock_send_templated_email.assert_called_once()
mock_admin_email.assert_called_once()
@less_console_noise_decorator
@patch("registrar.utility.email_invitations._get_requestor_email")
@patch("registrar.utility.email_invitations._send_portfolio_admin_addition_emails_to_portfolio_admins")
def test_send_email_success(self, mock_send_admin_emails, mock_get_requestor_email):
"""Test successful sending of admin addition emails."""
mock_get_requestor_email.return_value = "requestor@example.com"
mock_send_admin_emails.return_value = True
result = send_portfolio_admin_addition_emails(self.email, self.requestor, self.portfolio)
mock_get_requestor_email.assert_called_once_with(self.requestor, portfolio=self.portfolio)
mock_send_admin_emails.assert_called_once_with(self.email, "requestor@example.com", self.portfolio)
self.assertTrue(result)
@less_console_noise_decorator
@patch(
"registrar.utility.email_invitations._get_requestor_email",
side_effect=MissingEmailError("Requestor email missing"),
)
def test_missing_requestor_email_raises_exception(self, mock_get_requestor_email):
"""Test exception raised if requestor email is missing."""
with self.assertRaises(MissingEmailError):
send_portfolio_admin_addition_emails(self.email, self.requestor, self.portfolio)
@less_console_noise_decorator
@patch("registrar.utility.email_invitations._get_requestor_email")
@patch("registrar.utility.email_invitations._send_portfolio_admin_addition_emails_to_portfolio_admins")
def test_send_email_failure(self, mock_send_admin_emails, mock_get_requestor_email):
"""Test handling of failure in sending admin addition emails."""
mock_get_requestor_email.return_value = "requestor@example.com"
mock_send_admin_emails.return_value = False # Simulate failure
result = send_portfolio_admin_addition_emails(self.email, self.requestor, self.portfolio)
self.assertFalse(result)
mock_get_requestor_email.assert_called_once_with(self.requestor, portfolio=self.portfolio)
mock_send_admin_emails.assert_called_once_with(self.email, "requestor@example.com", self.portfolio)
class SendPortfolioAdminAdditionEmailsTests(unittest.TestCase):
"""Unit tests for _send_portfolio_admin_addition_emails_to_portfolio_admins function."""
def setUp(self):
"""Set up test data."""
self.email = "new.admin@example.com"
self.requestor_email = "requestor@example.com"
self.portfolio = MagicMock(spec=Portfolio)
self.portfolio.organization_name = "Test Organization"
# Mock portfolio admin users
self.admin_user1 = MagicMock(spec=User)
self.admin_user1.email = "admin1@example.com"
self.admin_user2 = MagicMock(spec=User)
self.admin_user2.email = "admin2@example.com"
self.portfolio_admin1 = MagicMock(spec=UserPortfolioPermission)
self.portfolio_admin1.user = self.admin_user1
self.portfolio_admin1.roles = [UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
self.portfolio_admin2 = MagicMock(spec=UserPortfolioPermission)
self.portfolio_admin2.user = self.admin_user2
self.portfolio_admin2.roles = [UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
@less_console_noise_decorator
@patch("registrar.utility.email_invitations.send_templated_email")
@patch("registrar.utility.email_invitations.UserPortfolioPermission.objects.filter")
def test_send_email_success(self, mock_filter, mock_send_templated_email):
"""Test successful sending of admin addition emails."""
mock_filter.return_value.exclude.return_value = [self.portfolio_admin1, self.portfolio_admin2]
mock_send_templated_email.return_value = None # No exception means success
result = _send_portfolio_admin_addition_emails_to_portfolio_admins(
self.email, self.requestor_email, self.portfolio
)
mock_filter.assert_called_once_with(
portfolio=self.portfolio, roles__contains=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
)
mock_send_templated_email.assert_any_call(
"emails/portfolio_admin_addition_notification.txt",
"emails/portfolio_admin_addition_notification_subject.txt",
to_address=self.admin_user1.email,
context={
"portfolio": self.portfolio,
"requestor_email": self.requestor_email,
"invited_email_address": self.email,
"portfolio_admin": self.admin_user1,
"date": date.today(),
},
)
mock_send_templated_email.assert_any_call(
"emails/portfolio_admin_addition_notification.txt",
"emails/portfolio_admin_addition_notification_subject.txt",
to_address=self.admin_user2.email,
context={
"portfolio": self.portfolio,
"requestor_email": self.requestor_email,
"invited_email_address": self.email,
"portfolio_admin": self.admin_user2,
"date": date.today(),
},
)
self.assertTrue(result)
@less_console_noise_decorator
@patch("registrar.utility.email_invitations.send_templated_email", side_effect=EmailSendingError)
@patch("registrar.utility.email_invitations.UserPortfolioPermission.objects.filter")
def test_send_email_failure(self, mock_filter, mock_send_templated_email):
"""Test handling of failure in sending admin addition emails."""
mock_filter.return_value.exclude.return_value = [self.portfolio_admin1, self.portfolio_admin2]
result = _send_portfolio_admin_addition_emails_to_portfolio_admins(
self.email, self.requestor_email, self.portfolio
)
self.assertFalse(result)
mock_filter.assert_called_once_with(
portfolio=self.portfolio, roles__contains=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
)
mock_send_templated_email.assert_any_call(
"emails/portfolio_admin_addition_notification.txt",
"emails/portfolio_admin_addition_notification_subject.txt",
to_address=self.admin_user1.email,
context={
"portfolio": self.portfolio,
"requestor_email": self.requestor_email,
"invited_email_address": self.email,
"portfolio_admin": self.admin_user1,
"date": date.today(),
},
)
mock_send_templated_email.assert_any_call(
"emails/portfolio_admin_addition_notification.txt",
"emails/portfolio_admin_addition_notification_subject.txt",
to_address=self.admin_user2.email,
context={
"portfolio": self.portfolio,
"requestor_email": self.requestor_email,
"invited_email_address": self.email,
"portfolio_admin": self.admin_user2,
"date": date.today(),
},
)
@less_console_noise_decorator
@patch("registrar.utility.email_invitations.UserPortfolioPermission.objects.filter")
def test_no_admins_to_notify(self, mock_filter):
"""Test case where there are no portfolio admins to notify."""
mock_filter.return_value.exclude.return_value = [] # No admins
result = _send_portfolio_admin_addition_emails_to_portfolio_admins(
self.email, self.requestor_email, self.portfolio
)
self.assertTrue(result) # No emails sent, but also no failures
mock_filter.assert_called_once_with(
portfolio=self.portfolio, roles__contains=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
)
class SendPortfolioAdminRemovalEmailsToAdminsTests(unittest.TestCase):
"""Unit tests for _send_portfolio_admin_removal_emails_to_portfolio_admins function."""
def setUp(self):
"""Set up test data."""
self.email = "removed.admin@example.com"
self.requestor_email = "requestor@example.com"
self.portfolio = MagicMock(spec=Portfolio)
self.portfolio.organization_name = "Test Organization"
# Mock portfolio admin users
self.admin_user1 = MagicMock(spec=User)
self.admin_user1.email = "admin1@example.com"
self.admin_user2 = MagicMock(spec=User)
self.admin_user2.email = "admin2@example.com"
self.portfolio_admin1 = MagicMock(spec=UserPortfolioPermission)
self.portfolio_admin1.user = self.admin_user1
self.portfolio_admin1.roles = [UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
self.portfolio_admin2 = MagicMock(spec=UserPortfolioPermission)
self.portfolio_admin2.user = self.admin_user2
self.portfolio_admin2.roles = [UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
@less_console_noise_decorator
@patch("registrar.utility.email_invitations.send_templated_email")
@patch("registrar.utility.email_invitations.UserPortfolioPermission.objects.filter")
def test_send_email_success(self, mock_filter, mock_send_templated_email):
"""Test successful sending of admin removal emails."""
mock_filter.return_value.exclude.return_value = [self.portfolio_admin1, self.portfolio_admin2]
mock_send_templated_email.return_value = None # No exception means success
result = _send_portfolio_admin_removal_emails_to_portfolio_admins(
self.email, self.requestor_email, self.portfolio
)
mock_filter.assert_called_once_with(
portfolio=self.portfolio, roles__contains=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
)
mock_send_templated_email.assert_any_call(
"emails/portfolio_admin_removal_notification.txt",
"emails/portfolio_admin_removal_notification_subject.txt",
to_address=self.admin_user1.email,
context={
"portfolio": self.portfolio,
"requestor_email": self.requestor_email,
"removed_email_address": self.email,
"portfolio_admin": self.admin_user1,
"date": date.today(),
},
)
mock_send_templated_email.assert_any_call(
"emails/portfolio_admin_removal_notification.txt",
"emails/portfolio_admin_removal_notification_subject.txt",
to_address=self.admin_user2.email,
context={
"portfolio": self.portfolio,
"requestor_email": self.requestor_email,
"removed_email_address": self.email,
"portfolio_admin": self.admin_user2,
"date": date.today(),
},
)
self.assertTrue(result)
@less_console_noise_decorator
@patch("registrar.utility.email_invitations.send_templated_email", side_effect=EmailSendingError)
@patch("registrar.utility.email_invitations.UserPortfolioPermission.objects.filter")
def test_send_email_failure(self, mock_filter, mock_send_templated_email):
"""Test handling of failure in sending admin removal emails."""
mock_filter.return_value.exclude.return_value = [self.portfolio_admin1, self.portfolio_admin2]
result = _send_portfolio_admin_removal_emails_to_portfolio_admins(
self.email, self.requestor_email, self.portfolio
)
self.assertFalse(result)
mock_filter.assert_called_once_with(
portfolio=self.portfolio, roles__contains=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
)
mock_send_templated_email.assert_any_call(
"emails/portfolio_admin_removal_notification.txt",
"emails/portfolio_admin_removal_notification_subject.txt",
to_address=self.admin_user1.email,
context={
"portfolio": self.portfolio,
"requestor_email": self.requestor_email,
"removed_email_address": self.email,
"portfolio_admin": self.admin_user1,
"date": date.today(),
},
)
mock_send_templated_email.assert_any_call(
"emails/portfolio_admin_removal_notification.txt",
"emails/portfolio_admin_removal_notification_subject.txt",
to_address=self.admin_user2.email,
context={
"portfolio": self.portfolio,
"requestor_email": self.requestor_email,
"removed_email_address": self.email,
"portfolio_admin": self.admin_user2,
"date": date.today(),
},
)
@less_console_noise_decorator
@patch("registrar.utility.email_invitations.UserPortfolioPermission.objects.filter")
def test_no_admins_to_notify(self, mock_filter):
"""Test case where there are no portfolio admins to notify."""
mock_filter.return_value.exclude.return_value = [] # No admins
result = _send_portfolio_admin_removal_emails_to_portfolio_admins(
self.email, self.requestor_email, self.portfolio
)
self.assertTrue(result) # No emails sent, but also no failures
mock_filter.assert_called_once_with(
portfolio=self.portfolio, roles__contains=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
)
class SendPortfolioAdminRemovalEmailsTests(unittest.TestCase):
"""Unit tests for send_portfolio_admin_removal_emails function."""
def setUp(self):
"""Set up test data."""
self.email = "removed.admin@example.com"
self.requestor = MagicMock(spec=User)
self.requestor.email = "requestor@example.com"
self.portfolio = MagicMock(spec=Portfolio)
self.portfolio.organization_name = "Test Organization"
@less_console_noise_decorator
@patch("registrar.utility.email_invitations._get_requestor_email")
@patch("registrar.utility.email_invitations._send_portfolio_admin_removal_emails_to_portfolio_admins")
def test_send_email_success(self, mock_send_removal_emails, mock_get_requestor_email):
"""Test successful execution of send_portfolio_admin_removal_emails."""
mock_get_requestor_email.return_value = self.requestor.email
mock_send_removal_emails.return_value = True # Simulating success
result = send_portfolio_admin_removal_emails(self.email, self.requestor, self.portfolio)
mock_get_requestor_email.assert_called_once_with(self.requestor, portfolio=self.portfolio)
mock_send_removal_emails.assert_called_once_with(self.email, self.requestor.email, self.portfolio)
self.assertTrue(result)
@less_console_noise_decorator
@patch("registrar.utility.email_invitations._get_requestor_email", side_effect=MissingEmailError("No email found"))
@patch("registrar.utility.email_invitations._send_portfolio_admin_removal_emails_to_portfolio_admins")
def test_missing_email_error(self, mock_send_removal_emails, mock_get_requestor_email):
"""Test handling of MissingEmailError when requestor has no email."""
with self.assertRaises(MissingEmailError) as context:
send_portfolio_admin_removal_emails(self.email, self.requestor, self.portfolio)
mock_get_requestor_email.assert_called_once_with(self.requestor, portfolio=self.portfolio)
mock_send_removal_emails.assert_not_called() # Should not proceed if email retrieval fails
self.assertEqual(
str(context.exception), "Can't send invitation email. No email is associated with your user account."
)
@less_console_noise_decorator
@patch("registrar.utility.email_invitations._get_requestor_email")
@patch(
"registrar.utility.email_invitations._send_portfolio_admin_removal_emails_to_portfolio_admins",
return_value=False,
)
def test_send_email_failure(self, mock_send_removal_emails, mock_get_requestor_email):
"""Test handling of failure when admin removal emails fail to send."""
mock_get_requestor_email.return_value = self.requestor.email
mock_send_removal_emails.return_value = False # Simulating failure
result = send_portfolio_admin_removal_emails(self.email, self.requestor, self.portfolio)
mock_get_requestor_email.assert_called_once_with(self.requestor, portfolio=self.portfolio)
mock_send_removal_emails.assert_called_once_with(self.email, self.requestor.email, self.portfolio)
self.assertFalse(result)

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

@ -849,7 +849,10 @@ class TestDomainManagers(TestDomainOverview):
# Verify that the invitation emails were sent
mock_send_portfolio_email.assert_called_once_with(
email="mayor@igorville.gov", requestor=self.user, portfolio=self.portfolio
email="mayor@igorville.gov",
requestor=self.user,
portfolio=self.portfolio,
is_admin_invitation=False,
)
mock_send_domain_email.assert_called_once()
call_args = mock_send_domain_email.call_args.kwargs
@ -903,7 +906,10 @@ class TestDomainManagers(TestDomainOverview):
# Verify that the invitation emails were sent
mock_send_portfolio_email.assert_called_once_with(
email="notauser@igorville.gov", requestor=self.user, portfolio=self.portfolio
email="notauser@igorville.gov",
requestor=self.user,
portfolio=self.portfolio,
is_admin_invitation=False,
)
mock_send_domain_email.assert_called_once()
call_args = mock_send_domain_email.call_args.kwargs
@ -1038,7 +1044,10 @@ class TestDomainManagers(TestDomainOverview):
# Verify that the invitation emails were sent
mock_send_portfolio_email.assert_called_once_with(
email="mayor@igorville.gov", requestor=self.user, portfolio=self.portfolio
email="mayor@igorville.gov",
requestor=self.user,
portfolio=self.portfolio,
is_admin_invitation=False,
)
mock_send_domain_email.assert_not_called()

932
src/registrar/tests/test_views_portfolio.py
View file

File diff suppressed because it is too large Load diff

152
src/registrar/utility/email_invitations.py
View file

@ -1,6 +1,9 @@
from datetime import date
from django.conf import settings
from registrar.models import Domain, DomainInvitation, UserDomainRole
from registrar.models.portfolio import Portfolio
from registrar.models.user_portfolio_permission import UserPortfolioPermission
from registrar.models.utility.portfolio_helper import UserPortfolioRoleChoices
from registrar.utility.errors import (
AlreadyDomainInvitedError,
AlreadyDomainManagerError,
@ -37,8 +40,8 @@ def send_domain_invitation_email(
OutsideOrgMemberError: If the requested_user is part of a different organization.
EmailSendingError: If there is an error while sending the email.
"""
domains = normalize_domains(domains)
requestor_email = get_requestor_email(requestor, domains)
domains = _normalize_domains(domains)
requestor_email = _get_requestor_email(requestor, domains=domains)
_validate_invitation(email, requested_user, domains, requestor, is_member_of_different_org)
@ -92,22 +95,27 @@ def send_emails_to_domain_managers(email: str, requestor_email, domain: Domain,
return all_emails_sent
def normalize_domains(domains: Domain | list[Domain]) -> list[Domain]:
def _normalize_domains(domains: Domain | list[Domain]) -> list[Domain]:
"""Ensures domains is always a list."""
return [domains] if isinstance(domains, Domain) else domains
def get_requestor_email(requestor, domains):
def _get_requestor_email(requestor, domains=None, portfolio=None):
"""Get the requestor's email or raise an error if it's missing.
If the requestor is staff, default email is returned.
Raises:
MissingEmailError
"""
if requestor.is_staff:
return settings.DEFAULT_FROM_EMAIL
if not requestor.email or requestor.email.strip() == "":
domain_names = ", ".join([domain.name for domain in domains])
raise MissingEmailError(email=requestor.email, domain=domain_names)
domain_names = None
if domains:
domain_names = ", ".join([domain.name for domain in domains])
raise MissingEmailError(email=requestor.email, domain=domain_names, portfolio=portfolio)
return requestor.email
@ -169,7 +177,7 @@ def send_invitation_email(email, requestor_email, domains, requested_user):
raise EmailSendingError(f"Could not send email invitation to {email} for domains: {domain_names}") from err
def send_portfolio_invitation_email(email: str, requestor, portfolio):
def send_portfolio_invitation_email(email: str, requestor, portfolio, is_admin_invitation):
"""
Sends a portfolio member invitation email to the specified address.
@ -179,21 +187,17 @@ def send_portfolio_invitation_email(email: str, requestor, portfolio):
email (str): Email address of the recipient
requestor (User): The user initiating the invitation.
portfolio (Portfolio): The portfolio object for which the invitation is being sent.
is_admin_invitation (boolean): boolean indicating if the invitation is an admin invitation
Returns:
Boolean indicating if all messages were sent successfully.
Raises:
MissingEmailError: If the requestor has no email associated with their account.
EmailSendingError: If there is an error while sending the email.
"""
# Default email address for staff
requestor_email = settings.DEFAULT_FROM_EMAIL
# Check if the requestor is staff and has an email
if not requestor.is_staff:
if not requestor.email or requestor.email.strip() == "":
raise MissingEmailError(email=email, portfolio=portfolio)
else:
requestor_email = requestor.email
requestor_email = _get_requestor_email(requestor, portfolio=portfolio)
try:
send_templated_email(
@ -210,3 +214,119 @@ def send_portfolio_invitation_email(email: str, requestor, portfolio):
raise EmailSendingError(
f"Could not sent email invitation to {email} for portfolio {portfolio}. Portfolio invitation not saved."
) from err
all_admin_emails_sent = True
# send emails to portfolio admins
if is_admin_invitation:
all_admin_emails_sent = _send_portfolio_admin_addition_emails_to_portfolio_admins(
email=email,
requestor_email=requestor_email,
portfolio=portfolio,
)
return all_admin_emails_sent
def send_portfolio_admin_addition_emails(email: str, requestor, portfolio: Portfolio):
"""
Notifies all portfolio admins of the provided portfolio of a newly invited portfolio admin
Returns:
Boolean indicating if all messages were sent successfully.
Raises:
MissingEmailError
"""
requestor_email = _get_requestor_email(requestor, portfolio=portfolio)
return _send_portfolio_admin_addition_emails_to_portfolio_admins(email, requestor_email, portfolio)
def _send_portfolio_admin_addition_emails_to_portfolio_admins(email: str, requestor_email, portfolio: Portfolio):
"""
Notifies all portfolio admins of the provided portfolio of a newly invited portfolio admin
Returns:
Boolean indicating if all messages were sent successfully.
"""
all_emails_sent = True
# Get each portfolio admin from list
user_portfolio_permissions = UserPortfolioPermission.objects.filter(
portfolio=portfolio, roles__contains=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
).exclude(user__email=email)
for user_portfolio_permission in user_portfolio_permissions:
# Send email to each portfolio_admin
user = user_portfolio_permission.user
try:
send_templated_email(
"emails/portfolio_admin_addition_notification.txt",
"emails/portfolio_admin_addition_notification_subject.txt",
to_address=user.email,
context={
"portfolio": portfolio,
"requestor_email": requestor_email,
"invited_email_address": email,
"portfolio_admin": user,
"date": date.today(),
},
)
except EmailSendingError:
logger.warning(
"Could not send email organization admin notification to %s " "for portfolio: %s",
user.email,
portfolio.organization_name,
exc_info=True,
)
all_emails_sent = False
return all_emails_sent
def send_portfolio_admin_removal_emails(email: str, requestor, portfolio: Portfolio):
"""
Notifies all portfolio admins of the provided portfolio of a removed portfolio admin
Returns:
Boolean indicating if all messages were sent successfully.
Raises:
MissingEmailError
"""
requestor_email = _get_requestor_email(requestor, portfolio=portfolio)
return _send_portfolio_admin_removal_emails_to_portfolio_admins(email, requestor_email, portfolio)
def _send_portfolio_admin_removal_emails_to_portfolio_admins(email: str, requestor_email, portfolio: Portfolio):
"""
Notifies all portfolio admins of the provided portfolio of a removed portfolio admin
Returns:
Boolean indicating if all messages were sent successfully.
"""
all_emails_sent = True
# Get each portfolio admin from list
user_portfolio_permissions = UserPortfolioPermission.objects.filter(
portfolio=portfolio, roles__contains=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
).exclude(user__email=email)
for user_portfolio_permission in user_portfolio_permissions:
# Send email to each portfolio_admin
user = user_portfolio_permission.user
try:
send_templated_email(
"emails/portfolio_admin_removal_notification.txt",
"emails/portfolio_admin_removal_notification_subject.txt",
to_address=user.email,
context={
"portfolio": portfolio,
"requestor_email": requestor_email,
"removed_email_address": email,
"portfolio_admin": user,
"date": date.today(),
},
)
except EmailSendingError:
logger.warning(
"Could not send email organization admin notification to %s " "for portfolio: %s",
user.email,
portfolio.organization_name,
exc_info=True,
)
all_emails_sent = False
return all_emails_sent

4
src/registrar/views/domain.py
View file

@ -1234,7 +1234,9 @@ class DomainAddUserView(DomainFormBaseView):
and requestor_can_update_portfolio
and not member_of_this_org
):
send_portfolio_invitation_email(email=requested_email, requestor=requestor, portfolio=domain_org)
send_portfolio_invitation_email(
email=requested_email, requestor=requestor, portfolio=domain_org, is_admin_invitation=False
)
portfolio_invitation, _ = PortfolioInvitation.objects.get_or_create(
email=requested_email, portfolio=domain_org, roles=[UserPortfolioRoleChoices.ORGANIZATION_MEMBER]
)

142
src/registrar/views/portfolios.py
View file

@ -15,7 +15,12 @@ from registrar.models.user_domain_role import UserDomainRole
from registrar.models.user_portfolio_permission import UserPortfolioPermission
from registrar.models.utility.portfolio_helper import UserPortfolioPermissionChoices, UserPortfolioRoleChoices
from registrar.utility.email import EmailSendingError
from registrar.utility.email_invitations import send_domain_invitation_email, send_portfolio_invitation_email
from registrar.utility.email_invitations import (
send_domain_invitation_email,
send_portfolio_admin_addition_emails,
send_portfolio_admin_removal_emails,
send_portfolio_invitation_email,
)
from registrar.utility.errors import MissingEmailError
from registrar.utility.enums import DefaultUserValues
from registrar.views.utility.mixins import PortfolioMemberPermission
@ -143,6 +148,19 @@ class PortfolioMemberDeleteView(PortfolioMemberPermission, View):
messages.error(request, error_message)
return redirect(reverse("member", kwargs={"pk": pk}))
# if member being removed is an admin
if UserPortfolioRoleChoices.ORGANIZATION_ADMIN in portfolio_member_permission.roles:
try:
# attempt to send notification emails of the removal to other portfolio admins
if not send_portfolio_admin_removal_emails(
email=portfolio_member_permission.user.email,
requestor=request.user,
portfolio=portfolio_member_permission.portfolio,
):
messages.warning(self.request, "Could not send email notification to existing organization admins.")
except Exception as e:
self._handle_exceptions(e)
# passed all error conditions
portfolio_member_permission.delete()
@ -154,6 +172,18 @@ class PortfolioMemberDeleteView(PortfolioMemberPermission, View):
messages.success(request, success_message)
return redirect(reverse("members"))
def _handle_exceptions(self, exception):
"""Handle exceptions raised during the process."""
if isinstance(exception, MissingEmailError):
messages.warning(self.request, "Could not send email notification to existing organization admins.")
logger.warning(
"Could not send email notification to existing organization admins.",
exc_info=True,
)
else:
logger.warning("Could not send email notification to existing organization admins.", exc_info=True)
messages.warning(self.request, "Could not send email notification to existing organization admins.")
class PortfolioMemberEditView(PortfolioMemberEditPermissionView, View):
@ -177,16 +207,33 @@ class PortfolioMemberEditView(PortfolioMemberEditPermissionView, View):
def post(self, request, pk):
portfolio_permission = get_object_or_404(UserPortfolioPermission, pk=pk)
user_initially_is_admin = UserPortfolioRoleChoices.ORGANIZATION_ADMIN in portfolio_permission.roles
user = portfolio_permission.user
form = self.form_class(request.POST, instance=portfolio_permission)
removing_admin_role_on_self = False
if form.is_valid():
# Check if user is removing their own admin or edit role
removing_admin_role_on_self = (
request.user == user
and user_initially_is_admin
and UserPortfolioRoleChoices.ORGANIZATION_ADMIN not in form.cleaned_data.get("role", [])
)
try:
if form.is_change_from_member_to_admin():
if not send_portfolio_admin_addition_emails(
email=portfolio_permission.user.email,
requestor=request.user,
portfolio=portfolio_permission.portfolio,
):
messages.warning(
self.request, "Could not send email notification to existing organization admins."
)
elif form.is_change_from_admin_to_member():
if not send_portfolio_admin_removal_emails(
email=portfolio_permission.user.email,
requestor=request.user,
portfolio=portfolio_permission.portfolio,
):
messages.warning(
self.request, "Could not send email notification to existing organization admins."
)
# Check if user is removing their own admin or edit role
removing_admin_role_on_self = request.user == user
except Exception as e:
self._handle_exceptions(e)
form.save()
messages.success(self.request, "The member access and permission changes have been saved.")
return redirect("member", pk=pk) if not removing_admin_role_on_self else redirect("home")
@ -200,6 +247,18 @@ class PortfolioMemberEditView(PortfolioMemberEditPermissionView, View):
},
)
def _handle_exceptions(self, exception):
"""Handle exceptions raised during the process."""
if isinstance(exception, MissingEmailError):
messages.warning(self.request, "Could not send email notification to existing organization admins.")
logger.warning(
"Could not send email notification to existing organization admins.",
exc_info=True,
)
else:
logger.warning("Could not send email notification to existing organization admins.", exc_info=True)
messages.warning(self.request, "Could not send email notification to existing organization admins.")
class PortfolioMemberDomainsView(PortfolioMemberDomainsPermissionView, View):
@ -380,6 +439,17 @@ class PortfolioInvitedMemberDeleteView(PortfolioMemberPermission, View):
"""
portfolio_invitation = get_object_or_404(PortfolioInvitation, pk=pk)
# if invitation being removed is an admin
if UserPortfolioRoleChoices.ORGANIZATION_ADMIN in portfolio_invitation.roles:
try:
# attempt to send notification emails of the removal to portfolio admins
if not send_portfolio_admin_removal_emails(
email=portfolio_invitation.email, requestor=request.user, portfolio=portfolio_invitation.portfolio
):
messages.warning(self.request, "Could not send email notification to existing organization admins.")
except Exception as e:
self._handle_exceptions(e)
portfolio_invitation.delete()
success_message = f"You've removed {portfolio_invitation.email} from the organization."
@ -390,6 +460,18 @@ class PortfolioInvitedMemberDeleteView(PortfolioMemberPermission, View):
messages.success(request, success_message)
return redirect(reverse("members"))
def _handle_exceptions(self, exception):
"""Handle exceptions raised during the process."""
if isinstance(exception, MissingEmailError):
messages.warning(self.request, "Could not send email notification to existing organization admins.")
logger.warning(
"Could not send email notification to existing organization admins.",
exc_info=True,
)
else:
logger.warning("Could not send email notification to existing organization admins.", exc_info=True)
messages.warning(self.request, "Could not send email notification to existing organization admins.")
class PortfolioInvitedMemberEditView(PortfolioMemberEditPermissionView, View):
@ -413,6 +495,27 @@ class PortfolioInvitedMemberEditView(PortfolioMemberEditPermissionView, View):
portfolio_invitation = get_object_or_404(PortfolioInvitation, pk=pk)
form = self.form_class(request.POST, instance=portfolio_invitation)
if form.is_valid():
try:
if form.is_change_from_member_to_admin():
if not send_portfolio_admin_addition_emails(
email=portfolio_invitation.email,
requestor=request.user,
portfolio=portfolio_invitation.portfolio,
):
messages.warning(
self.request, "Could not send email notification to existing organization admins."
)
elif form.is_change_from_admin_to_member():
if not send_portfolio_admin_removal_emails(
email=portfolio_invitation.email,
requestor=request.user,
portfolio=portfolio_invitation.portfolio,
):
messages.warning(
self.request, "Could not send email notification to existing organization admins."
)
except Exception as e:
self._handle_exceptions(e)
form.save()
messages.success(self.request, "The member access and permission changes have been saved.")
return redirect("invitedmember", pk=pk)
@ -426,6 +529,18 @@ class PortfolioInvitedMemberEditView(PortfolioMemberEditPermissionView, View):
},
)
def _handle_exceptions(self, exception):
"""Handle exceptions raised during the process."""
if isinstance(exception, MissingEmailError):
messages.warning(self.request, "Could not send email notification to existing organization admins.")
logger.warning(
"Could not send email notification to existing organization admins.",
exc_info=True,
)
else:
logger.warning("Could not send email notification to existing organization admins.", exc_info=True)
messages.warning(self.request, "Could not send email notification to existing organization admins.")
class PortfolioInvitedMemberDomainsView(PortfolioMemberDomainsPermissionView, View):
@ -781,12 +896,19 @@ class PortfolioAddMemberView(PortfolioMembersPermissionView, FormMixin):
requested_email = form.cleaned_data["email"]
requestor = self.request.user
portfolio = form.cleaned_data["portfolio"]
is_admin_invitation = UserPortfolioRoleChoices.ORGANIZATION_ADMIN in form.cleaned_data["roles"]
requested_user = User.objects.filter(email=requested_email).first()
permission_exists = UserPortfolioPermission.objects.filter(user=requested_user, portfolio=portfolio).exists()
try:
if not requested_user or not permission_exists:
send_portfolio_invitation_email(email=requested_email, requestor=requestor, portfolio=portfolio)
if not send_portfolio_invitation_email(
email=requested_email,
requestor=requestor,
portfolio=portfolio,
is_admin_invitation=is_admin_invitation,
):
messages.warning(self.request, "Could not send email notification to existing organization admins.")
portfolio_invitation = form.save()
# if user exists for email, immediately retrieve portfolio invitation upon creation
if requested_user is not None:
@ -809,7 +931,7 @@ class PortfolioAddMemberView(PortfolioMembersPermissionView, FormMixin):
portfolio,
exc_info=True,
)
messages.warning(self.request, "Could not send portfolio email invitation.")
messages.error(self.request, "Could not send organization invitation email.")
elif isinstance(exception, MissingEmailError):
messages.error(self.request, str(exception))
logger.error(