Mercurial > libervia-desktop

comparison tests/unit/test_plugin_calls.py @ 500:b0f70be331c5

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
tests: unit test for "Calls" plugins: rel 424
author Goffi <goffi@goffi.org>
date Wed, 04 Oct 2023 22:54:42 +0200
parents
children f6b8300e8234
comparison
equal deleted inserted replaced
499:f387992d8e37 500:b0f70be331c5
1 #!/usr/bin/env python3
2
3 # Libervia: an XMPP client
4 # Copyright (C) 2009-2023 Jérôme Poisson (goffi@goffi.org)
5
6 # This program is free software: you can redistribute it and/or modify
7 # it under the terms of the GNU Affero General Public License as published by
8 # the Free Software Foundation, either version 3 of the License, or
9 # (at your option) any later version.
10
11 # This program is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 # GNU Affero General Public License for more details.
15
16 # You should have received a copy of the GNU Affero General Public License
17 # along with this program. If not, see <http://www.gnu.org/licenses/>.
18
19 from unittest.mock import AsyncMock, MagicMock
20
21 from gi.repository import Gst
22 from libervia.backend.core import exceptions
23 from libervia.backend.tools.common import data_format
24 import pytest
25
26 from libervia.desktop_kivy import G
27 from libervia.desktop_kivy.plugins import plugin_wid_calls
28
29
30 @pytest.fixture
31 def host(monkeypatch):
32 host = MagicMock()
33 host.a_bridge = AsyncMock()
34 monkeypatch.setattr(G, "_host", host, raising=False)
35 return host
36
37
38 @pytest.fixture(scope="function")
39 def webrtc():
40 """Fixture for WebRTC instantiation."""
41 host_mock = MagicMock()
42 profile = "test_profile"
43 instance = plugin_wid_calls.WebRTC(host_mock, profile)
44
45 instance._set_media_types = MagicMock()
46 instance.start_pipeline = MagicMock()
47 instance.webrtc = MagicMock()
48 instance.webrtc.emit = MagicMock()
49
50 instance.GstSdp_SDPMessage_new_from_text = MagicMock()
51 instance.GstWebRTC_WebRTCSessionDescription_new = MagicMock()
52 instance.Gst_Promise_new_with_change_func = MagicMock()
53
54 return instance
55
56
57 class TestWebRtc:
58 def test_get_payload_types(self, webrtc):
59 """The method can identify the correct payload types for video and audio."""
60 fake_sdpmsg = MagicMock()
61 fake_media = MagicMock()
62 fake_caps = MagicMock()
63 fake_structure = MagicMock()
64
65 # This side effect will return 'fake_video_encoding' first, then
66 # 'fake_audio_encoding'.
67 fake_structure.__getitem__.side_effect = [
68 "fake_video_encoding",
69 "fake_audio_encoding",
70 ]
71 fake_caps.get_structure.return_value = fake_structure
72 fake_media.get_format.side_effect = ["webrtc-datachannel", "10", "20"]
73 fake_media.get_caps_from_media.return_value = fake_caps
74 fake_sdpmsg.get_media.return_value = fake_media
75 fake_sdpmsg.medias_len.return_value = 1
76 fake_media.formats_len.return_value = 3
77
78 result = webrtc.get_payload_types(
79 fake_sdpmsg, "fake_video_encoding", "fake_audio_encoding"
80 )
81 expected_result = {"fake_video_encoding": 10, "fake_audio_encoding": 20}
82
83 assert result == expected_result
84
85 def test_on_accepted_call(self, webrtc):
86 """The method correctly sets the remote SDP upon acceptance of an outgoing call."""
87 sdp_str = "mock_sdp_string"
88 profile_str = "test_profile"
89
90 webrtc.on_accepted_call(sdp_str, profile_str)
91
92 # remote description must be set
93 assert webrtc.webrtc.emit.call_count == 1
94 assert webrtc.webrtc.emit.call_args[0][0] == "set-remote-description"
95
96 @pytest.mark.asyncio
97 async def test_answer_call(self, webrtc, monkeypatch):
98 """The method correctly answers an incoming call."""
99 mock_setup_call = AsyncMock()
100
101 def mock_get_payload_types(sdpmsg, video_encoding, audio_encoding):
102 return {"VP8": 96, "OPUS": 97}
103
104 monkeypatch.setattr(webrtc, "setup_call", mock_setup_call)
105 monkeypatch.setattr(webrtc, "get_payload_types", mock_get_payload_types)
106
107 sdp_str = "mock_sdp_string"
108 profile_str = "mock_profile"
109
110 await webrtc.answer_call(sdp_str, profile_str)
111
112 mock_setup_call.assert_called_once_with("responder", audio_pt=97, video_pt=96)
113
114 # remote description must be set
115 assert webrtc.webrtc.emit.call_count == 1
116 assert webrtc.webrtc.emit.call_args[0][0] == "set-remote-description"
117
118 def test_on_remote_decodebin_stream_video(self, webrtc, monkeypatch):
119 """The method correctly handles video streams from the remote decodebin."""
120 mock_pipeline = MagicMock()
121 monkeypatch.setattr(webrtc, "pipeline", mock_pipeline)
122
123 mock_pad = MagicMock()
124 mock_caps = MagicMock()
125 mock_structure = MagicMock()
126
127 mock_pad.has_current_caps.return_value = True
128 mock_pad.get_current_caps.return_value = mock_caps
129 mock_caps.__len__.return_value = 1
130 mock_caps.__getitem__.return_value = mock_structure
131 mock_structure.get_name.return_value = "video/x-h264"
132 # We use non-standard resolution as example to trigger the workaround
133 mock_structure.get_int.side_effect = lambda x: MagicMock(
134 value=990 if x == "width" else 557
135 )
136
137 webrtc.on_remote_decodebin_stream(None, mock_pad)
138
139 assert webrtc._remote_video_pad == mock_pad
140 mock_pipeline.add.assert_called()
141 mock_pipeline.set_state.assert_called()
142 mock_pad.link.assert_called()
143
144 def test_on_remote_decodebin_stream_audio(self, webrtc, monkeypatch):
145 """The method correctly handles audio streams from the remote decodebin."""
146 mock_pipeline = MagicMock()
147 monkeypatch.setattr(webrtc, "pipeline", mock_pipeline)
148
149 mock_pad = MagicMock()
150 mock_caps = MagicMock()
151 mock_structure = MagicMock()
152
153 mock_pad.has_current_caps.return_value = True
154 mock_pad.get_current_caps.return_value = mock_caps
155 mock_caps.__len__.return_value = 1
156 mock_caps.__getitem__.return_value = mock_structure
157 mock_structure.get_name.return_value = "audio/x-raw"
158
159 webrtc.on_remote_decodebin_stream(None, mock_pad)
160
161 mock_pipeline.add.assert_called()
162 mock_pipeline.set_state.assert_called()
163 mock_pad.link.assert_called()
164
165 @pytest.mark.asyncio
166 async def test_setup_call_correct_role(self, host, webrtc, monkeypatch):
167 """Roles are set in setup_call."""
168 monkeypatch.setattr(Gst, "parse_launch", MagicMock())
169 monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))
170
171 await webrtc.setup_call("initiator")
172 assert webrtc.role == "initiator"
173
174 await webrtc.setup_call("responder")
175 assert webrtc.role == "responder"
176
177 with pytest.raises(AssertionError):
178 await webrtc.setup_call("invalid_role")
179
180 @pytest.mark.asyncio
181 async def test_setup_call_test_mode(self, host, webrtc, monkeypatch):
182 """Test mode use fake video and audio in setup_call."""
183 monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))
184 monkeypatch.setattr(webrtc, "test_mode", True)
185 await webrtc.setup_call("initiator")
186 assert "videotestsrc" in webrtc.gst_pipe_desc
187 assert "audiotestsrc" in webrtc.gst_pipe_desc
188
189 @pytest.mark.asyncio
190 async def test_setup_call_normal_mode(self, host, webrtc, monkeypatch):
191 """Normal mode use real video and audio in setup_call."""
192 monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))
193 monkeypatch.setattr(webrtc, "test_mode", False)
194 await webrtc.setup_call("initiator")
195 assert "v4l2src" in webrtc.gst_pipe_desc
196 assert "pulsesrc" in webrtc.gst_pipe_desc
197
198 @pytest.mark.asyncio
199 async def test_setup_call_with_stun_and_turn(self, host, webrtc, monkeypatch):
200 """STUN and TURN server configurations are done in setup_call."""
201 mock_pipeline = MagicMock()
202 mock_parse_launch = MagicMock()
203 mock_parse_launch.return_value = mock_pipeline
204 monkeypatch.setattr(Gst, "parse_launch", mock_parse_launch)
205
206 mock_pipeline.get_by_name.return_value = webrtc.webrtc
207
208 mock_external_disco = [
209 {"type": "stun", "transport": "udp", "host": "stun.host", "port": "3478"},
210 {
211 "type": "turn",
212 "transport": "udp",
213 "host": "turn.host",
214 "port": "3478",
215 "username": "user",
216 "password": "pass",
217 },
218 ]
219
220 monkeypatch.setattr(
221 data_format, "deserialise", MagicMock(return_value=mock_external_disco)
222 )
223
224 mock_emit = AsyncMock()
225 monkeypatch.setattr(webrtc.webrtc, "emit", mock_emit)
226
227 mock_set_property = AsyncMock()
228 monkeypatch.setattr(webrtc.webrtc, "set_property", mock_set_property)
229
230 await webrtc.setup_call("initiator")
231
232 G.host.a_bridge.external_disco_get.assert_called_once_with("", webrtc.profile)
233 mock_set_property.assert_any_call("stun-server", "stun://stun.host:3478")
234 mock_emit.assert_called_once_with(
235 "add-turn-server", "turn://user:pass@turn.host:3478"
236 )
237
238 @pytest.mark.asyncio
239 async def test_setup_call_gstreamer_pipeline_failure(self, webrtc, monkeypatch):
240 """Test setup_call method handling Gstreamer pipeline failure."""
241 monkeypatch.setattr(Gst, "parse_launch", lambda _: None)
242 with pytest.raises(exceptions.InternalError):
243 await webrtc.setup_call("initiator")