model: Add stream id helper methods.

This commit adds two helper methods - _get_stream_from_id and
_get_all_stream_ids. These two helper methods help in preparation for
adding stream and subscription property accessor methods in the next
commit.

Tests added.

Author: theViz343

tests/conftest.py

@@ -11,6 +11,7 @@
     CustomProfileField,
     Message,
     MessageType,
+    Stream,
     Subscription,
 )
 from zulipterminal.config.keys import (
@@ -359,6 +360,61 @@ def streams_fixture(
     return deepcopy(streams)
 
 
+@pytest.fixture
+def unsubscribed_streams_fixture() -> Dict[int, Subscription]:
+    unsubscribed_streams: Dict[int, Subscription] = {}
+    for i in range(3, 5):
+        unsubscribed_streams[i] = {
+            "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,
+        }
+    return deepcopy(unsubscribed_streams)
+
+
+@pytest.fixture
+def never_subscribed_streams_fixture() -> Dict[int, Stream]:
+    never_subscribed_streams: Dict[int, Stream] = {}
+    for i in range(5, 7):
+        never_subscribed_streams[i] = {
+            "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),
@@ -1453,7 +1509,7 @@ def user_id(logged_on_user: Dict[str, Any]) -> int:
 
 
 @pytest.fixture
-def stream_dict(streams_fixture: List[Subscription]) -> Dict[int, Subscription]:
+def stream_dict(streams_fixture: List[Dict[str, Any]]) -> Dict[int, Any]:
     return {stream["stream_id"]: stream for stream in streams_fixture}
 
 

tests/model/test_model.py

@@ -1700,6 +1700,122 @@ def test__update_users_data_from_initial_data(
         assert model.user_dict == user_dict
         assert model.users == user_list
 
+    @pytest.mark.parametrize(
+        "stream_id, expected_value",
+        [
+            case(
+                1000,
+                {
+                    "name": "Some general stream",
+                    "date_created": None,
+                    "invite_only": False,
+                    "color": "#baf",  # Color in '#xxxxxx' format
+                    "pin_to_top": False,
+                    "stream_id": 1000,
+                    "is_muted": False,
+                    "audible_notifications": False,
+                    "description": "General Stream",
+                    "rendered_description": "General Stream",
+                    "is_old_stream": True,
+                    "desktop_notifications": False,
+                    "stream_weekly_traffic": 0,
+                    "push_notifications": False,
+                    "email_address": "[email protected]",
+                    "message_retention_days": None,
+                    "subscribers": [1001, 11, 12],
+                    "history_public_to_subscribers": True,
+                },
+            ),
+            case(
+                3,
+                {
+                    "name": "Stream 3",
+                    "date_created": 1472047127,
+                    "invite_only": False,
+                    "color": "#b0a5fd",
+                    "pin_to_top": False,
+                    "stream_id": 3,
+                    "is_muted": False,
+                    "audible_notifications": False,
+                    "description": "A description of stream 3",
+                    "rendered_description": "A description of stream 3",
+                    "desktop_notifications": False,
+                    "stream_weekly_traffic": 0,
+                    "push_notifications": False,
+                    "message_retention_days": 33,
+                    "email_address": "[email protected]",
+                    "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,
+                },
+            ),
+            case(
+                5,
+                {
+                    "name": "Stream 5",
+                    "date_created": 1472047129,
+                    "invite_only": False,
+                    "stream_id": 5,
+                    "description": "A description of stream 5",
+                    "rendered_description": "A description of stream 5",
+                    "stream_weekly_traffic": 0,
+                    "message_retention_days": 35,
+                    "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,
+                },
+            ),
+        ],
+    )
+    def test__get_stream_from_id(
+        self,
+        model,
+        stream_id,
+        expected_value,
+        stream_dict,
+        unsubscribed_streams_fixture,
+        never_subscribed_streams_fixture,
+    ):
+        model.stream_dict = stream_dict
+        model._unsubscribed_streams = unsubscribed_streams_fixture
+        model._never_subscribed_streams = never_subscribed_streams_fixture
+        assert model._get_stream_from_id(stream_id) == expected_value
+
+    def test__get_stream_from_id__nonexistent_stream(
+        self,
+        model,
+        stream_dict,
+        unsubscribed_streams_fixture,
+        never_subscribed_streams_fixture,
+        stream_id=231,  # id 231 does not belong to any stream
+    ):
+        model.stream_dict = stream_dict
+        model._unsubscribed_streams = unsubscribed_streams_fixture
+        model._never_subscribed_streams = never_subscribed_streams_fixture
+        with pytest.raises(RuntimeError):
+            model._get_stream_from_id(stream_id)
+
+    def test__get_all_stream_ids(
+        self,
+        model,
+        stream_dict,
+        unsubscribed_streams_fixture,
+        never_subscribed_streams_fixture,
+        expected_value=[1000, 99, 999, 1, 2, 3, 4, 5, 6],
+    ):
+        model.stream_dict = stream_dict
+        model._unsubscribed_streams = unsubscribed_streams_fixture
+        model._never_subscribed_streams = never_subscribed_streams_fixture
+        assert model._get_all_stream_ids() == expected_value
+
     @pytest.mark.parametrize("muted", powerset([99, 1000]))
     @pytest.mark.parametrize("visual_notification_enabled", powerset([99, 1000]))
     def test__subscribe_to_streams(

zulipterminal/model.py

@@ -1236,6 +1236,22 @@ def _register_non_subscribed_streams(
             stream["stream_id"]: stream for stream in never_subscribed_streams
         }
 
+    def _get_stream_from_id(self, stream_id: int) -> Union[Subscription, Stream]:
+        if stream_id in self.stream_dict:
+            return self.stream_dict[stream_id]
+        elif stream_id in self._unsubscribed_streams:
+            return self._unsubscribed_streams[stream_id]
+        elif stream_id in self._never_subscribed_streams:
+            return self._never_subscribed_streams[stream_id]
+        else:
+            raise RuntimeError(f"Stream with id {stream_id} does not exist!")
+
+    def _get_all_stream_ids(self) -> List[int]:
+        id_list = list(self.stream_dict)
+        id_list.extend(stream_id for stream_id in self._unsubscribed_streams)
+        id_list.extend(stream_id for stream_id in self._never_subscribed_streams)
+        return id_list
+
     def _subscribe_to_streams(self, subscriptions: List[Subscription]) -> None:
         def make_reduced_stream_data(stream: Subscription) -> StreamData:
             # stream_id has been changed to id.