Mercurial > libervia-backend
view tests/unit/test_plugin_xep_0215.py @ 4042:877145b4ba01
core: don't use `meta_` prefix anymore for `action_extra` in `action_new` signal.
author | Goffi <goffi@goffi.org> |
---|---|
date | Mon, 15 May 2023 16:20:50 +0200 |
parents | 9496f28dadff |
children | 4b842c1fb686 |
line wrap: on
line source
#!/usr/bin/env python3 # Libervia: an XMPP client # Copyright (C) 2009-2023 Jérôme Poisson (goffi@goffi.org) # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. from twisted.internet import defer from pytest_twisted import ensureDeferred as ed from unittest.mock import MagicMock from sat.plugins.plugin_xep_0215 import XEP_0215 from sat.tools import xml_tools from twisted.words.protocols.jabber import jid class TestXEP0215: def test_parse_services(self, host): """Services are parsed correctly""" xep_0215 = XEP_0215(host) services_elt = xml_tools.parse( """ <services xmlns="urn:xmpp:extdisco:2"> <service host="test.example" type="stun" port="1234" /> <service host="example.org" type="turn" port="5678" restricted="true"> <x xmlns="jabber:x:data" type="result"> <field type="hidden" var="FORM_TYPE"> <value>https://test.example/some_extension</value> </field> <field type="text-single" var="some_key"> <value>some_value</value> </field> <field type="boolean" var="some_bool"> <value>0</value> </field> <field type="list-multi" var="multiple_values"> <value>value_1</value> <value>value_2</value> </field> </x> </service> </services>" """ ) services = xep_0215.parse_services(services_elt) expected_services = [ { "host": "test.example", "type": "stun", "port": 1234, }, { "host": "example.org", "type": "turn", "port": 5678, "restricted": True, "extended": [ { "fields": [ { "name": "some_key", "type": "text-single", "value": "some_value", "values": ["some_value"], }, { "name": "some_bool", "type": "boolean", "value": "0", "values": ["0"], }, { "name": "multiple_values", "type": "list-multi", "value": "value_1", "values": ["value_1", "value_2"], }, ], "namespace": "https://test.example/some_extension", } ], }, ] assert services == expected_services @ed async def test_get_external_services(self, host, client): xep_0215 = XEP_0215(host) client._xep_0215_services = {} iq_result = MagicMock() iq_result.send.return_value = defer.succeed(iq_result) client.IQ.return_value = iq_result iq_result_elt = xml_tools.parse( """ <iq type="result"> <services xmlns="urn:xmpp:extdisco:2"> <service host="test.example" type="stun" port="1234" /> <service host="example.org" type="turn" port="5678" restricted="true" /> </services> </iq> """ ) iq_result.send.return_value = defer.succeed(iq_result_elt) services = await xep_0215.get_external_services(client) expected_services = [ { "host": "test.example", "type": "stun", "port": 1234, }, { "host": "example.org", "type": "turn", "port": 5678, "restricted": True, }, ] assert services == expected_services @ed async def test_request_credentials(self, host, client): xep_0215 = XEP_0215(host) iq_result = MagicMock() iq_result.send.return_value = defer.succeed(iq_result) client.IQ.return_value = iq_result iq_result_elt = xml_tools.parse( """ <iq type="result"> <credentials xmlns="urn:xmpp:extdisco:2"> <service host="test.example" type="stun" port="1234" username="user1" password="pass1" restricted="true"/> </credentials> </iq> """ ) iq_result.send.return_value = defer.succeed(iq_result_elt) credentials = await xep_0215.request_credentials(client, "test.example", "stun") expected_credentials = [ { "host": "test.example", "type": "stun", "port": 1234, "username": "user1", "password": "pass1", "restricted": True, } ] assert credentials == expected_credentials def test_services_push(self, host, client): xep_0215 = XEP_0215(host) client._xep_0215_services = { jid.JID("test.example"): [ { "host": "test.example", "type": "stun", "port": 1234, }, { "host": "example.org", "type": "turn", "port": 5678, }, ], } iq_elt = xml_tools.parse( """ <iq type="set" from="test.example"> <services xmlns="urn:xmpp:extdisco:2"> <service action="add" host="example.net" type="stun" port="2345" /> <service action="modify" host="test.example" type="stun" port="1234" expires="2023-04-10T12:34:56Z" /> <service action="delete" host="example.org" type="turn" port="5678" /> </services> </iq> """ ) xep_0215.on_services_push(iq_elt, client) expected_services = [ { "host": "test.example", "type": "stun", "port": 1234, "expires": 1681130096.0, }, { "host": "example.net", "type": "stun", "port": 2345, }, ] assert client._xep_0215_services[jid.JID("test.example")] == expected_services