Mercurial > libervia-backend

comparison tests/unit/test_plugin_xep_0215.py @ 4034:9496f28dadff

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
tests (unit): Add test for plugin XEP-0215: rel 418
author Goffi <goffi@goffi.org>
date Fri, 07 Apr 2023 15:17:24 +0200
parents
children 4b842c1fb686
comparison
equal deleted inserted replaced
4033:5a42c7842556 4034:9496f28dadff
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 twisted.internet import defer
20 from pytest_twisted import ensureDeferred as ed
21 from unittest.mock import MagicMock
22 from sat.plugins.plugin_xep_0215 import XEP_0215
23 from sat.tools import xml_tools
24 from twisted.words.protocols.jabber import jid
25
26
27 class TestXEP0215:
28 def test_parse_services(self, host):
29 """Services are parsed correctly"""
30 xep_0215 = XEP_0215(host)
31
32 services_elt = xml_tools.parse(
33 """
34 <services xmlns="urn:xmpp:extdisco:2">
35 <service host="test.example" type="stun" port="1234" />
36 <service host="example.org" type="turn" port="5678" restricted="true">
37 <x xmlns="jabber:x:data" type="result">
38 <field type="hidden" var="FORM_TYPE">
39 <value>https://test.example/some_extension</value>
40 </field>
41 <field type="text-single" var="some_key">
42 <value>some_value</value>
43 </field>
44 <field type="boolean" var="some_bool">
45 <value>0</value>
46 </field>
47 <field type="list-multi" var="multiple_values">
48 <value>value_1</value>
49 <value>value_2</value>
50 </field>
51 </x>
52 </service>
53 </services>"
54 """
55 )
56
57 services = xep_0215.parse_services(services_elt)
58
59 expected_services = [
60 {
61 "host": "test.example",
62 "type": "stun",
63 "port": 1234,
64 },
65 {
66 "host": "example.org",
67 "type": "turn",
68 "port": 5678,
69 "restricted": True,
70 "extended": [
71 {
72 "fields": [
73 {
74 "name": "some_key",
75 "type": "text-single",
76 "value": "some_value",
77 "values": ["some_value"],
78 },
79 {
80 "name": "some_bool",
81 "type": "boolean",
82 "value": "0",
83 "values": ["0"],
84 },
85 {
86 "name": "multiple_values",
87 "type": "list-multi",
88 "value": "value_1",
89 "values": ["value_1", "value_2"],
90 },
91 ],
92 "namespace": "https://test.example/some_extension",
93 }
94 ],
95 },
96 ]
97
98 assert services == expected_services
99
100 @ed
101 async def test_get_external_services(self, host, client):
102 xep_0215 = XEP_0215(host)
103 client._xep_0215_services = {}
104
105 iq_result = MagicMock()
106 iq_result.send.return_value = defer.succeed(iq_result)
107
108 client.IQ.return_value = iq_result
109
110 iq_result_elt = xml_tools.parse(
111 """
112 <iq type="result">
113 <services xmlns="urn:xmpp:extdisco:2">
114 <service host="test.example" type="stun" port="1234" />
115 <service host="example.org" type="turn" port="5678"
116 restricted="true" />
117 </services>
118 </iq>
119 """
120 )
121 iq_result.send.return_value = defer.succeed(iq_result_elt)
122
123 services = await xep_0215.get_external_services(client)
124
125 expected_services = [
126 {
127 "host": "test.example",
128 "type": "stun",
129 "port": 1234,
130 },
131 {
132 "host": "example.org",
133 "type": "turn",
134 "port": 5678,
135 "restricted": True,
136 },
137 ]
138
139 assert services == expected_services
140
141 @ed
142 async def test_request_credentials(self, host, client):
143 xep_0215 = XEP_0215(host)
144
145 iq_result = MagicMock()
146 iq_result.send.return_value = defer.succeed(iq_result)
147
148 client.IQ.return_value = iq_result
149
150 iq_result_elt = xml_tools.parse(
151 """
152 <iq type="result">
153 <credentials xmlns="urn:xmpp:extdisco:2">
154 <service host="test.example" type="stun" port="1234" username="user1"
155 password="pass1" restricted="true"/>
156 </credentials>
157 </iq>
158 """
159 )
160 iq_result.send.return_value = defer.succeed(iq_result_elt)
161
162 credentials = await xep_0215.request_credentials(client, "test.example", "stun")
163
164 expected_credentials = [
165 {
166 "host": "test.example",
167 "type": "stun",
168 "port": 1234,
169 "username": "user1",
170 "password": "pass1",
171 "restricted": True,
172 }
173 ]
174
175 assert credentials == expected_credentials
176
177 def test_services_push(self, host, client):
178 xep_0215 = XEP_0215(host)
179
180 client._xep_0215_services = {
181 jid.JID("test.example"): [
182 {
183 "host": "test.example",
184 "type": "stun",
185 "port": 1234,
186 },
187 {
188 "host": "example.org",
189 "type": "turn",
190 "port": 5678,
191 },
192 ],
193 }
194
195 iq_elt = xml_tools.parse(
196 """
197 <iq type="set" from="test.example">
198 <services xmlns="urn:xmpp:extdisco:2">
199 <service action="add" host="example.net" type="stun" port="2345" />
200 <service action="modify" host="test.example" type="stun" port="1234"
201 expires="2023-04-10T12:34:56Z" />
202 <service action="delete" host="example.org" type="turn" port="5678" />
203 </services>
204 </iq>
205 """
206 )
207
208 xep_0215.on_services_push(iq_elt, client)
209
210 expected_services = [
211 {
212 "host": "test.example",
213 "type": "stun",
214 "port": 1234,
215 "expires": 1681130096.0,
216 },
217 {
218 "host": "example.net",
219 "type": "stun",
220 "port": 2345,
221 },
222 ]
223
224 assert client._xep_0215_services[jid.JID("test.example")] == expected_services