Mercurial > libervia-backend

comparison tests/unit/test_plugin_xep_0272.py @ 4246:5eb13251fd75

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
tests (unit/XEP-0272): XEP-0272 tests: fix 429
author Goffi <goffi@goffi.org>
date Wed, 15 May 2024 17:35:16 +0200
parents
children f1d0cde61af7
comparison
equal deleted inserted replaced
4245:a7d4007a8fa5 4246:5eb13251fd75
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 == []