ringcentral.websocket.web_socket_client_test

  1import asyncio
  2import contextlib
  3import json
  4import unittest
  5import uuid
  6from unittest import mock
  7
  8from .events import WebSocketEvents
  9from .web_socket_client import WebSocketClient
 10
 11
 12class FakeResponse:
 13    def __init__(self, payload):
 14        self._payload = payload
 15
 16    def json_dict(self):
 17        return self._payload
 18
 19
 20class FakePlatform:
 21    def post(self, path, body=None):
 22        return FakeResponse({
 23            "uri": "wss://fake-websocket",
 24            "ws_access_token": "fake-ws-token",
 25        })
 26
 27
 28class FakeWebSocket:
 29    def __init__(self):
 30        self.sent = []
 31        self.received = []
 32        self.closed = False
 33        self._fail_next_recv = False
 34        self._receive_error = None
 35        self._frames = asyncio.Queue()
 36
 37    def fail_next_receive(self, error):
 38        self._fail_next_recv = True
 39        self._receive_error = error
 40
 41    def push(self, frame):
 42        self._frames.put_nowait(frame)
 43
 44    def has_pending_frames(self):
 45        return self._frames.qsize() > 0
 46
 47    async def recv(self):
 48        if self._fail_next_recv:
 49            self._fail_next_recv = False
 50            raise self._receive_error
 51        frame = await self._frames.get()
 52        if frame is None:
 53            raise ConnectionResetError("connection closed")
 54        self.received.append(frame)
 55        return frame
 56
 57    async def send(self, message):
 58        self.sent.append(message)
 59
 60    async def close(self):
 61        self.closed = True
 62        self._frames.put_nowait(None)
 63
 64
 65class GatedSendWebSocket(FakeWebSocket):
 66    def __init__(self):
 67        super().__init__()
 68        self.send_started = asyncio.Event()
 69        self.send_gate = asyncio.Event()
 70
 71    async def send(self, message):
 72        self.send_started.set()
 73        await self.send_gate.wait()
 74        await FakeWebSocket.send(self, message)
 75
 76
 77class FlakySendWebSocket(FakeWebSocket):
 78    def __init__(self, error):
 79        super().__init__()
 80        self._send_error = error
 81
 82    async def send(self, message):
 83        if self._send_error is not None:
 84            error = self._send_error
 85            self._send_error = None
 86            raise error
 87        await FakeWebSocket.send(self, message)
 88
 89
 90class RecordingHandler:
 91    def __init__(self, error=None):
 92        self.calls = []
 93        self.error = error
 94
 95    def __call__(self, *args):
 96        self.calls.append(args)
 97        if self.error is not None:
 98            raise self.error
 99
100
101EVENT_FILTERS = ["/restapi/v1.0/account/~/extension/~/presence"]
102OTHER_FILTERS = ["/restapi/v1.0/account/~/extension/~/message-store"]
103
104
105def subscription_creation_frame(message_id, status=200):
106    return json.dumps([
107        {
108            "type": "ClientRequest",
109            "messageId": message_id,
110            "status": status,
111        },
112        {
113            "uri": "/restapi/v1.0/subscription/fake-subscription",
114            "id": "fake-subscription",
115            "status": "Active",
116            "eventFilters": EVENT_FILTERS,
117            "deliveryMode": {"transportType": "WebSocket"},
118        },
119    ])
120
121
122def subscription_removal_frame(message_id):
123    return json.dumps([
124        {
125            "type": "ClientRequest",
126            "messageId": message_id,
127            "status": 200,
128        },
129    ])
130
131
132def server_notification_frame():
133    return json.dumps([
134        {
135            "type": "ServerNotification",
136            "messageId": str(uuid.uuid4()),
137        },
138        {
139            "uri": "/restapi/v1.0/subscription/fake-subscription",
140            "event": {"/restapi/v1.0/account/~/extension/~/presence": {"activeCalls": []}},
141        },
142    ])
143
144
145class WebSocketClientTest(unittest.IsolatedAsyncioTestCase):
146    def setUp(self):
147        self.client = WebSocketClient(FakePlatform())
148
149    async def start_receive_loop(self):
150        fake_socket = FakeWebSocket()
151
152        async def connect(uri):
153            return fake_socket
154
155        patcher = mock.patch("websockets.connect", connect)
156        patcher.start()
157        self.addCleanup(patcher.stop)
158        connection_task = asyncio.create_task(self.client.create_new_connection())
159        return fake_socket, connection_task
160
161    async def start_receive_loop_with(self, fake_socket):
162        async def connect(uri):
163            return fake_socket
164
165        patcher = mock.patch("websockets.connect", connect)
166        patcher.start()
167        self.addCleanup(patcher.stop)
168        connection_task = asyncio.create_task(self.client.create_new_connection())
169        fake_socket.push("[/heartbeat]")
170        await self.wait_until(lambda: self.client._is_ready)
171        return connection_task
172
173    async def deliver_creation_response(self, fake_socket, status=200):
174        request = json.loads(fake_socket.sent[0])
175        fake_socket.push(subscription_creation_frame(request[0]["messageId"], status=status))
176
177    async def wait_until(self, predicate, limit=100):
178        for _ in range(limit):
179            if predicate():
180                return
181            await asyncio.sleep(0)
182        self.fail("condition was not met before the spin limit")
183
184    async def wait_until_frames_delivered(self, fake_socket):
185        await self.wait_until(lambda: not fake_socket.has_pending_frames())
186
187    async def stop_receive_loop(self, connection_task):
188        connection_task.cancel()
189        with contextlib.suppress(asyncio.CancelledError):
190            await connection_task
191
192    async def test_failing_handler_does_not_stop_delivery_to_remaining_handlers(self):
193        failing = RecordingHandler(error=RuntimeError("handler failure"))
194        healthy = RecordingHandler()
195        receive_errors = RecordingHandler()
196        connection_errors = RecordingHandler()
197        self.client.on(WebSocketEvents.receiveMessage, failing)
198        self.client.on(WebSocketEvents.receiveMessage, healthy)
199        self.client.on(WebSocketEvents.receiveMessageError, receive_errors)
200        self.client.on(WebSocketEvents.createConnectionError, connection_errors)
201
202        fake_socket, connection_task = await self.start_receive_loop()
203        fake_socket.push("[/heartbeat]")
204        fake_socket.push("raw-message")
205        await self.wait_until_frames_delivered(fake_socket)
206        await self.stop_receive_loop(connection_task)
207
208        self.assertEqual(failing.calls, [("raw-message",)])
209        self.assertEqual(healthy.calls, [("raw-message",)])
210        self.assertEqual(len(receive_errors.calls), 1)
211        self.assertEqual(len(receive_errors.calls[0]), 1)
212        self.assertIs(receive_errors.calls[0][0], failing.error)
213        self.assertEqual(connection_errors.calls, [])
214
215    async def test_failing_handler_without_error_observers_does_not_stop_reception(self):
216        failing = RecordingHandler(error=RuntimeError("handler failure"))
217        healthy = RecordingHandler()
218        self.client.on(WebSocketEvents.receiveMessage, failing)
219        self.client.on(WebSocketEvents.receiveMessage, healthy)
220
221        fake_socket, connection_task = await self.start_receive_loop()
222        fake_socket.push("[/heartbeat]")
223        fake_socket.push("raw-message")
224        await self.wait_until_frames_delivered(fake_socket)
225        await self.stop_receive_loop(connection_task)
226
227        self.assertEqual(healthy.calls, [("raw-message",)])
228
229    async def test_future_messages_continue_and_failed_handler_remains_registered(self):
230        failing = RecordingHandler(error=RuntimeError("handler failure"))
231        healthy = RecordingHandler()
232        receive_errors = RecordingHandler()
233        self.client.on(WebSocketEvents.receiveMessage, failing)
234        self.client.on(WebSocketEvents.receiveMessage, healthy)
235        self.client.on(WebSocketEvents.receiveMessageError, receive_errors)
236
237        fake_socket, connection_task = await self.start_receive_loop()
238        fake_socket.push("[/heartbeat]")
239        fake_socket.push("first-message")
240        fake_socket.push("second-message")
241        await self.wait_until_frames_delivered(fake_socket)
242        await self.stop_receive_loop(connection_task)
243
244        self.assertEqual(failing.calls, [("first-message",), ("second-message",)])
245        self.assertEqual(healthy.calls, [("first-message",), ("second-message",)])
246        self.assertEqual(
247            [call[0] for call in receive_errors.calls],
248            [failing.error, failing.error],
249        )
250
251    async def test_each_failing_handler_produces_its_own_error_event(self):
252        first = RecordingHandler(error=RuntimeError("first failure"))
253        second = RecordingHandler(error=ValueError("second failure"))
254        healthy = RecordingHandler()
255        receive_errors = RecordingHandler()
256        self.client.on(WebSocketEvents.receiveMessage, first)
257        self.client.on(WebSocketEvents.receiveMessage, healthy)
258        self.client.on(WebSocketEvents.receiveMessage, second)
259        self.client.on(WebSocketEvents.receiveMessageError, receive_errors)
260
261        fake_socket, connection_task = await self.start_receive_loop()
262        fake_socket.push("[/heartbeat]")
263        fake_socket.push("raw-message")
264        await self.wait_until_frames_delivered(fake_socket)
265        await self.stop_receive_loop(connection_task)
266
267        self.assertEqual(first.calls, [("raw-message",)])
268        self.assertEqual(second.calls, [("raw-message",)])
269        self.assertEqual(healthy.calls, [("raw-message",)])
270        self.assertEqual(
271            [call[0] for call in receive_errors.calls],
272            [first.error, second.error],
273        )
274
275    async def test_subscription_parser_failures_are_isolated_and_reception_continues(self):
276        raw_messages = RecordingHandler()
277        receive_errors = RecordingHandler()
278        notifications = RecordingHandler()
279        subscription_created = RecordingHandler()
280        connection_errors = RecordingHandler()
281        connection_created = RecordingHandler()
282        self.client.on(WebSocketEvents.receiveMessage, raw_messages)
283        self.client.on(WebSocketEvents.receiveMessageError, receive_errors)
284        self.client.on(WebSocketEvents.receiveSubscriptionNotification, notifications)
285        self.client.on(WebSocketEvents.subscriptionCreated, subscription_created)
286        self.client.on(WebSocketEvents.createConnectionError, connection_errors)
287        self.client.on(WebSocketEvents.connectionCreated, connection_created)
288
289        fake_socket, connection_task = await self.start_receive_loop()
290        fake_socket.push("[/heartbeat]")
291        await self.wait_until(lambda: len(connection_created.calls) == 1)
292
293        await self.client.create_subscription(events=EVENT_FILTERS)
294
295        request = json.loads(fake_socket.sent[0])
296        creation_frame = subscription_creation_frame(request[0]["messageId"])
297        notification_frame = server_notification_frame()
298        fake_socket.push(creation_frame)
299        fake_socket.push("not-json")
300        fake_socket.push("{}")
301        fake_socket.push("[]")
302        fake_socket.push(notification_frame)
303        await self.wait_until_frames_delivered(fake_socket)
304        await self.stop_receive_loop(connection_task)
305
306        self.assertEqual(len(subscription_created.calls), 1)
307        subscription = subscription_created.calls[0][0]
308        self.assertEqual(subscription.get_subscription_info()[1]["id"], "fake-subscription")
309
310        self.assertEqual(raw_messages.calls, [
311            (creation_frame,),
312            ("not-json",),
313            ("{}",),
314            ("[]",),
315            (notification_frame,),
316        ])
317        self.assertEqual(len(receive_errors.calls), 3)
318        self.assertIsInstance(receive_errors.calls[0][0], json.JSONDecodeError)
319        self.assertIsInstance(receive_errors.calls[1][0], KeyError)
320        self.assertIsInstance(receive_errors.calls[2][0], IndexError)
321        self.assertEqual(len(notifications.calls), 1)
322        self.assertEqual(notifications.calls[0][0], json.loads(notification_frame))
323        self.assertEqual(connection_errors.calls, [])
324
325    async def test_failing_error_observer_does_not_block_other_error_observers_or_reception(self):
326        failing_receive = RecordingHandler(error=RuntimeError("handler failure"))
327        healthy_receive = RecordingHandler()
328        failing_error_observer = RecordingHandler(error=RuntimeError("observer failure"))
329        recording_error_observer = RecordingHandler()
330        self.client.on(WebSocketEvents.receiveMessage, failing_receive)
331        self.client.on(WebSocketEvents.receiveMessage, healthy_receive)
332        self.client.on(WebSocketEvents.receiveMessageError, failing_error_observer)
333        self.client.on(WebSocketEvents.receiveMessageError, recording_error_observer)
334
335        fake_socket, connection_task = await self.start_receive_loop()
336        fake_socket.push("[/heartbeat]")
337        fake_socket.push("first-message")
338        fake_socket.push("second-message")
339        await self.wait_until_frames_delivered(fake_socket)
340        await self.stop_receive_loop(connection_task)
341
342        self.assertEqual(recording_error_observer.calls, [
343            (failing_receive.error,),
344            (failing_receive.error,),
345        ])
346        self.assertEqual(failing_receive.calls, [("first-message",), ("second-message",)])
347        self.assertEqual(healthy_receive.calls, [("first-message",), ("second-message",)])
348
349    async def test_unrelated_event_dispatch_semantics_remain_unchanged(self):
350        failing = RecordingHandler(error=RuntimeError("unrelated failure"))
351        healthy = RecordingHandler()
352        self.client.on(WebSocketEvents.subscriptionCreated, failing)
353        self.client.on(WebSocketEvents.subscriptionCreated, healthy)
354
355        with self.assertRaises(RuntimeError):
356            self.client.trigger(WebSocketEvents.subscriptionCreated, self.client)
357
358        self.assertEqual(failing.calls, [(self.client,)])
359        self.assertEqual(healthy.calls, [])
360
361    async def test_create_subscription_requires_events_argument(self):
362        fake_socket, connection_task = await self.start_receive_loop()
363        fake_socket.push("[/heartbeat]")
364        await self.wait_until(lambda: self.client._is_ready)
365
366        with self.assertRaises(TypeError):
367            await self.client.create_subscription()
368
369        await self.stop_receive_loop(connection_task)
370        self.assertEqual(fake_socket.sent, [])
371
372    async def test_create_subscription_rejects_none_and_empty_events(self):
373        creation_errors = RecordingHandler()
374        created = RecordingHandler()
375        notifications = RecordingHandler()
376        self.client.on(WebSocketEvents.createSubscriptionError, creation_errors)
377        self.client.on(WebSocketEvents.subscriptionCreated, created)
378        self.client.on(WebSocketEvents.receiveSubscriptionNotification, notifications)
379
380        fake_socket, connection_task = await self.start_receive_loop()
381        fake_socket.push("[/heartbeat]")
382        await self.wait_until(lambda: self.client._is_ready)
383
384        for invalid_events in (None, []):
385            with self.assertRaises(Exception) as ctx:
386                await self.client.create_subscription(events=invalid_events)
387            self.assertEqual(str(ctx.exception), "Events are undefined")
388
389        fake_socket.push(server_notification_frame())
390        await self.wait_until_frames_delivered(fake_socket)
391
392        self.assertEqual(fake_socket.sent, [])
393        self.assertEqual(created.calls, [])
394        self.assertEqual(len(creation_errors.calls), 2)
395        for call in creation_errors.calls:
396            self.assertEqual(str(call[0]), "Events are undefined")
397        self.assertEqual(notifications.calls, [])
398        await self.stop_receive_loop(connection_task)
399
400    async def test_second_creation_while_first_is_pending_is_rejected(self):
401        creation_errors = RecordingHandler()
402        created = RecordingHandler()
403        notifications = RecordingHandler()
404        self.client.on(WebSocketEvents.createSubscriptionError, creation_errors)
405        self.client.on(WebSocketEvents.subscriptionCreated, created)
406        self.client.on(WebSocketEvents.receiveSubscriptionNotification, notifications)
407
408        fake_socket = GatedSendWebSocket()
409        connection_task = await self.start_receive_loop_with(fake_socket)
410
411        first_task = asyncio.create_task(self.client.create_subscription(events=EVENT_FILTERS))
412        await self.wait_until(lambda: fake_socket.send_started.is_set())
413
414        with self.assertRaises(Exception) as ctx:
415            await self.client.create_subscription(events=OTHER_FILTERS)
416        self.assertEqual(
417            str(ctx.exception),
418            "WebSocket subscription creation is already in progress; "
419            "wait for subscriptionCreated or createSubscriptionError before retrying",
420        )
421        self.assertEqual(len(creation_errors.calls), 1)
422        self.assertEqual(str(creation_errors.calls[0][0]), str(ctx.exception))
423
424        fake_socket.send_gate.set()
425        await first_task
426
427        self.assertEqual(len(fake_socket.sent), 1)
428        request = json.loads(fake_socket.sent[0])
429        self.assertEqual(request[0]["method"], "POST")
430        self.assertEqual(request[1]["eventFilters"], EVENT_FILTERS)
431
432        fake_socket.push(subscription_creation_frame(request[0]["messageId"]))
433        await self.wait_until(lambda: len(created.calls) == 1)
434
435        fake_socket.push(server_notification_frame())
436        await self.wait_until_frames_delivered(fake_socket)
437        self.assertEqual(len(notifications.calls), 1)
438
439        await self.stop_receive_loop(connection_task)
440        self.assertEqual(len(created.calls), 1)
441        self.assertEqual(len(creation_errors.calls), 1)
442
443    async def test_second_creation_after_confirmed_creation_is_rejected(self):
444        creation_errors = RecordingHandler()
445        created = RecordingHandler()
446        updated = RecordingHandler()
447        notifications = RecordingHandler()
448        self.client.on(WebSocketEvents.createSubscriptionError, creation_errors)
449        self.client.on(WebSocketEvents.subscriptionCreated, created)
450        self.client.on(WebSocketEvents.subscriptionUpdated, updated)
451        self.client.on(WebSocketEvents.receiveSubscriptionNotification, notifications)
452
453        fake_socket, connection_task = await self.start_receive_loop()
454        fake_socket.push("[/heartbeat]")
455        await self.wait_until(lambda: self.client._is_ready)
456
457        await self.client.create_subscription(events=EVENT_FILTERS)
458        await self.deliver_creation_response(fake_socket)
459        await self.wait_until(lambda: len(created.calls) == 1)
460        subscription = created.calls[0][0]
461
462        with self.assertRaises(Exception) as ctx:
463            await self.client.create_subscription(events=OTHER_FILTERS)
464        self.assertEqual(
465            str(ctx.exception),
466            "A WebSocket subscription already exists; use update_subscription() "
467            "to change its events or remove_subscription() before creating another",
468        )
469        self.assertEqual(len(creation_errors.calls), 1)
470        self.assertEqual(str(creation_errors.calls[0][0]), str(ctx.exception))
471
472        self.assertEqual(len(fake_socket.sent), 1)
473        self.assertEqual(json.loads(fake_socket.sent[0])[0]["method"], "POST")
474        self.assertEqual(len(created.calls), 1)
475        self.assertEqual(updated.calls, [])
476        self.assertEqual(subscription.get_subscription_info()[1]["id"], "fake-subscription")
477
478        fake_socket.push(server_notification_frame())
479        await self.wait_until_frames_delivered(fake_socket)
480        self.assertEqual(len(notifications.calls), 1)
481
482        await self.stop_receive_loop(connection_task)
483
484    async def test_retry_after_send_failure_uses_new_events_without_duplicate_listeners(self):
485        creation_errors = RecordingHandler()
486        send_errors = RecordingHandler()
487        created = RecordingHandler()
488        notifications = RecordingHandler()
489        self.client.on(WebSocketEvents.createSubscriptionError, creation_errors)
490        self.client.on(WebSocketEvents.sendMessageError, send_errors)
491        self.client.on(WebSocketEvents.subscriptionCreated, created)
492        self.client.on(WebSocketEvents.receiveSubscriptionNotification, notifications)
493
494        fake_socket = FlakySendWebSocket(RuntimeError("send failure"))
495        connection_task = await self.start_receive_loop_with(fake_socket)
496
497        with self.assertRaises(Exception):
498            await self.client.create_subscription(events=EVENT_FILTERS)
499        self.assertEqual(len(send_errors.calls), 1)
500        self.assertEqual(len(creation_errors.calls), 1)
501        self.assertEqual(fake_socket.sent, [])
502
503        await self.client.create_subscription(events=OTHER_FILTERS)
504
505        self.assertEqual(len(fake_socket.sent), 1)
506        request = json.loads(fake_socket.sent[0])
507        self.assertEqual(request[1]["eventFilters"], OTHER_FILTERS)
508        fake_socket.push(subscription_creation_frame(request[0]["messageId"]))
509        await self.wait_until(lambda: len(created.calls) == 1)
510
511        fake_socket.push(server_notification_frame())
512        await self.wait_until_frames_delivered(fake_socket)
513        self.assertEqual(len(notifications.calls), 1)
514
515        await self.stop_receive_loop(connection_task)
516        self.assertEqual(len(created.calls), 1)
517
518    async def test_retry_after_rejected_creation_succeeds_without_duplicate_listeners(self):
519        creation_errors = RecordingHandler()
520        created = RecordingHandler()
521        notifications = RecordingHandler()
522        self.client.on(WebSocketEvents.createSubscriptionError, creation_errors)
523        self.client.on(WebSocketEvents.subscriptionCreated, created)
524        self.client.on(WebSocketEvents.receiveSubscriptionNotification, notifications)
525
526        fake_socket, connection_task = await self.start_receive_loop()
527        fake_socket.push("[/heartbeat]")
528        await self.wait_until(lambda: self.client._is_ready)
529
530        await self.client.create_subscription(events=EVENT_FILTERS)
531        await self.deliver_creation_response(fake_socket, status=403)
532        await self.wait_until(lambda: len(creation_errors.calls) == 1)
533        self.assertEqual(
534            str(creation_errors.calls[0][0]),
535            "WebSocket subscription creation failed with status 403",
536        )
537        self.assertEqual(created.calls, [])
538
539        await self.client.create_subscription(events=OTHER_FILTERS)
540
541        self.assertEqual(len(fake_socket.sent), 2)
542        retry_request = json.loads(fake_socket.sent[1])
543        self.assertEqual(retry_request[0]["method"], "POST")
544        self.assertEqual(retry_request[1]["eventFilters"], OTHER_FILTERS)
545        fake_socket.push(subscription_creation_frame(retry_request[0]["messageId"]))
546        await self.wait_until(lambda: len(created.calls) == 1)
547
548        fake_socket.push(server_notification_frame())
549        await self.wait_until_frames_delivered(fake_socket)
550        self.assertEqual(len(notifications.calls), 1)
551
552        await self.stop_receive_loop(connection_task)
553        self.assertEqual(len(created.calls), 1)
554        self.assertEqual(len(creation_errors.calls), 1)
555
556    async def test_creation_after_removal_reuses_client_subscription_without_duplicate_listeners(self):
557        created = RecordingHandler()
558        removed = RecordingHandler()
559        notifications = RecordingHandler()
560        self.client.on(WebSocketEvents.subscriptionCreated, created)
561        self.client.on(WebSocketEvents.subscriptionRemoved, removed)
562        self.client.on(WebSocketEvents.receiveSubscriptionNotification, notifications)
563
564        fake_socket, connection_task = await self.start_receive_loop()
565        fake_socket.push("[/heartbeat]")
566        await self.wait_until(lambda: self.client._is_ready)
567
568        await self.client.create_subscription(events=EVENT_FILTERS)
569        await self.deliver_creation_response(fake_socket)
570        await self.wait_until(lambda: len(created.calls) == 1)
571        subscription = created.calls[0][0]
572
573        await self.client.remove_subscription(subscription)
574        removal_request = json.loads(fake_socket.sent[1])
575        self.assertEqual(removal_request[0]["method"], "DELETE")
576        fake_socket.push(subscription_removal_frame(removal_request[0]["messageId"]))
577        await self.wait_until(lambda: len(removed.calls) == 1)
578
579        fake_socket.push(server_notification_frame())
580        await self.wait_until_frames_delivered(fake_socket)
581        self.assertEqual(notifications.calls, [])
582
583        await self.client.create_subscription(events=OTHER_FILTERS)
584
585        self.assertEqual(len(fake_socket.sent), 3)
586        recreated_request = json.loads(fake_socket.sent[2])
587        self.assertEqual(recreated_request[0]["method"], "POST")
588        self.assertEqual(recreated_request[1]["eventFilters"], OTHER_FILTERS)
589        fake_socket.push(subscription_creation_frame(recreated_request[0]["messageId"]))
590        await self.wait_until(lambda: len(created.calls) == 2)
591
592        fake_socket.push(server_notification_frame())
593        await self.wait_until_frames_delivered(fake_socket)
594        self.assertEqual(len(notifications.calls), 1)
595
596        await self.stop_receive_loop(connection_task)
597        self.assertEqual(len(created.calls), 2)
598        self.assertEqual(len(removed.calls), 1)
599
600    async def test_successful_creation_returns_none_and_delivers_subscription_via_event(self):
601        created = RecordingHandler()
602        self.client.on(WebSocketEvents.subscriptionCreated, created)
603
604        fake_socket, connection_task = await self.start_receive_loop()
605        fake_socket.push("[/heartbeat]")
606        await self.wait_until(lambda: self.client._is_ready)
607
608        result = await self.client.create_subscription(events=EVENT_FILTERS)
609        self.assertIsNone(result)
610
611        await self.deliver_creation_response(fake_socket)
612        await self.wait_until(lambda: len(created.calls) == 1)
613
614        self.assertEqual(len(created.calls), 1)
615        subscription = created.calls[0][0]
616        self.assertEqual(subscription.get_subscription_info()[1]["id"], "fake-subscription")
617
618        await self.stop_receive_loop(connection_task)
619
620    async def test_recv_failure_after_handshake_reports_receive_error_and_cleans_up(self):
621        receive_error = RuntimeError("receive failure")
622        receive_errors = RecordingHandler()
623        connection_errors = RecordingHandler()
624        self.client.on(WebSocketEvents.receiveMessageError, receive_errors)
625        self.client.on(WebSocketEvents.createConnectionError, connection_errors)
626
627        sockets = []
628
629        def on_connected(*_):
630            sockets[0].fail_next_receive(receive_error)
631
632        self.client.on(WebSocketEvents.connectionCreated, on_connected)
633
634        fake_socket, connection_task = await self.start_receive_loop()
635        fake_socket.push("[/heartbeat]")
636        sockets.append(fake_socket)
637
638        result = await connection_task
639
640        self.assertIsNone(result)
641        self.assertEqual(len(receive_errors.calls), 1)
642        self.assertEqual(len(receive_errors.calls[0]), 1)
643        self.assertIs(receive_errors.calls[0][0], receive_error)
644        self.assertEqual(connection_errors.calls, [])
645        self.assertFalse(self.client._is_ready)
646        await self.wait_until(lambda: self.client._heartbeat_task.done())
647        self.assertTrue(self.client._heartbeat_task.cancelled())
648
649    async def test_receive_loop_cancellation_marks_client_not_ready_and_cancels_heartbeat(self):
650        receive_errors = RecordingHandler()
651        connection_errors = RecordingHandler()
652        self.client.on(WebSocketEvents.receiveMessageError, receive_errors)
653        self.client.on(WebSocketEvents.createConnectionError, connection_errors)
654
655        fake_socket, connection_task = await self.start_receive_loop()
656        fake_socket.push("[/heartbeat]")
657        await self.wait_until(lambda: self.client._is_ready)
658        self.assertTrue(self.client._is_ready)
659
660        await self.stop_receive_loop(connection_task)
661
662        self.assertFalse(self.client._is_ready)
663        await self.wait_until(lambda: self.client._heartbeat_task.done())
664        self.assertTrue(self.client._heartbeat_task.cancelled())
665        self.assertEqual(receive_errors.calls, [])
666        self.assertEqual(connection_errors.calls, [])
667
668    async def test_intentional_close_performs_cleanup_without_receive_error(self):
669        receive_errors = RecordingHandler()
670        connection_errors = RecordingHandler()
671        close_errors = RecordingHandler()
672        self.client.on(WebSocketEvents.receiveMessageError, receive_errors)
673        self.client.on(WebSocketEvents.createConnectionError, connection_errors)
674        self.client.on(WebSocketEvents.closeConnectionError, close_errors)
675
676        fake_socket, connection_task = await self.start_receive_loop()
677        fake_socket.push("[/heartbeat]")
678        await self.wait_until(lambda: self.client._is_ready)
679
680        await self.client.close_connection()
681        result = await connection_task
682
683        self.assertIsNone(result)
684        self.assertTrue(fake_socket.closed)
685        self.assertFalse(self.client._is_ready)
686        await self.wait_until(lambda: self.client._heartbeat_task.done())
687        self.assertTrue(self.client._heartbeat_task.cancelled())
688        self.assertEqual(receive_errors.calls, [])
689        self.assertEqual(connection_errors.calls, [])
690        self.assertEqual(close_errors.calls, [])
691
692    async def test_receive_failure_on_reconnected_client_is_reported_after_intentional_close(self):
693        receive_error = RuntimeError("receive failure after reconnect")
694        receive_errors = RecordingHandler()
695        connection_errors = RecordingHandler()
696        close_errors = RecordingHandler()
697        self.client.on(WebSocketEvents.receiveMessageError, receive_errors)
698        self.client.on(WebSocketEvents.createConnectionError, connection_errors)
699        self.client.on(WebSocketEvents.closeConnectionError, close_errors)
700
701        sockets = []
702
703        def fail_latest_socket(*_):
704            sockets[-1].fail_next_receive(receive_error)
705
706        self.client.on(WebSocketEvents.connectionCreated, fail_latest_socket)
707
708        first_socket, connection_task = await self.start_receive_loop()
709        sockets.append(first_socket)
710        first_socket.push("[/heartbeat]")
711        await self.wait_until(lambda: self.client._is_ready)
712
713        await self.client.close_connection()
714        await connection_task
715        self.assertTrue(first_socket.closed)
716
717        second_socket = FakeWebSocket()
718        sockets.append(second_socket)
719
720        async def connect_again(uri):
721            return second_socket
722
723        with mock.patch("websockets.connect", connect_again):
724            recovery_task = asyncio.create_task(self.client.recover_connection())
725            second_socket.push("[/heartbeat]")
726            await self.wait_until(lambda: self.client._is_ready)
727            result = await recovery_task
728
729        self.assertIsNone(result)
730        self.assertEqual(len(receive_errors.calls), 1)
731        self.assertEqual(len(receive_errors.calls[0]), 1)
732        self.assertIs(receive_errors.calls[0][0], receive_error)
733        self.assertEqual(connection_errors.calls, [])
734        self.assertEqual(close_errors.calls, [])
735        self.assertFalse(self.client._is_ready)
736        await self.wait_until(lambda: self.client._heartbeat_task.done())
737        self.assertTrue(self.client._heartbeat_task.cancelled())
738
739    async def test_connection_establishment_failure_retains_create_connection_error(self):
740        connect_error = RuntimeError("connect failure")
741        receive_errors = RecordingHandler()
742        connection_errors = RecordingHandler()
743        self.client.on(WebSocketEvents.receiveMessageError, receive_errors)
744        self.client.on(WebSocketEvents.createConnectionError, connection_errors)
745
746        async def connect(uri):
747            raise connect_error
748
749        with mock.patch("websockets.connect", connect):
750            with self.assertRaises(RuntimeError):
751                await self.client.create_new_connection()
752
753        self.assertTrue(connection_errors.calls)
754        for call in connection_errors.calls:
755            self.assertIs(call[0], connect_error)
756        self.assertEqual(receive_errors.calls, [])
757
758    async def test_initial_handshake_failure_retains_create_connection_error(self):
759        handshake_error = RuntimeError("handshake failure")
760        receive_errors = RecordingHandler()
761        connection_errors = RecordingHandler()
762        self.client.on(WebSocketEvents.receiveMessageError, receive_errors)
763        self.client.on(WebSocketEvents.createConnectionError, connection_errors)
764
765        fake_socket = FakeWebSocket()
766        fake_socket.fail_next_receive(handshake_error)
767
768        async def connect(uri):
769            return fake_socket
770
771        with mock.patch("websockets.connect", connect):
772            with self.assertRaises(RuntimeError):
773                await self.client.create_new_connection()
774
775        self.assertTrue(connection_errors.calls)
776        for call in connection_errors.calls:
777            self.assertIs(call[0], handshake_error)
778        self.assertEqual(receive_errors.calls, [])
779        self.assertFalse(self.client._is_ready)
780        self.assertIsNone(self.client._heartbeat_task)
781
782
783if __name__ == "__main__":
784    unittest.main()