diff --git a/src/registrar/admin.py b/src/registrar/admin.py index 31c75e05e..928ead442 100644 --- a/src/registrar/admin.py +++ b/src/registrar/admin.py @@ -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) diff --git a/src/registrar/fixtures/fixtures_domains.py b/src/registrar/fixtures/fixtures_domains.py index 4606024d0..8855194f8 100644 --- a/src/registrar/fixtures/fixtures_domains.py +++ b/src/registrar/fixtures/fixtures_domains.py @@ -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): diff --git a/src/registrar/fixtures/fixtures_portfolios.py b/src/registrar/fixtures/fixtures_portfolios.py index 2a391fb16..b93b9efdd 100644 --- a/src/registrar/fixtures/fixtures_portfolios.py +++ b/src/registrar/fixtures/fixtures_portfolios.py @@ -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}") diff --git a/src/registrar/fixtures/fixtures_requests.py b/src/registrar/fixtures/fixtures_requests.py index c4d824b37..6eee6438f 100644 --- a/src/registrar/fixtures/fixtures_requests.py +++ b/src/registrar/fixtures/fixtures_requests.py @@ -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 diff --git a/src/registrar/fixtures/fixtures_suborganizations.py b/src/registrar/fixtures/fixtures_suborganizations.py index af7e02804..787ce8f75 100644 --- a/src/registrar/fixtures/fixtures_suborganizations.py +++ b/src/registrar/fixtures/fixtures_suborganizations.py @@ -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): diff --git a/src/registrar/fixtures/fixtures_user_portfolio_permissions.py b/src/registrar/fixtures/fixtures_user_portfolio_permissions.py index 5f9fd64ef..e2c84f817 100644 --- a/src/registrar/fixtures/fixtures_user_portfolio_permissions.py +++ b/src/registrar/fixtures/fixtures_user_portfolio_permissions.py @@ -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): diff --git a/src/registrar/fixtures/fixtures_users.py b/src/registrar/fixtures/fixtures_users.py index 977bf0858..876bc9fb5 100644 --- a/src/registrar/fixtures/fixtures_users.py +++ b/src/registrar/fixtures/fixtures_users.py @@ -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) diff --git a/src/registrar/forms/portfolio.py b/src/registrar/forms/portfolio.py index c9ef280b0..2725224f1 100644 --- a/src/registrar/forms/portfolio.py +++ b/src/registrar/forms/portfolio.py @@ -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): """ diff --git a/src/registrar/management/commands/remove_unused_portfolios.py b/src/registrar/management/commands/remove_unused_portfolios.py index 4940cc16f..859318a45 100644 --- a/src/registrar/management/commands/remove_unused_portfolios.py +++ b/src/registrar/management/commands/remove_unused_portfolios.py @@ -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 diff --git a/src/registrar/models/utility/portfolio_helper.py b/src/registrar/models/utility/portfolio_helper.py index 2c7b733d5..5feae1cc1 100644 --- a/src/registrar/models/utility/portfolio_helper.py +++ b/src/registrar/models/utility/portfolio_helper.py @@ -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(): diff --git a/src/registrar/templates/emails/portfolio_admin_addition_notification.txt b/src/registrar/templates/emails/portfolio_admin_addition_notification.txt new file mode 100644 index 000000000..b8953aa67 --- /dev/null +++ b/src/registrar/templates/emails/portfolio_admin_addition_notification.txt @@ -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 . + + +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: +Learn about .gov + +The .gov registry is a part of the Cybersecurity and Infrastructure Security Agency +(CISA) +{% endautoescape %} diff --git a/src/registrar/templates/emails/portfolio_admin_addition_notification_subject.txt b/src/registrar/templates/emails/portfolio_admin_addition_notification_subject.txt new file mode 100644 index 000000000..3d6b2a140 --- /dev/null +++ b/src/registrar/templates/emails/portfolio_admin_addition_notification_subject.txt @@ -0,0 +1 @@ +An admin was invited to your .gov organization \ No newline at end of file diff --git a/src/registrar/templates/emails/portfolio_admin_removal_notification.txt b/src/registrar/templates/emails/portfolio_admin_removal_notification.txt new file mode 100644 index 000000000..6a536aa49 --- /dev/null +++ b/src/registrar/templates/emails/portfolio_admin_removal_notification.txt @@ -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 . + +---------------------------------------------------------------- + +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: +Learn about .gov + +The .gov registry is a part of the Cybersecurity and Infrastructure Security Agency +(CISA) +{% endautoescape %} diff --git a/src/registrar/templates/emails/portfolio_admin_removal_notification_subject.txt b/src/registrar/templates/emails/portfolio_admin_removal_notification_subject.txt new file mode 100644 index 000000000..e250b17f8 --- /dev/null +++ b/src/registrar/templates/emails/portfolio_admin_removal_notification_subject.txt @@ -0,0 +1 @@ +An admin was removed from your .gov organization \ No newline at end of file diff --git a/src/registrar/tests/test_admin.py b/src/registrar/tests/test_admin.py index 387319663..28a407036 100644 --- a/src/registrar/tests/test_admin.py +++ b/src/registrar/tests/test_admin.py @@ -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 diff --git a/src/registrar/tests/test_email_invitations.py b/src/registrar/tests/test_email_invitations.py index 1377dec42..77a8c402f 100644 --- a/src/registrar/tests/test_email_invitations.py +++ b/src/registrar/tests/test_email_invitations.py @@ -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) diff --git a/src/registrar/tests/test_views_domain.py b/src/registrar/tests/test_views_domain.py index dc5bff27a..27dff5e3a 100644 --- a/src/registrar/tests/test_views_domain.py +++ b/src/registrar/tests/test_views_domain.py @@ -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() diff --git a/src/registrar/tests/test_views_portfolio.py b/src/registrar/tests/test_views_portfolio.py index b50c9a36f..0c7c56e74 100644 --- a/src/registrar/tests/test_views_portfolio.py +++ b/src/registrar/tests/test_views_portfolio.py @@ -1643,10 +1643,33 @@ class TestPortfolio(WebTest): # Assert that the toggleable alert ID exists self.assertContains(response, '
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 diff --git a/src/registrar/views/domain.py b/src/registrar/views/domain.py index 297cb689a..24673ac4f 100644 --- a/src/registrar/views/domain.py +++ b/src/registrar/views/domain.py @@ -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] ) diff --git a/src/registrar/views/portfolios.py b/src/registrar/views/portfolios.py index beb04d2c7..0f93ec8e1 100644 --- a/src/registrar/views/portfolios.py +++ b/src/registrar/views/portfolios.py @@ -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(