view tests/unit/test_plugin_xep_0215.py @ 4167:319a0e47dc8b

plugin ad-hoc D-Bus: fix deprecated use of python-dbus: the plugin was using python-dbus which is deprecated in the backend in the favor of TxDBus. Methods calls have been updated, and the plugin works again, but there seems to be still some issues (warnings in the logs). Those will be fixed later.
author Goffi <goffi@goffi.org>
date Fri, 01 Dec 2023 15:22:55 +0100
parents 4b842c1fb686
children 2246eeeccc74
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 libervia.backend.plugins.plugin_xep_0215 import XEP_0215
from libervia.backend.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