1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128
|
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo.tests import JsonRpcException
from odoo.addons.base.tests.common import HttpCaseWithUserDemo
from odoo.addons.bus.models.bus import channel_with_db, json_dump
class TestWebsocketController(HttpCaseWithUserDemo):
def test_websocket_peek(self):
result = self.make_jsonrpc_request('/websocket/peek_notifications', {
'channels': [],
'last': 0,
'is_first_poll': True,
})
# Response containing channels/notifications is retrieved and is
# conform to excpectations.
self.assertIsNotNone(result)
channels = result.get('channels')
self.assertIsNotNone(channels)
self.assertIsInstance(channels, list)
notifications = result.get('notifications')
self.assertIsNotNone(notifications)
self.assertIsInstance(notifications, list)
result = self.make_jsonrpc_request('/websocket/peek_notifications', {
'channels': [],
'last': 0,
'is_first_poll': False,
})
# Reponse is received as long as the session is valid.
self.assertIsNotNone(result)
def test_websocket_peek_session_expired_login(self):
session = self.authenticate(None, None)
# first rpc should be fine
self.make_jsonrpc_request('/websocket/peek_notifications', {
'channels': [],
'last': 0,
'is_first_poll': True,
})
self.authenticate('admin', 'admin')
# rpc with outdated session should lead to error.
headers = {'Cookie': f'session_id={session.sid};'}
with self.assertRaises(JsonRpcException, msg='odoo.http.SessionExpiredException'):
self.make_jsonrpc_request('/websocket/peek_notifications', {
'channels': [],
'last': 0,
'is_first_poll': False,
}, headers=headers)
def test_websocket_peek_session_expired_logout(self):
session = self.authenticate('demo', 'demo')
# first rpc should be fine
self.make_jsonrpc_request('/websocket/peek_notifications', {
'channels': [],
'last': 0,
'is_first_poll': True,
})
self.url_open('/web/session/logout')
# rpc with outdated session should lead to error.
headers = {'Cookie': f'session_id={session.sid};'}
with self.assertRaises(JsonRpcException, msg='odoo.http.SessionExpiredException'):
self.make_jsonrpc_request('/websocket/peek_notifications', {
'channels': [],
'last': 0,
'is_first_poll': False,
}, headers=headers)
def test_on_websocket_closed(self):
session = self.authenticate("demo", "demo")
headers = {"Cookie": f"session_id={session.sid};"}
self.env["bus.presence"]._update_presence(
inactivity_period=0, identity_field="user_id", identity_value=self.user_demo.id
)
self.env.cr.precommit.run() # trigger the creation of bus.bus records
self.env["bus.bus"].search([]).unlink()
self.make_jsonrpc_request("/websocket/on_closed", {}, headers=headers)
self.env.cr.precommit.run() # trigger the creation of bus.bus records
message = self.make_jsonrpc_request(
"/websocket/peek_notifications",
{
"channels": [f"odoo-presence-res.partner_{self.partner_demo.id}"],
"last": 0,
"is_first_poll": True,
},
headers=headers,
)["notifications"][0]["message"]
self.assertEqual(message["type"], "bus.bus/im_status_updated")
self.assertEqual(message["payload"]["partner_id"], self.partner_demo.id)
self.assertEqual(message["payload"]["im_status"], "offline")
def test_receive_missed_presences_on_peek_notifications(self):
session = self.authenticate("demo", "demo")
headers = {"Cookie": f"session_id={session.sid};"}
self.env["bus.presence"].create({"user_id": self.user_demo.id, "status": "online"})
self.env.cr.precommit.run() # trigger the creation of bus.bus records
# First request will get notifications and trigger the creation
# of the missed presences one.
last_id = self.env["bus.bus"]._bus_last_id()
self.make_jsonrpc_request(
"/websocket/peek_notifications",
{
"channels": [f"odoo-presence-res.partner_{self.partner_demo.id}"],
"last": last_id,
"is_first_poll": True,
},
headers=headers,
)
self.env.cr.precommit.run() # trigger the creation of bus.bus records
notification = self.make_jsonrpc_request(
"/websocket/peek_notifications",
{
"channels": [f"odoo-presence-res.partner_{self.partner_demo.id}"],
"last": last_id,
"is_first_poll": True,
},
headers=headers,
)["notifications"][0]
bus_record = self.env["bus.bus"].search([("id", "=", int(notification["id"]))])
self.assertEqual(
bus_record.channel, json_dump(channel_with_db(self.env.cr.dbname, self.partner_demo))
)
self.assertEqual(notification["message"]["type"], "bus.bus/im_status_updated")
self.assertEqual(notification["message"]["payload"]["partner_id"], self.partner_demo.id)
self.assertEqual(notification["message"]["payload"]["im_status"], "online")
|