archive: Add functions for accessing web-public streams.

These will be helpful for variable upcoming projects to support the
web-public streams feature.

zerver/lib/actions.py

@@ -3918,11 +3918,40 @@ def decode_email_address(email: Text) -> Optional[Tuple[Text, Text]]:
     stream_name = re.sub("%\d{4}", lambda x: unichr(int(x.group(0)[1:])), encoded_stream_name)
     return stream_name, token
 
+SubHelperT = Tuple[List[Dict[str, Any]], List[Dict[str, Any]], List[Dict[str, Any]]]
+
+def get_web_public_subs(realm: Realm) -> SubHelperT:
+    color_idx = 0
+
+    def get_next_color() -> str:
+        nonlocal color_idx
+        color = STREAM_ASSIGNMENT_COLORS[color_idx]
+        color_idx = (color_idx + 1) % len(STREAM_ASSIGNMENT_COLORS)
+        return color
+
+    subscribed = [
+        {'name': stream.name,
+         'in_home_view': True,
+         'invite_only': False,
+         'color': get_next_color(),
+         'desktop_notifications': True,
+         'audible_notifications': True,
+         'push_notifications': False,
+         'pin_to_top': False,
+         'stream_id': stream.id,
+         'description': stream.description,
+         'is_old_stream': is_old_stream(stream.date_created),
+         'stream_weekly_traffic': get_average_weekly_stream_traffic(stream.id,
+                                                                    stream.date_created,
+                                                                    {}),
+         'email_address': ''}
+        for stream in Stream.objects.filter(realm=realm, is_web_public=True, deactivated=False)]
+    return (subscribed, [], [])
+
 # In general, it's better to avoid using .values() because it makes
 # the code pretty ugly, but in this case, it has significant
 # performance impact for loading / for users with large numbers of
 # subscriptions, so it's worth optimizing.
-SubHelperT = Tuple[List[Dict[str, Any]], List[Dict[str, Any]], List[Dict[str, Any]]]
 def gather_subscriptions_helper(user_profile: UserProfile,
                                 include_subscribers: bool=True) -> SubHelperT:
     sub_dicts = get_stream_subscriptions_for_user(user_profile).values(
@@ -4487,6 +4516,11 @@ def get_occupied_streams(realm: Realm) -> QuerySet:
 
     return Stream.objects.filter(id__in=stream_ids, realm=realm, deactivated=False)
 
+def get_web_public_streams(realm: Realm) -> List[Dict[str, Any]]:
+    query = Stream.objects.filter(realm=realm, deactivated=False, is_web_public=True)
+    streams = [(row.to_dict()) for row in query]
+    return streams
+
 def do_get_streams(user_profile: UserProfile, include_public: bool=True,
                    include_subscribed: bool=True, include_all_active: bool=False,
                    include_default: bool=False) -> List[Dict[str, Any]]:

zerver/tests/test_archive.py

@@ -2,6 +2,9 @@
 from django.http import HttpResponse
 from zerver.lib.test_classes import ZulipTestCase
 from zerver.lib.actions import do_change_stream_web_public
+from zerver.lib.actions import get_web_public_streams, get_web_public_subs, \
+    do_deactivate_stream
+from zerver.models import get_realm
 
 class GlobalPublicStreamTest(ZulipTestCase):
 
@@ -45,3 +48,36 @@ class GlobalPublicStreamTest(ZulipTestCase):
         self.assert_in_success_response(["Test Message 1"], result)
         result = send_msg_and_get_result('/me goes testing.')
         self.assert_in_success_response(["goes testing."], result)
+
+    def test_get_web_public_streams(self) -> None:
+        realm = get_realm("zulip")
+        public_streams = get_web_public_streams(realm)
+        self.assert_length(public_streams, 1)
+        public_stream = public_streams[0]
+        self.assertEqual(public_stream['name'], "Rome")
+
+        public_subs, public_unsubs, public_neversubs = get_web_public_subs(realm)
+        self.assert_length(public_subs, 1)
+        public_sub = public_subs[0]
+        self.assertEqual(public_sub['name'], "Rome")
+        self.assert_length(public_unsubs, 0)
+        self.assert_length(public_neversubs, 0)
+
+        # Now add a second public stream
+        test_stream = self.make_stream('Test Public Archives')
+        do_change_stream_web_public(test_stream, True)
+        public_streams = get_web_public_streams(realm)
+        self.assert_length(public_streams, 2)
+        public_subs, public_unsubs, public_neversubs = get_web_public_subs(realm)
+        self.assert_length(public_subs, 2)
+        self.assert_length(public_unsubs, 0)
+        self.assert_length(public_neversubs, 0)
+        self.assertNotEqual(public_subs[0]['color'], public_subs[1]['color'])
+
+        do_deactivate_stream(test_stream)
+        public_streams = get_web_public_streams(realm)
+        self.assert_length(public_streams, 1)
+        public_subs, public_unsubs, public_neversubs = get_web_public_subs(realm)
+        self.assert_length(public_subs, 1)
+        self.assert_length(public_unsubs, 0)
+        self.assert_length(public_neversubs, 0)