Odoo18-Base/addons/mail/tests/discuss/test_bus_presence.py

99 lines
5.1 KiB
Python
Raw Permalink Normal View History

2025-01-06 10:57:38 +07:00
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import json
try:
import websocket as ws
except ImportError:
ws = None
from odoo.tests import tagged, new_test_user
from odoo.addons.bus.tests.common import WebsocketCase
from odoo.addons.mail.tests.common import MailCommon
from odoo.addons.bus.models.bus import channel_with_db, json_dump
@tagged("post_install", "-at_install")
class TestBusPresence(WebsocketCase, MailCommon):
def _receive_presence(self, sender, recipient):
self._reset_bus()
sent_from_user = isinstance(sender, self.env.registry["res.users"])
receive_to_user = isinstance(recipient, self.env.registry["res.users"])
if receive_to_user:
session = self.authenticate(recipient.login, recipient.login)
auth_cookie = f"session_id={session.sid};"
else:
self.authenticate(None, None)
auth_cookie = f"{recipient._cookie_name}={recipient._format_auth_cookie()};"
websocket = self.websocket_connect(cookie=auth_cookie, timeout=1)
sender_bus_target = sender.partner_id if sent_from_user else sender
self.subscribe(
websocket,
[f"odoo-presence-{sender_bus_target._name}_{sender_bus_target.id}"],
self.env["bus.bus"]._bus_last_id(),
)
self.env["bus.presence"].create(
{"user_id" if sent_from_user else "guest_id": sender.id, "status": "online"}
)
self.trigger_notification_dispatching([(sender_bus_target, "presence")])
notifications = json.loads(websocket.recv())
self._close_websockets()
bus_record = self.env["bus.bus"].search([("id", "=", int(notifications[0]["id"]))])
self.assertEqual(
bus_record.channel,
json_dump(channel_with_db(self.env.cr.dbname, (sender_bus_target, "presence"))),
)
self.assertEqual(notifications[0]["message"]["type"], "bus.bus/im_status_updated")
self.assertEqual(notifications[0]["message"]["payload"]["im_status"], "online")
self.assertEqual(
notifications[0]["message"]["payload"]["partner_id" if sent_from_user else "guest_id"],
sender_bus_target.id,
)
def test_receive_presences_as_guest(self):
guest = self.env["mail.guest"].create({"name": "Guest"})
bob = new_test_user(self.env, login="bob_user", groups="base.group_user")
# Guest should not receive users's presence: no common channel.
with self.assertRaises(ws._exceptions.WebSocketTimeoutException):
self._receive_presence(sender=bob, recipient=guest)
channel = self.env["discuss.channel"].channel_create(group_id=None, name="General")
channel.add_members(guest_ids=[guest.id], partner_ids=[bob.partner_id.id])
# Now that they share a channel, guest should receive users's presence.
self._receive_presence(sender=bob, recipient=guest)
other_guest = self.env["mail.guest"].create({"name": "OtherGuest"})
# Guest should not receive guest's presence: no common channel.
with self.assertRaises(ws._exceptions.WebSocketTimeoutException):
self._receive_presence(sender=other_guest, recipient=guest)
channel.add_members(guest_ids=[other_guest.id])
# Now that they share a channel, guest should receive guest's presence.
self._receive_presence(sender=other_guest, recipient=guest)
def test_receive_presences_as_portal(self):
portal = new_test_user(self.env, login="portal_user", groups="base.group_portal")
bob = new_test_user(self.env, login="bob_user", groups="base.group_user")
# Portal should not receive users's presence: no common channel.
with self.assertRaises(ws._exceptions.WebSocketTimeoutException):
self._receive_presence(sender=bob, recipient=portal)
channel = self.env["discuss.channel"].channel_create(group_id=None, name="General")
channel.add_members(partner_ids=[portal.partner_id.id, bob.partner_id.id])
# Now that they share a channel, portal should receive users's presence.
self._receive_presence(sender=bob, recipient=portal)
guest = self.env["mail.guest"].create({"name": "Guest"})
# Portal should not receive guest's presence: no common channel.
with self.assertRaises(ws._exceptions.WebSocketTimeoutException):
self._receive_presence(sender=guest, recipient=portal)
channel.add_members(guest_ids=[guest.id])
# Now that they share a channel, portal should receive guest's presence.
self._receive_presence(sender=guest, recipient=portal)
def test_receive_presences_as_internal(self):
internal = new_test_user(self.env, login="internal_user", groups="base.group_user")
guest = self.env["mail.guest"].create({"name": "Guest"})
# Internal can access guest's presence regardless of their channels.
self._receive_presence(sender=guest, recipient=internal)
# Internal can access users's presence regardless of their channels.
bob = new_test_user(self.env, login="bob_user", groups="base.group_user")
self._receive_presence(sender=bob, recipient=internal)