# HG changeset patch # User Goffi # Date 1685360291 -7200 # Node ID e807a5434f82c1a15dc230a55563984940e5cbd3 # Parent 1c4f4aa36d98d5f59daea10627c744315c9a85ce tests (units): tests for plugin XEP-0167: fix 420 diff -r 1c4f4aa36d98 -r e807a5434f82 tests/unit/test_plugin_xep_0167.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/tests/unit/test_plugin_xep_0167.py Mon May 29 13:38:11 2023 +0200 @@ -0,0 +1,377 @@ +#!/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 . + +import base64 +from unittest.mock import MagicMock, patch + +from pytest import fixture +from pytest import raises +from twisted.words.protocols.jabber import jid + +from sat.plugins.plugin_xep_0166 import XEP_0166 +from sat.plugins.plugin_xep_0167 import XEP_0167, mapping +from sat.plugins.plugin_xep_0167.constants import NS_JINGLE_RTP, NS_JINGLE_RTP_INFO +from sat.tools import xml_tools +from sat.tools.common import data_format + + +@fixture(autouse=True) +def no_application_register(monkeypatch): + """Do not register the application in XEP-0166""" + monkeypatch.setattr(XEP_0166, "register_application", lambda *a, **kw: None) + + +class TestXEP0167Mapping: + @fixture(scope="class", autouse=True) + def set_mapping_host(self, host): + mapping.host = host + + def test_senders_to_sdp(self): + """Senders are mapped to SDP attribute""" + assert mapping.senders_to_sdp("both", {"role": "initiator"}) == "a=sendrecv" + assert mapping.senders_to_sdp("none", {"role": "initiator"}) == "a=inactive" + assert mapping.senders_to_sdp("initiator", {"role": "initiator"}) == "a=sendonly" + assert mapping.senders_to_sdp("responder", {"role": "initiator"}) == "a=recvonly" + + def test_generate_sdp_from_session(self): + """SDP is correctly generated from session data""" + session = { + "local_jid": jid.JID("toto@example.org/test"), + "metadata": {}, + "contents": { + "audio": { + "application_data": { + "media": "audio", + "local_data": { + "payload_types": { + 96: { + "name": "opus", + "clockrate": 48000, + "parameters": {"sprop-stereo": "1"}, + } + } + }, + }, + "transport_data": { + "local_ice_data": { + "ufrag": "ufrag", + "pwd": "pwd", + "candidates": [ + { + "foundation": "1", + "component_id": 1, + "transport": "UDP", + "priority": 1, + "address": "10.0.0.1", + "port": 12345, + "type": "host", + } + ], + } + }, + "senders": "both", + } + }, + } + + expected_sdp = ( + "v=0\r\n" + f"o={base64.b64encode('toto@example.org/test'.encode()).decode()} 1 1 IN IP4 0.0.0.0\r\n" + "s=-\r\n" + "t=0 0\r\n" + "a=sendrecv\r\n" + "a=msid-semantic:WMS *\r\n" + "m=audio 9999 UDP/TLS/RTP/SAVPF 96\r\n" + "c=IN IP4 0.0.0.0\r\n" + "a=mid:audio\r\n" + "a=rtpmap:96 opus/48000\r\n" + "a=fmtp:96 sprop-stereo=1\r\n" + "a=ice-ufrag:ufrag\r\n" + "a=ice-pwd:pwd\r\n" + "a=candidate:1 1 UDP 1 10.0.0.1 12345 typ host\r\n" + ) + + assert mapping.generate_sdp_from_session(session, True) == expected_sdp + + def test_parse_sdp(self): + """SDP is correctly parsed to session data""" + sdp = ( + "v=0\r\n" + "o=toto@example.org/test 1 1 IN IP4 0.0.0.0\r\n" + "s=-\r\n" + "t=0 0\r\n" + "a=sendrecv\r\n" + "a=msid-semantic:WMS *\r\n" + "m=audio 9999 UDP/TLS/RTP/SAVPF 96\r\n" + "c=IN IP4 0.0.0.0\r\n" + "a=mid:audio\r\n" + "a=rtpmap:96 opus/48000\r\n" + "a=fmtp:96 sprop-stereo=1\r\n" + "a=ice-ufrag:ufrag\r\n" + "a=ice-pwd:pwd\r\n" + "a=candidate:1 1 UDP 1 10.0.0.1 12345 typ host\r\n" + ) + + expected_session = { + "audio": { + "application_data": { + "media": "audio", + "payload_types": { + 96: { + "id": 96, + "name": "opus", + "clockrate": 48000, + "parameters": {"sprop-stereo": "1"}, + } + }, + }, + "transport_data": { + "port": 9999, + "pwd": "pwd", + "ufrag": "ufrag", + "candidates": [ + { + "foundation": "1", + "component_id": 1, + "transport": "UDP", + "priority": 1, + "address": "10.0.0.1", + "port": 12345, + "type": "host", + } + ], + }, + "id": "audio", + }, + "metadata": {}, + } + + assert mapping.parse_sdp(sdp) == expected_session + + def test_build_description(self): + """ element is generated from media data""" + session = {"metadata": {}} + + media_data = { + "payload_types": { + 96: { + "channels": "2", + "clockrate": "48000", + "id": "96", + "maxptime": "60", + "name": "opus", + "ptime": "20", + "parameters": {"sprop-stereo": "1"}, + } + }, + "bandwidth": "AS:40000", + "rtcp-mux": True, + "encryption": [ + { + "tag": "1", + "crypto-suite": "AES_CM_128_HMAC_SHA1_80", + "key-params": "inline:DPSKYRle84Ua2MjbScjadpCvFQH5Tutuls2N4/xx", + "session-params": "", + } + ], + } + + description_element = mapping.build_description("audio", media_data, session) + + # Assertions + assert description_element.name == "description" + assert description_element.uri == NS_JINGLE_RTP + assert description_element["media"] == "audio" + + # Payload types + payload_types = list(description_element.elements(NS_JINGLE_RTP, "payload-type")) + assert len(payload_types) == 1 + assert payload_types[0].name == "payload-type" + assert payload_types[0]["id"] == "96" + assert payload_types[0]["channels"] == "2" + assert payload_types[0]["clockrate"] == "48000" + assert payload_types[0]["maxptime"] == "60" + assert payload_types[0]["name"] == "opus" + assert payload_types[0]["ptime"] == "20" + + # Parameters + parameters = list(payload_types[0].elements(NS_JINGLE_RTP, "parameter")) + assert len(parameters) == 1 + assert parameters[0].name == "parameter" + assert parameters[0]["name"] == "sprop-stereo" + assert parameters[0]["value"] == "1" + + # Bandwidth + bandwidth = list(description_element.elements(NS_JINGLE_RTP, "bandwidth")) + assert len(bandwidth) == 1 + assert bandwidth[0]["type"] == "AS:40000" + + # RTCP-mux + rtcp_mux = list(description_element.elements(NS_JINGLE_RTP, "rtcp-mux")) + assert len(rtcp_mux) == 1 + + # Encryption + encryption = list(description_element.elements(NS_JINGLE_RTP, "encryption")) + assert len(encryption) == 1 + assert encryption[0]["required"] == "1" + crypto = list(encryption[0].elements("crypto")) + assert len(crypto) == 1 + assert crypto[0]["tag"] == "1" + assert crypto[0]["crypto-suite"] == "AES_CM_128_HMAC_SHA1_80" + assert ( + crypto[0]["key-params"] == "inline:DPSKYRle84Ua2MjbScjadpCvFQH5Tutuls2N4/xx" + ) + assert crypto[0]["session-params"] == "" + + def test_parse_description(self): + """Parsing to a dict is successful""" + description_element = xml_tools.parse( + """ + + + + + + + + + + + """ + ) + + parsed_data = mapping.parse_description(description_element) + + # Assertions + assert parsed_data["payload_types"] == { + 96: { + "channels": "2", + "clockrate": "48000", + "maxptime": "60", + "name": "opus", + "ptime": "20", + "parameters": {"sprop-stereo": "1"}, + } + } + assert parsed_data["bandwidth"] == "AS:40000" + assert parsed_data["rtcp-mux"] is True + assert parsed_data["encryption_required"] is True + assert parsed_data["encryption"] == [ + { + "tag": "1", + "crypto-suite": "AES_CM_128_HMAC_SHA1_80", + "key-params": "inline:DPSKYRle84Ua2MjbScjadpCvFQH5Tutuls2N4/xx", + "session-params": "", + } + ] + + +class TestXEP0167: + def test_jingle_session_info(self, host, client): + """Bridge's call_info method is called with correct parameters.""" + xep_0167 = XEP_0167(host) + session = {"id": "123"} + mock_call_info = MagicMock() + host.bridge.call_info = mock_call_info + + jingle_elt = xml_tools.parse( + """ + + + + """ + ) + + xep_0167.jingle_session_info(client, "mute", session, "content_name", jingle_elt) + + mock_call_info.assert_called_with( + session["id"], + "mute", + data_format.serialise({"name": "mute_name"}), + client.profile, + ) + + def test_jingle_session_info_invalid_actions(self, host, client): + """When receiving invalid actions, no further action is taken.""" + xep_0167 = XEP_0167(host) + session = {"id": "123"} + mock_call_info = MagicMock() + host.bridge.call_info = mock_call_info + + jingle_elt = xml_tools.parse( + """ + + + + """ + ) + + xep_0167.jingle_session_info( + client, "invalid", session, "content_name", jingle_elt + ) + mock_call_info.assert_not_called() + + def test_send_info(self, host, client): + """A jingle element with the correct info is created and sent.""" + xep_0167 = XEP_0167(host) + session_id = "123" + extra = {"name": "test"} + + iq_elt = xml_tools.parse( + """ + + + + + + """ + ) + jingle_elt = iq_elt.firstChildElement() + mock_send = MagicMock() + iq_elt.send = mock_send + + with patch.object( + xep_0167._j, "build_session_info", return_value=(iq_elt, jingle_elt) + ): + xep_0167.send_info(client, session_id, "mute", extra) + + info_elt = jingle_elt.firstChildElement() + assert info_elt.name == "active" + assert info_elt.uri == NS_JINGLE_RTP_INFO + mock_send.assert_called() + + def test_send_info_invalid_actions(self, host, client): + """When trying to send invalid actions, an error is raised.""" + xep_0167 = XEP_0167(host) + session_id = "123" + extra = {"name": "test"} + + with raises(ValueError, match="Unkown info type 'invalid_action'"): + xep_0167.send_info(client, session_id, "invalid_action", extra)