diff --git a/examples/native_sync/message_reactions.py b/examples/native_sync/message_reactions.py index d04e820b..df67f5a0 100644 --- a/examples/native_sync/message_reactions.py +++ b/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" diff --git a/pubnub/endpoints/access/grant.py b/pubnub/endpoints/access/grant.py index 126decb6..396cda30 100644 --- a/pubnub/endpoints/access/grant.py +++ b/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() diff --git a/pubnub/endpoints/message_actions/get_message_actions.py b/pubnub/endpoints/message_actions/get_message_actions.py index 765680b6..c1b1a810 100644 --- a/pubnub/endpoints/message_actions/get_message_actions.py +++ b/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: diff --git a/pubnub/endpoints/objects_v2/objects_endpoint.py b/pubnub/endpoints/objects_v2/objects_endpoint.py index 65ca1922..f62eea50 100644 --- a/pubnub/endpoints/objects_v2/objects_endpoint.py +++ b/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 = [] diff --git a/pubnub/event_engine/models/states.py b/pubnub/event_engine/models/states.py index 6d40b3e5..59303fe8 100644 --- a/pubnub/event_engine/models/states.py +++ b/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__) diff --git a/pubnub/managers.py b/pubnub/managers.py index 08e86877..69833669 100644 --- a/pubnub/managers.py +++ b/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 diff --git a/pubnub/models/consumer/history.py b/pubnub/models/consumer/history.py index 9d421a44..96d7628b 100644 --- a/pubnub/models/consumer/history.py +++ b/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) diff --git a/pubnub/models/consumer/pubsub.py b/pubnub/models/consumer/pubsub.py index c6af8462..7d9ca6d4 100644 --- a/pubnub/models/consumer/pubsub.py +++ b/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): diff --git a/pubnub/models/subscription.py b/pubnub/models/subscription.py index fe2e47b0..8d4bbaa8 100644 --- a/pubnub/models/subscription.py +++ b/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) diff --git a/pubnub/pubnub_core.py b/pubnub/pubnub_core.py index 0538bb36..5768662d 100644 --- a/pubnub/pubnub_core.py +++ b/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. diff --git a/pubnub/workers.py b/pubnub/workers.py index f815617f..8bf0389f 100644 --- a/pubnub/workers.py +++ b/pubnub/workers.py @@ -40,6 +40,12 @@ def _get_url_for_file_event_message(self, channel, extracted_message): .file_name(extracted_message["file"]["name"])\ .file_id(extracted_message["file"]["id"]).get_complete_url() + @staticmethod + def _set_origin(result, channel, subscription): + # subscription and subscription set listeners match App Context events by these attributes + result.channel = channel + result.subscription = subscription + def _process_message(self, message_input): if self._pubnub.config.cipher_key is None: return message_input, None @@ -134,6 +140,7 @@ def _process_incoming_payload(self, message: SubscribeMessage): event=message.payload['event'], data=message.payload['data'] ) + self._set_origin(channel_result, channel, subscription_match) self.announce(channel_result) return channel_result @@ -142,6 +149,7 @@ def _process_incoming_payload(self, message: SubscribeMessage): event=message.payload['event'], data=message.payload['data'] ) + self._set_origin(uuid_result, channel, subscription_match) self.announce(uuid_result) return uuid_result @@ -150,6 +158,7 @@ def _process_incoming_payload(self, message: SubscribeMessage): event=message.payload['event'], data=message.payload['data'] ) + self._set_origin(membership_result, channel, subscription_match) self.announce(membership_result) return membership_result @@ -199,6 +208,7 @@ def _process_incoming_payload(self, message: SubscribeMessage): message_action['uuid'] = publisher message_action_result = PNMessageActionResult(message_action, subscription=subscription_match, channel=channel) + message_action_result.event = extracted_message.get('event') self._listener_manager.announce_message_action(message_action_result) else: diff --git a/setup.py b/setup.py index c3be7e7c..c7a8f58c 100644 --- a/setup.py +++ b/setup.py @@ -34,7 +34,7 @@ 'httpx>=0.28,<1.0', 'h2>=4.3', 'requests>=2.32.2', - 'aiohttp>3.10.11', + 'aiohttp>=3.10.11', 'cbor2>=5.6' ], zip_safe=False, diff --git a/tests/integrational/fixtures/native_sync/objects_v2/channel_members/channel_members_with_include_object.json b/tests/integrational/fixtures/native_sync/objects_v2/channel_members/channel_members_with_include_object.json index a7edece5..fbd9c70a 100644 --- a/tests/integrational/fixtures/native_sync/objects_v2/channel_members/channel_members_with_include_object.json +++ b/tests/integrational/fixtures/native_sync/objects_v2/channel_members/channel_members_with_include_object.json @@ -126,7 +126,7 @@ { "request": { "method": "PATCH", - "uri": "https://ps.pndsn.com/v2/objects/{PN_KEY_SUBSCRIBE}/channels/somechannel/uuids?include=custom%2Cstatus%2Ctype%2CtotalCount%2Cuuid%2Cuuid.custom%2Cuuid.type%2Cuuid.status", + "uri": "https://ps.pndsn.com/v2/objects/{PN_KEY_SUBSCRIBE}/channels/somechannel/uuids?include=custom%2Cstatus%2Ctype%2CtotalCount%2Cuuid%2Cuuid.custom%2Cuuid.type%2Cuuid.status&count=True", "body": { "pickle": "gASVeQAAAAAAAACMdXsic2V0IjogW3sidXVpZCI6IHsiaWQiOiAic29tZXV1aWQifSwgInR5cGUiOiAiUUEiLCAic3RhdHVzIjogImFjdGl2ZSIsICJjdXN0b20iOiB7ImlzQ3VzdG9tIjogdHJ1ZX19XSwgImRlbGV0ZSI6IFtdfZQu" }, @@ -187,7 +187,7 @@ { "request": { "method": "GET", - "uri": "https://ps.pndsn.com/v2/objects/{PN_KEY_SUBSCRIBE}/channels/somechannel/uuids?include=custom%2Cstatus%2Ctype%2CtotalCount%2Cuuid%2Cuuid.custom%2Cuuid.type%2Cuuid.status", + "uri": "https://ps.pndsn.com/v2/objects/{PN_KEY_SUBSCRIBE}/channels/somechannel/uuids?include=custom%2Cstatus%2Ctype%2CtotalCount%2Cuuid%2Cuuid.custom%2Cuuid.type%2Cuuid.status&count=True", "body": "", "headers": { "host": [ @@ -362,7 +362,7 @@ { "request": { "method": "GET", - "uri": "https://ps.pndsn.com/v2/objects/{PN_KEY_SUBSCRIBE}/channels/somechannel/uuids?include=custom%2Cstatus%2Ctype%2CtotalCount%2Cuuid%2Cuuid.custom%2Cuuid.type%2Cuuid.status", + "uri": "https://ps.pndsn.com/v2/objects/{PN_KEY_SUBSCRIBE}/channels/somechannel/uuids?include=custom%2Cstatus%2Ctype%2CtotalCount%2Cuuid%2Cuuid.custom%2Cuuid.type%2Cuuid.status&count=True", "body": "", "headers": { "host": [ diff --git a/tests/integrational/fixtures/native_sync/objects_v2/memberships/channel_memberships_with_include_object.json b/tests/integrational/fixtures/native_sync/objects_v2/memberships/channel_memberships_with_include_object.json index 97323eb1..66396bb1 100644 --- a/tests/integrational/fixtures/native_sync/objects_v2/memberships/channel_memberships_with_include_object.json +++ b/tests/integrational/fixtures/native_sync/objects_v2/memberships/channel_memberships_with_include_object.json @@ -126,7 +126,7 @@ { "request": { "method": "PATCH", - "uri": "https://ps.pndsn.com/v2/objects/{PN_KEY_SUBSCRIBE}/uuids/someuuid/channels?include=custom%2Cstatus%2Ctype%2CtotalCount%2Cchannel%2Cchannel.custom%2Cchannel.type%2Cchannel.status", + "uri": "https://ps.pndsn.com/v2/objects/{PN_KEY_SUBSCRIBE}/uuids/someuuid/channels?include=custom%2Cstatus%2Ctype%2CtotalCount%2Cchannel%2Cchannel.custom%2Cchannel.type%2Cchannel.status&count=True", "body": { "pickle": "gASVgwAAAAAAAACMf3sic2V0IjogW3siY2hhbm5lbCI6IHsiaWQiOiAic29tZWNoYW5uZWwifSwgImN1c3RvbSI6IHsiaXNEZWZhdWx0Q2hhbm5lbCI6IHRydWV9LCAic3RhdHVzIjogIk9GRiIsICJ0eXBlIjogIjEifV0sICJkZWxldGUiOiBbXX2ULg==" }, @@ -187,7 +187,7 @@ { "request": { "method": "GET", - "uri": "https://ps.pndsn.com/v2/objects/{PN_KEY_SUBSCRIBE}/uuids/someuuid/channels?include=custom%2Cstatus%2Ctype%2CtotalCount%2Cchannel%2Cchannel.custom%2Cchannel.type%2Cchannel.status", + "uri": "https://ps.pndsn.com/v2/objects/{PN_KEY_SUBSCRIBE}/uuids/someuuid/channels?include=custom%2Cstatus%2Ctype%2CtotalCount%2Cchannel%2Cchannel.custom%2Cchannel.type%2Cchannel.status&count=True", "body": "", "headers": { "host": [ @@ -362,7 +362,7 @@ { "request": { "method": "GET", - "uri": "https://ps.pndsn.com/v2/objects/{PN_KEY_SUBSCRIBE}/uuids/someuuid/channels?include=custom%2Cstatus%2Ctype%2CtotalCount%2Cchannel%2Cchannel.custom%2Cchannel.type%2Cchannel.status", + "uri": "https://ps.pndsn.com/v2/objects/{PN_KEY_SUBSCRIBE}/uuids/someuuid/channels?include=custom%2Cstatus%2Ctype%2CtotalCount%2Cchannel%2Cchannel.custom%2Cchannel.type%2Cchannel.status&count=True", "body": "", "headers": { "host": [ diff --git a/tests/unit/test_get_message_actions_params.py b/tests/unit/test_get_message_actions_params.py new file mode 100644 index 00000000..f6ca7b6d --- /dev/null +++ b/tests/unit/test_get_message_actions_params.py @@ -0,0 +1,60 @@ +import pytest + +from pubnub.pubnub import PubNub +from tests.helper import pnconf_copy + +pubnub = PubNub(pnconf_copy()) + + +def test_start_and_end_are_both_sent(): + params = pubnub.get_message_actions().channel('ch').start('300').end('200').custom_params() + + assert params['start'] == '300' + assert params['end'] == '200' + + +def test_start_and_end_are_both_sent_with_named_arguments(): + params = pubnub.get_message_actions(channel='ch', start='300', end='200').custom_params() + + assert params['start'] == '300' + assert params['end'] == '200' + + +def test_end_only(): + params = pubnub.get_message_actions().channel('ch').end('200').custom_params() + + assert 'start' not in params + assert params['end'] == '200' + + +def test_start_only(): + params = pubnub.get_message_actions().channel('ch').start('300').custom_params() + + assert params['start'] == '300' + assert 'end' not in params + + +def test_no_bounds_and_default_limit(): + params = pubnub.get_message_actions().channel('ch').custom_params() + + assert params == {} + + +def test_limit_accepts_str(): + params = pubnub.get_message_actions().channel('ch').limit('50').custom_params() + + assert params['limit'] == '50' + + +def test_limit_accepts_int(): + params = pubnub.get_message_actions().channel('ch').limit(50).custom_params() + + assert params['limit'] == '50' + + +def test_limit_rejects_other_types(): + with pytest.raises(AssertionError): + pubnub.get_message_actions().channel('ch').limit(5.5) + + with pytest.raises(AssertionError): + pubnub.get_message_actions().channel('ch').limit(True) diff --git a/tests/unit/test_message_worker_events.py b/tests/unit/test_message_worker_events.py new file mode 100644 index 00000000..96aeba1b --- /dev/null +++ b/tests/unit/test_message_worker_events.py @@ -0,0 +1,117 @@ +from types import SimpleNamespace + +from pubnub.models.consumer.objects_v2.channel import PNChannelMetadataResult +from pubnub.models.consumer.objects_v2.memberships import PNMembershipResult +from pubnub.models.consumer.objects_v2.uuid import PNUUIDMetadataResult +from pubnub.models.consumer.pubsub import PNMessageActionResult +from pubnub.models.server.subscribe import SubscribeMessage +from pubnub.workers import BaseMessageWorker + + +class RecordingListenerManager: + def __init__(self): + self.announced = [] + + def announce_channel(self, result): + self.announced.append(('channel', result)) + + def announce_uuid(self, result): + self.announced.append(('uuid', result)) + + def announce_membership(self, result): + self.announced.append(('membership', result)) + + def announce_message_action(self, result): + self.announced.append(('message_action', result)) + + +def make_worker(): + worker = BaseMessageWorker(SimpleNamespace(config=SimpleNamespace(cipher_key=None))) + worker._listener_manager = RecordingListenerManager() + return worker + + +def envelope(channel, payload, message_type, subscription=None): + json_input = { + 'c': channel, + 'd': payload, + 'f': 0, + 'i': 'publisher-1', + 'k': 'sub-key', + 'p': {'t': '17000000000000000', 'r': 1}, + 'e': message_type, + } + if subscription is not None: + json_input['b'] = subscription + return SubscribeMessage.from_json(json_input) + + +def action_payload(event): + return { + 'source': 'actions', + 'version': '1.0', + 'event': event, + 'data': { + 'type': 'reaction', + 'value': 'smiley_face', + 'messageTimetoken': '16999999999999999', + 'actionTimetoken': '17000000000000000', + 'uuid': 'user-1', + }, + } + + +def test_message_action_event_kind_added(): + worker = make_worker() + + worker._process_incoming_payload(envelope('ch1', action_payload('added'), BaseMessageWorker.TYPE_MESSAGE_ACTION)) + + kind, result = worker._listener_manager.announced[0] + assert kind == 'message_action' + assert isinstance(result, PNMessageActionResult) + assert result.event == 'added' + assert result.type == 'reaction' + assert result.value == 'smiley_face' + assert result.channel == 'ch1' + + +def test_message_action_event_kind_removed(): + worker = make_worker() + + worker._process_incoming_payload(envelope('ch1', action_payload('removed'), BaseMessageWorker.TYPE_MESSAGE_ACTION)) + + _, result = worker._listener_manager.announced[0] + assert result.event == 'removed' + + +def test_message_action_result_defaults_event_to_none(): + result = PNMessageActionResult(action_payload('added')['data']) + + assert result.event is None + + +def test_app_context_results_carry_channel_and_subscription(): + worker = make_worker() + cases = [ + ('channel', PNChannelMetadataResult), + ('uuid', PNUUIDMetadataResult), + ('membership', PNMembershipResult), + ] + + for object_type, result_class in cases: + payload = {'source': 'objects', 'version': '2.0', 'event': 'set', 'type': object_type, 'data': {'id': 'x'}} + worker._process_incoming_payload(envelope('ch1', payload, BaseMessageWorker.TYPE_OBJECT)) + worker._process_incoming_payload(envelope('ch2', payload, BaseMessageWorker.TYPE_OBJECT, + subscription='group1')) + + announced = worker._listener_manager.announced + assert [kind for kind, _ in announced] == ['channel', 'channel', 'uuid', 'uuid', 'membership', 'membership'] + for index, (_, result_class) in enumerate(cases): + direct = announced[index * 2][1] + via_group = announced[index * 2 + 1][1] + assert isinstance(direct, result_class) + assert direct.event == 'set' + assert direct.channel == 'ch1' + assert direct.subscription is None + assert via_group.channel == 'ch2' + assert via_group.subscription == 'group1' diff --git a/tests/unit/test_sdk_fixes.py b/tests/unit/test_sdk_fixes.py new file mode 100644 index 00000000..27562e27 --- /dev/null +++ b/tests/unit/test_sdk_fixes.py @@ -0,0 +1,230 @@ +import base64 +from types import SimpleNamespace +from unittest.mock import Mock + +import cbor2 +import pytest + +from pubnub.event_engine.models import states +from pubnub.event_engine.statemachine import StateMachine +from pubnub.exceptions import PubNubException +from pubnub.models.consumer.history import PNFetchMessageItem, PNFetchMessagesResult +from pubnub.models.consumer.objects_v2.common import MemberIncludes, MembershipIncludes +from pubnub.pubnub import PubNub +from pubnub.pubnub_asyncio import EventEngineSubscriptionManager +from tests.helper import pnconf_copy, pnconf_pam_copy + + +# PubNubCore.reconnect() and disconnect() + + +def test_reconnect_delegates_to_subscription_manager(): + pubnub = PubNub(pnconf_copy()) + pubnub._subscription_manager = Mock(spec=['reconnect', 'disconnect']) + + pubnub.reconnect() + + pubnub._subscription_manager.reconnect.assert_called_once_with() + + +def test_disconnect_delegates_to_subscription_manager(): + pubnub = PubNub(pnconf_copy()) + pubnub._subscription_manager = Mock(spec=['reconnect', 'disconnect']) + + pubnub.disconnect() + + pubnub._subscription_manager.disconnect.assert_called_once_with() + + +def test_disconnect_without_manager_support_raises_pubnub_exception(): + pubnub = PubNub(pnconf_copy()) + pubnub._subscription_manager = Mock(spec=['reconnect']) + + with pytest.raises(PubNubException): + pubnub.disconnect() + + +def test_reconnect_without_subscription_manager_raises_clear_error(): + config = pnconf_copy() + config.enable_subscribe = False + pubnub = PubNub(config) + + with pytest.raises(Exception, match="Subscription manager is not enabled"): + pubnub.reconnect() + + +# Event engine context before the first subscribe + + +def test_state_machine_initial_context_has_empty_channels_and_groups(): + machine = StateMachine(states.UnsubscribedState, name="subscribe") + + assert machine.get_context().channels == [] + assert machine.get_context().groups == [] + + +def test_contexts_do_not_share_channel_lists(): + first = states.PNContext() + second = states.PNContext() + + first.channels.append('ch1') + + assert second.channels == [] + + +def make_event_engine_manager(): + config = pnconf_copy() + config.enable_presence_heartbeat = False + pubnub = SimpleNamespace(config=config) + return EventEngineSubscriptionManager(pubnub) + + +def test_event_engine_manager_before_first_subscribe(): + manager = make_event_engine_manager() + try: + assert manager.get_subscribed_channels() == [] + assert manager.get_subscribed_channel_groups() == [] + + manager.unsubscribe_all() + + assert manager.event_engine.get_state_name() == 'UnsubscribedState' + assert manager.get_subscribed_channels() == [] + finally: + manager.loop.close() + + +def test_disconnect_with_event_engine_manager_raises_pubnub_exception(): + manager = make_event_engine_manager() + try: + pubnub = PubNub(pnconf_copy()) + pubnub._subscription_manager = manager + + with pytest.raises(PubNubException): + pubnub.disconnect() + finally: + manager.loop.close() + + +# Objects If-Match header and total count + + +def test_if_match_header_does_not_leak_to_later_requests(): + pubnub = PubNub(pnconf_copy()) + + with_etag = pubnub.set_channel_metadata().channel('ch1').if_matches_etag('etag-1') + later = pubnub.set_channel_metadata().channel('ch1') + other_endpoint = pubnub.get_all_channel_metadata() + + assert with_etag._custom_headers['If-Match'] == 'etag-1' + assert with_etag.request_headers()['If-Match'] == 'etag-1' + assert 'If-Match' not in later.request_headers() + assert 'If-Match' not in other_endpoint.request_headers() + + +def test_if_match_header_is_per_instance(): + pubnub = PubNub(pnconf_copy()) + + first = pubnub.set_uuid_metadata().if_matches_etag('etag-1') + second = pubnub.set_uuid_metadata().if_matches_etag('etag-2') + + assert first._custom_headers['If-Match'] == 'etag-1' + assert second._custom_headers['If-Match'] == 'etag-2' + + +def test_membership_includes_total_count_requests_count(): + pubnub = PubNub(pnconf_copy()) + + params = pubnub.get_memberships(uuid='user-1', include=MembershipIncludes(total_count=True)).custom_params() + + assert params['count'] is True + + +def test_member_includes_total_count_requests_count(): + pubnub = PubNub(pnconf_copy()) + + params = pubnub.get_channel_members(channel='ch1', include=MemberIncludes(total_count=True)).custom_params() + + assert params['count'] is True + + +def test_includes_without_total_count_do_not_request_count(): + pubnub = PubNub(pnconf_copy()) + + params = pubnub.get_memberships(uuid='user-1', include=MembershipIncludes(custom=True)).custom_params() + + assert 'count' not in params + + +def test_deprecated_include_total_count_still_requests_count(): + pubnub = PubNub(pnconf_copy()) + + params = pubnub.get_memberships(uuid='user-1', include_total_count=True).custom_params() + + assert params['count'] is True + + +# parse_token with malformed input + + +def encode_token(payload): + return base64.b64encode(cbor2.dumps(payload)).decode('utf-8') + + +@pytest.mark.parametrize('token', [ + None, + '', + 123, + 'abc', + 'not a token!!', + 'qEF2AkF0GmFLd-NDdHRsGQWgQ3Jlc6VEY2hhbqFjY2gx', + base64.b64encode(b'not cbor at all').decode('utf-8'), + encode_token(42), + encode_token(['v', 't']), + encode_token({'v': 2}), + encode_token({'v': 2, 't': 1, 'ttl': 1, 'meta': {}, 'res': ['chan'], 'pat': {}}), +]) +def test_parse_token_raises_pubnub_exception_for_invalid_tokens(token): + pubnub = PubNub(pnconf_copy()) + + with pytest.raises(PubNubException): + pubnub.parse_token(token) + + +# Grant flags + + +def test_grant_accepts_delete_as_the_only_flag(): + pubnub = PubNub(pnconf_pam_copy()) + + pubnub.grant().channels('ch1').delete(True).validate_params() + + +def test_grant_without_flags_still_fails(): + pubnub = PubNub(pnconf_pam_copy()) + + with pytest.raises(PubNubException, match="At least one flag should be specified"): + pubnub.grant().channels('ch1').validate_params() + + +# fetch_messages custom message type + + +def test_fetch_messages_parses_custom_message_type(): + envelope = { + 'channels': { + 'ch1': [ + {'message': 'with type', 'timetoken': '17000000000000000', 'custom_message_type': 'text-message'}, + {'message': 'without type', 'timetoken': '17000000000000001'}, + ] + } + } + + result = PNFetchMessagesResult.from_json(envelope) + + with_type, without_type = result.channels['ch1'] + assert with_type.custom_message_type == 'text-message' + assert without_type.custom_message_type is None + + +def test_fetch_message_item_defaults_custom_message_type_to_none(): + assert PNFetchMessageItem('message', '17000000000000000').custom_message_type is None diff --git a/tests/unit/test_subscription_registry.py b/tests/unit/test_subscription_registry.py new file mode 100644 index 00000000..8eec7bbe --- /dev/null +++ b/tests/unit/test_subscription_registry.py @@ -0,0 +1,363 @@ +from pubnub.callbacks import SubscribeCallback +from pubnub.models.consumer.objects_v2.channel import PNChannelMetadataResult +from pubnub.models.consumer.objects_v2.memberships import PNMembershipResult +from pubnub.models.consumer.objects_v2.uuid import PNUUIDMetadataResult +from pubnub.models.consumer.pubsub import PNFileMessageResult, PNMessageResult +from pubnub.models.subscription import ( + PNSubscriptionRegistry, PNSubscriptionType, PubNubChannel, PubNubChannelGroup, PubNubSubscriptionSet, + PubNubUserMetadata +) + + +class FakeSubscriptionManager: + def __init__(self): + self._timetoken = 0 + self.subscribe_operations = [] + self.unsubscribe_operations = [] + + def adapt_subscribe_builder(self, subscribe_operation): + self.subscribe_operations.append(subscribe_operation) + + def adapt_unsubscribe_builder(self, unsubscribe_operation): + self.unsubscribe_operations.append(unsubscribe_operation) + + +class FakePubNub: + def __init__(self): + self._subscription_manager = FakeSubscriptionManager() + self._subscription_registry = PNSubscriptionRegistry(self) + self.listeners = [] + + def add_listener(self, listener): + self.listeners.append(listener) + + def channel(self, name): + return PubNubChannel(self, name) + + def subscription_set(self, subscriptions): + return PubNubSubscriptionSet(self, subscriptions) + + +class RecordingCallback(SubscribeCallback): + def __init__(self): + self.calls = [] + + def status(self, pubnub, status): + self.calls.append(('status', pubnub, status)) + + def message(self, pubnub, message): + self.calls.append(('message', pubnub, message)) + + def presence(self, pubnub, presence): + self.calls.append(('presence', pubnub, presence)) + + def channel(self, pubnub, channel): + self.calls.append(('channel', pubnub, channel)) + + def uuid(self, pubnub, uuid): + self.calls.append(('uuid', pubnub, uuid)) + + def membership(self, pubnub, membership): + self.calls.append(('membership', pubnub, membership)) + + def file(self, pubnub, file_message): + self.calls.append(('file', pubnub, file_message)) + + +def message_on(channel, text='hello'): + return PNMessageResult(message=text, subscription=None, channel=channel, timetoken=17000000000000000) + + +def app_context_event(result_class, channel): + result = result_class(event='set', data={'id': channel}) + result.channel = channel + result.subscription = None + return result + + +def file_event(channel): + return PNFileMessageResult(message='file', subscription=None, channel=channel, timetoken=17000000000000000, + publisher='user', file_url='https://example.com/f', file_id='id', file_name='f.txt') + + +def registry_callback(pubnub): + callback = pubnub._subscription_registry.subscription_registry_callback + assert callback in pubnub.listeners + return callback + + +# Subscription set unsubscribe + + +def test_set_unsubscribe_removes_only_its_channels(): + pubnub = FakePubNub() + registry = pubnub._subscription_registry + pubnub.channel('ch3').subscription().subscribe() + subscription_set = pubnub.subscription_set([ + pubnub.channel('ch1').subscription(), + pubnub.channel('ch2').subscription() + ]) + subscription_set.subscribe() + + subscription_set.unsubscribe() + + operation = pubnub._subscription_manager.unsubscribe_operations[-1] + assert sorted(operation.channels) == ['ch1', 'ch2'] + assert operation.channel_groups == [] + assert registry.get_subscribed_channels() == ['ch3'] + assert subscription_set not in registry.get_all_listeners() + + +def test_set_unsubscribe_keeps_channel_needed_by_another_subscription(): + pubnub = FakePubNub() + registry = pubnub._subscription_registry + standalone = pubnub.channel('ch1').subscription() + standalone.subscribe() + subscription_set = pubnub.subscription_set([ + pubnub.channel('ch1').subscription(), + pubnub.channel('ch2').subscription() + ]) + subscription_set.subscribe() + + subscription_set.unsubscribe() + + operation = pubnub._subscription_manager.unsubscribe_operations[-1] + assert operation.channels == ['ch2'] + assert registry.get_subscribed_channels() == ['ch1'] + assert standalone in registry.get_all_listeners() + assert subscription_set not in registry.get_all_listeners() + + +def test_set_unsubscribe_with_channel_groups(): + pubnub = FakePubNub() + registry = pubnub._subscription_registry + subscription_set = pubnub.subscription_set([ + pubnub.channel('ch1').subscription(), + PubNubChannelGroup(pubnub, 'group1').subscription() + ]) + subscription_set.subscribe() + + subscription_set.unsubscribe() + + operation = pubnub._subscription_manager.unsubscribe_operations[-1] + assert operation.channels == ['ch1'] + assert operation.channel_groups == ['group1'] + assert registry.get_subscribed_channels() == [] + assert registry.get_subscribed_channel_groups() == [] + + +def test_unsubscribe_with_presence_removes_presence_channel(): + pubnub = FakePubNub() + registry = pubnub._subscription_registry + subscription = pubnub.channel('ch1').subscription(with_presence=True) + subscription.subscribe() + + subscription.unsubscribe() + + operation = pubnub._subscription_manager.unsubscribe_operations[-1] + assert sorted(operation.channels) == ['ch1', 'ch1-pnpres'] + assert registry.get_subscribed_channels() == [] + + +def test_unsubscribed_channel_is_not_subscribed_again(): + pubnub = FakePubNub() + first = pubnub.channel('ch1').subscription() + first.subscribe() + first.unsubscribe() + + pubnub.channel('ch2').subscription().subscribe() + + assert pubnub._subscription_manager.subscribe_operations[-1].channels == ['ch2'] + + +def test_unsubscribe_without_changes_sends_nothing(): + pubnub = FakePubNub() + pubnub.channel('ch1').subscription().subscribe() + duplicate = pubnub.channel('ch1').subscription() + duplicate.subscribe() + + duplicate.unsubscribe() + + assert pubnub._subscription_manager.unsubscribe_operations == [] + assert pubnub._subscription_registry.get_subscribed_channels() == ['ch1'] + + +# unsubscribe_all + + +def test_unsubscribe_all_keeps_registry_usable(): + pubnub = FakePubNub() + registry = pubnub._subscription_registry + pubnub.channel('ch1').subscription().subscribe() + PubNubChannelGroup(pubnub, 'group1').subscription().subscribe() + + registry.unsubscribe_all() + + operation = pubnub._subscription_manager.unsubscribe_operations[-1] + assert operation.channels == ['ch1'] + assert operation.channel_groups == ['group1'] + assert registry.channels == {} + assert registry.channel_groups == {} + assert registry.get_subscribed_channels() == [] + + pubnub.channel('ch2').subscription().subscribe() + assert pubnub._subscription_manager.subscribe_operations[-1].channels == ['ch2'] + assert registry.get_subscribed_channels() == ['ch2'] + + +def test_unsubscribe_all_twice_is_safe(): + pubnub = FakePubNub() + registry = pubnub._subscription_registry + pubnub.channel('ch1').subscription().subscribe() + + registry.unsubscribe_all() + registry.unsubscribe_all() + + assert len(pubnub._subscription_manager.unsubscribe_operations) == 1 + + +def test_unsubscribe_after_unsubscribe_all_does_not_raise(): + pubnub = FakePubNub() + subscription = pubnub.channel('ch1').subscription() + subscription.subscribe() + pubnub._subscription_registry.unsubscribe_all() + + subscription.unsubscribe() + + assert len(pubnub._subscription_manager.unsubscribe_operations) == 1 + + +# User metadata subscription type + + +def test_user_metadata_subscription_is_a_channel(): + pubnub = FakePubNub() + subscription = PubNubUserMetadata(pubnub, 'user-1').subscription() + + assert subscription._type == PNSubscriptionType.CHANNEL + + subscription.subscribe() + assert pubnub._subscription_registry.get_subscribed_channels() == ['user-1'] + assert pubnub._subscription_registry.get_subscribed_channel_groups() == [] + + +# Listener dispatch + + +def test_app_context_and_file_events_reach_subscription_listeners(): + pubnub = FakePubNub() + received = [] + subscription = pubnub.channel('ch1').subscription() + subscription.on_channel_metadata = lambda event: received.append(('channel_metadata', event)) + subscription.on_user_metadata = lambda event: received.append(('user_metadata', event)) + subscription.on_membership = lambda event: received.append(('membership', event)) + subscription.on_file = lambda event: received.append(('file', event)) + subscription.subscribe() + callback = registry_callback(pubnub) + + channel_event = app_context_event(PNChannelMetadataResult, 'ch1') + uuid_event = app_context_event(PNUUIDMetadataResult, 'ch1') + membership_event = app_context_event(PNMembershipResult, 'ch1') + file_message = file_event('ch1') + callback.channel(pubnub, channel_event) + callback.uuid(pubnub, uuid_event) + callback.membership(pubnub, membership_event) + callback.file(pubnub, file_message) + + assert received == [ + ('channel_metadata', channel_event), + ('user_metadata', uuid_event), + ('membership', membership_event), + ('file', file_message), + ] + + +def test_app_context_events_for_other_channels_are_not_delivered(): + pubnub = FakePubNub() + received = [] + subscription = pubnub.channel('ch1').subscription() + subscription.on_channel_metadata = received.append + subscription.on_file = received.append + subscription.subscribe() + callback = registry_callback(pubnub) + + callback.channel(pubnub, app_context_event(PNChannelMetadataResult, 'other')) + callback.file(pubnub, file_event('other')) + + assert received == [] + + +def test_app_context_event_without_channel_does_not_raise(): + pubnub = FakePubNub() + received = [] + subscription = pubnub.channel('ch1').subscription() + subscription.on_channel_metadata = received.append + subscription.subscribe() + callback = registry_callback(pubnub) + + callback.channel(pubnub, PNChannelMetadataResult(event='set', data={'id': 'ch1'})) + + assert received == [] + + +def test_events_reach_subscription_set_listeners(): + pubnub = FakePubNub() + received = [] + subscription_set = pubnub.subscription_set([ + pubnub.channel('ch1').subscription(), + pubnub.channel('ch2').subscription() + ]) + subscription_set.on_membership = received.append + subscription_set.on_file = received.append + subscription_set.subscribe() + callback = registry_callback(pubnub) + + membership_event = app_context_event(PNMembershipResult, 'ch2') + file_message = file_event('ch1') + callback.membership(pubnub, membership_event) + callback.file(pubnub, file_message) + + assert received == [membership_event, file_message] + + +def test_subscribe_callback_added_to_subscription_gets_pubnub_argument(): + pubnub = FakePubNub() + listener = RecordingCallback() + subscription = pubnub.channel('ch1').subscription() + subscription.add_listener(listener) + subscription.subscribe() + callback = registry_callback(pubnub) + + message = message_on('ch1') + channel_event = app_context_event(PNChannelMetadataResult, 'ch1') + uuid_event = app_context_event(PNUUIDMetadataResult, 'ch1') + membership_event = app_context_event(PNMembershipResult, 'ch1') + file_message = file_event('ch1') + callback.message(pubnub, message) + callback.channel(pubnub, channel_event) + callback.uuid(pubnub, uuid_event) + callback.membership(pubnub, membership_event) + callback.file(pubnub, file_message) + + assert listener.calls == [ + ('message', pubnub, message), + ('channel', pubnub, channel_event), + ('uuid', pubnub, uuid_event), + ('membership', pubnub, membership_event), + ('file', pubnub, file_message), + ] + + +def test_subscription_on_message_still_gets_event_only(): + pubnub = FakePubNub() + received = [] + subscription = pubnub.channel('ch1').subscription() + subscription.on_message = received.append + subscription.subscribe() + callback = registry_callback(pubnub) + + message = message_on('ch1') + callback.message(pubnub, message) + callback.message(pubnub, message_on('other')) + + assert received == [message]