|
4246
|
1 #!/usr/bin/env python3 |
|
|
2 |
|
|
3 # Libervia: an XMPP client |
|
|
4 # Copyright (C) 2009-2024 Jérôme Poisson (goffi@goffi.org) |
|
|
5 |
|
|
6 # This program is free software: you can redistribute it and/or modify |
|
|
7 # it under the terms of the GNU Affero General Public License as published by |
|
|
8 # the Free Software Foundation, either version 3 of the License, or |
|
|
9 # (at your option) any later version. |
|
|
10 |
|
|
11 # This program is distributed in the hope that it will be useful, |
|
|
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of |
|
|
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
|
|
14 # GNU Affero General Public License for more details. |
|
|
15 |
|
|
16 # You should have received a copy of the GNU Affero General Public License |
|
|
17 # along with this program. If not, see <http://www.gnu.org/licenses/>. |
|
|
18 |
|
|
19 from unittest.mock import MagicMock, patch |
|
|
20 from pytest_twisted import ensureDeferred as ed |
|
|
21 |
|
|
22 from pytest import fixture |
|
|
23 from twisted.internet import defer |
|
|
24 from twisted.words.protocols.jabber import jid |
|
|
25 from twisted.words.xish import domish |
|
|
26 |
|
|
27 from libervia.backend.plugins.plugin_xep_0272 import ( |
|
|
28 NS_MUJI, |
|
|
29 PRESENCE_MUJI, |
|
|
30 XEP_0272, |
|
|
31 XEP_0272_handler, |
|
|
32 ) |
|
|
33 from libervia.backend.tools.common import data_format |
|
|
34 |
|
|
35 TEST_ROOM_JID = jid.JID("room@example.com/user") |
|
|
36 |
|
|
37 |
|
|
38 |
|
|
39 @fixture |
|
|
40 def xep_0272(host) -> XEP_0272: |
|
|
41 """Fixture for initializing XEP-0272 plugin.""" |
|
|
42 host.plugins = { |
|
|
43 "XEP-0045": MagicMock(), |
|
|
44 "XEP-0166": MagicMock(), |
|
|
45 "XEP-0167": MagicMock(), |
|
|
46 "XEP-0249": MagicMock(), |
|
|
47 } |
|
|
48 return XEP_0272(host) |
|
|
49 |
|
|
50 |
|
|
51 class TestXEP0272: |
|
|
52 def test_get_handler(self, host, client): |
|
|
53 """XEP_0272_handler is instantiated and returned.""" |
|
|
54 xep_0272 = XEP_0272(host) |
|
|
55 handler = xep_0272.get_handler(client) |
|
|
56 assert isinstance(handler, XEP_0272_handler) |
|
|
57 assert handler.plugin_parent == xep_0272 |
|
|
58 |
|
|
59 def test_on_muji_request_preparing_state(self, host, client): |
|
|
60 """try_to_finish_preparation is called when preparing_state is True.""" |
|
|
61 xep_0272 = XEP_0272(host) |
|
|
62 presence_elt = domish.Element((None, "presence")) |
|
|
63 presence_elt["from"] = TEST_ROOM_JID.full() |
|
|
64 muji_elt = presence_elt.addElement((NS_MUJI, "muji")) |
|
|
65 muji_elt.addElement("preparing") |
|
|
66 |
|
|
67 muji_data = { |
|
|
68 "preparing_jids": set() |
|
|
69 } |
|
|
70 with patch.object( |
|
|
71 xep_0272._muc, "get_room_user_jid", return_value=TEST_ROOM_JID) : |
|
|
72 with patch.object(xep_0272, "get_muji_data", return_value=muji_data): |
|
|
73 with patch.object( |
|
|
74 xep_0272, "try_to_finish_preparation" |
|
|
75 ) as mock_try_to_finish_preparation: |
|
|
76 xep_0272.on_muji_request(presence_elt, client) |
|
|
77 mock_try_to_finish_preparation.assert_called_once() |
|
|
78 |
|
|
79 def test_on_muji_request_not_preparing_state(self, host, client): |
|
|
80 """try_to_finish_preparation is not called when preparing_state is False.""" |
|
|
81 xep_0272 = XEP_0272(host) |
|
|
82 presence_elt = domish.Element((None, "presence")) |
|
|
83 presence_elt["from"] = "room@example.com/user" |
|
|
84 presence_elt.addElement((NS_MUJI, "muji")) |
|
|
85 |
|
|
86 muji_data = { |
|
|
87 "done_preparing": True, |
|
|
88 "preparing_jids": set(), |
|
|
89 "to_call": set() |
|
|
90 } |
|
|
91 with patch.object(xep_0272, "get_muji_data", return_value=muji_data): |
|
|
92 with patch.object( |
|
|
93 xep_0272, "try_to_finish_preparation" |
|
|
94 ) as mock_try_to_finish_preparation: |
|
|
95 xep_0272.on_muji_request(presence_elt, client) |
|
|
96 mock_try_to_finish_preparation.assert_not_called() |
|
|
97 |
|
|
98 def test_on_muji_request_own_jid(self, host, client): |
|
|
99 """try_to_finish_preparation is not called when the presence is from own JID.""" |
|
|
100 xep_0272 = XEP_0272(host) |
|
|
101 presence_elt = domish.Element((None, "presence")) |
|
|
102 presence_elt["from"] = "room@example.com/user" |
|
|
103 presence_elt.addElement((NS_MUJI, "muji")) |
|
|
104 |
|
|
105 client.jid = jid.JID("room@example.com/user") |
|
|
106 with patch.object( |
|
|
107 xep_0272, "try_to_finish_preparation" |
|
|
108 ) as mock_try_to_finish_preparation: |
|
|
109 xep_0272.on_muji_request(presence_elt, client) |
|
|
110 mock_try_to_finish_preparation.assert_not_called() |
|
|
111 |
|
|
112 def test_try_to_finish_preparation(self, host, client): |
|
|
113 """try_to_finish_preparation sets done_preparing to True.""" |
|
|
114 xep_0272 = XEP_0272(host) |
|
|
115 room = MagicMock() |
|
|
116 muji_data = {"preparing_jids": set(), "to_call": set()} |
|
|
117 |
|
|
118 with patch.object(xep_0272, "get_muji_data", return_value=muji_data): |
|
|
119 xep_0272.try_to_finish_preparation(client, room, muji_data) |
|
|
120 assert muji_data["done_preparing"] is True |
|
|
121 |
|
|
122 @ed |
|
|
123 async def test_call_group_data_set(self, host, client): |
|
|
124 """call_group_data_set sends correct presence and muji data.""" |
|
|
125 xep_0272 = XEP_0272(host) |
|
|
126 room_jid = jid.JID("room@example.com") |
|
|
127 call_data = {"sdp": "sdp_data"} |
|
|
128 |
|
|
129 with patch.object( |
|
|
130 xep_0272, |
|
|
131 "generate_presence_and_muji", |
|
|
132 return_value=(MagicMock(), MagicMock()), |
|
|
133 ): |
|
|
134 with patch.object( |
|
|
135 client, "a_send", return_value=defer.succeed(None) |
|
|
136 ) as mock_a_send: |
|
|
137 await xep_0272.call_group_data_set(client, room_jid, call_data) |
|
|
138 mock_a_send.assert_called() |
|
|
139 |
|
|
140 @ed |
|
|
141 async def test_start_preparation(self, host, client): |
|
|
142 """start_preparation sends correct presence and muji data.""" |
|
|
143 xep_0272 = XEP_0272(host) |
|
|
144 room = MagicMock() |
|
|
145 |
|
|
146 with patch.object( |
|
|
147 xep_0272, |
|
|
148 "generate_presence_and_muji", |
|
|
149 return_value=(MagicMock(), MagicMock()), |
|
|
150 ): |
|
|
151 with patch.object( |
|
|
152 client, "a_send", return_value=defer.succeed(None) |
|
|
153 ) as mock_a_send: |
|
|
154 await xep_0272.start_preparation(client, room) |
|
|
155 mock_a_send.assert_called() |
|
|
156 |
|
|
157 |
|
|
158 class TestXEP0272Handler: |
|
|
159 def test_connectionInitialized(self, host, client): |
|
|
160 """connectionInitialized adds MUJI presence observer.""" |
|
|
161 xep_0272 = XEP_0272(host) |
|
|
162 handler = XEP_0272_handler(xep_0272) |
|
|
163 handler.parent = MagicMock() |
|
|
164 handler.xmlstream = MagicMock() |
|
|
165 |
|
|
166 with patch.object(handler.xmlstream, "addObserver") as mock_addObserver: |
|
|
167 handler.connectionInitialized() |
|
|
168 mock_addObserver.assert_called_once_with( |
|
|
169 PRESENCE_MUJI, xep_0272.on_muji_request, client=handler.parent |
|
|
170 ) |
|
|
171 |
|
|
172 def test_getDiscoInfo(self): |
|
|
173 """getDiscoInfo returns MUJI feature.""" |
|
|
174 handler = XEP_0272_handler(MagicMock()) |
|
|
175 info = handler.getDiscoInfo(MagicMock(), MagicMock()) |
|
|
176 assert len(info) == 1 |
|
|
177 assert info[0] == NS_MUJI |
|
|
178 |
|
|
179 def test_getDiscoItems(self): |
|
|
180 """getDiscoItems returns empty list.""" |
|
|
181 handler = XEP_0272_handler(MagicMock()) |
|
|
182 items = handler.getDiscoItems(MagicMock(), MagicMock()) |
|
|
183 assert items == [] |
