subs: Fix clearing unread counts when leaving private streams.

Because we use access_stream_by_id here, and that checks for an active
subscription to interact with a private stream, this didn't work.

The correct fix to add an option to active_stream_by_id to accept an
argument indicating whether we need an active subscription; for this
use case, we definitely do not.

zerver/lib/streams.py

@@ -30,7 +30,8 @@ def access_stream_for_delete(user_profile: UserProfile, stream_id: int) -> Strea
     return stream
 
 def access_stream_common(user_profile: UserProfile, stream: Stream,
-                         error: Text) -> Tuple[Recipient, Subscription]:
+                         error: Text,
+                         require_active: bool=True) -> Tuple[Recipient, Subscription]:
     """Common function for backend code where the target use attempts to
     access the target stream, returning all the data fetched along the
     way.  If that user does not have permission to access that stream,
@@ -46,7 +47,7 @@ def access_stream_common(user_profile: UserProfile, stream: Stream,
     try:
         sub = Subscription.objects.get(user_profile=user_profile,
                                        recipient=recipient,
-                                       active=True)
+                                       active=require_active)
     except Subscription.DoesNotExist:
         sub = None
 
@@ -62,14 +63,17 @@ def access_stream_common(user_profile: UserProfile, stream: Stream,
     # an error.
     raise JsonableError(error)
 
-def access_stream_by_id(user_profile: UserProfile, stream_id: int) -> Tuple[Stream, Recipient, Subscription]:
+def access_stream_by_id(user_profile: UserProfile,
+                        stream_id: int,
+                        require_active: bool=True) -> Tuple[Stream, Recipient, Subscription]:
     error = _("Invalid stream id")
     try:
         stream = Stream.objects.get(id=stream_id)
     except Stream.DoesNotExist:
         raise JsonableError(error)
 
-    (recipient, sub) = access_stream_common(user_profile, stream, error)
+    (recipient, sub) = access_stream_common(user_profile, stream, error,
+                                            require_active=require_active)
     return (stream, recipient, sub)
 
 def check_stream_name_available(realm: Realm, name: Text) -> None:

zerver/tests/test_subs.py

@@ -631,7 +631,7 @@ class StreamAdminTest(ZulipTestCase):
         are on.
         """
         result = self.attempt_unsubscribe_of_principal(
-            query_count=19, is_admin=True, is_subbed=True, invite_only=True,
+            query_count=21, is_admin=True, is_subbed=True, invite_only=True,
             other_user_subbed=True)
         json = self.assert_json_success(result)
         self.assertEqual(len(json["removed"]), 1)
@@ -2319,14 +2319,18 @@ class SubscriptionAPITest(ZulipTestCase):
         random_user = self.example_user("hamlet")
         (stream1, _) = create_stream_if_needed(realm, "stream1", invite_only=False)
         (stream2, _) = create_stream_if_needed(realm, "stream2", invite_only=False)
+        (private, _) = create_stream_if_needed(realm, "private_stream", invite_only=True)
 
         self.subscribe(user, "stream1")
         self.subscribe(user, "stream2")
+        self.subscribe(user, "private_stream")
         self.subscribe(random_user, "stream1")
         self.subscribe(random_user, "stream2")
+        self.subscribe(random_user, "private_stream")
 
         self.send_stream_message(random_user.email, "stream1", "test", "test")
         self.send_stream_message(random_user.email, "stream2", "test", "test")
+        self.send_stream_message(random_user.email, "private_stream", "test", "test")
 
         def get_unread_stream_data() -> List[Dict[str, Any]]:
             raw_unread_data = get_raw_unread_data(user)
@@ -2334,14 +2338,17 @@ class SubscriptionAPITest(ZulipTestCase):
             return aggregated_data['streams']
 
         result = get_unread_stream_data()
-        self.assert_length(result, 2)
+        self.assert_length(result, 3)
         self.assertEqual(result[0]['stream_id'], stream1.id)
         self.assertEqual(result[1]['stream_id'], stream2.id)
+        self.assertEqual(result[2]['stream_id'], private.id)
 
         # Unsubscribing should mark all the messages in stream2 as read
         self.unsubscribe(user, "stream2")
+        self.unsubscribe(user, "private_stream")
 
         self.subscribe(user, "stream2")
+        self.subscribe(user, "private_stream")
         result = get_unread_stream_data()
         self.assert_length(result, 1)
         self.assertEqual(result[0]['stream_id'], stream1.id)

zerver/worker/queue_processors.py

@@ -517,5 +517,9 @@ class DeferredWorker(QueueProcessingWorker):
             user_profile = get_user_profile_by_id(event['user_profile_id'])
 
             for stream_id in event['stream_ids']:
-                (stream, recipient, sub) = access_stream_by_id(user_profile, stream_id)
+                # Since the user just unsubscribed, we don't require
+                # an active Subscription object (otherwise, private
+                # streams would never be accessible)
+                (stream, recipient, sub) = access_stream_by_id(user_profile, stream_id,
+                                                               require_active=False)
                 do_mark_stream_messages_as_read(user_profile, stream)