pengyuthon/odoo_source
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[FIX] bus: fix race in websocket tests

Browse Source 
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) <aku@odoo.com>
Signed-off-by: Matthieu Stockbauer (tsm) <tsm@odoo.com>
This commit is contained in:
tsm-odoo
2023-10-14 02:26:53 +00:00
parent 3ee7cb6680
commit e9446a511e
2 changed files with 105 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
addons/bus/tests/common.py
+44 -1
View File
@@ -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:
addons/bus/tests/test_websocket_caryall.py
+61 -136
View File
@@ -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)):