diff --git a/tests/unit/test_plugin_xep_0176.py b/tests/unit/test_plugin_xep_0176.py
new file mode 100644
index 00000000..ec7c3986
--- /dev/null
+++ b/tests/unit/test_plugin_xep_0176.py
@@ -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 .
+
+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(
+ """
+
+
+
+
+ """
+ )
+
+ 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):
+ """ 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):
+ """ 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("")
+
+ 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"]
+ )