File: test_websocket_controller.py

package info (click to toggle)
odoo 18.0.0%2Bdfsg-2
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 878,716 kB
  • sloc: javascript: 927,937; python: 685,670; xml: 388,524; sh: 1,033; sql: 415; makefile: 26
file content (128 lines) | stat: -rw-r--r-- 5,615 bytes parent folder | download
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")