feat(notifications): Notify assigned team (#34704)

src/sentry/notifications/notifications/activity/assigned.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
-from typing import Any, Mapping
+from collections import defaultdict
+from typing import Any, Mapping, MutableMapping
 
-from sentry.models import Activity, Organization, Team, User
+from sentry.models import Activity, NotificationSetting, Organization, Team, User
+from sentry.notifications.types import GroupSubscriptionReason
+from sentry.types.integrations import ExternalProviders
 
 from .base import GroupActivityNotification
 
@@ -18,33 +21,34 @@ def _get_team_option(assignee_id: int, organization: Organization) -> Team | Non
     return Team.objects.filter(id=assignee_id, organization=organization).first()
 
 
+def is_team_assignee(activity: Activity) -> bool:
+    assignee_type: str | None = activity.data.get("assigneeType")
+    return assignee_type == "team"
+
+
 def get_assignee_str(activity: Activity, organization: Organization) -> str:
     """Get a human-readable version of the assignment's target."""
 
     assignee_id = activity.data.get("assignee")
-    assignee_type = activity.data.get("assigneeType", "user")
     assignee_email: str | None = activity.data.get("assigneeEmail")
 
-    if assignee_type == "user":
-        # TODO(mgaeta): Refactor GroupAssigneeManager to not make IDs into strings.
-        if str(activity.user_id) == str(assignee_id):
-            return "themselves"
-
-        assignee_user = _get_user_option(assignee_id)
-        if assignee_user:
-            assignee: str = assignee_user.get_display_name()
-            return assignee
-        if assignee_email:
-            return assignee_email
-        return "an unknown user"
-
-    if assignee_type == "team":
+    if is_team_assignee(activity):
         assignee_team = _get_team_option(assignee_id, organization)
         if assignee_team:
             return f"the {assignee_team.slug} team"
         return "an unknown team"
 
-    raise NotImplementedError("Unknown Assignee Type")
+    # TODO(mgaeta): Refactor GroupAssigneeManager to not make IDs into strings.
+    if str(activity.user_id) == str(assignee_id):
+        return "themselves"
+
+    assignee_user = _get_user_option(assignee_id)
+    if assignee_user:
+        assignee: str = assignee_user.get_display_name()
+        return assignee
+    if assignee_email:
+        return assignee_email
+    return "an unknown user"
 
 
 class AssignedActivityNotification(GroupActivityNotification):
@@ -68,3 +72,30 @@ class AssignedActivityNotification(GroupActivityNotification):
             author, assignee = assignee, author
 
         return f"Issue assigned to {assignee} by {author}"
+
+    def get_participants_with_group_subscription_reason(
+        self,
+    ) -> Mapping[ExternalProviders, Mapping[Team | User, int]]:
+        """Hack to tack on the assigned team to the list of users subscribed to the group."""
+        users_by_provider = super().get_participants_with_group_subscription_reason()
+        if is_team_assignee(self.activity):
+            assignee_id = self.activity.data.get("assignee")
+            assignee_team = _get_team_option(assignee_id, self.organization)
+
+            if assignee_team:
+                teams_by_provider = NotificationSetting.objects.filter_to_accepting_recipients(
+                    parent=self.project,
+                    recipients=[assignee_team],
+                    type=self.notification_setting_type,
+                )
+                actors_by_provider: MutableMapping[
+                    ExternalProviders,
+                    MutableMapping[Team | User, int],
+                ] = defaultdict(dict)
+                actors_by_provider.update({**users_by_provider})
+                for provider, teams in teams_by_provider.items():
+                    for team in teams:
+                        actors_by_provider[provider][team] = GroupSubscriptionReason.assigned
+                return actors_by_provider
+
+        return users_by_provider

tests/sentry/notifications/notifications/test_assigned.py

@@ -0,0 +1,96 @@
+import responses
+from django.core import mail
+
+from sentry.models import (
+    Identity,
+    IdentityProvider,
+    IdentityStatus,
+    NotificationSetting,
+    UserOption,
+)
+from sentry.notifications.types import NotificationSettingOptionValues, NotificationSettingTypes
+from sentry.testutils import APITestCase
+from sentry.testutils.helpers import get_attachment, install_slack, link_team
+from sentry.types.integrations import ExternalProviders
+
+
+class AssignedNotificationAPITest(APITestCase):
+    def setUp(self):
+        super().setUp()
+
+        self.integration = install_slack(self.organization)
+
+        self.login_as(self.user)
+
+    @responses.activate
+    def test_sends_assignment_notification(self):
+        """
+        Test that an email AND Slack notification are sent with
+        the expected values when an issue is assigned.
+        """
+
+        UserOption.objects.create(user=self.user, key="self_notifications", value="1")
+        NotificationSetting.objects.update_settings(
+            ExternalProviders.SLACK,
+            NotificationSettingTypes.WORKFLOW,
+            NotificationSettingOptionValues.ALWAYS,
+            user=self.user,
+        )
+
+        Identity.objects.create(
+            external_id="UXXXXXXX1",
+            idp=IdentityProvider.objects.create(type="slack", external_id="TXXXXXXX1", config={}),
+            user=self.user,
+            status=IdentityStatus.VALID,
+            scopes=[],
+        )
+
+        url = f"/api/0/issues/{self.group.id}/"
+        with self.tasks():
+            response = self.client.put(url, format="json", data={"assignedTo": self.user.username})
+        assert response.status_code == 200, response.content
+
+        msg = mail.outbox[0]
+        # check the txt version
+        assert f"assigned {self.group.qualified_short_id} to themselves" in msg.body
+        # check the html version
+        assert f"{self.group.qualified_short_id}</a> to themselves</p>" in msg.alternatives[0][0]
+
+        attachment, text = get_attachment()
+
+        assert text == f"Issue assigned to {self.user.get_display_name()} by themselves"
+        assert attachment["title"] == self.group.title
+        assert self.project.slug in attachment["footer"]
+
+    @responses.activate
+    def test_sends_assignment_notification_team(self):
+        link_team(
+            team=self.team,
+            integration=self.integration,
+            channel_id="CXXXXXXX1",
+            channel_name="#javascript",
+        )
+        NotificationSetting.objects.update_settings(
+            ExternalProviders.SLACK,
+            NotificationSettingTypes.WORKFLOW,
+            NotificationSettingOptionValues.ALWAYS,
+            team=self.team,
+        )
+
+        url = f"/api/0/issues/{self.group.id}/"
+        with self.tasks():
+            response = self.client.put(
+                url, format="json", data={"assignedTo": f"team:{self.team.id}"}
+            )
+        assert response.status_code == 200, response.content
+
+        # No email version for teams.
+        assert not len(mail.outbox)
+
+        attachment, text = get_attachment()
+
+        assert (
+            text == f"Issue assigned to the {self.team.name} team by {self.user.get_display_name()}"
+        )
+        assert attachment["title"] == self.group.title
+        assert self.project.slug in attachment["footer"]

tests/sentry/notifications/test_notifications.py

@@ -127,33 +127,6 @@ class ActivityNotificationTest(APITestCase):
             == f"{self.project.slug} | <http://testserver/settings/account/notifications/workflow/?referrer=note_activity-slack-user|Notification Settings>"
         )
 
-    @responses.activate
-    def test_sends_assignment_notification(self):
-        """
-        Test that an email AND Slack notification are sent with
-        the expected values when an issue is assigned.
-        """
-
-        url = f"/api/0/issues/{self.group.id}/"
-        with self.tasks():
-            response = self.client.put(url, format="json", data={"assignedTo": self.user.username})
-        assert response.status_code == 200, response.content
-
-        msg = mail.outbox[0]
-        # check the txt version
-        assert f"assigned {self.short_id} to themselves" in msg.body
-        # check the html version
-        assert f"{self.short_id}</a> to themselves</p>" in msg.alternatives[0][0]
-
-        attachment, text = get_attachment()
-
-        assert text == f"Issue assigned to {self.name} by themselves"
-        assert attachment["title"] == self.group.title
-        assert (
-            attachment["footer"]
-            == f"{self.project.slug} | <http://testserver/settings/account/notifications/workflow/?referrer=assigned_activity-slack-user|Notification Settings>"
-        )
-
     @responses.activate
     def test_sends_unassignment_notification(self):
         """