tests (unit): tests for plugin XEP-0176:
fix 419
This commit is contained in:
Goffi
parent
5e6ca719ec
commit
bd675233cb
1 changed files with 300 additions and 0 deletions
300
tests/unit/test_plugin_xep_0176.py
Normal file
300
tests/unit/test_plugin_xep_0176.py
Normal file
|
|
@ -0,0 +1,300 @@
|
|||
#!/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/>.
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
from pytest import fixture
|
||||
from pytest_twisted import ensureDeferred as ed
|
||||
|
||||
from sat.plugins.plugin_xep_0166 import XEP_0166
|
||||
from sat.plugins.plugin_xep_0176 import NS_JINGLE_ICE_UDP, XEP_0176
|
||||
from sat.tools import xml_tools
|
||||
|
||||
|
||||
@fixture(autouse=True)
|
||||
def no_transport_register(monkeypatch):
|
||||
"""Do not register the transport in XEP-0166"""
|
||||
monkeypatch.setattr(XEP_0166, "register_transport", lambda *a, **kw: None)
|
||||
|
||||
|
||||
class TestXEP0176:
|
||||
def create_mock_session(self, content_name):
|
||||
return {
|
||||
"id": "test-session-id",
|
||||
"contents": {
|
||||
content_name: {
|
||||
"application_data": {"media": "audio"},
|
||||
"transport_data": {
|
||||
"local_ice_data": {
|
||||
"ufrag": "testufrag",
|
||||
"pwd": "testpwd",
|
||||
"candidates": [
|
||||
{
|
||||
"id": "candidate_1",
|
||||
"component_id": 1,
|
||||
"foundation": "1",
|
||||
"address": "192.0.2.1",
|
||||
"port": 1234,
|
||||
"priority": 1,
|
||||
"transport": "udp",
|
||||
"type": "host",
|
||||
}
|
||||
],
|
||||
}
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
def test_build_transport(self, host, monkeypatch):
|
||||
"""ICE data is correctly transformed into transport element"""
|
||||
xep_0176 = XEP_0176(host)
|
||||
|
||||
ice_data = {
|
||||
"ufrag": "user1",
|
||||
"pwd": "password1",
|
||||
"candidates": [
|
||||
{
|
||||
"component_id": 1,
|
||||
"foundation": "1",
|
||||
"address": "192.168.0.1",
|
||||
"port": 1234,
|
||||
"priority": 100,
|
||||
"transport": "udp",
|
||||
"type": "host",
|
||||
"generation": "0",
|
||||
"network": "0",
|
||||
},
|
||||
{
|
||||
"component_id": 2,
|
||||
"foundation": "1",
|
||||
"address": "192.168.0.2",
|
||||
"port": 5678,
|
||||
"priority": 100,
|
||||
"transport": "udp",
|
||||
"type": "host",
|
||||
"generation": "0",
|
||||
"network": "0",
|
||||
"rel_addr": "10.0.0.1",
|
||||
"rel_port": 9012,
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
transport_elt = xep_0176.build_transport(ice_data)
|
||||
|
||||
assert transport_elt.name == "transport"
|
||||
assert transport_elt.uri == NS_JINGLE_ICE_UDP
|
||||
assert transport_elt.getAttribute("ufrag") == "user1"
|
||||
assert transport_elt.getAttribute("pwd") == "password1"
|
||||
|
||||
candidates = list(transport_elt.elements(NS_JINGLE_ICE_UDP, "candidate"))
|
||||
|
||||
assert len(candidates) == len(ice_data["candidates"])
|
||||
|
||||
for i, candidate_elt in enumerate(candidates):
|
||||
ice_candidate = ice_data["candidates"][i]
|
||||
assert (
|
||||
int(candidate_elt.getAttribute("component"))
|
||||
== ice_candidate["component_id"]
|
||||
)
|
||||
assert candidate_elt.getAttribute("foundation") == ice_candidate["foundation"]
|
||||
assert candidate_elt.getAttribute("ip") == ice_candidate["address"]
|
||||
assert int(candidate_elt.getAttribute("port")) == ice_candidate["port"]
|
||||
assert (
|
||||
int(candidate_elt.getAttribute("priority")) == ice_candidate["priority"]
|
||||
)
|
||||
assert candidate_elt.getAttribute("protocol") == ice_candidate["transport"]
|
||||
assert candidate_elt.getAttribute("type") == ice_candidate["type"]
|
||||
assert candidate_elt.getAttribute("generation") == str(
|
||||
ice_candidate.get("generation", "0")
|
||||
)
|
||||
assert candidate_elt.getAttribute("network") == str(
|
||||
ice_candidate.get("network", "0")
|
||||
)
|
||||
|
||||
if "rel_addr" in ice_candidate:
|
||||
assert candidate_elt.getAttribute("rel-addr") == ice_candidate["rel_addr"]
|
||||
assert (
|
||||
int(candidate_elt.getAttribute("rel-port"))
|
||||
== ice_candidate["rel_port"]
|
||||
)
|
||||
else:
|
||||
assert candidate_elt.getAttribute("rel-addr") is None
|
||||
assert candidate_elt.getAttribute("rel-port") is None
|
||||
|
||||
def test_parse_transport(self, host):
|
||||
"""Transport element is correctly parsed into ICE data"""
|
||||
xep_0176 = XEP_0176(host)
|
||||
|
||||
transport_elt = xml_tools.parse(
|
||||
"""
|
||||
<transport xmlns="urn:xmpp:jingle:transports:ice-udp:1"
|
||||
pwd="password1"
|
||||
ufrag="user1">
|
||||
<candidate component="1"
|
||||
foundation="1"
|
||||
generation="0"
|
||||
id="uuid1"
|
||||
ip="192.168.0.1"
|
||||
network="0"
|
||||
port="1234"
|
||||
priority="100"
|
||||
protocol="udp"
|
||||
type="host" />
|
||||
<candidate component="2"
|
||||
foundation="1"
|
||||
generation="0"
|
||||
id="uuid2"
|
||||
ip="192.168.0.2"
|
||||
network="0"
|
||||
port="5678"
|
||||
priority="100"
|
||||
protocol="udp"
|
||||
type="host"
|
||||
rel-addr="10.0.0.1"
|
||||
rel-port="9012" />
|
||||
</transport>
|
||||
"""
|
||||
)
|
||||
|
||||
ice_data = xep_0176.parse_transport(transport_elt)
|
||||
|
||||
assert transport_elt.getAttribute("ufrag") == "user1"
|
||||
assert transport_elt.getAttribute("pwd") == "password1"
|
||||
|
||||
candidates = list(transport_elt.elements(NS_JINGLE_ICE_UDP, "candidate"))
|
||||
assert len(candidates) == len(ice_data["candidates"])
|
||||
|
||||
for i, candidate_elt in enumerate(candidates):
|
||||
ice_candidate = ice_data["candidates"][i]
|
||||
assert (
|
||||
int(candidate_elt.getAttribute("component"))
|
||||
== ice_candidate["component_id"]
|
||||
)
|
||||
assert candidate_elt.getAttribute("foundation") == ice_candidate["foundation"]
|
||||
assert candidate_elt.getAttribute("ip") == ice_candidate["address"]
|
||||
assert int(candidate_elt.getAttribute("port")) == ice_candidate["port"]
|
||||
assert (
|
||||
int(candidate_elt.getAttribute("priority")) == ice_candidate["priority"]
|
||||
)
|
||||
assert candidate_elt.getAttribute("protocol") == ice_candidate["transport"]
|
||||
assert candidate_elt.getAttribute("type") == ice_candidate["type"]
|
||||
assert candidate_elt.getAttribute("generation") == str(
|
||||
ice_candidate.get("generation", "0")
|
||||
)
|
||||
assert candidate_elt.getAttribute("network") == str(
|
||||
ice_candidate.get("network", "0")
|
||||
)
|
||||
|
||||
if "rel_addr" in ice_candidate:
|
||||
assert candidate_elt.getAttribute("rel-addr") == ice_candidate["rel_addr"]
|
||||
assert (
|
||||
int(candidate_elt.getAttribute("rel-port"))
|
||||
== ice_candidate["rel_port"]
|
||||
)
|
||||
else:
|
||||
assert candidate_elt.getAttribute("rel-addr") is None
|
||||
assert candidate_elt.getAttribute("rel-port") is None
|
||||
|
||||
@ed
|
||||
async def test_jingle_session_init(self, host, client):
|
||||
"""<transport/> element is built on initiator side during init"""
|
||||
xep_0176 = XEP_0176(host)
|
||||
|
||||
content_name = "test-content"
|
||||
session = self.create_mock_session(content_name)
|
||||
|
||||
transport_elt = await xep_0176.jingle_session_init(client, session, content_name)
|
||||
|
||||
expected_transport_elt = xep_0176.build_transport(
|
||||
session["contents"][content_name]["transport_data"]["local_ice_data"]
|
||||
)
|
||||
|
||||
assert transport_elt.toXml() == expected_transport_elt.toXml()
|
||||
|
||||
@ed
|
||||
async def test_jingle_handler(self, host, client):
|
||||
"""<transport/> element is built on responder side during init"""
|
||||
xep_0176 = XEP_0176(host)
|
||||
|
||||
content_name = "test-content"
|
||||
action = "session-initiate"
|
||||
session = self.create_mock_session(content_name)
|
||||
transport_elt = xml_tools.parse("<transport/>")
|
||||
|
||||
returned_transport_elt = await xep_0176.jingle_handler(
|
||||
client, action, session, content_name, transport_elt
|
||||
)
|
||||
|
||||
expected_transport_elt = xep_0176.build_transport(
|
||||
session["contents"][content_name]["transport_data"]["local_ice_data"]
|
||||
)
|
||||
|
||||
assert returned_transport_elt.toXml() == expected_transport_elt.toXml()
|
||||
|
||||
@ed
|
||||
async def test_ice_candidates_add(self, host, client):
|
||||
"""local new ICE candidates are added, IQ is sent, bridge signal emitted"""
|
||||
xep_0176 = XEP_0176(host)
|
||||
|
||||
content_name = "test-content"
|
||||
session_id = "test-session-id"
|
||||
media_ice_data_s = {
|
||||
"audio": {
|
||||
# we use different "ufrag" and "pwd" than in "create_mock_session" to
|
||||
# trigger an ICE restart
|
||||
"ufrag": "new_testufrag",
|
||||
"pwd": "new_testpwd",
|
||||
"candidates": [
|
||||
{
|
||||
"component_id": 2,
|
||||
"foundation": "2",
|
||||
"address": "192.0.2.2",
|
||||
"port": 1235,
|
||||
"priority": 2,
|
||||
"transport": "udp",
|
||||
"type": "host",
|
||||
}
|
||||
],
|
||||
}
|
||||
}
|
||||
|
||||
mock_session = self.create_mock_session(content_name)
|
||||
xep_0176._j.get_session = MagicMock(return_value=mock_session)
|
||||
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()
|
||||
|
||||
await xep_0176.ice_candidates_add(client, session_id, media_ice_data_s)
|
||||
|
||||
xep_0176._j.build_action.assert_called()
|
||||
iq_elt.send.assert_called()
|
||||
xep_0176.host.bridge.ice_restart.assert_called()
|
||||
|
||||
# Checking that local ICE data is updated correctly
|
||||
updated_local_ice_data = mock_session["contents"][content_name]["transport_data"][
|
||||
"local_ice_data"
|
||||
]
|
||||
assert updated_local_ice_data["ufrag"] == "new_testufrag"
|
||||
assert updated_local_ice_data["pwd"] == "new_testpwd"
|
||||
assert (
|
||||
updated_local_ice_data["candidates"]
|
||||
== media_ice_data_s["audio"]["candidates"]
|
||||
)
|
||||
Loading…
Reference in a new issue