From e9446a511ed2fe0bde0c2a597021bb5915f1869a Mon Sep 17 00:00:00 2001 From: tsm-odoo Date: Thu, 12 Oct 2023 10:31:48 +0200 Subject: [PATCH] [FIX] bus: fix race in websocket tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Before this commit, websocket tests were sometimes failing with the `no result to fetch` error. This is due to a race between the `subscribe` event being received which itself leads to dispatching bus notifications interferring with the rest of the test (the connection is shared). In order to solve this issue, the `WebsocketCase` now exposes a `subscribe` method that waits for this dispatching to be done in order to ensure it will not interfere with the rest of the test. Some tests were also failing because the subscription was done with 0 as the last notification id. This is an issue since this could lead to notifications being sent while they are not expected. In order to solve this issue, `last_notification_id` is now correctly passed when subscribing to channel notifications. fixes runbot-24226 closes odoo/odoo#138649 X-original-commit: 3256885fd08d33a67d7165fa33d373fd039ef7af Signed-off-by: Alexandre Kühn (aku) Signed-off-by: Matthieu Stockbauer (tsm) --- addons/bus/tests/common.py | 45 ++++- addons/bus/tests/test_websocket_caryall.py | 197 +++++++-------------- 2 files changed, 105 insertions(+), 137 deletions(-) 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)):