Mercurial > libervia-backend

comparison tests/unit/test_plugin_xep_0176.py @ 4046:0e3ce379aae3

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
tests (unit): tests for plugin XEP-0176: fix 419
author Goffi <goffi@goffi.org>
date Mon, 15 May 2023 16:23:50 +0200
parents
children 4b842c1fb686
comparison
equal deleted inserted replaced
4045:ae756bf7c3e8 4046:0e3ce379aae3
1 #!/usr/bin/env python3
2
3 # Libervia: an XMPP client
4 # Copyright (C) 2009-2023 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 AsyncMock, MagicMock
20
21 from pytest import fixture
22 from pytest_twisted import ensureDeferred as ed
23
24 from sat.plugins.plugin_xep_0166 import XEP_0166
25 from sat.plugins.plugin_xep_0176 import NS_JINGLE_ICE_UDP, XEP_0176
26 from sat.tools import xml_tools
27
28
29 @fixture(autouse=True)
30 def no_transport_register(monkeypatch):
31 """Do not register the transport in XEP-0166"""
32 monkeypatch.setattr(XEP_0166, "register_transport", lambda *a, **kw: None)
33
34
35 class TestXEP0176:
36 def create_mock_session(self, content_name):
37 return {
38 "id": "test-session-id",
39 "contents": {
40 content_name: {
41 "application_data": {"media": "audio"},
42 "transport_data": {
43 "local_ice_data": {
44 "ufrag": "testufrag",
45 "pwd": "testpwd",
46 "candidates": [
47 {
48 "id": "candidate_1",
49 "component_id": 1,
50 "foundation": "1",
51 "address": "192.0.2.1",
52 "port": 1234,
53 "priority": 1,
54 "transport": "udp",
55 "type": "host",
56 }
57 ],
58 }
59 },
60 }
61 },
62 }
63
64 def test_build_transport(self, host, monkeypatch):
65 """ICE data is correctly transformed into transport element"""
66 xep_0176 = XEP_0176(host)
67
68 ice_data = {
69 "ufrag": "user1",
70 "pwd": "password1",
71 "candidates": [
72 {
73 "component_id": 1,
74 "foundation": "1",
75 "address": "192.168.0.1",
76 "port": 1234,
77 "priority": 100,
78 "transport": "udp",
79 "type": "host",
80 "generation": "0",
81 "network": "0",
82 },
83 {
84 "component_id": 2,
85 "foundation": "1",
86 "address": "192.168.0.2",
87 "port": 5678,
88 "priority": 100,
89 "transport": "udp",
90 "type": "host",
91 "generation": "0",
92 "network": "0",
93 "rel_addr": "10.0.0.1",
94 "rel_port": 9012,
95 },
96 ],
97 }
98
99 transport_elt = xep_0176.build_transport(ice_data)
100
101 assert transport_elt.name == "transport"
102 assert transport_elt.uri == NS_JINGLE_ICE_UDP
103 assert transport_elt.getAttribute("ufrag") == "user1"
104 assert transport_elt.getAttribute("pwd") == "password1"
105
106 candidates = list(transport_elt.elements(NS_JINGLE_ICE_UDP, "candidate"))
107
108 assert len(candidates) == len(ice_data["candidates"])
109
110 for i, candidate_elt in enumerate(candidates):
111 ice_candidate = ice_data["candidates"][i]
112 assert (
113 int(candidate_elt.getAttribute("component"))
114 == ice_candidate["component_id"]
115 )
116 assert candidate_elt.getAttribute("foundation") == ice_candidate["foundation"]
117 assert candidate_elt.getAttribute("ip") == ice_candidate["address"]
118 assert int(candidate_elt.getAttribute("port")) == ice_candidate["port"]
119 assert (
120 int(candidate_elt.getAttribute("priority")) == ice_candidate["priority"]
121 )
122 assert candidate_elt.getAttribute("protocol") == ice_candidate["transport"]
123 assert candidate_elt.getAttribute("type") == ice_candidate["type"]
124 assert candidate_elt.getAttribute("generation") == str(
125 ice_candidate.get("generation", "0")
126 )
127 assert candidate_elt.getAttribute("network") == str(
128 ice_candidate.get("network", "0")
129 )
130
131 if "rel_addr" in ice_candidate:
132 assert candidate_elt.getAttribute("rel-addr") == ice_candidate["rel_addr"]
133 assert (
134 int(candidate_elt.getAttribute("rel-port"))
135 == ice_candidate["rel_port"]
136 )
137 else:
138 assert candidate_elt.getAttribute("rel-addr") is None
139 assert candidate_elt.getAttribute("rel-port") is None
140
141 def test_parse_transport(self, host):
142 """Transport element is correctly parsed into ICE data"""
143 xep_0176 = XEP_0176(host)
144
145 transport_elt = xml_tools.parse(
146 """
147 <transport xmlns="urn:xmpp:jingle:transports:ice-udp:1"
148 pwd="password1"
149 ufrag="user1">
150 <candidate component="1"
151 foundation="1"
152 generation="0"
153 id="uuid1"
154 ip="192.168.0.1"
155 network="0"
156 port="1234"
157 priority="100"
158 protocol="udp"
159 type="host" />
160 <candidate component="2"
161 foundation="1"
162 generation="0"
163 id="uuid2"
164 ip="192.168.0.2"
165 network="0"
166 port="5678"
167 priority="100"
168 protocol="udp"
169 type="host"
170 rel-addr="10.0.0.1"
171 rel-port="9012" />
172 </transport>
173 """
174 )
175
176 ice_data = xep_0176.parse_transport(transport_elt)
177
178 assert transport_elt.getAttribute("ufrag") == "user1"
179 assert transport_elt.getAttribute("pwd") == "password1"
180
181 candidates = list(transport_elt.elements(NS_JINGLE_ICE_UDP, "candidate"))
182 assert len(candidates) == len(ice_data["candidates"])
183
184 for i, candidate_elt in enumerate(candidates):
185 ice_candidate = ice_data["candidates"][i]
186 assert (
187 int(candidate_elt.getAttribute("component"))
188 == ice_candidate["component_id"]
189 )
190 assert candidate_elt.getAttribute("foundation") == ice_candidate["foundation"]
191 assert candidate_elt.getAttribute("ip") == ice_candidate["address"]
192 assert int(candidate_elt.getAttribute("port")) == ice_candidate["port"]
193 assert (
194 int(candidate_elt.getAttribute("priority")) == ice_candidate["priority"]
195 )
196 assert candidate_elt.getAttribute("protocol") == ice_candidate["transport"]
197 assert candidate_elt.getAttribute("type") == ice_candidate["type"]
198 assert candidate_elt.getAttribute("generation") == str(
199 ice_candidate.get("generation", "0")
200 )
201 assert candidate_elt.getAttribute("network") == str(
202 ice_candidate.get("network", "0")
203 )
204
205 if "rel_addr" in ice_candidate:
206 assert candidate_elt.getAttribute("rel-addr") == ice_candidate["rel_addr"]
207 assert (
208 int(candidate_elt.getAttribute("rel-port"))
209 == ice_candidate["rel_port"]
210 )
211 else:
212 assert candidate_elt.getAttribute("rel-addr") is None
213 assert candidate_elt.getAttribute("rel-port") is None
214
215 @ed
216 async def test_jingle_session_init(self, host, client):
217 """<transport/> element is built on initiator side during init"""
218 xep_0176 = XEP_0176(host)
219
220 content_name = "test-content"
221 session = self.create_mock_session(content_name)
222
223 transport_elt = await xep_0176.jingle_session_init(client, session, content_name)
224
225 expected_transport_elt = xep_0176.build_transport(
226 session["contents"][content_name]["transport_data"]["local_ice_data"]
227 )
228
229 assert transport_elt.toXml() == expected_transport_elt.toXml()
230
231 @ed
232 async def test_jingle_handler(self, host, client):
233 """<transport/> element is built on responder side during init"""
234 xep_0176 = XEP_0176(host)
235
236 content_name = "test-content"
237 action = "session-initiate"
238 session = self.create_mock_session(content_name)
239 transport_elt = xml_tools.parse("<transport/>")
240
241 returned_transport_elt = await xep_0176.jingle_handler(
242 client, action, session, content_name, transport_elt
243 )
244
245 expected_transport_elt = xep_0176.build_transport(
246 session["contents"][content_name]["transport_data"]["local_ice_data"]
247 )
248
249 assert returned_transport_elt.toXml() == expected_transport_elt.toXml()
250
251 @ed
252 async def test_ice_candidates_add(self, host, client):
253 """local new ICE candidates are added, IQ is sent, bridge signal emitted"""
254 xep_0176 = XEP_0176(host)
255
256 content_name = "test-content"
257 session_id = "test-session-id"
258 media_ice_data_s = {
259 "audio": {
260 # we use different "ufrag" and "pwd" than in "create_mock_session" to
261 # trigger an ICE restart
262 "ufrag": "new_testufrag",
263 "pwd": "new_testpwd",
264 "candidates": [
265 {
266 "component_id": 2,
267 "foundation": "2",
268 "address": "192.0.2.2",
269 "port": 1235,
270 "priority": 2,
271 "transport": "udp",
272 "type": "host",
273 }
274 ],
275 }
276 }
277
278 mock_session = self.create_mock_session(content_name)
279 xep_0176._j.get_session = MagicMock(return_value=mock_session)
280 xep_0176._j.build_action = MagicMock()
281 iq_elt = AsyncMock()
282 xep_0176._j.build_action.return_value = (iq_elt, None)
283 xep_0176.host.bridge.ice_restart = MagicMock()
284
285 await xep_0176.ice_candidates_add(client, session_id, media_ice_data_s)
286
287 xep_0176._j.build_action.assert_called()
288 iq_elt.send.assert_called()
289 xep_0176.host.bridge.ice_restart.assert_called()
290
291 # Checking that local ICE data is updated correctly
292 updated_local_ice_data = mock_session["contents"][content_name]["transport_data"][
293 "local_ice_data"
294 ]
295 assert updated_local_ice_data["ufrag"] == "new_testufrag"
296 assert updated_local_ice_data["pwd"] == "new_testpwd"
297 assert (
298 updated_local_ice_data["candidates"]
299 == media_ice_data_s["audio"]["candidates"]
300 )