libervia-backend/tests/unit/test_plugin_xep_0338.py

#!/usr/bin/env python3

# Libervia: an XMPP client
# Copyright (C) 2009-2026 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 pytest import fixture
from twisted.words.protocols.jabber import jid
from twisted.words.xish import domish

from libervia.backend.plugins.plugin_xep_0166 import XEP_0166, NS_JINGLE
from libervia.backend.plugins.plugin_xep_0338 import NS_JINGLE_GROUPING
from libervia.backend.tools.xml_tools import parse


@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 TestXEP0338:
    def test_parse_sdp(self, host):
        """'group' attribute in SDP is correctly parsed"""
        xep_0338 = host.plugins["XEP-0338"]

        call_data = {}
        metadata = {}
        media_type = "video"
        application_data = {}
        transport_data = {}

        # SDP: a=group:BUNDLE audio video
        attribute = "group"
        parts = ["BUNDLE", "audio", "video"]

        xep_0338._parse_sdp_a_trigger(
            attribute,
            parts,
            call_data,
            metadata,
            media_type,
            application_data,
            transport_data,
        )

        assert metadata == {"group": {"BUNDLE": ["audio", "video"]}}

    def test_generate_sdp(self, host):
        """'group' attribute in SDP is correctly generated"""
        xep_0338 = host.plugins["XEP-0338"]

        session = {"metadata": {"group": {"BUNDLE": ["audio", "video"]}}}
        sdp_lines = []
        local = True

        xep_0338._generate_sdp_session_trigger(session, local, sdp_lines)

        assert sdp_lines == ["a=group:BUNDLE audio video"]

    def test_group_building(self, host, client):
        """<group> element are built from session in session init trigger"""
        xep_0338 = host.plugins["XEP-0338"]

        session = {
            "contents": {"audio": {}},
            "jingle_elt": domish.Element((None, "jingle")),
            "metadata": {"group": {"BUNDLE": ["audio", "video"]}},
        }
        content_name = "audio"
        media = "audio"
        media_data = {}
        desc_elt = domish.Element((None, "description"))

        xep_0338._jingle_session_init_trigger(
            client, session, content_name, media, media_data, desc_elt
        )

        group_elts = list(session["jingle_elt"].elements(NS_JINGLE_GROUPING, "group"))
        assert len(group_elts) == 1
        group_elt = group_elts[0]
        assert group_elt["semantics"] == "BUNDLE"
        content_names = [
            content_elt["name"]
            for content_elt in group_elt.elements(NS_JINGLE_GROUPING, "content")
        ]
        assert content_names == ["audio", "video"]

    async def test_group_parsing(self, host, client):
        """<group> elements are correctly parsed in jingle_handler trigger"""
        xep_0338 = host.plugins["XEP-0338"]

        action = xep_0338._j.A_SESSION_INITIATE
        session = {
            "contents": ["audio", "video"],
            "metadata": {},
        }

        raw_xml = """
        <jingle xmlns='urn:xmpp:jingle:1'
                action='session-initiate'
                initiator='user@example.org/orchard'
                sid='a73sjjvkla37jfea'>
          <group xmlns='urn:xmpp:jingle:apps:grouping:0' semantics='BUNDLE'>
            <content name='audio'/>
            <content name='video'/>
          </group>
          <content creator='initiator' name='audio'>
              <description xmlns='urn:xmpp:jingle:apps:rtp:1' media='audio'/>
          </content>
          <content creator='initiator' name='video'>
              <description xmlns='urn:xmpp:jingle:apps:rtp:1' media='video'/>
          </content>
        </jingle>
        """
        session["jingle_elt"] = parse(raw_xml)

        for content_elt in session["jingle_elt"].elements("urn:xmpp:jingle:1", "content"):
            content_name = content_elt["name"]
            desc_elt = next(
                content_elt.elements("urn:xmpp:jingle:apps:rtp:1", "description")
            )

            await xep_0338._jingle_handler_trigger(
                client, action, session, content_name, desc_elt
            )

        group_elts = list(session["jingle_elt"].elements(NS_JINGLE_GROUPING, "group"))
        assert len(group_elts) == 1
        group_elt = group_elts[0]
        assert group_elt["semantics"] == "BUNDLE"
        content_names = [
            content_elt["name"]
            for content_elt in group_elt.elements(NS_JINGLE_GROUPING, "content")
        ]
        assert content_names == ["audio", "video"]

    async def test_content_add_bundle_group(self, host, client, client_IQ):
        """A BUNDLE group is declared with all the contents sharing the transport"""
        xep_0167 = host.plugins["XEP-0167"]
        client.IQ = client_IQ()
        session_id = "123"
        session = {
            "id": session_id,
            "local_jid": client.jid,
            "peer_jid": jid.JID("room@conference.test.example"),
            "role": "initiator",
            # an existing content already using the transport (e.g. a single transport
            # shared by all the streams of the session)
            "contents": {
                "0-0": {
                    "application_data": {"media": "audio"},
                    "transport_data": {
                        "local_ice_data": {"ufrag": "uf1", "pwd": "pw1", "candidates": []}
                    },
                }
            },
        }
        client.jingle_sessions = {session_id: session}
        lines = [
            "v=0",
            "o=- 0 0 IN IP4 127.0.0.1",
            "s=-",
            "t=0 0",
            "m=audio 9 UDP/TLS/RTP/SAVPF 111",
            "c=IN IP4 0.0.0.0",
            "a=rtcp-mux",
            "a=mid:1-0",
            "a=sendonly",
            "a=rtpmap:111 opus/48000/2",
            "a=ice-ufrag:uf1",
            "a=ice-pwd:pw1",
            "m=video 9 UDP/TLS/RTP/SAVPF 96",
            "c=IN IP4 0.0.0.0",
            "a=rtcp-mux",
            "a=mid:1-1",
            "a=sendonly",
            "a=rtpmap:96 VP8/90000",
            "a=ice-ufrag:uf1",
            "a=ice-pwd:pw1",
        ]
        added = await xep_0167.call_update(
            client, session_id, {"sdp": "\r\n".join(lines)}
        )
        assert added == ["1-0", "1-1"]
        # the BUNDLE group is declared for the added contents
        iq_elt = client.IQ.sent_iqs[0]
        jingle_elt = list(iq_elt.elements(NS_JINGLE, "jingle"))[0]
        group_elts = list(jingle_elt.elements(NS_JINGLE_GROUPING, "group"))
        assert len(group_elts) == 1
        group_elt = group_elts[0]
        assert group_elt["semantics"] == "BUNDLE"
        content_names = [
            c["name"] for c in group_elt.elements(NS_JINGLE_GROUPING, "content")
        ]
        # the existing content sharing the transport is included in the group
        assert content_names == ["0-0", "1-0", "1-1"]

    def test_sdp_bundle_group(self, host, client):
        """The stream answer SDP uses the stream's own BUNDLE group, not the uplink's"""
        xep_0167 = host.plugins["XEP-0167"]
        session = {
            "id": "123",
            "local_jid": client.jid,
            "role": "responder",
            "peer_metadata": {"group": {"BUNDLE": ["video0", "audio1"]}},
            "contents": {
                "1-0": {
                    "application_data": {
                        "media": "audio",
                        "peer_data": {
                            "payload_types": {111: {"name": "opus", "clockrate": 48000}}
                        },
                    },
                    "transport_data": {
                        "peer_ice_data": {"ufrag": "uf", "pwd": "pw", "candidates": []}
                    },
                    "senders": "responder",
                },
                "1-1": {
                    "application_data": {
                        "media": "video",
                        "peer_data": {
                            "payload_types": {96: {"name": "VP8", "clockrate": 90000}}
                        },
                    },
                    "transport_data": {
                        "peer_ice_data": {"ufrag": "uf", "pwd": "pw", "candidates": []}
                    },
                    "senders": "responder",
                },
            },
        }
        sdps = []
        session["content_accept_cb"] = lambda *a: sdps.append(a[2])
        xep_0167.on_content_change_trigger(
            client, XEP_0166.A_CONTENT_ACCEPT, session, ["1-0", "1-1"]
        )
        assert len(sdps) == 1
        assert "a=group:BUNDLE 1-0 1-1" in sdps[0]
        assert "video0" not in sdps[0]

    def test_content_change_sdp_no_bundle_group(self, host, client):
        """Accepted contents with different transports are not bundled"""
        xep_0167 = host.plugins["XEP-0167"]
        session = {
            "id": "123",
            "local_jid": client.jid,
            "role": "responder",
            "peer_metadata": {"group": {"BUNDLE": ["video0", "audio1"]}},
            "contents": {
                "1-0": {
                    "application_data": {
                        "media": "audio",
                        "peer_data": {
                            "payload_types": {111: {"name": "opus", "clockrate": 48000}}
                        },
                    },
                    "transport_data": {
                        "peer_ice_data": {
                            "ufrag": "uf1",
                            "pwd": "pw1",
                            "candidates": [],
                        }
                    },
                    "senders": "responder",
                },
                "1-1": {
                    "application_data": {
                        "media": "video",
                        "peer_data": {
                            "payload_types": {96: {"name": "VP8", "clockrate": 90000}}
                        },
                    },
                    "transport_data": {
                        "peer_ice_data": {
                            "ufrag": "uf2",
                            "pwd": "pw2",
                            "candidates": [],
                        }
                    },
                    "senders": "responder",
                },
            },
        }
        sdps = []
        session["content_accept_cb"] = lambda *a: sdps.append(a[2])
        xep_0167.on_content_change_trigger(
            client, XEP_0166.A_CONTENT_ACCEPT, session, ["1-0", "1-1"]
        )
        assert len(sdps) == 1
        assert "a=group:" not in sdps[0]