ab26a56c1d
- 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
94 lines
3.6 KiB
Python
94 lines
3.6 KiB
Python
#!/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()
|