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
|
# Copyright 2016 ACSONE SA/NV
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
import json
from unittest import mock
from odoo import exceptions
from odoo.tests import common
from odoo.addons.bus.models.bus import json_dump
from ..models.res_users import DANGER, DEFAULT, INFO, SUCCESS, WARNING
class TestResUsers(common.TransactionCase):
def test_notify_success(self):
bus_bus = self.env["bus.bus"]
domain = [
("channel", "=", json_dump(self.env.user.notify_success_channel_name))
]
existing = bus_bus.search(domain)
test_msg = {"message": "message", "title": "title", "sticky": True}
self.env.user.notify_success(**test_msg)
news = bus_bus.search(domain) - existing
self.assertEqual(1, len(news))
test_msg.update({"type": SUCCESS})
self.assertDictEqual(test_msg, json.loads(news.message))
def test_notify_danger(self):
bus_bus = self.env["bus.bus"]
domain = [("channel", "=", json_dump(self.env.user.notify_danger_channel_name))]
existing = bus_bus.search(domain)
test_msg = {"message": "message", "title": "title", "sticky": True}
self.env.user.notify_danger(**test_msg)
news = bus_bus.search(domain) - existing
self.assertEqual(1, len(news))
test_msg.update({"type": DANGER})
self.assertDictEqual(test_msg, json.loads(news.message))
def test_notify_warning(self):
bus_bus = self.env["bus.bus"]
domain = [
("channel", "=", json_dump(self.env.user.notify_warning_channel_name))
]
existing = bus_bus.search(domain)
test_msg = {"message": "message", "title": "title", "sticky": True}
self.env.user.notify_warning(**test_msg)
news = bus_bus.search(domain) - existing
self.assertEqual(1, len(news))
test_msg.update({"type": WARNING})
self.assertDictEqual(test_msg, json.loads(news.message))
def test_notify_info(self):
bus_bus = self.env["bus.bus"]
domain = [("channel", "=", json_dump(self.env.user.notify_info_channel_name))]
existing = bus_bus.search(domain)
test_msg = {"message": "message", "title": "title", "sticky": True}
self.env.user.notify_info(**test_msg)
news = bus_bus.search(domain) - existing
self.assertEqual(1, len(news))
test_msg.update({"type": INFO})
self.assertDictEqual(test_msg, json.loads(news.message))
def test_notify_default(self):
bus_bus = self.env["bus.bus"]
domain = [
("channel", "=", json_dump(self.env.user.notify_default_channel_name))
]
existing = bus_bus.search(domain)
test_msg = {"message": "message", "title": "title", "sticky": True}
self.env.user.notify_default(**test_msg)
news = bus_bus.search(domain) - existing
self.assertEqual(1, len(news))
test_msg.update({"type": DEFAULT})
self.assertDictEqual(test_msg, json.loads(news.message))
def test_notify_many(self):
# check that the notification of a list of users is done with
# a single call to the bus
with mock.patch("odoo.addons.bus.models.bus.ImBus.sendmany") as mockedSendMany:
users = self.env.user.search([(1, "=", 1)])
self.assertTrue(len(users) > 1)
users.notify_warning(message="message")
self.assertEqual(1, mockedSendMany.call_count)
# call_args is a tuple with args (tuple) & kwargs (dict). We check
# positional arguments (args), hence the [0].
pos_call_args = mockedSendMany.call_args[0]
# Ensure the first positional argument is a list with as many
# elements as we had users.
first_pos_call_args = pos_call_args[0]
self.assertIsInstance(first_pos_call_args, list)
self.assertEqual(len(users), len(first_pos_call_args))
def test_notify_other_user(self):
other_user = self.env.ref("base.user_demo")
other_user_model = self.env["res.users"].with_user(other_user)
with self.assertRaises(exceptions.UserError):
other_user_model.browse(self.env.uid).notify_info(message="hello")
def test_notify_admin_allowed_other_user(self):
other_user = self.env.ref("base.user_demo")
other_user.notify_info(message="hello")
|
