--- /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 <http://www.gnu.org/licenses/>.
+
+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):
+        """<description> 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 <description> to a dict is successful"""
+        description_element = xml_tools.parse(
+            """
+            <description xmlns="urn:xmpp:jingle:apps:rtp:1" media="audio">
+                <payload-type id="96" channels="2" clockrate="48000" maxptime="60" name="opus" ptime="20">
+                    <parameter name="sprop-stereo" value="1" />
+                </payload-type>
+                <bandwidth type="AS:40000" />
+                <rtcp-mux />
+                <encryption required="1">
+                    <crypto tag="1" crypto-suite="AES_CM_128_HMAC_SHA1_80" key-params="inline:DPSKYRle84Ua2MjbScjadpCvFQH5Tutuls2N4/xx" session-params="" />
+                </encryption>
+            </description>
+            """
+        )
+
+        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(
+            """
+            <jingle xmlns='urn:xmpp:jingle:1'
+                action='session-info'
+                initiator='client1@example.org'
+                sid='a73sjjvkla37jfea'>
+                <mute xmlns="urn:xmpp:jingle:apps:rtp:info:1" name="mute_name"/>
+            </jingle>
+            """
+        )
+
+        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(
+            """
+            <jingle xmlns='urn:xmpp:jingle:1'
+                action='session-info'
+                initiator='client1@example.org'
+                sid='a73sjjvkla37jfea'>
+                <invalid xmlns="urn:xmpp:jingle:apps:rtp:info:1" name="invalid_name"/>
+            </jingle>
+            """
+        )
+
+        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(
+            """
+            <iq from='client1@example.org'
+                id='yh3gr714'
+                to='client2@example.net'
+                type='set'>
+                <jingle xmlns='urn:xmpp:jingle:1'
+                        action='session-info'
+                        initiator='client1@example.org'
+                        sid='a73sjjvkla37jfea'>
+                    <active xmlns='urn:xmpp:jingle:apps:rtp:info:1'/>
+                </jingle>
+            </iq>
+            """
+        )
+        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)