ref(update-groups): Move is_subscribed to new method

src/sentry/api/helpers/group_index/update.py

@@ -638,29 +638,10 @@ def update_groups(
             result["isBookmarked"], group_list, group_ids, project_lookup, acting_user
         )
 
-    # TODO(dcramer): we could make these more efficient by first
-    # querying for rich rows are present (if N > 2), flipping the flag
-    # on those rows, and then creating the missing rows
     if result.get("isSubscribed") in (True, False):
-        is_subscribed = result["isSubscribed"]
-        for group in group_list:
-            # NOTE: Subscribing without an initiating event (assignment,
-            # commenting, etc.) clears out the previous subscription reason
-            # to avoid showing confusing messaging as a result of this
-            # action. It'd be jarring to go directly from "you are not
-            # subscribed" to "you were subscribed due since you were
-            # assigned" just by clicking the "subscribe" button (and you
-            # may no longer be assigned to the issue anyway.)
-            GroupSubscription.objects.create_or_update(
-                user_id=acting_user.id,
-                group=group,
-                project=project_lookup[group.project_id],
-                values={"is_active": is_subscribed, "reason": GroupSubscriptionReason.unknown},
-            )
-
-        result["subscriptionDetails"] = {
-            "reason": SUBSCRIPTION_REASON_MAP.get(GroupSubscriptionReason.unknown, "unknown")
-        }
+        result["subscriptionDetails"] = handle_is_subscribed(
+            result["isSubscribed"], group_list, project_lookup, acting_user
+        )
 
     if "isPublic" in result:
         result["shareId"] = handle_is_public(
@@ -700,6 +681,33 @@ def update_groups(
     return Response(result)
 
 
+def handle_is_subscribed(
+    is_subscribed: bool,
+    group_list: Sequence[Group],
+    project_lookup: dict[int, Any],
+    acting_user: User,
+) -> dict[str, str]:
+    # TODO(dcramer): we could make these more efficient by first
+    # querying for which `GroupSubscription` rows are present (if N > 2),
+    # flipping the flag on those rows, and then creating the missing rows
+    for group in group_list:
+        # NOTE: Subscribing without an initiating event (assignment,
+        # commenting, etc.) clears out the previous subscription reason
+        # to avoid showing confusing messaging as a result of this
+        # action. It'd be jarring to go directly from "you are not
+        # subscribed" to "you were subscribed since you were
+        # assigned" just by clicking the "subscribe" button (and you
+        # may no longer be assigned to the issue anyway).
+        GroupSubscription.objects.create_or_update(
+            user_id=acting_user.id,
+            group=group,
+            project=project_lookup[group.project_id],
+            values={"is_active": is_subscribed, "reason": GroupSubscriptionReason.unknown},
+        )
+
+    return {"reason": SUBSCRIPTION_REASON_MAP.get(GroupSubscriptionReason.unknown, "unknown")}
+
+
 def handle_is_bookmarked(
     is_bookmarked: bool,
     group_list: Sequence[Group],

tests/sentry/api/helpers/test_group_index.py

@@ -12,6 +12,7 @@
     handle_has_seen,
     handle_is_bookmarked,
     handle_is_public,
+    handle_is_subscribed,
 )
 from sentry.api.issue_search import parse_search_query
 from sentry.models import (
@@ -22,6 +23,7 @@
     GroupSeen,
     GroupShare,
     GroupStatus,
+    GroupSubscription,
     GroupSubStatus,
     add_group_to_inbox,
 )
@@ -212,6 +214,31 @@ def test_ignore_with_substatus_until_escalating(self, send_robust):
         assert not GroupInbox.objects.filter(group=group).exists()
 
 
+class TestHandleIsSubscribed(TestCase):
+    def setUp(self) -> None:
+        self.group = self.create_group()
+        self.group_list = [self.group]
+        self.project_lookup = {self.group.project_id: self.group.project}
+
+    def test_is_subscribed(self) -> None:
+        resp = handle_is_subscribed(True, self.group_list, self.project_lookup, self.user)
+
+        assert GroupSubscription.objects.filter(group=self.group, user_id=self.user.id).exists()
+        assert resp["reason"] == "unknown"
+
+    def test_is_subscribed_updates(self) -> None:
+        GroupSubscription.objects.create(
+            group=self.group, project=self.group.project, user_id=self.user.id, is_active=False
+        )
+
+        resp = handle_is_subscribed(True, self.group_list, self.project_lookup, self.user)
+
+        subscription = GroupSubscription.objects.filter(group=self.group, user_id=self.user.id)
+        assert subscription.exists()
+        assert subscription.first().is_active
+        assert resp["reason"] == "unknown"
+
+
 class TestHandleIsBookmarked(TestCase):
     def setUp(self):
         self.group = self.create_group()