model/api_types: Add data structures to support non-subscribed streams.

This commit creates data structures, similar to stream_dict, to support
unsubscribed and never-subscribed streams. These streams are available
in the fetched initial data which is used to populate the new data
structures using the _register_non_subscribed_streams function.

Test updated and fixtures added.

Author: theViz343

tests/conftest.py

@@ -11,6 +11,8 @@
     CustomProfileField,
     Message,
     MessageType,
+    Stream,
+    Subscription,
 )
 from zulipterminal.config.keys import (
     ZT_TO_URWID_CMD_MAPPING,
@@ -339,6 +341,66 @@ def streams_fixture(
     return deepcopy(streams)
 
 
+@pytest.fixture
+def unsubscribed_streams_fixture() -> List[Subscription]:
+    unsubscribed_streams: List[Subscription] = []
+    for i in range(3, 5):
+        unsubscribed_streams.append(
+            {
+                "name": f"Stream {i}",
+                "date_created": 1472047124 + i,
+                "invite_only": False,
+                "color": "#b0a5fd",
+                "pin_to_top": False,
+                "stream_id": i,
+                "is_muted": False,
+                "audible_notifications": False,
+                "description": f"A description of stream {i}",
+                "rendered_description": f"A description of stream {i}",
+                "desktop_notifications": False,
+                "stream_weekly_traffic": 0,
+                "push_notifications": False,
+                "message_retention_days": i + 30,
+                "email_address": f"stream{i}@example.com",
+                "email_notifications": False,
+                "wildcard_mentions_notify": False,
+                "subscribers": [1001, 11, 12],
+                "history_public_to_subscribers": True,
+                "is_announcement_only": True,
+                "stream_post_policy": 0,
+                "is_web_public": True,
+                "first_message_id": None,
+                "role": 1,
+            }
+        )
+    return deepcopy(unsubscribed_streams)
+
+
+@pytest.fixture
+def never_subscribed_streams_fixture() -> List[Stream]:
+    never_subscribed_streams: List[Stream] = []
+    for i in range(5, 7):
+        never_subscribed_streams.append(
+            {
+                "name": f"Stream {i}",
+                "date_created": 1472047124 + i,
+                "invite_only": False,
+                "stream_id": i,
+                "description": f"A description of stream {i}",
+                "rendered_description": f"A description of stream {i}",
+                "stream_weekly_traffic": 0,
+                "message_retention_days": i + 30,
+                "subscribers": [1001, 11, 12],
+                "history_public_to_subscribers": True,
+                "is_announcement_only": True,
+                "stream_post_policy": 0,
+                "is_web_public": True,
+                "first_message_id": None,
+            }
+        )
+    return deepcopy(never_subscribed_streams)
+
+
 @pytest.fixture
 def realm_emojis() -> Dict[str, Dict[str, Any]]:
     # Omitting source_url, author_id (server version 3.0),
@@ -873,6 +935,8 @@ def initial_data(
     logged_on_user: Dict[str, Any],
     users_fixture: List[Dict[str, Any]],
     streams_fixture: List[Dict[str, Any]],
+    unsubscribed_streams_fixture: List[Subscription],
+    never_subscribed_streams_fixture: List[Stream],
     realm_emojis: Dict[str, Dict[str, Any]],
     custom_profile_fields_fixture: List[Dict[str, Union[str, int]]],
 ) -> Dict[str, Any]:
@@ -884,24 +948,7 @@ def initial_data(
         "email": logged_on_user["email"],
         "user_id": logged_on_user["user_id"],
         "realm_name": "Test Organization Name",
-        "unsubscribed": [
-            {
-                "audible_notifications": False,
-                "description": "announce",
-                "stream_id": 7,
-                "is_old_stream": True,
-                "desktop_notifications": False,
-                "pin_to_top": False,
-                "stream_weekly_traffic": 0,
-                "invite_only": False,
-                "name": "announce",
-                "push_notifications": False,
-                "email_address": "",
-                "color": "#bfd56f",
-                "is_muted": False,
-                "history_public_to_subscribers": True,
-            }
-        ],
+        "unsubscribed": unsubscribed_streams_fixture,
         "result": "success",
         "queue_id": "1522420755:786",
         "realm_users": users_fixture,
@@ -950,24 +997,7 @@ def initial_data(
         "subscriptions": streams_fixture,
         "msg": "",
         "max_message_id": 552761,
-        "never_subscribed": [
-            {
-                "invite_only": False,
-                "description": "Announcements from the Zulip GCI Mentors",
-                "stream_id": 87,
-                "name": "GCI announce",
-                "is_old_stream": True,
-                "stream_weekly_traffic": 0,
-            },
-            {
-                "invite_only": False,
-                "description": "General discussion",
-                "stream_id": 74,
-                "name": "GCI general",
-                "is_old_stream": True,
-                "stream_weekly_traffic": 0,
-            },
-        ],
+        "never_subscribed": never_subscribed_streams_fixture,
         "unread_msgs": {
             "pms": [
                 {"sender_id": 1, "unread_message_ids": [1, 2]},
@@ -1437,6 +1467,26 @@ def stream_dict(streams_fixture: List[Dict[str, Any]]) -> Dict[int, Any]:
     return {stream["stream_id"]: stream for stream in streams_fixture}
 
 
+@pytest.fixture
+def unsubscribed_streams(
+    unsubscribed_streams_fixture: List[Subscription],
+) -> Dict[int, Subscription]:
+    return {
+        unsubscribed_stream["stream_id"]: unsubscribed_stream
+        for unsubscribed_stream in unsubscribed_streams_fixture
+    }
+
+
+@pytest.fixture
+def never_subscribed_streams(
+    never_subscribed_streams_fixture: List[Stream],
+) -> Dict[int, Stream]:
+    return {
+        never_subscribed_stream["stream_id"]: never_subscribed_stream
+        for never_subscribed_stream in never_subscribed_streams_fixture
+    }
+
+
 @pytest.fixture(
     params=[
         {

tests/model/test_model.py

@@ -63,13 +63,17 @@ def test_init(
         realm_emojis_data,
         zulip_emoji,
         stream_dict,
+        unsubscribed_streams,
+        never_subscribed_streams,
     ):
         assert hasattr(model, "controller")
         assert hasattr(model, "client")
         assert model.narrow == []
         assert model._have_last_message == {}
         assert model.stream_id is None
         assert model.stream_dict == stream_dict
+        assert model._unsubscribed_streams == unsubscribed_streams
+        assert model._never_subscribed_streams == never_subscribed_streams
         assert model.recipients == frozenset()
         assert model.index == initial_index
         assert model.last_unread_pm is None

zulipterminal/api_types.py

@@ -204,21 +204,33 @@ class Message(TypedDict, total=False):
 
 
 ###############################################################################
-# In "subscriptions" response from:
+# In "subscriptions", "unsubscribed", and "never_subscribed" responses from:
 #   https://zulip.com/api/register-queue
 # Also directly from:
 #   https://zulip.com/api/get-events#subscription-add
 #   https://zulip.com/api/get-subscriptions (unused)
 
 
-class Subscription(TypedDict):
+class Stream(TypedDict):
     stream_id: int
     name: str
     description: str
     rendered_description: str
     date_created: int  # NOTE: new in Zulip 4.0 / ZFL 30
     invite_only: bool
     subscribers: List[int]
+
+    is_announcement_only: bool  # Deprecated in Zulip 3.0 -> stream_post_policy
+    stream_post_policy: int  # NOTE: new in Zulip 3.0 / ZFL 1
+
+    is_web_public: bool
+    message_retention_days: Optional[int]  # NOTE: new in Zulip 3.0 / ZFL 17
+    history_public_to_subscribers: bool
+    first_message_id: Optional[int]
+    stream_weekly_traffic: Optional[int]
+
+
+class Subscription(Stream):
     desktop_notifications: Optional[bool]
     email_notifications: Optional[bool]
     wildcard_mentions_notify: Optional[bool]
@@ -229,16 +241,8 @@ class Subscription(TypedDict):
 
     is_muted: bool
 
-    is_announcement_only: bool  # Deprecated in Zulip 3.0 -> stream_post_policy
-    stream_post_policy: int  # NOTE: new in Zulip 3.0 / ZFL 1
-
-    is_web_public: bool
     role: int  # NOTE: new in Zulip 4.0 / ZFL 31
     color: str
-    message_retention_days: Optional[int]  # NOTE: new in Zulip 3.0 / ZFL 17
-    history_public_to_subscribers: bool
-    first_message_id: Optional[int]
-    stream_weekly_traffic: Optional[int]
 
     # Deprecated fields
     # in_home_view: bool  # Replaced by is_muted in Zulip 2.1; still present in updates

zulipterminal/model.py

@@ -46,6 +46,7 @@
     PrivateMessageUpdateRequest,
     RealmEmojiData,
     RealmUser,
+    Stream,
     StreamComposition,
     StreamMessageUpdateRequest,
     Subscription,
@@ -177,11 +178,18 @@ def __init__(self, controller: Any) -> None:
         self._update_users_data_from_initial_data()
 
         self.stream_dict: Dict[int, Any] = {}
+        self._unsubscribed_streams: Dict[int, Subscription] = {}
+        self._never_subscribed_streams: Dict[int, Stream] = {}
         self.muted_streams: Set[int] = set()
         self.pinned_streams: List[StreamData] = []
         self.unpinned_streams: List[StreamData] = []
         self.visual_notified_streams: Set[int] = set()
 
+        self._register_non_subscribed_streams(
+            unsubscribed_streams=self.initial_data["unsubscribed"],
+            never_subscribed_streams=self.initial_data["never_subscribed"],
+        )
+
         self._subscribe_to_streams(self.initial_data["subscriptions"])
 
         # NOTE: The date_created field of stream has been added in feature
@@ -1260,6 +1268,19 @@ def user_name_from_id(self, user_id: int) -> str:
 
         return self.user_dict[user_email]["full_name"]
 
+    def _register_non_subscribed_streams(
+        self,
+        unsubscribed_streams: List[Subscription],
+        never_subscribed_streams: List[Stream],
+    ) -> None:
+        self._unsubscribed_streams = {
+            subscription["stream_id"]: subscription
+            for subscription in unsubscribed_streams
+        }
+        self._never_subscribed_streams = {
+            stream["stream_id"]: stream for stream in never_subscribed_streams
+        }
+
     def _subscribe_to_streams(self, subscriptions: List[Subscription]) -> None:
         def make_reduced_stream_data(stream: Subscription) -> StreamData:
             # stream_id has been changed to id.