Mercurial > libervia-backend
changeset 4034:9496f28dadff
tests (unit): Add test for plugin XEP-0215:
rel 418
author | Goffi <goffi@goffi.org> |
---|---|
date | Fri, 07 Apr 2023 15:17:24 +0200 (20 months ago) |
parents | 5a42c7842556 |
children | 9c76678a39e2 |
files | tests/unit/conftest.py tests/unit/test_plugin_xep_0215.py |
diffstat | 2 files changed, 225 insertions(+), 0 deletions(-) [+] |
line wrap: on
line diff
--- a/tests/unit/conftest.py Fri Apr 07 15:16:39 2023 +0200 +++ b/tests/unit/conftest.py Fri Apr 07 15:17:24 2023 +0200 @@ -112,6 +112,7 @@ def client(): client = MagicMock() client.jid = jid.JID("test_user@test.example/123") + client.server_jid = jid.JID("test.example") client.pubsub_service = jid.JID("pubsub.test.example") client.pubsub_client = AsyncMock() return client
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/tests/unit/test_plugin_xep_0215.py Fri Apr 07 15:17:24 2023 +0200 @@ -0,0 +1,224 @@ +#!/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