Fix SDK bugs found by the live docs snippet tests

examples/native_sync/message_reactions.py

@@ -192,10 +192,11 @@ def main() -> None:
     assert len(messages[0].actions) >= 2, "Unexpected number of actions in history"
 
     # Step 4: Retrieve all reactions for the message
-    # We use a time window around the message timetoken to fetch reactions
-    # The window is 1000 time units before and after the message
-    start_timetoken = str(int(message_timetoken) - 1000)
-    end_timetoken = str(int(message_timetoken) + 1000)
+    # The range is made of action timetokens. start is the newer bound (exclusive) and end is the older bound
+    # (inclusive), so start must be greater than end. Reactions are added after the message, hence the message
+    # timetoken is a safe older bound and the newest reaction plus one is the newer bound.
+    start_timetoken = str(int(second_reaction.action_timetoken) + 1)
+    end_timetoken = str(message_timetoken)
     reactions = get_reactions(pubnub, channel, start_timetoken, end_timetoken, "100")
     print(f"Reactions found: {len(reactions.actions)}")
     assert len(reactions.actions) >= 2, "Unexpected number of reactions"

pubnub/endpoints/access/grant.py

@@ -160,7 +160,7 @@ def validate_params(self):
                                   )
 
         if self._write is None and self._read is None and self._manage is None and self._get is None \
-                and self._update is None and self._join is None:
+                and self._update is None and self._join is None and self._delete is None:
             raise PubNubException(pn_error=PNERR_PAM_NO_FLAGS)
 
         self.validate_categories()

pubnub/endpoints/message_actions/get_message_actions.py

@@ -37,8 +37,8 @@ def end(self, end: str) -> 'GetMessageActions':
         return self
 
     def limit(self, limit: str) -> 'GetMessageActions':
-        assert isinstance(limit, str)
-        self._limit = limit
+        assert isinstance(limit, (str, int)) and not isinstance(limit, bool)
+        self._limit = str(limit)
         return self
 
     def custom_params(self):
@@ -47,7 +47,7 @@ def custom_params(self):
         if self._start is not None:
             params['start'] = self._start
 
-        if self._end is not None and self._start is None:
+        if self._end is not None:
             params['end'] = self._end
 
         if self._limit != GetMessageActions.MAX_LIMIT:

pubnub/endpoints/objects_v2/objects_endpoint.py

@@ -19,7 +19,7 @@ class ObjectsEndpoint(Endpoint):
 
     __if_matches_etag: str = None
 
-    _custom_headers: dict = {}
+    _custom_headers: dict = None
 
     def __init__(self, pubnub):
         Endpoint.__init__(self, pubnub)
@@ -42,7 +42,9 @@ def validate_specific_params(self):
 
     def if_matches_etag(self, etag: str):
         self.__if_matches_etag = etag
-        self._custom_headers.update({"If-Match": etag})
+        # copy so the header stays on this request and never lands on a shared dict
+        self._custom_headers = dict(self._custom_headers or {})
+        self._custom_headers["If-Match"] = etag
         return self
 
     def encoded_params(self):
@@ -67,8 +69,8 @@ def custom_params(self):
             if self._limit:
                 params["limit"] = int(self._limit)
 
-            if self._include_total_count:
-                params["count"] = bool(self._include_total_count)
+            if self._include_total_count or getattr(self._includes, 'total_count', False):
+                params["count"] = True
 
             if self._sort_keys:
                 joined_sort_params_array = []

pubnub/event_engine/models/states.py

@@ -16,6 +16,12 @@ class PNContext(dict):
     reason: PubNubException
     with_presence: bool = False
 
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        # read by get_subscribed_channels() and unsubscribe_all() before the first subscribe
+        self.channels = []
+        self.groups = []
+
     def update(self, context):
         super().update(context.__dict__)
 

pubnub/managers.py

@@ -2,6 +2,7 @@
 from abc import abstractmethod, ABCMeta
 
 import base64
+import binascii
 import random
 
 from cbor2 import loads
@@ -401,6 +402,14 @@ def get_token(self):
     def parse_token(cls, token):
         token = cls.unwrap_token(token)
 
+        try:
+            return cls._parse_unwrapped_token(token)
+        except (KeyError, TypeError, AttributeError, ValueError):
+            raise PubNubException(pn_error=PNERR_INVALID_ACCESS_TOKEN,
+                                  errormsg="token payload is missing required fields or has the wrong shape")
+
+    @staticmethod
+    def _parse_unwrapped_token(token):
         parsed_token = {
             "version": token["v"],
             "timestamp": token["t"],
@@ -444,13 +453,23 @@ def parse_token(cls, token):
 
     @staticmethod
     def unwrap_token(token):
+        if not isinstance(token, str) or len(token) == 0:
+            raise PubNubException(pn_error=PNERR_INVALID_ACCESS_TOKEN, errormsg="token must be a non-empty string")
+
         token = token.replace("_", "/").replace("-", "+")
-        byte_array = base64.b64decode(token)
+
+        try:
+            byte_array = base64.b64decode(token)
+        except (binascii.Error, ValueError):
+            raise PubNubException(pn_error=PNERR_INVALID_ACCESS_TOKEN, errormsg="token is not valid base64")
 
         try:
             unwrapped_obj = loads(byte_array)
             decoded_obj = utils.decode_utf8_dict(unwrapped_obj)
-
-            return decoded_obj
         except Exception:
             raise PubNubException(pn_error=PNERR_INVALID_ACCESS_TOKEN)
+
+        if not isinstance(decoded_obj, dict):
+            raise PubNubException(pn_error=PNERR_INVALID_ACCESS_TOKEN, errormsg="token payload is not a map")
+
+        return decoded_obj

pubnub/models/consumer/history.py

@@ -92,6 +92,8 @@ def from_json(cls, json_input, include_message_actions=False, start_timetoken=No
                     message.uuid = item['uuid']
                 if 'message_type' in item:
                     message.message_type = item['message_type']
+                if 'custom_message_type' in item:
+                    message.custom_message_type = item['custom_message_type']
 
                 if 'meta' in item:
                     message.meta = item['meta']
@@ -118,6 +120,7 @@ def __init__(self, message, timetoken, meta=None, actions=None, error: Exception
         self.timetoken = timetoken
         self.actions = actions
         self.error = error
+        self.custom_message_type = None
 
     def __str__(self):
         return "Fetch message item with tt: %s and content: %s" % (self.timetoken, self.message)

pubnub/models/consumer/pubsub.py

@@ -91,6 +91,8 @@ def __init__(self, result, *, subscription=None, channel=None):
         super(PNMessageActionResult, self).__init__(result)
         self.subscription = subscription
         self.channel = channel
+        # 'added' or 'removed', set by the subscribe message worker from the event payload
+        self.event = None
 
 
 class PNPublishResult(object):

pubnub/models/subscription.py

@@ -40,10 +40,12 @@ def wildcard_match(name, subscription):
             return any([subscription_item.is_matching_listener(message)
                         for subscription_item in self.get_subscription_items()])
         else:
+            # App Context results built outside the message worker have no channel or subscription
+            channel = getattr(message, 'channel', None)
             if self._type == PNSubscriptionType.CHANNEL:
-                return message.channel == self.name or wildcard_match(message.channel, self.name)
+                return channel is not None and (channel == self.name or wildcard_match(channel, self.name))
             else:
-                return message.subscription == self.name
+                return getattr(message, 'subscription', None) == self.name
 
     def presence(self, presence):
         if not hasattr(self, 'on_presence') or not (hasattr(self, 'with_presence') and self.with_presence):
@@ -64,6 +66,22 @@ def signal(self, signal):
         if self.is_matching_listener(signal) and hasattr(self, 'on_signal'):
             self.on_signal(signal)
 
+    def channel_metadata(self, channel_metadata):
+        if self.is_matching_listener(channel_metadata) and hasattr(self, 'on_channel_metadata'):
+            self.on_channel_metadata(channel_metadata)
+
+    def user_metadata(self, user_metadata):
+        if self.is_matching_listener(user_metadata) and hasattr(self, 'on_user_metadata'):
+            self.on_user_metadata(user_metadata)
+
+    def membership(self, membership):
+        if self.is_matching_listener(membership) and hasattr(self, 'on_membership'):
+            self.on_membership(membership)
+
+    def file_message(self, file_message):
+        if self.is_matching_listener(file_message) and hasattr(self, 'on_file'):
+            self.on_file(file_message)
+
 
 class PNSubscribeCapable:
     def subscribe(self, timetoken: Optional[int] = None, region: Optional[str] = None):
@@ -122,7 +140,7 @@ def __init__(self, pubnub_instance, channel: str) -> None:
 
 
 class PubNubUserMetadata(PNSubscribable):
-    _types = PNSubscriptionType.CHANNEL
+    _type = PNSubscriptionType.CHANNEL
 
     def __init__(self, pubnub_instance, user_id: str) -> None:
         super().__init__(pubnub_instance, user_id)
@@ -153,17 +171,23 @@ def __add_subscription(self, subscription: PubNubSubscription, subscription_set:
 
         for name in subscription.get_names_with_presence():
             if name not in subscription_list:
-                subscription_list[name] = subscriptions
+                # each name gets its own list, a shared one would be emptied by the first removal
+                subscription_list[name] = list(subscriptions)
                 names_added.append(name)
             else:
                 subscription_list[name].extend(subscriptions)
         return names_added
 
-    def __remove_subscription(self, subscription: PubNubSubscription):
+    def __remove_subscription(self, subscription: PubNubSubscription, subscription_set: PubNubSubscriptionSet = None):
         names_removed = {'channels': [],
                          'groups': []}
 
-        self.subscriptions.remove(subscription)
+        if subscription in self.subscriptions:
+            self.subscriptions.remove(subscription)
+
+        subscriptions = [subscription]
+        if subscription_set:
+            subscriptions.append(subscription_set)
 
         if subscription._type == PNSubscriptionType.CHANNEL:
             subscription_list = self.channels
@@ -173,10 +197,17 @@ def __remove_subscription(self, subscription: PubNubSubscription):
             removed = names_removed['groups']
 
         for name in subscription.get_names_with_presence():
-            if name in subscription_list and subscription in subscription_list[name]:
-                subscription_list[name].remove(subscription)
-                if len(subscription_list[name]) == 0:
-                    removed.append(name)
+            if name not in subscription_list:
+                continue
+            listeners = subscription_list[name]
+            if not any(item in listeners for item in subscriptions):
+                continue
+            for item in subscriptions:
+                if item in listeners:
+                    listeners.remove(item)
+            if len(listeners) == 0:
+                del subscription_list[name]
+                removed.append(name)
 
         return names_removed
 
@@ -214,7 +245,7 @@ def remove(self, subscription: Union[PubNubSubscription, PubNubSubscriptionSet])
 
         if isinstance(subscription, PubNubSubscriptionSet):
             for subscription_part in subscription.subscriptions:
-                names_changed = self.__remove_subscription(subscription_part)
+                names_changed = self.__remove_subscription(subscription_part, subscription)
                 channels_changed += names_changed['channels']
                 groups_changed += names_changed['groups']
         else:
@@ -224,10 +255,10 @@ def remove(self, subscription: Union[PubNubSubscription, PubNubSubscriptionSet])
 
         self.with_presence = any(sub.with_presence for sub in self.subscriptions)
 
-        if names_changed:
+        if channels_changed or groups_changed:
             unsubscribe_operation = UnsubscribeOperation(channels=channels_changed, channel_groups=groups_changed)
             self.pubnub._subscription_manager.adapt_unsubscribe_builder(unsubscribe_operation)
-        return names_changed
+        return {'channels': channels_changed, 'groups': groups_changed}
 
     def get_subscribed_channels(self):
         return list(self.channels.keys())
@@ -261,13 +292,16 @@ def remove_listener(self, listener):
         self.global_listeners.remove(listener)
 
     def unsubscribe_all(self):
-        unsubscribe_operation = UnsubscribeOperation(
-            channels=list(self.channels.keys()),
-            channel_groups=list(self.channel_groups.keys())
-        )
+        channels = list(self.channels.keys())
+        channel_groups = list(self.channel_groups.keys())
+        self.channels = {}
+        self.channel_groups = {}
+        self.subscriptions = []
+        self.with_presence = False
+        if not channels and not channel_groups:
+            return
+        unsubscribe_operation = UnsubscribeOperation(channels=channels, channel_groups=channel_groups)
         self.pubnub._subscription_manager.adapt_unsubscribe_builder(unsubscribe_operation)
-        self.channels = []
-        self.channel_groups = []
 
     def unsubscribe(self, channels=None, groups=None):
         presence_channels = []
@@ -296,37 +330,38 @@ def __init__(self, subscription_registry: PNSubscriptionRegistry) -> None:
         self.subscription_registry = subscription_registry
         super().__init__()
 
+    def _dispatch(self, pubnub, callback_name, emitter_name, event):
+        # Listeners added with subscription.add_listener() are SubscribeCallback instances and take
+        # (pubnub, event). Subscriptions and subscription sets take only the event.
+        for listener in self.subscription_registry.get_all_listeners():
+            if isinstance(listener, SubscribeCallback):
+                getattr(listener, callback_name)(pubnub, event)
+            else:
+                getattr(listener, emitter_name)(event)
+
     def status(self, _, status):
         pass
 
-    def presence(self, _, presence):
-        for listener in self.subscription_registry.get_all_listeners():
-            listener.presence(presence)
+    def presence(self, pubnub, presence):
+        self._dispatch(pubnub, 'presence', 'presence', presence)
 
-    def message(self, _, message):
-        for listener in self.subscription_registry.get_all_listeners():
-            listener.message(message)
+    def message(self, pubnub, message):
+        self._dispatch(pubnub, 'message', 'message', message)
 
-    def signal(self, _, signal):
-        for listener in self.subscription_registry.get_all_listeners():
-            listener.signal(signal)
+    def signal(self, pubnub, signal):
+        self._dispatch(pubnub, 'signal', 'signal', signal)
 
-    def channel(self, _, channel):
-        for listener in self.subscription_registry.get_all_listeners():
-            listener.channel(channel)
+    def channel(self, pubnub, channel):
+        self._dispatch(pubnub, 'channel', 'channel_metadata', channel)
 
     def uuid(self, pubnub, uuid):
-        for listener in self.subscription_registry.get_all_listeners():
-            listener.uuid(uuid)
+        self._dispatch(pubnub, 'uuid', 'user_metadata', uuid)
 
-    def membership(self, _, membership):
-        for listener in self.subscription_registry.get_all_listeners():
-            listener.membership(membership)
+    def membership(self, pubnub, membership):
+        self._dispatch(pubnub, 'membership', 'membership', membership)
 
-    def message_action(self, _, message_action):
-        for listener in self.subscription_registry.get_all_listeners():
-            listener.message_action(message_action)
+    def message_action(self, pubnub, message_action):
+        self._dispatch(pubnub, 'message_action', 'message_action', message_action)
 
-    def file(self, _, file_message):
-        for listener in self.subscription_registry.get_all_listeners():
-            listener.file_message(file_message)
+    def file(self, pubnub, file_message):
+        self._dispatch(pubnub, 'file', 'file_message', file_message)

pubnub/pubnub_core.py

@@ -406,7 +406,25 @@ def unsubscribe_all(self) -> None:
         return self._subscription_registry.unsubscribe_all()
 
     def reconnect(self) -> None:
-        return self._subscription_registry.reconnect()
+        """Restart the subscribe loop for the current subscriptions.
+
+        The event engine subscription manager (asyncio default) handles reconnection
+        itself, so there this call does nothing.
+        """
+        self._validate_subscribe_manager_enabled()
+        return self._subscription_manager.reconnect()
+
+    def disconnect(self) -> None:
+        """Stop the subscribe loop and heartbeat timer without removing the subscriptions.
+
+        Raises:
+            PubNubException: If the subscription manager in use does not support disconnect.
+        """
+        self._validate_subscribe_manager_enabled()
+        if not hasattr(self._subscription_manager, 'disconnect'):
+            raise PubNubException(
+                errormsg="disconnect() is not supported by %s" % type(self._subscription_manager).__name__)
+        return self._subscription_manager.disconnect()
 
     def heartbeat(self) -> Heartbeat:
         """Send a heartbeat signal to the PubNub network.