--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/tests/unit/test_plugin_calls.py	Wed Oct 04 22:54:42 2023 +0200
@@ -0,0 +1,243 @@
+#!/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 gi.repository import Gst
+from libervia.backend.core import exceptions
+from libervia.backend.tools.common import data_format
+import pytest
+
+from libervia.desktop_kivy import G
+from libervia.desktop_kivy.plugins import plugin_wid_calls
+
+
+@pytest.fixture
+def host(monkeypatch):
+    host = MagicMock()
+    host.a_bridge = AsyncMock()
+    monkeypatch.setattr(G, "_host", host, raising=False)
+    return host
+
+
+@pytest.fixture(scope="function")
+def webrtc():
+    """Fixture for WebRTC instantiation."""
+    host_mock = MagicMock()
+    profile = "test_profile"
+    instance = plugin_wid_calls.WebRTC(host_mock, profile)
+
+    instance._set_media_types = MagicMock()
+    instance.start_pipeline = MagicMock()
+    instance.webrtc = MagicMock()
+    instance.webrtc.emit = MagicMock()
+
+    instance.GstSdp_SDPMessage_new_from_text = MagicMock()
+    instance.GstWebRTC_WebRTCSessionDescription_new = MagicMock()
+    instance.Gst_Promise_new_with_change_func = MagicMock()
+
+    return instance
+
+
+class TestWebRtc:
+    def test_get_payload_types(self, webrtc):
+        """The method can identify the correct payload types for video and audio."""
+        fake_sdpmsg = MagicMock()
+        fake_media = MagicMock()
+        fake_caps = MagicMock()
+        fake_structure = MagicMock()
+
+        # This side effect will return 'fake_video_encoding' first, then
+        # 'fake_audio_encoding'.
+        fake_structure.__getitem__.side_effect = [
+            "fake_video_encoding",
+            "fake_audio_encoding",
+        ]
+        fake_caps.get_structure.return_value = fake_structure
+        fake_media.get_format.side_effect = ["webrtc-datachannel", "10", "20"]
+        fake_media.get_caps_from_media.return_value = fake_caps
+        fake_sdpmsg.get_media.return_value = fake_media
+        fake_sdpmsg.medias_len.return_value = 1
+        fake_media.formats_len.return_value = 3
+
+        result = webrtc.get_payload_types(
+            fake_sdpmsg, "fake_video_encoding", "fake_audio_encoding"
+        )
+        expected_result = {"fake_video_encoding": 10, "fake_audio_encoding": 20}
+
+        assert result == expected_result
+
+    def test_on_accepted_call(self, webrtc):
+        """The method correctly sets the remote SDP upon acceptance of an outgoing call."""
+        sdp_str = "mock_sdp_string"
+        profile_str = "test_profile"
+
+        webrtc.on_accepted_call(sdp_str, profile_str)
+
+        # remote description must be set
+        assert webrtc.webrtc.emit.call_count == 1
+        assert webrtc.webrtc.emit.call_args[0][0] == "set-remote-description"
+
+    @pytest.mark.asyncio
+    async def test_answer_call(self, webrtc, monkeypatch):
+        """The method correctly answers an incoming call."""
+        mock_setup_call = AsyncMock()
+
+        def mock_get_payload_types(sdpmsg, video_encoding, audio_encoding):
+            return {"VP8": 96, "OPUS": 97}
+
+        monkeypatch.setattr(webrtc, "setup_call", mock_setup_call)
+        monkeypatch.setattr(webrtc, "get_payload_types", mock_get_payload_types)
+
+        sdp_str = "mock_sdp_string"
+        profile_str = "mock_profile"
+
+        await webrtc.answer_call(sdp_str, profile_str)
+
+        mock_setup_call.assert_called_once_with("responder", audio_pt=97, video_pt=96)
+
+        # remote description must be set
+        assert webrtc.webrtc.emit.call_count == 1
+        assert webrtc.webrtc.emit.call_args[0][0] == "set-remote-description"
+
+    def test_on_remote_decodebin_stream_video(self, webrtc, monkeypatch):
+        """The method correctly handles video streams from the remote decodebin."""
+        mock_pipeline = MagicMock()
+        monkeypatch.setattr(webrtc, "pipeline", mock_pipeline)
+
+        mock_pad = MagicMock()
+        mock_caps = MagicMock()
+        mock_structure = MagicMock()
+
+        mock_pad.has_current_caps.return_value = True
+        mock_pad.get_current_caps.return_value = mock_caps
+        mock_caps.__len__.return_value = 1
+        mock_caps.__getitem__.return_value = mock_structure
+        mock_structure.get_name.return_value = "video/x-h264"
+        # We use non-standard resolution as example to trigger the workaround
+        mock_structure.get_int.side_effect = lambda x: MagicMock(
+            value=990 if x == "width" else 557
+        )
+
+        webrtc.on_remote_decodebin_stream(None, mock_pad)
+
+        assert webrtc._remote_video_pad == mock_pad
+        mock_pipeline.add.assert_called()
+        mock_pipeline.set_state.assert_called()
+        mock_pad.link.assert_called()
+
+    def test_on_remote_decodebin_stream_audio(self, webrtc, monkeypatch):
+        """The method correctly handles audio streams from the remote decodebin."""
+        mock_pipeline = MagicMock()
+        monkeypatch.setattr(webrtc, "pipeline", mock_pipeline)
+
+        mock_pad = MagicMock()
+        mock_caps = MagicMock()
+        mock_structure = MagicMock()
+
+        mock_pad.has_current_caps.return_value = True
+        mock_pad.get_current_caps.return_value = mock_caps
+        mock_caps.__len__.return_value = 1
+        mock_caps.__getitem__.return_value = mock_structure
+        mock_structure.get_name.return_value = "audio/x-raw"
+
+        webrtc.on_remote_decodebin_stream(None, mock_pad)
+
+        mock_pipeline.add.assert_called()
+        mock_pipeline.set_state.assert_called()
+        mock_pad.link.assert_called()
+
+    @pytest.mark.asyncio
+    async def test_setup_call_correct_role(self, host, webrtc, monkeypatch):
+        """Roles are set in setup_call."""
+        monkeypatch.setattr(Gst, "parse_launch", MagicMock())
+        monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))
+
+        await webrtc.setup_call("initiator")
+        assert webrtc.role == "initiator"
+
+        await webrtc.setup_call("responder")
+        assert webrtc.role == "responder"
+
+        with pytest.raises(AssertionError):
+            await webrtc.setup_call("invalid_role")
+
+    @pytest.mark.asyncio
+    async def test_setup_call_test_mode(self, host, webrtc, monkeypatch):
+        """Test mode use fake video and audio in setup_call."""
+        monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))
+        monkeypatch.setattr(webrtc, "test_mode", True)
+        await webrtc.setup_call("initiator")
+        assert "videotestsrc" in webrtc.gst_pipe_desc
+        assert "audiotestsrc" in webrtc.gst_pipe_desc
+
+    @pytest.mark.asyncio
+    async def test_setup_call_normal_mode(self, host, webrtc, monkeypatch):
+        """Normal mode use real video and audio in setup_call."""
+        monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))
+        monkeypatch.setattr(webrtc, "test_mode", False)
+        await webrtc.setup_call("initiator")
+        assert "v4l2src" in webrtc.gst_pipe_desc
+        assert "pulsesrc" in webrtc.gst_pipe_desc
+
+    @pytest.mark.asyncio
+    async def test_setup_call_with_stun_and_turn(self, host, webrtc, monkeypatch):
+        """STUN and TURN server configurations are done in setup_call."""
+        mock_pipeline = MagicMock()
+        mock_parse_launch = MagicMock()
+        mock_parse_launch.return_value = mock_pipeline
+        monkeypatch.setattr(Gst, "parse_launch", mock_parse_launch)
+
+        mock_pipeline.get_by_name.return_value = webrtc.webrtc
+
+        mock_external_disco = [
+            {"type": "stun", "transport": "udp", "host": "stun.host", "port": "3478"},
+            {
+                "type": "turn",
+                "transport": "udp",
+                "host": "turn.host",
+                "port": "3478",
+                "username": "user",
+                "password": "pass",
+            },
+        ]
+
+        monkeypatch.setattr(
+            data_format, "deserialise", MagicMock(return_value=mock_external_disco)
+        )
+
+        mock_emit = AsyncMock()
+        monkeypatch.setattr(webrtc.webrtc, "emit", mock_emit)
+
+        mock_set_property = AsyncMock()
+        monkeypatch.setattr(webrtc.webrtc, "set_property", mock_set_property)
+
+        await webrtc.setup_call("initiator")
+
+        G.host.a_bridge.external_disco_get.assert_called_once_with("", webrtc.profile)
+        mock_set_property.assert_any_call("stun-server", "stun://stun.host:3478")
+        mock_emit.assert_called_once_with(
+            "add-turn-server", "turn://user:pass@turn.host:3478"
+        )
+
+    @pytest.mark.asyncio
+    async def test_setup_call_gstreamer_pipeline_failure(self, webrtc, monkeypatch):
+        """Test setup_call method handling Gstreamer pipeline failure."""
+        monkeypatch.setattr(Gst, "parse_launch", lambda _: None)
+        with pytest.raises(exceptions.InternalError):
+            await webrtc.setup_call("initiator")