From a1fe7c5dc968045f3a1df8255ace36b398efa3fb Mon Sep 17 00:00:00 2001 From: Mateusz Wiktor Date: Thu, 12 Nov 2026 13:26:09 +0100 Subject: [PATCH 1/3] fix: subscription sets, listeners and message action events * PubNubSubscriptionSet.unsubscribe() now unsubscribes the channels of the set and removes the set from the registry. It sent an empty unsubscribe before. * unsubscribe_all() resets the registry to dicts. It set a list, so the next subscribe() raised TypeError and a second unsubscribe_all() AttributeError. * subscriptions and sets get channel_metadata, user_metadata, membership and file_message methods. The registry callback called methods that did not exist, so the sync message worker raised on the first App Context or file event. * subscription.add_listener() calls SubscribeCallback listeners with (pubnub, event) like the client does. It raised TypeError. * PubNubUserMetadata sets _type, it set the unused _types. * message action events expose event ("added" or "removed"). Co-Authored-By: Claude Sonnet 5.5 --- pubnub/models/consumer/pubsub.py | 2 + pubnub/models/subscription.py | 119 +++++--- pubnub/workers.py | 10 + tests/unit/test_message_worker_events.py | 117 ++++++++ tests/unit/test_subscription_registry.py | 363 +++++++++++++++++++++++ 5 files changed, 569 insertions(+), 42 deletions(-) create mode 100644 tests/unit/test_message_worker_events.py create mode 100644 tests/unit/test_subscription_registry.py 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/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/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_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] From 6cfcb852e47e19df62faf244a14bff14c421c399 Mon Sep 17 00:00:00 2001 From: Mateusz Wiktor Date: Thu, 12 Nov 2026 13:26:09 +0100 Subject: [PATCH 2/3] fix: message actions, reconnect, App Context, token parsing and packaging * get_message_actions sends end when start is set, limit accepts int or str. * pubnub.reconnect() calls the subscription manager and no longer raises AttributeError. Added pubnub.disconnect(). * asyncio get_subscribed_channels(), get_subscribed_channel_groups() and unsubscribe_all() work before the first subscribe (PNContext starts empty). * if_matches_etag() no longer leaks the If-Match header into later requests (the headers dict was shared by the class). * total_count=True in MembershipIncludes and MemberIncludes requests a count. The two include_object cassettes now record count=True. * parse_token() raises PubNubException for every malformed token. * grant() counts delete as a flag. * fetch_messages items parse custom_message_type. * setup.py: aiohttp>=3.10.11, the old range did not overlap the dev pin. Adds offline regression tests in tests/unit. Co-Authored-By: Claude Sonnet 5.5 --- pubnub/endpoints/access/grant.py | 2 +- .../message_actions/get_message_actions.py | 6 +- .../endpoints/objects_v2/objects_endpoint.py | 10 +- pubnub/event_engine/models/states.py | 6 + pubnub/managers.py | 25 +- pubnub/models/consumer/history.py | 3 + pubnub/pubnub_core.py | 20 +- setup.py | 2 +- .../channel_members_with_include_object.json | 6 +- ...annel_memberships_with_include_object.json | 6 +- tests/unit/test_get_message_actions_params.py | 60 +++++ tests/unit/test_sdk_fixes.py | 230 ++++++++++++++++++ 12 files changed, 357 insertions(+), 19 deletions(-) create mode 100644 tests/unit/test_get_message_actions_params.py create mode 100644 tests/unit/test_sdk_fixes.py 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/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/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_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 From bb77e3057d94daf48d18dc27e3452203f1a46c32 Mon Sep 17 00:00:00 2001 From: Mateusz Wiktor Date: Thu, 12 Nov 2026 14:02:41 +0100 Subject: [PATCH 3/3] fix: use a valid start and end in the message reactions example The example passed start below end. The SDK dropped end, so the call worked by accident. Now that get_message_actions sends both bounds the service answers HTTP 400 "Start must be greater than or equal to end". start is the newer bound (exclusive) and end the older bound (inclusive). Co-Authored-By: Claude Sonnet 5.5 --- examples/native_sync/message_reactions.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) 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"