diff --git a/addons/bus/tests/common.py b/addons/bus/tests/common.py index c11959e4b05..e332c2ec348 100644 --- a/addons/bus/tests/common.py +++ b/addons/bus/tests/common.py @@ -1,5 +1,6 @@ # Part of Odoo. See LICENSE file for full copyright and licensing details. +import json import struct from threading import Event import unittest @@ -12,7 +13,8 @@ except ImportError: import odoo.tools from odoo.tests import HOST, common -from ..websocket import CloseCode, WebsocketConnectionHandler +from ..websocket import CloseCode, Websocket, WebsocketConnectionHandler +from ..models.bus import dispatch, hashable, channel_with_db class WebsocketCase(common.HttpCase): @@ -76,6 +78,47 @@ class WebsocketCase(common.HttpCase): self._websockets.add(ws) return ws + def subscribe(self, websocket, channels=None, last=None, wait_for_dispatch=True): + """ Subscribe the websocket to the given channels. + + :param websocket: The websocket of the client. + :param channels: The list of channels to subscribe to. + :param last: The last notification id the client received. + :param wait_for_dispatch: Whether to wait for the notification + dispatching trigerred by the subscription. + """ + dispatch_bus_notification_done = Event() + original_dispatch_bus_notifications = Websocket._dispatch_bus_notifications + + def _mocked_dispatch_bus_notifications(self, *args): + original_dispatch_bus_notifications(self, *args) + dispatch_bus_notification_done.set() + + with patch.object(Websocket, '_dispatch_bus_notifications', _mocked_dispatch_bus_notifications): + sub = {'event_name': 'subscribe', 'data': { + 'channels': channels or [], + }} + if last: + sub['data']['last'] = last + websocket.send(json.dumps(sub)) + if wait_for_dispatch: + dispatch_bus_notification_done.wait(timeout=5) + + def trigger_notification_dispatching(self, channels): + """ Notify the websockets subscribed to the given channels that new + notifications are available. Usefull since the bus is not able to do + it during tests. + """ + channels = [ + hashable(channel_with_db(self.registry.db_name, c)) + if isinstance(c, str) else c for c in channels + ] + websockets = set() + for channel in channels: + websockets.update(dispatch._channels_to_ws.get(hashable(channel), [])) + for websocket in websockets: + websocket.trigger_notification_dispatching() + def wait_remaining_websocket_connections(self): """ Wait for the websocket connections to terminate. """ for event in self._websocket_events: diff --git a/addons/bus/tests/test_websocket_caryall.py b/addons/bus/tests/test_websocket_caryall.py index d329c9f38ff..3e26e661c9d 100644 --- a/addons/bus/tests/test_websocket_caryall.py +++ b/addons/bus/tests/test_websocket_caryall.py @@ -8,6 +8,7 @@ from datetime import timedelta from freezegun import freeze_time from threading import Event from unittest.mock import patch +from weakref import WeakSet try: from websocket._exceptions import WebSocketBadStatusException except ImportError: @@ -50,18 +51,18 @@ class TestWebsocketCaryall(WebsocketCase): self.assertEqual(events, ['open', 'close']) def test_instances_weak_set(self): - gc.collect() - first_ws = self.websocket_connect() - second_ws = self.websocket_connect() - self.assertEqual(len(Websocket._instances), 2) - first_ws.close(CloseCode.CLEAN) - second_ws.close(CloseCode.CLEAN) - self.wait_remaining_websocket_connections() - # serve_forever_patch prevent websocket instances from being - # collected. Stop it now. - self._serve_forever_patch.stop() - gc.collect() - self.assertEqual(len(Websocket._instances), 0) + with patch.object(Websocket, "_instances", WeakSet()): + first_ws = self.websocket_connect() + second_ws = self.websocket_connect() + self.assertEqual(len(Websocket._instances), 2) + first_ws.close(CloseCode.CLEAN) + second_ws.close(CloseCode.CLEAN) + self.wait_remaining_websocket_connections() + # serve_forever_patch prevent websocket instances from being + # collected. Stop it now. + self._serve_forever_patch.stop() + gc.collect() + self.assertEqual(len(Websocket._instances), 0) def test_timeout_manager_no_response_timeout(self): with freeze_time('2022-08-19') as frozen_time: @@ -117,7 +118,7 @@ class TestWebsocketCaryall(WebsocketCase): # The session with whom the websocket connected has been # deleted. WebSocket should disconnect in order for the # session to be updated. - websocket.send(json.dumps({'event_name': 'subscribe'})) + self.subscribe(websocket, wait_for_dispatch=False) self.assert_close_with_code(websocket, CloseCode.SESSION_EXPIRED) def test_user_logout_incoming_message(self): @@ -128,117 +129,59 @@ class TestWebsocketCaryall(WebsocketCase): # The session with whom the websocket connected has been # deleted. WebSocket should disconnect in order for the # session to be updated. - websocket.send(json.dumps({'event_name': 'subscribe'})) + self.subscribe(websocket, wait_for_dispatch=False) self.assert_close_with_code(websocket, CloseCode.SESSION_EXPIRED) def test_user_logout_outgoing_message(self): - subscribe_done_event = Event() - original_subscribe = Websocket.subscribe - odoo_ws = None - - def patched_subscribe(self, *args): - nonlocal odoo_ws - odoo_ws = self - original_subscribe(self, *args) - subscribe_done_event.set() - new_test_user(self.env, login='test_user', password='Password!1') user_session = self.authenticate('test_user', 'Password!1') websocket = self.websocket_connect(cookie=f'session_id={user_session.sid};') - with patch.object(Websocket, 'subscribe', patched_subscribe): - websocket.send(json.dumps({ - 'event_name': 'subscribe', - 'data': {'channels': ['channel1'], 'last': 0} - })) - subscribe_done_event.wait(timeout=5) - self.url_open('/web/session/logout') - # Simulate postgres notify. The session with whom the websocket - # connected has been deleted. WebSocket should be closed without - # receiving the message. - self.env['bus.bus']._sendone('channel1', 'notif type', 'message') - odoo_ws.trigger_notification_dispatching() - self.assert_close_with_code(websocket, CloseCode.SESSION_EXPIRED) + self.subscribe(websocket, ['channel1'], self.env['bus.bus']._bus_last_id()) + self.url_open('/web/session/logout') + # Simulate postgres notify. The session with whom the websocket + # connected has been deleted. WebSocket should be closed without + # receiving the message. + self.env['bus.bus']._sendone('channel1', 'notif type', 'message') + self.trigger_notification_dispatching(["channel1"]) + self.assert_close_with_code(websocket, CloseCode.SESSION_EXPIRED) def test_channel_subscription_disconnect(self): - subscribe_done_event = Event() - original_subscribe = dispatch.subscribe - - def patched_subscribe(*args): - original_subscribe(*args) - subscribe_done_event.set() - - with patch.object(dispatch, 'subscribe', patched_subscribe): - websocket = self.websocket_connect() - websocket.send(json.dumps({ - 'event_name': 'subscribe', - 'data': {'channels': ['my_channel'], 'last': 0} - })) - subscribe_done_event.wait(timeout=5) - # channel is added as expected to the channel to websocket map. - self.assertIn((self.env.registry.db_name, 'my_channel'), dispatch._channels_to_ws) - websocket.close(CloseCode.CLEAN) - self.wait_remaining_websocket_connections() - # channel is removed as expected when removing the last - # websocket that was listening to this channel. - self.assertNotIn((self.env.registry.db_name, 'my_channel'), dispatch._channels_to_ws) + websocket = self.websocket_connect() + self.subscribe(websocket, ['my_channel'], self.env['bus.bus']._bus_last_id()) + # channel is added as expected to the channel to websocket map. + self.assertIn((self.env.registry.db_name, 'my_channel'), dispatch._channels_to_ws) + websocket.close(CloseCode.CLEAN) + self.wait_remaining_websocket_connections() + # channel is removed as expected when removing the last + # websocket that was listening to this channel. + self.assertNotIn((self.env.registry.db_name, 'my_channel'), dispatch._channels_to_ws) def test_channel_subscription_update(self): - subscribe_done_event = Event() - original_subscribe = dispatch.subscribe - - def patched_subscribe(*args): - original_subscribe(*args) - subscribe_done_event.set() - - with patch.object(dispatch, 'subscribe', patched_subscribe): - websocket = self.websocket_connect() - websocket.send(json.dumps({ - 'event_name': 'subscribe', - 'data': {'channels': ['my_channel'], 'last': 0} - })) - subscribe_done_event.wait(timeout=5) - subscribe_done_event.clear() - # channel is added as expected to the channel to websocket map. - self.assertIn((self.env.registry.db_name, 'my_channel'), dispatch._channels_to_ws) - websocket.send(json.dumps({ - 'event_name': 'subscribe', - 'data': {'channels': ['my_channel_2'], 'last': 0} - })) - subscribe_done_event.wait(timeout=5) - # channel is removed as expected when updating the subscription. - self.assertNotIn((self.env.registry.db_name, 'my_channel'), dispatch._channels_to_ws) + websocket = self.websocket_connect() + self.subscribe(websocket, ['my_channel'], self.env['bus.bus']._bus_last_id()) + # channel is added as expected to the channel to websocket map. + self.assertIn((self.env.registry.db_name, 'my_channel'), dispatch._channels_to_ws) + self.subscribe(websocket, ['my_channel_2'], self.env['bus.bus']._bus_last_id()) + # channel is removed as expected when updating the subscription. + self.assertNotIn((self.env.registry.db_name, 'my_channel'), dispatch._channels_to_ws) def test_trigger_notification(self): - original_subscribe = Websocket.subscribe - odoo_ws = None - - def patched_subscribe(self, *args): - nonlocal odoo_ws - odoo_ws = self - original_subscribe(self, *args) - - with patch.object(Websocket, 'subscribe', patched_subscribe): - websocket = self.websocket_connect() - self.env['bus.bus']._sendone('my_channel', 'notif_type', 'message') - websocket.send(json.dumps({ - 'event_name': 'subscribe', - 'data': {'channels': ['my_channel'], 'last': 0} - })) - - notifications = json.loads(websocket.recv()) - self.assertEqual(1, len(notifications)) - self.assertEqual(notifications[0]['message']['type'], 'notif_type') - self.assertEqual(notifications[0]['message']['payload'], 'message') - - self.env['bus.bus']._sendone('my_channel', 'notif_type', 'another_message') - odoo_ws.trigger_notification_dispatching() - - notifications = json.loads(websocket.recv()) - # First notification has been received, we should only receive - # the second one. - self.assertEqual(1, len(notifications)) - self.assertEqual(notifications[0]['message']['type'], 'notif_type') - self.assertEqual(notifications[0]['message']['payload'], 'another_message') + websocket = self.websocket_connect() + self.subscribe(websocket, ['my_channel'], self.env['bus.bus']._bus_last_id()) + self.env['bus.bus']._sendone('my_channel', 'notif_type', 'message') + self.trigger_notification_dispatching(["my_channel"]) + notifications = json.loads(websocket.recv()) + self.assertEqual(1, len(notifications)) + self.assertEqual(notifications[0]['message']['type'], 'notif_type') + self.assertEqual(notifications[0]['message']['payload'], 'message') + self.env['bus.bus']._sendone('my_channel', 'notif_type', 'another_message') + self.trigger_notification_dispatching(["my_channel"]) + notifications = json.loads(websocket.recv()) + # First notification has been received, we should only receive + # the second one. + self.assertEqual(1, len(notifications)) + self.assertEqual(notifications[0]['message']['type'], 'notif_type') + self.assertEqual(notifications[0]['message']['payload'], 'another_message') def test_opening_websocket_connection_during_tests(self): # During tests, browsers can't open websocket connections. @@ -252,40 +195,22 @@ class TestWebsocketCaryall(WebsocketCase): self.websocket_connect() def test_subscribe_higher_last_notification_id(self): - subscribe_done_event = Event() server_last_notification_id = self.env['bus.bus'].sudo().search([], limit=1, order='id desc').id or 0 client_last_notification_id = server_last_notification_id + 1 - def subscribe_side_effect(_, last): - # Last notification id given by the client is higher than - # the one known by the server, should default to 0. - self.assertEqual(last, 0) - subscribe_done_event.set() - - with patch.object(Websocket, 'subscribe', side_effect=subscribe_side_effect): + with patch.object(Websocket, 'subscribe', side_effect=Websocket.subscribe, autospec=True) as mock: websocket = self.websocket_connect() - websocket.send(json.dumps({ - 'event_name': 'subscribe', - 'data': {'channels': ['my_channel'], 'last': client_last_notification_id} - })) - subscribe_done_event.wait() + self.subscribe(websocket, ['my_channel'], client_last_notification_id) + self.assertEqual(mock.call_args[0][2], 0) def test_subscribe_lower_last_notification_id(self): - subscribe_done_event = Event() server_last_notification_id = self.env['bus.bus'].sudo().search([], limit=1, order='id desc').id or 0 client_last_notification_id = server_last_notification_id - 1 - def subscribe_side_effect(_, last): - self.assertEqual(last, client_last_notification_id) - subscribe_done_event.set() - - with patch.object(Websocket, 'subscribe', side_effect=subscribe_side_effect): + with patch.object(Websocket, 'subscribe', side_effect=Websocket.subscribe, autospec=True) as mock: websocket = self.websocket_connect() - websocket.send(json.dumps({ - 'event_name': 'subscribe', - 'data': {'channels': ['my_channel'], 'last': client_last_notification_id} - })) - subscribe_done_event.wait() + self.subscribe(websocket, ['my_channel'], client_last_notification_id) + self.assertEqual(mock.call_args[0][2], client_last_notification_id) def test_no_cursor_when_no_callback_for_lifecycle_event(self): with patch.object(Websocket, '_event_callbacks', defaultdict(set)):