lint

src/registrar/tests/test_views_member_domains_json.py

@@ -94,6 +94,12 @@ def setUpClass(cls):
         DomainInvitation.objects.create(
             email=cls.invited_member_email, domain=cls.domain2, status=DomainInvitation.DomainInvitationStatus.INVITED
         )
+        DomainInvitation.objects.create(
+            email=cls.invited_member_email, domain=cls.domain3, status=DomainInvitation.DomainInvitationStatus.CANCELED
+        )
+        DomainInvitation.objects.create(
+            email=cls.invited_member_email, domain=cls.domain4, status=DomainInvitation.DomainInvitationStatus.RETRIEVED
+        )
 
     @classmethod
     def tearDownClass(cls):

src/registrar/tests/test_views_portfolio.py

@@ -14,6 +14,7 @@
     Suborganization,
     AllowedEmail,
 )
+from registrar.models.domain_invitation import DomainInvitation
 from registrar.models.portfolio_invitation import PortfolioInvitation
 from registrar.models.user_group import UserGroup
 from registrar.models.user_portfolio_permission import UserPortfolioPermission
@@ -25,6 +26,7 @@
 import boto3_mocking  # type: ignore
 from django.test import Client
 import logging
+import json
 
 logger = logging.getLogger(__name__)
 
@@ -1927,7 +1929,7 @@ def setUpClass(cls):
         cls.portfolio = Portfolio.objects.create(creator=cls.user, organization_name="Test Portfolio")
 
         # Assign permissions to the user making requests
-        UserPortfolioPermission.objects.create(
+        cls.portfolio_permission = UserPortfolioPermission.objects.create(
             user=cls.user,
             portfolio=cls.portfolio,
             roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN],
@@ -2106,10 +2108,15 @@ class TestPortfolioMemberDomainsEditView(TestPortfolioMemberDomainsView):
     @classmethod
     def setUpClass(cls):
         super().setUpClass()
+        cls.url = reverse("member-domains-edit", kwargs={"pk": cls.portfolio_permission.pk})
+        names = ["1.gov", "2.gov", "3.gov"]
+        Domain.objects.bulk_create([Domain(name=name) for name in names])
 
     @classmethod
     def tearDownClass(cls):
         super().tearDownClass()
+        UserDomainRole.objects.all().delete()
+        Domain.objects.all().delete()
 
     @less_console_noise_decorator
     @override_flag("organization_feature", active=True)
@@ -2162,15 +2169,132 @@ def test_member_domains_edit_not_found(self):
         # Make sure the response is not found
         self.assertEqual(response.status_code, 404)
 
+    @less_console_noise_decorator
+    @override_flag("organization_feature", active=True)
+    @override_flag("organization_members", active=True)
+    def test_post_with_valid_added_domains(self):
+        """Test that domains can be successfully added."""
+        self.client.force_login(self.user)
+
+        data = {
+            "added_domains": json.dumps([1, 2, 3]),  # Mock domain IDs
+        }
+        response = self.client.post(self.url, data)
+
+        # Check that the UserDomainRole objects were created
+        self.assertEqual(UserDomainRole.objects.filter(user=self.user).count(), 3)
+
+        # Check for a success message and a redirect
+        self.assertRedirects(response, reverse("member-domains", kwargs={"pk": self.portfolio_permission.pk}))
+        messages = list(response.wsgi_request._messages)
+        self.assertEqual(len(messages), 1)
+        self.assertEqual(str(messages[0]), "The domain assignment changes have been saved.")
+
+    @less_console_noise_decorator
+    @override_flag("organization_feature", active=True)
+    @override_flag("organization_members", active=True)
+    def test_post_with_valid_removed_domains(self):
+        """Test that domains can be successfully removed."""
+        self.client.force_login(self.user)
+
+        # Create some UserDomainRole objects
+        domains = [1, 2, 3]
+        UserDomainRole.objects.bulk_create([UserDomainRole(domain_id=domain, user=self.user) for domain in domains])
+
+        data = {
+            "removed_domains": json.dumps([1, 2]),
+        }
+        response = self.client.post(self.url, data)
+
+        # Check that the UserDomainRole objects were deleted
+        self.assertEqual(UserDomainRole.objects.filter(user=self.user).count(), 1)
+        self.assertEqual(UserDomainRole.objects.filter(domain_id=3, user=self.user).count(), 1)
+
+        # Check for a success message and a redirect
+        self.assertRedirects(response, reverse("member-domains", kwargs={"pk": self.portfolio_permission.pk}))
+        messages = list(response.wsgi_request._messages)
+        self.assertEqual(len(messages), 1)
+        self.assertEqual(str(messages[0]), "The domain assignment changes have been saved.")
+
+        UserDomainRole.objects.all().delete()
+
+    @less_console_noise_decorator
+    @override_flag("organization_feature", active=True)
+    @override_flag("organization_members", active=True)
+    def test_post_with_invalid_added_domains_data(self):
+        """Test that an error is returned for invalid added domains data."""
+        self.client.force_login(self.user)
+
+        data = {
+            "added_domains": "json-statham",
+        }
+        response = self.client.post(self.url, data)
+
+        # Check that no UserDomainRole objects were created
+        self.assertEqual(UserDomainRole.objects.filter(user=self.user).count(), 0)
+
+        # Check for an error message and a redirect
+        self.assertRedirects(response, reverse("member-domains", kwargs={"pk": self.portfolio_permission.pk}))
+        messages = list(response.wsgi_request._messages)
+        self.assertEqual(len(messages), 1)
+        self.assertEqual(str(messages[0]), "Invalid data for added domains.")
+
+    @less_console_noise_decorator
+    @override_flag("organization_feature", active=True)
+    @override_flag("organization_members", active=True)
+    def test_post_with_invalid_removed_domains_data(self):
+        """Test that an error is returned for invalid removed domains data."""
+        self.client.force_login(self.user)
+
+        data = {
+            "removed_domains": "not-a-json",
+        }
+        response = self.client.post(self.url, data)
+
+        # Check that no UserDomainRole objects were deleted
+        self.assertEqual(UserDomainRole.objects.filter(user=self.user).count(), 0)
+
+        # Check for an error message and a redirect
+        self.assertRedirects(response, reverse("member-domains", kwargs={"pk": self.portfolio_permission.pk}))
+        messages = list(response.wsgi_request._messages)
+        self.assertEqual(len(messages), 1)
+        self.assertEqual(str(messages[0]), "Invalid data for removed domains.")
+
+    @less_console_noise_decorator
+    @override_flag("organization_feature", active=True)
+    @override_flag("organization_members", active=True)
+    def test_post_with_no_changes(self):
+        """Test that no changes message is displayed when no changes are made."""
+        self.client.force_login(self.user)
+
+        response = self.client.post(self.url, {})
+
+        # Check that no UserDomainRole objects were created or deleted
+        self.assertEqual(UserDomainRole.objects.filter(user=self.user).count(), 0)
+
+        # Check for an info message and a redirect
+        self.assertRedirects(response, reverse("member-domains", kwargs={"pk": self.portfolio_permission.pk}))
+        messages = list(response.wsgi_request._messages)
+        self.assertEqual(len(messages), 1)
+        self.assertEqual(str(messages[0]), "No changes detected.")
+
 
 class TestPortfolioInvitedMemberEditDomainsView(TestPortfolioInvitedMemberDomainsView):
     @classmethod
     def setUpClass(cls):
         super().setUpClass()
+        cls.url = reverse("invitedmember-domains-edit", kwargs={"pk": cls.invitation.pk})
+        names = ["1.gov", "2.gov", "3.gov"]
+        Domain.objects.bulk_create([Domain(name=name) for name in names])
 
     @classmethod
     def tearDownClass(cls):
         super().tearDownClass()
+        Domain.objects.all().delete()
+
+    def tearDown(self):
+        return super().tearDown()
+        DomainInvitation.objects.all().delete()
 
     @less_console_noise_decorator
     @override_flag("organization_feature", active=True)
@@ -2222,6 +2346,171 @@ def test_member_domains_edit_not_found(self):
         # Make sure the response is not found
         self.assertEqual(response.status_code, 404)
 
+    @less_console_noise_decorator
+    @override_flag("organization_feature", active=True)
+    @override_flag("organization_members", active=True)
+    def test_post_with_valid_added_domains(self):
+        """Test adding new domains successfully."""
+        self.client.force_login(self.user)
+
+        data = {
+            "added_domains": json.dumps([1, 2, 3]),  # Mock domain IDs
+        }
+        response = self.client.post(self.url, data)
+
+        # Check that the DomainInvitation objects were created
+        self.assertEqual(
+            DomainInvitation.objects.filter(
+                email="[email protected]", status=DomainInvitation.DomainInvitationStatus.INVITED
+            ).count(),
+            3,
+        )
+
+        # Check for a success message and a redirect
+        self.assertRedirects(response, reverse("invitedmember-domains", kwargs={"pk": self.invitation.pk}))
+        messages = list(response.wsgi_request._messages)
+        self.assertEqual(len(messages), 1)
+        self.assertEqual(str(messages[0]), "The domain assignment changes have been saved.")
+
+    @less_console_noise_decorator
+    @override_flag("organization_feature", active=True)
+    @override_flag("organization_members", active=True)
+    def test_post_with_existing_and_new_added_domains(self):
+        """Test updating existing and adding new invitations."""
+        self.client.force_login(self.user)
+
+        # Create existing invitations
+        DomainInvitation.objects.bulk_create(
+            [
+                DomainInvitation(
+                    domain_id=1, email="[email protected]", status=DomainInvitation.DomainInvitationStatus.CANCELED
+                ),
+                DomainInvitation(
+                    domain_id=2, email="[email protected]", status=DomainInvitation.DomainInvitationStatus.INVITED
+                ),
+            ]
+        )
+
+        data = {
+            "added_domains": json.dumps([1, 2, 3]),
+        }
+        response = self.client.post(self.url, data)
+
+        # Check that status for domain_id=1 was updated to INVITED
+        self.assertEqual(
+            DomainInvitation.objects.get(domain_id=1, email="[email protected]").status,
+            DomainInvitation.DomainInvitationStatus.INVITED,
+        )
+
+        # Check that domain_id=3 was created as INVITED
+        self.assertTrue(
+            DomainInvitation.objects.filter(
+                domain_id=3, email="[email protected]", status=DomainInvitation.DomainInvitationStatus.INVITED
+            ).exists()
+        )
+
+        # Check for a success message and a redirect
+        self.assertRedirects(response, reverse("invitedmember-domains", kwargs={"pk": self.invitation.pk}))
+
+    @less_console_noise_decorator
+    @override_flag("organization_feature", active=True)
+    @override_flag("organization_members", active=True)
+    def test_post_with_valid_removed_domains(self):
+        """Test removing domains successfully."""
+        self.client.force_login(self.user)
+
+        # Create existing invitations
+        DomainInvitation.objects.bulk_create(
+            [
+                DomainInvitation(
+                    domain_id=1, email="[email protected]", status=DomainInvitation.DomainInvitationStatus.INVITED
+                ),
+                DomainInvitation(
+                    domain_id=2, email="[email protected]", status=DomainInvitation.DomainInvitationStatus.INVITED
+                ),
+            ]
+        )
+
+        data = {
+            "removed_domains": json.dumps([1]),
+        }
+        response = self.client.post(self.url, data)
+
+        # Check that the status for domain_id=1 was updated to CANCELED
+        self.assertEqual(
+            DomainInvitation.objects.get(domain_id=1, email="[email protected]").status,
+            DomainInvitation.DomainInvitationStatus.CANCELED,
+        )
+
+        # Check that domain_id=2 remains INVITED
+        self.assertEqual(
+            DomainInvitation.objects.get(domain_id=2, email="[email protected]").status,
+            DomainInvitation.DomainInvitationStatus.INVITED,
+        )
+
+        # Check for a success message and a redirect
+        self.assertRedirects(response, reverse("invitedmember-domains", kwargs={"pk": self.invitation.pk}))
+
+    @less_console_noise_decorator
+    @override_flag("organization_feature", active=True)
+    @override_flag("organization_members", active=True)
+    def test_post_with_invalid_added_domains_data(self):
+        """Test handling of invalid JSON for added domains."""
+        self.client.force_login(self.user)
+
+        data = {
+            "added_domains": "not-a-json",
+        }
+        response = self.client.post(self.url, data)
+
+        # Check that no DomainInvitation objects were created
+        self.assertEqual(DomainInvitation.objects.count(), 0)
+
+        # Check for an error message and a redirect
+        self.assertRedirects(response, reverse("invitedmember-domains", kwargs={"pk": self.invitation.pk}))
+        messages = list(response.wsgi_request._messages)
+        self.assertEqual(len(messages), 1)
+        self.assertEqual(str(messages[0]), "Invalid data for added domains.")
+
+    @less_console_noise_decorator
+    @override_flag("organization_feature", active=True)
+    @override_flag("organization_members", active=True)
+    def test_post_with_invalid_removed_domains_data(self):
+        """Test handling of invalid JSON for removed domains."""
+        self.client.force_login(self.user)
+
+        data = {
+            "removed_domains": "json-sudeikis",
+        }
+        response = self.client.post(self.url, data)
+
+        # Check that no DomainInvitation objects were updated
+        self.assertEqual(DomainInvitation.objects.count(), 0)
+
+        # Check for an error message and a redirect
+        self.assertRedirects(response, reverse("invitedmember-domains", kwargs={"pk": self.invitation.pk}))
+        messages = list(response.wsgi_request._messages)
+        self.assertEqual(len(messages), 1)
+        self.assertEqual(str(messages[0]), "Invalid data for removed domains.")
+
+    @less_console_noise_decorator
+    @override_flag("organization_feature", active=True)
+    @override_flag("organization_members", active=True)
+    def test_post_with_no_changes(self):
+        """Test the case where no changes are made."""
+        self.client.force_login(self.user)
+
+        response = self.client.post(self.url, {})
+
+        # Check that no DomainInvitation objects were created or updated
+        self.assertEqual(DomainInvitation.objects.count(), 0)
+
+        # Check for an info message and a redirect
+        self.assertRedirects(response, reverse("invitedmember-domains", kwargs={"pk": self.invitation.pk}))
+        messages = list(response.wsgi_request._messages)
+        self.assertEqual(len(messages), 1)
+        self.assertEqual(str(messages[0]), "No changes detected.")
+
 
 class TestRequestingEntity(WebTest):
     """The requesting entity page is a domain request form that only exists

src/registrar/views/member_domains_json.py

@@ -90,7 +90,9 @@ def get_domain_ids_from_request(self, request):
                 domain_info_ids = DomainInformation.objects.filter(portfolio=portfolio).values_list(
                     "domain_id", flat=True
                 )
-                domain_invitations = DomainInvitation.objects.filter(email=email, status=DomainInvitation.DomainInvitationStatus.INVITED).values_list("domain_id", flat=True)
+                domain_invitations = DomainInvitation.objects.filter(
+                    email=email, status=DomainInvitation.DomainInvitationStatus.INVITED
+                ).values_list("domain_id", flat=True)
                 return domain_info_ids.intersection(domain_invitations)
         else:
             domain_infos = DomainInformation.objects.filter(portfolio=portfolio)