erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

plugin XEP-0166: support all content actions + fixes:

Browse source 
- all content actions are now handled
- XEP-0167: the new contents of an offered SDP are added with a single content-add, the
  contents sharing a single transport are bundled (XEP-0338)
- XEP-0167: the generated SDP declares the BUNDLE group of its own contents
- XEP-0176: one transport-info IQ per content. ICE data are keyed by content names.
- fix session-level ICE attributes and fingerprint
This commit is contained in:
Goffi 2026-10-06 19:27:53 +02:00
commit ab26a56c1d
15 changed files with 1804 additions and 261 deletions
Download patch file Download diff file Expand all files Collapse all files

1
tests/unit/conftest.py
View file

@ -141,6 +141,7 @@ def client_IQ():
@fixture
def client():
client = MagicMock()
client.is_component = False
client.jid = jid.JID("test_user@test.example/123")
client.server_jid = jid.JID("test.example")
client.pubsub_service = jid.JID("pubsub.test.example")

94
tests/unit/test_plugin_xep_0166.py Normal file
View file

@ -0,0 +1,94 @@
#!/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 unittest.mock import MagicMock
from libervia.backend.plugins.plugin_xep_0166 import XEP_0166
from libervia.backend.tools.common import data_format
class TestXEP0166ContentChangeSignal:
def _get_session(self, content_name="video"):
mock_application = MagicMock()
mock_application.handler.get_content_change_payload.return_value = {
"media": "video",
}
return {
"id": "123",
"added_content_names": [content_name],
"contents": {
content_name: {
"senders": "initiator",
"application": mock_application,
}
},
}
def test_emit_content_change(self, host, client):
"""The signal carries the session data and the application extra data"""
xep_0166 = XEP_0166(host)
mock_signal = MagicMock()
host.bridge.jingle_content_change = mock_signal
session = self._get_session()
xep_0166._emit_content_change(client, XEP_0166.A_CONTENT_ADD, session)
args = mock_signal.call_args.args
assert args[1] == client.profile
info = data_format.deserialise(args[0])
assert info["session_id"] == "123"
assert info["action"] == XEP_0166.A_CONTENT_ADD
assert info["changes"] == [
{"content": "video", "senders": "initiator", "payload": {"media": "video"}}
]
session["contents"]["video"][
"application"
].handler.get_content_change_payload.assert_called_once_with(
client, XEP_0166.A_CONTENT_ADD, session, "video"
)
def test_emit_content_change_multi_content(self, host, client):
"""A stanza with several contents emits a single signal with all the changes"""
xep_0166 = XEP_0166(host)
mock_signal = MagicMock()
host.bridge.jingle_content_change = mock_signal
session = self._get_session("video")
session["added_content_names"] = ["video", "audio"]
session["contents"]["audio"] = session["contents"]["video"]
xep_0166._emit_content_change(client, XEP_0166.A_CONTENT_ADD, session)
assert mock_signal.call_count == 1
info = data_format.deserialise(mock_signal.call_args.args[0])
changes = [change["content"] for change in info["changes"]]
assert changes == ["video", "audio"]
def test_emit_content_change_unknown_content(self, host, client):
"""When the content is unknown, no signal is emitted and no exception raised"""
xep_0166 = XEP_0166(host)
mock_signal = MagicMock()
host.bridge.jingle_content_change = mock_signal
session = {
"id": "123",
"added_content_names": ["video"],
"contents": {},
}
xep_0166._emit_content_change(client, XEP_0166.A_CONTENT_ADD, session)
mock_signal.assert_not_called()

398
tests/unit/test_plugin_xep_0167.py
View file

@ -22,13 +22,19 @@ from unittest.mock import MagicMock, patch
from pytest import fixture
from pytest import raises
from twisted.words.protocols.jabber import jid
from twisted.words.xish import domish
from libervia.backend.plugins.plugin_xep_0166 import XEP_0166
from libervia.backend.plugins.plugin_xep_0166 import XEP_0166, NS_JINGLE
from libervia.backend.plugins.plugin_xep_0166.models import (
ContentChangePayload,
TransportData,
)
from libervia.backend.plugins.plugin_xep_0167 import XEP_0167, mapping
from libervia.backend.plugins.plugin_xep_0167.constants import (
NS_JINGLE_RTP,
NS_JINGLE_RTP_INFO,
)
from libervia.backend.plugins.plugin_xep_0176 import NS_JINGLE_ICE_UDP
from libervia.backend.tools import xml_tools
from libervia.backend.tools.common import data_format
@ -40,10 +46,6 @@ def no_application_register(monkeypatch):
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"
@ -112,6 +114,68 @@ class TestXEP0167Mapping:
assert mapping.generate_sdp_from_session(session, True) == expected_sdp
def test_generate_content_add_sdp(self):
"""The m-line of a content added by the peer is correctly generated"""
content_data = {
"application_data": {
"media": "video",
"peer_data": {
"payload_types": {
96: {
"name": "VP8",
"clockrate": 90000,
"parameters": {"max-fs": "12288"},
}
},
"rtcp-mux": True,
},
},
"transport_data": {
"peer_ice_data": {
"ufrag": "ufrag",
"pwd": "pwd",
"fingerprint": {
"hash": "sha-256",
"setup": "actpass",
"fingerprint": "AB:CD:EF",
},
"candidates": [
{
"foundation": "1",
"component_id": 1,
"transport": "UDP",
"priority": 1,
"address": "10.0.0.2",
"port": 54321,
"type": "host",
}
],
}
},
"senders": "initiator",
}
session = {"role": "responder"}
expected_sdp = (
"m=video 9 UDP/TLS/RTP/SAVPF 96\r\n"
"a=mid:video\r\n"
"c=IN IP4 0.0.0.0\r\n"
"a=sendonly\r\n"
"a=rtpmap:96 VP8/90000\r\n"
"a=fmtp:96 max-fs=12288\r\n"
"a=rtcp-mux\r\n"
"a=fingerprint:sha-256 AB:CD:EF\r\n"
"a=setup:actpass\r\n"
"a=ice-ufrag:ufrag\r\n"
"a=ice-pwd:pwd\r\n"
"a=candidate:1 1 UDP 1 10.0.0.2 54321 typ host\r\n"
)
assert (
mapping.generate_content_add_sdp(content_data, session, "video")
== expected_sdp
)
def test_parse_sdp(self):
"""SDP is correctly parsed to session data"""
sdp = (
@ -170,6 +234,68 @@ class TestXEP0167Mapping:
assert mapping.parse_sdp(sdp, "responder") == expected_session
def test_parse_sdp_session_level_ice(self):
"""Session-level transport attributes apply to all media
unless overridden at the media level
media
"""
sdp = (
"v=0\r\no=- 0 0 IN IP4 0.0.0.0\r\nt=0 0\r\n"
"a=fingerprint:sha-256 00:11:22\r\n"
"a=ice-pwd:spwd\r\n"
"m=audio 1 UDP/TLS/RTP/SAVPF 96\r\n"
"a=fingerprint:sha-256 33:44:55\r\n"
"a=ice-ufrag:sufrag\r\n"
"m=video 2 UDP/TLS/RTP/SAVPF 97\r\n"
"a=ice-ufrag:sufrag\r\n"
)
parsed = mapping.parse_sdp(sdp, "initiator")
# audio: the media-level fingerprint wins
assert parsed["audio"]["transport_data"]["fingerprint"] == {
"hash": "sha-256",
"fingerprint": "33:44:55",
}
# video: no media-level fingerprint, the session one applies (the audio
# override must not have polluted it)
assert parsed["video"]["transport_data"]["fingerprint"] == {
"hash": "sha-256",
"fingerprint": "00:11:22",
}
# the session-level ice-pwd applies to both media
assert parsed["audio"]["transport_data"]["pwd"] == "spwd"
assert parsed["video"]["transport_data"]["pwd"] == "spwd"
def test_parse_mline(self):
"""Test parse_mline on a standalone m-line block"""
mline = (
"m=audio 49170 UDP/TLS/RTP/SAVPF 0 96\r\n"
"c=IN IP4 0.0.0.0\r\n"
"a=rtcp-mux\r\n"
"a=sendonly\r\n"
"a=rtpmap:96 opus/48000/2\r\n"
"a=ice-ufrag:abc123\r\n"
"a=ice-pwd:def456\r\n"
"a=fingerprint:sha-256 00:11:22\r\n"
)
media, media_data = mapping.parse_mline(mline, "responder")
assert media == "audio"
assert media_data["application_data"]["senders"] == "responder"
assert media_data["application_data"]["payload_types"][96]["name"] == "opus"
assert media_data["transport_data"]["port"] == 49170
assert media_data["transport_data"]["ufrag"] == "abc123"
assert media_data["transport_data"]["pwd"] == "def456"
assert media_data["transport_data"]["fingerprint"]["fingerprint"] == "00:11:22"
# session attributes (as forwarded by parse_sdp) are applied to the media:
# with no media-level direction, the session one is used (a media-level
# a=sendonly would override it, remapped with the role)
mline_no_dir = mline.replace("a=sendonly\r\n", "")
media, media_data = mapping.parse_mline(
mline_no_dir, "responder", {"sendonly": "initiator"}
)
assert media_data["application_data"]["senders"] == "initiator"
def test_build_description(self):
"""<description> element is generated from media data"""
session = {"metadata": {}}
@ -374,3 +500,265 @@ class TestXEP0167:
with raises(ValueError, match="Unkown info type 'invalid_action'"):
xep_0167.send_info(client, session_id, "invalid_action", extra)
def test_content_change_payload_content_add(self, host, client):
"""For a content-add, the offered m-line SDP is returned in the payload"""
xep_0167 = XEP_0167(host)
session = {
"id": "123",
"role": "responder",
"contents": {
"video": {
"application_data": {
"media": "video",
"peer_data": {
"payload_types": {96: {"name": "VP8", "clockrate": 90000}},
},
},
"transport_data": {
"peer_ice_data": {
"ufrag": "uf",
"pwd": "pw",
"candidates": [],
}
},
"senders": "initiator",
}
},
}
payload = xep_0167.get_content_change_payload(
client, XEP_0166.A_CONTENT_ADD, session, "video"
)
assert payload.media == "video"
assert payload.sdp
assert "m=video 9 UDP/TLS/RTP/SAVPF 96" in payload.sdp
assert "a=rtpmap:96 VP8/90000" in payload.sdp
assert "a=ice-ufrag:uf" in payload.sdp
def test_content_change_payload_content_modify(self, host, client):
"""For a content-modify, the media and new direction are returned (no SDP)"""
xep_0167 = XEP_0167(host)
session = {
"id": "123",
"role": "responder",
"contents": {
"video": {
"application_data": {"media": "video"},
"senders": "both",
}
},
}
payload = xep_0167.get_content_change_payload(
client, XEP_0166.A_CONTENT_MODIFY, session, "video"
)
assert payload.model_dump(exclude_none=True) == {
"media": "video",
"direction": "sendrecv",
}
async def test_content_modify(self, host, client, client_IQ):
"""The modified senders attribute is updated and acknowledged"""
xep_0166 = XEP_0166(host)
client.IQ = client_IQ()
mock_application = MagicMock()
mock_application.handler.get_content_change_payload.return_value = (
ContentChangePayload()
)
session = {
"id": "123",
"local_jid": jid.JID("client1@example.org/1"),
"peer_jid": jid.JID("client2@example.net/2"),
"role": "responder",
"contents": {
"video": {
"senders": "initiator",
"creator": XEP_0166.ROLE_INITIATOR,
"application": mock_application,
"transport": MagicMock(),
}
},
}
mock_send = MagicMock()
client.send = mock_send
request = domish.Element((None, "iq"))
request["type"] = "set"
jingle_elt = request.addElement("jingle", NS_JINGLE)
jingle_elt["sid"] = "123"
jingle_elt["action"] = XEP_0166.A_CONTENT_MODIFY
content_elt = jingle_elt.addElement("content", NS_JINGLE)
content_elt["name"] = "video"
content_elt["senders"] = "both"
await xep_0166.on_content_modify(client, request, jingle_elt, session)
# the direction is updated
assert session["contents"]["video"]["senders"] == "both"
assert session["modified_content_names"] == ["video"]
# no contrary content-modify is sent, only the IQ result
stanzas = [call.args[0] for call in mock_send.call_args_list]
assert stanzas[0]["type"] == "result"
async def test_call_update(self, host, client, client_IQ):
"""New m-lines of the offered SDP are added with a content-add IQ"""
xep_0167 = XEP_0167(host)
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",
"contents": {},
}
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",
]
added = await xep_0167.call_update(
client, session_id, {"sdp": "\r\n".join(lines)}
)
assert added == ["1-0"]
# the new content is sent with a content-add IQ
iq_elt = client.IQ.sent_iqs[0]
jingle_elt = list(iq_elt.elements(NS_JINGLE, "jingle"))[0]
assert jingle_elt["action"] == XEP_0166.A_CONTENT_ADD
content_elts = list(jingle_elt.elements(NS_JINGLE, "content"))
assert [c["name"] for c in content_elts] == ["1-0"]
def test_content_change_payload_content_accept(self, host, client):
"""For a content-accept, the answer m-line is generated"""
xep_0167 = XEP_0167(host)
# the real ICE-UDP transport handler, so the whole parsing chain is tested
# (XEP-0176 parsing + XEP-0320 fingerprint)
transport = TransportData(NS_JINGLE_ICE_UDP, host.plugins["XEP-0176"], 100)
desc_elt = domish.Element(
(NS_JINGLE_RTP, "description"), attribs={"media": "video"}
)
pt_elt = desc_elt.addElement("payload-type")
pt_elt["id"] = "96"
pt_elt["name"] = "VP8"
pt_elt["clockrate"] = "90000"
# the peer transport element, parsed by the XEP-0176 handler (fingerprint
# included by the XEP-0320 trigger)
peer_transport_elt = domish.Element(
(NS_JINGLE_ICE_UDP, "transport"), attribs={"ufrag": "uf", "pwd": "pw"}
)
fingerprint_elt = peer_transport_elt.addElement(
("urn:xmpp:jingle:apps:dtls:0", "fingerprint"), content="AB:CD"
)
fingerprint_elt["hash"] = "sha-256"
fingerprint_elt["setup"] = "actpass"
session = {
"id": "123",
"role": "responder",
"contents": {
"video": {
"application_data": {"media": "video"},
"transport": transport,
"transport_data": {},
"desc_elt": desc_elt,
"transport_elt": peer_transport_elt,
"senders": "responder",
}
},
}
payload = xep_0167.get_content_change_payload(
client, XEP_0166.A_CONTENT_ACCEPT, session, "video"
)
assert payload.media == "video"
# the peer data are parsed and stored in the session
assert session["contents"]["video"]["application_data"]["peer_data"]
assert session["contents"]["video"]["transport_data"]["peer_ice_data"]
assert payload.sdp
assert "m=video 9 UDP/TLS/RTP/SAVPF 96" in payload.sdp
# the direction is strict (senders=responder, we are the responder: the peer receives)
assert "a=recvonly" in payload.sdp
assert "a=fingerprint:sha-256 AB:CD" in payload.sdp
assert "a=ice-ufrag:uf" in payload.sdp
def test_call_content_accept(self, host, client, client_IQ):
"""A content-accept IQ is sent, built from the frontend answer m-line"""
xep_0167 = XEP_0167(host)
transport_elt = domish.Element(
(NS_JINGLE_ICE_UDP, "transport"), attribs={"ufrag": "uf", "pwd": "pw"}
)
mock_transport = MagicMock()
mock_transport.handler.build_transport = MagicMock(return_value=transport_elt)
client.IQ = client_IQ()
session = {
"id": "123",
"added_content_names": ["video"],
"local_jid": jid.JID("client1@example.org/1"),
"peer_jid": jid.JID("client2@example.net/2"),
"role": "responder",
"contents": {
"video": {
"application_data": {"media": "video"},
"transport": mock_transport,
"transport_data": {},
"senders": "initiator",
}
},
}
client.jingle_sessions = {"123": session}
host.get_client = MagicMock(return_value=client)
mock_send = MagicMock()
client.send = mock_send
mline = (
"m=video 9 UDP/TLS/RTP/SAVPF 96\r\n"
"c=IN IP4 0.0.0.0\r\n"
"a=recvonly\r\n"
"a=rtpmap:96 VP8/90000\r\n"
"a=rtcp-mux\r\n"
"a=fingerprint:sha-256 AA:BB\r\n"
"a=setup:active\r\n"
"a=ice-ufrag:uf\r\n"
"a=ice-pwd:pw\r\n"
"a=candidate:1 1 UDP 1 10.0.0.1 12345 typ host\r\n"
)
xep_0167._call_content_accept("123", "video", mline, client.profile)
assert "added_content_names" not in session
# local data are stored in the session
assert session["contents"]["video"]["application_data"]["local_data"]
assert (
session["contents"]["video"]["transport_data"]["local_ice_data"]["ufrag"]
== "uf"
)
sent_iq = mock_send.call_args.args[0]
jingle_elt = sent_iq.firstChildElement()
assert jingle_elt["action"] == XEP_0166.A_CONTENT_ACCEPT
content_elt = jingle_elt.firstChildElement()
assert content_elt["name"] == "video"
assert content_elt["creator"] == XEP_0166.ROLE_INITIATOR
assert content_elt["senders"] == "initiator"
desc_elts = list(content_elt.elements(NS_JINGLE_RTP, "description"))
assert len(desc_elts) == 1
desc_elt = desc_elts[0]
assert desc_elt["media"] == "video"
payload_elts = list(desc_elt.elements(NS_JINGLE_RTP, "payload-type"))
assert len(payload_elts) == 1
assert payload_elts[0]["name"] == "VP8"
transport_elts = content_elt.elements(NS_JINGLE_ICE_UDP, "transport")
transport_elt = next(transport_elts)
assert transport_elt["ufrag"] == "uf"
assert transport_elt["pwd"] == "pw"

7
tests/unit/test_plugin_xep_0176.py
View file

@ -251,7 +251,7 @@ class TestXEP0176:
assert returned_transport_elt.toXml() == expected_transport_elt.toXml()
@ed
async def test_ice_candidates_add(self, host, client):
async def test_ice_candidates_add(self, host, client, monkeypatch):
"""local new ICE candidates are added, IQ is sent, bridge signal emitted"""
xep_0176 = XEP_0176(host)
@ -278,8 +278,9 @@ class TestXEP0176:
}
mock_session = self.create_mock_session(content_name)
xep_0176._j.get_session = MagicMock(return_value=mock_session)
xep_0176._j.build_action = MagicMock()
get_session = MagicMock(return_value=mock_session)
monkeypatch.setattr(xep_0176._j, "get_session", get_session)
monkeypatch.setattr(xep_0176._j, "build_action", MagicMock())
iq_elt = AsyncMock()
xep_0176._j.build_action.return_value = (iq_elt, None)
xep_0176.host.bridge.ice_restart = MagicMock()

27
tests/unit/test_plugin_xep_0272.py
View file

@ -47,7 +47,8 @@ def xep_0272(host) -> XEP_0272:
"XEP-0045": MagicMock(),
"XEP-0166": MagicMock(),
"XEP-0167": MagicMock(),
"XEP-0249": MagicMock(),
"XEP-0402": MagicMock(),
"XEP-0482": MagicMock(),
}
return XEP_0272(host)
@ -236,6 +237,30 @@ class TestXEP0272:
xep_0272.try_to_finish_preparation(client, room, muji_data)
mock_call_group_setup.assert_not_called()
@ed
async def test_call_group_end(self, host, client):
"""call_group_end removes the MUJI presence data and the room bookmark."""
xep_0272 = XEP_0272(host)
room_jid = jid.JID("room@example.com")
room = MagicMock()
room.roomJID = room_jid
muji_data = {"to_call": {jid.JID("peer@example.com")}, "done_preparing": True}
with (
patch.object(xep_0272._muc, "get_room", return_value=room),
patch.object(xep_0272, "get_muji_data", return_value=muji_data),
patch.object(
client.presence, "send", return_value=defer.succeed(None)
) as mock_presence_send,
patch.object(
xep_0272._bookmarks, "remove", return_value=defer.succeed(None)
) as mock_remove,
):
await xep_0272.call_group_end(client, room_jid)
mock_presence_send.assert_called_once()
assert muji_data == {"to_call": set()}
mock_remove.assert_called_once_with(client, room_jid)
@ed
async def test_call_group_data_set(self, host, client):
"""call_group_data_set sends correct presence and muji data."""

175
tests/unit/test_plugin_xep_0338.py
View file

@ -17,16 +17,25 @@
# 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_0338 import NS_JINGLE_GROUPING, XEP_0338
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 = XEP_0338(host)
xep_0338 = host.plugins["XEP-0338"]
call_data = {}
metadata = {}
@ -52,7 +61,7 @@ class TestXEP0338:
def test_generate_sdp(self, host):
"""'group' attribute in SDP is correctly generated"""
xep_0338 = XEP_0338(host)
xep_0338 = host.plugins["XEP-0338"]
session = {"metadata": {"group": {"BUNDLE": ["audio", "video"]}}}
sdp_lines = []
@ -64,7 +73,7 @@ class TestXEP0338:
def test_group_building(self, host, client):
"""<group> element are built from session in session init trigger"""
xep_0338 = XEP_0338(host)
xep_0338 = host.plugins["XEP-0338"]
session = {
"contents": {"audio": {}},
@ -92,7 +101,7 @@ class TestXEP0338:
async def test_group_parsing(self, host, client):
"""<group> elements are correctly parsed in jingle_handler trigger"""
xep_0338 = XEP_0338(host)
xep_0338 = host.plugins["XEP-0338"]
action = xep_0338._j.A_SESSION_INITIATE
session = {
@ -138,3 +147,159 @@ class TestXEP0338:
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]