# HG changeset patch # User Goffi # Date 1696452882 -7200 # Node ID b0f70be331c544652eb3c7e596fcd7b578b8d54f # Parent f387992d8e37412a7ce91ad113cfe561faebe53e tests: unit test for "Calls" plugins: rel 424 diff -r f387992d8e37 -r b0f70be331c5 tests/unit/test_plugin_calls.py --- /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 . + +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")