Mercurial > libervia-desktop-kivy

comparison tests/unit/test_plugin_calls.py @ 509:f0ce49b360c8

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
calls: move webrtc code to core: WebRTC code which can be used in several frontends has been factorized and moved to common `frontends.tools`. Test have been updated consequently. rel 426
author Goffi <goffi@goffi.org>
date Wed, 01 Nov 2023 13:41:07 +0100
parents f6b8300e8234
children
comparison
equal deleted inserted replaced
508:d87b9a6b0b69 509:f0ce49b360c8
16 # You should have received a copy of the GNU Affero General Public License 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/>. 17 # along with this program. If not, see <http://www.gnu.org/licenses/>.
18 18
19 from unittest.mock import AsyncMock, MagicMock 19 from unittest.mock import AsyncMock, MagicMock
20 20
21 from gi.repository import Gst
22 from libervia.backend.core import exceptions
23 from libervia.backend.tools.common import data_format 21 from libervia.backend.tools.common import data_format
24 import pytest 22 import pytest
25 23
26 from libervia.desktop_kivy import G 24 from libervia.desktop_kivy import G
27 from libervia.desktop_kivy.plugins import plugin_wid_calls 25 from libervia.desktop_kivy.plugins import plugin_wid_calls
34 host.app.expand = lambda s: s 32 host.app.expand = lambda s: s
35 monkeypatch.setattr(G, "_host", host, raising=False) 33 monkeypatch.setattr(G, "_host", host, raising=False)
36 return host 34 return host
37 35
38 36
39 @pytest.fixture(scope="function")
40 def webrtc():
41 """Fixture for WebRTC instantiation."""
42 host_mock = MagicMock()
43 profile = "test_profile"
44 instance = plugin_wid_calls.WebRTC(host_mock, profile)
45
46 instance._set_media_types = MagicMock()
47 instance.start_pipeline = MagicMock()
48 instance.webrtc = MagicMock()
49 instance.webrtc.emit = MagicMock()
50
51 instance.GstSdp_SDPMessage_new_from_text = MagicMock()
52 instance.GstWebRTC_WebRTCSessionDescription_new = MagicMock()
53 instance.Gst_Promise_new_with_change_func = MagicMock()
54
55 return instance
56
57
58 @pytest.fixture 37 @pytest.fixture
59 def calls(monkeypatch, host): 38 def calls(monkeypatch, host):
60 """Fixture for Call UI instantiation.""" 39 """Fixture for Call UI instantiation."""
61 for attr in ("header_box", "local_video", "remote_video", "screen_manager"): 40 for attr in ("header_box", "local_video", "remote_video", "screen_manager"):
62 monkeypatch.setattr( 41 monkeypatch.setattr(plugin_wid_calls.Calls, attr, MagicMock())
63 plugin_wid_calls.Calls, 42 calls = plugin_wid_calls.Calls(host, "test_peer@example.org", ["test_profile"])
64 attr,
65 MagicMock()
66 )
67 calls = plugin_wid_calls.Calls(
68 host,
69 "test_peer@example.org",
70 ["test_profile"]
71 )
72 calls.jid_selector = MagicMock() 43 calls.jid_selector = MagicMock()
73 calls.header_input = MagicMock() 44 calls.header_input = MagicMock()
74 calls.header_input.text = "fake_jid@domain" 45 calls.header_input.text = "fake_jid@domain"
75 calls.webrtc = MagicMock() 46 calls.webrtc = MagicMock()
76 calls.webrtc.setup_call = AsyncMock() 47 calls.webrtc.setup_call = AsyncMock()
78 calls.end_call = AsyncMock() 49 calls.end_call = AsyncMock()
79 50
80 return calls 51 return calls
81 52
82 53
83 class TestWebRtc:
84 def test_get_payload_types(self, webrtc):
85 """The method can identify the correct payload types for video and audio."""
86 fake_sdpmsg = MagicMock()
87 fake_media = MagicMock()
88 fake_caps = MagicMock()
89 fake_structure = MagicMock()
90
91 # This side effect will return 'fake_video_encoding' first, then
92 # 'fake_audio_encoding'.
93 fake_structure.__getitem__.side_effect = [
94 "fake_video_encoding",
95 "fake_audio_encoding",
96 ]
97 fake_caps.get_structure.return_value = fake_structure
98 fake_media.get_format.side_effect = ["webrtc-datachannel", "10", "20"]
99 fake_media.get_caps_from_media.return_value = fake_caps
100 fake_sdpmsg.get_media.return_value = fake_media
101 fake_sdpmsg.medias_len.return_value = 1
102 fake_media.formats_len.return_value = 3
103
104 result = webrtc.get_payload_types(
105 fake_sdpmsg, "fake_video_encoding", "fake_audio_encoding"
106 )
107 expected_result = {"fake_video_encoding": 10, "fake_audio_encoding": 20}
108
109 assert result == expected_result
110
111 def test_on_accepted_call(self, webrtc):
112 """The method correctly sets the remote SDP upon acceptance of an outgoing call."""
113 sdp_str = "mock_sdp_string"
114 profile_str = "test_profile"
115
116 webrtc.on_accepted_call(sdp_str, profile_str)
117
118 # remote description must be set
119 assert webrtc.webrtc.emit.call_count == 1
120 assert webrtc.webrtc.emit.call_args[0][0] == "set-remote-description"
121
122 @pytest.mark.asyncio
123 async def test_answer_call(self, webrtc, monkeypatch):
124 """The method correctly answers an incoming call."""
125 mock_setup_call = AsyncMock()
126
127 def mock_get_payload_types(sdpmsg, video_encoding, audio_encoding):
128 return {"VP8": 96, "OPUS": 97}
129
130 monkeypatch.setattr(webrtc, "setup_call", mock_setup_call)
131 monkeypatch.setattr(webrtc, "get_payload_types", mock_get_payload_types)
132
133 sdp_str = "mock_sdp_string"
134 profile_str = "mock_profile"
135
136 await webrtc.answer_call(sdp_str, profile_str)
137
138 mock_setup_call.assert_called_once_with("responder", audio_pt=97, video_pt=96)
139
140 # remote description must be set
141 assert webrtc.webrtc.emit.call_count == 1
142 assert webrtc.webrtc.emit.call_args[0][0] == "set-remote-description"
143
144 def test_on_remote_decodebin_stream_video(self, webrtc, monkeypatch):
145 """The method correctly handles video 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 = "video/x-h264"
158 # We use non-standard resolution as example to trigger the workaround
159 mock_structure.get_int.side_effect = lambda x: MagicMock(
160 value=990 if x == "width" else 557
161 )
162
163 webrtc.on_remote_decodebin_stream(None, mock_pad)
164
165 assert webrtc._remote_video_pad == mock_pad
166 mock_pipeline.add.assert_called()
167 mock_pipeline.set_state.assert_called()
168 mock_pad.link.assert_called()
169
170 def test_on_remote_decodebin_stream_audio(self, webrtc, monkeypatch):
171 """The method correctly handles audio streams from the remote decodebin."""
172 mock_pipeline = MagicMock()
173 monkeypatch.setattr(webrtc, "pipeline", mock_pipeline)
174
175 mock_pad = MagicMock()
176 mock_caps = MagicMock()
177 mock_structure = MagicMock()
178
179 mock_pad.has_current_caps.return_value = True
180 mock_pad.get_current_caps.return_value = mock_caps
181 mock_caps.__len__.return_value = 1
182 mock_caps.__getitem__.return_value = mock_structure
183 mock_structure.get_name.return_value = "audio/x-raw"
184
185 webrtc.on_remote_decodebin_stream(None, mock_pad)
186
187 mock_pipeline.add.assert_called()
188 mock_pipeline.set_state.assert_called()
189 mock_pad.link.assert_called()
190
191 @pytest.mark.asyncio
192 async def test_setup_call_correct_role(self, host, webrtc, monkeypatch):
193 """Roles are set in setup_call."""
194 monkeypatch.setattr(Gst, "parse_launch", MagicMock())
195 monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))
196
197 await webrtc.setup_call("initiator")
198 assert webrtc.role == "initiator"
199
200 await webrtc.setup_call("responder")
201 assert webrtc.role == "responder"
202
203 with pytest.raises(AssertionError):
204 await webrtc.setup_call("invalid_role")
205
206 @pytest.mark.asyncio
207 async def test_setup_call_test_mode(self, host, webrtc, monkeypatch):
208 """Test mode use fake video and audio in setup_call."""
209 monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))
210 monkeypatch.setattr(webrtc, "test_mode", True)
211 await webrtc.setup_call("initiator")
212 assert "videotestsrc" in webrtc.gst_pipe_desc
213 assert "audiotestsrc" in webrtc.gst_pipe_desc
214
215 @pytest.mark.asyncio
216 async def test_setup_call_normal_mode(self, host, webrtc, monkeypatch):
217 """Normal mode use real video and audio in setup_call."""
218 monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))
219 monkeypatch.setattr(webrtc, "test_mode", False)
220 await webrtc.setup_call("initiator")
221 assert "v4l2src" in webrtc.gst_pipe_desc
222 assert "pulsesrc" in webrtc.gst_pipe_desc
223
224 @pytest.mark.asyncio
225 async def test_setup_call_with_stun_and_turn(self, host, webrtc, monkeypatch):
226 """STUN and TURN server configurations are done in setup_call."""
227 mock_pipeline = MagicMock()
228 mock_parse_launch = MagicMock()
229 mock_parse_launch.return_value = mock_pipeline
230 monkeypatch.setattr(Gst, "parse_launch", mock_parse_launch)
231
232 mock_pipeline.get_by_name.return_value = webrtc.webrtc
233
234 mock_external_disco = [
235 {"type": "stun", "transport": "udp", "host": "stun.host", "port": "3478"},
236 {
237 "type": "turn",
238 "transport": "udp",
239 "host": "turn.host",
240 "port": "3478",
241 "username": "user",
242 "password": "pass",
243 },
244 ]
245
246 monkeypatch.setattr(
247 data_format, "deserialise", MagicMock(return_value=mock_external_disco)
248 )
249
250 mock_emit = AsyncMock()
251 monkeypatch.setattr(webrtc.webrtc, "emit", mock_emit)
252
253 mock_set_property = AsyncMock()
254 monkeypatch.setattr(webrtc.webrtc, "set_property", mock_set_property)
255
256 await webrtc.setup_call("initiator")
257
258 G.host.a_bridge.external_disco_get.assert_called_once_with("", webrtc.profile)
259 mock_set_property.assert_any_call("stun-server", "stun://stun.host:3478")
260 mock_emit.assert_called_once_with(
261 "add-turn-server", "turn://user:pass@turn.host:3478"
262 )
263
264 @pytest.mark.asyncio
265 async def test_setup_call_gstreamer_pipeline_failure(self, webrtc, monkeypatch):
266 """Test setup_call method handling Gstreamer pipeline failure."""
267 monkeypatch.setattr(Gst, "parse_launch", lambda _: None)
268 with pytest.raises(exceptions.InternalError):
269 await webrtc.setup_call("initiator")
270
271
272 class TestCalls: 54 class TestCalls:
273
274 @pytest.mark.asyncio 55 @pytest.mark.asyncio
275 async def test_toggle_call_sid_none(self, monkeypatch, calls): 56 async def test_toggle_call_sid_none(self, monkeypatch, calls):
276 """Call is started when there is not sid set.""" 57 """Call is started when there is not sid set."""
277 monkeypatch.setattr(calls.webrtc, "sid", None) 58 monkeypatch.setattr(calls.webrtc, "sid", None)
278 59
291 72
292 calls.end_call.assert_called_once_with({"reason": "terminated"}, calls.profile) 73 calls.end_call.assert_called_once_with({"reason": "terminated"}, calls.profile)
293 host.a_bridge.call_end.assert_called_once_with("test_sid", "", calls.profile) 74 host.a_bridge.call_end.assert_called_once_with("test_sid", "", calls.profile)
294 assert calls.in_call == False 75 assert calls.in_call == False
295 76
296
297 @pytest.mark.asyncio 77 @pytest.mark.asyncio
298 async def test_on_incoming_call_sid_none(self, monkeypatch, host, calls): 78 async def test_on_incoming_call_sid_none(self, monkeypatch, host, calls):
299 """Incoming call is accepted if no ongoing call.""" 79 """Incoming call is accepted if no ongoing call."""
300 monkeypatch.setattr(calls.webrtc, "sid", None) 80 monkeypatch.setattr(calls.webrtc, "sid", None)
301 fake_action_id = "fake_action_id" 81 fake_action_id = "fake_action_id"
305 await calls.on_incoming_call(fake_action_data, fake_action_id, fake_profile) 85 await calls.on_incoming_call(fake_action_data, fake_action_id, fake_profile)
306 86
307 assert calls.in_call == True 87 assert calls.in_call == True
308 assert calls.webrtc.sid == "test_sid" 88 assert calls.webrtc.sid == "test_sid"
309 host.a_bridge.action_launch.assert_called_once_with( 89 host.a_bridge.action_launch.assert_called_once_with(
310 fake_action_id, 90 fake_action_id, data_format.serialise({"cancelled": False}), fake_profile
311 data_format.serialise({"cancelled": False}),
312 fake_profile
313 ) 91 )
314 92
315 @pytest.mark.asyncio 93 @pytest.mark.asyncio
316 async def test_on_incoming_call_sid_set(self, monkeypatch, host, calls): 94 async def test_on_incoming_call_sid_set(self, monkeypatch, host, calls):
317 """Incoming call is ignored if there's an ongoing call.""" 95 """Incoming call is ignored if there's an ongoing call."""
327 host.a_bridge.action_launch.assert_not_called() 105 host.a_bridge.action_launch.assert_not_called()
328 106
329 @pytest.mark.asyncio 107 @pytest.mark.asyncio
330 async def test_on_call_setup_initiator(self, calls): 108 async def test_on_call_setup_initiator(self, calls):
331 """Correct method called if role is 'initiator'.""" 109 """Correct method called if role is 'initiator'."""
332 setup_data = { 110 setup_data = {"role": "initiator", "sdp": "fake_sdp"}
333 "role": "initiator",
334 "sdp": "fake_sdp"
335 }
336 profile = "fake_profile" 111 profile = "fake_profile"
337 112
338 await calls.on_call_setup(setup_data, profile) 113 await calls.on_call_setup(setup_data, profile)
339 114
340 calls.webrtc.on_accepted_call.assert_called_once_with(setup_data["sdp"], profile) 115 calls.webrtc.on_accepted_call.assert_called_once_with(setup_data["sdp"], profile)
341 116
342 @pytest.mark.asyncio 117 @pytest.mark.asyncio
343 async def test_on_call_setup_responder(self, monkeypatch, calls): 118 async def test_on_call_setup_responder(self, monkeypatch, calls):
344 """Correct method called if role is 'responder'.""" 119 """Correct method called if role is 'responder'."""
345 monkeypatch.setattr( 120 monkeypatch.setattr(calls.webrtc, "answer_call", AsyncMock())
346 calls.webrtc, 121 setup_data = {"role": "responder", "sdp": "fake_sdp"}
347 "answer_call",
348 AsyncMock()
349 )
350 setup_data = {
351 "role": "responder",
352 "sdp": "fake_sdp"
353 }
354 profile = "fake_profile" 122 profile = "fake_profile"
355 123
356 await calls.on_call_setup(setup_data, profile) 124 await calls.on_call_setup(setup_data, profile)
357 125
358 calls.webrtc.answer_call.assert_called_once_with(setup_data["sdp"], profile) 126 calls.webrtc.answer_call.assert_called_once_with(setup_data["sdp"], profile)
359 calls.webrtc.on_accepted_call.assert_not_called() 127 calls.webrtc.on_accepted_call.assert_not_called()
360 128
361 @pytest.mark.asyncio 129 @pytest.mark.asyncio
362 async def test_on_call_setup_invalid_role(self, calls): 130 async def test_on_call_setup_invalid_role(self, calls):
363 """Nothing is called if role is neither 'initiator' nor 'responder'.""" 131 """Nothing is called if role is neither 'initiator' nor 'responder'."""
364 setup_data = { 132 setup_data = {"role": "invalid_role", "sdp": "fake_sdp"}
365 "role": "invalid_role",
366 "sdp": "fake_sdp"
367 }
368 profile = "fake_profile" 133 profile = "fake_profile"
369 134
370 await calls.on_call_setup(setup_data, profile) 135 await calls.on_call_setup(setup_data, profile)
371 136
372 calls.webrtc.answer_call.assert_not_called() 137 calls.webrtc.answer_call.assert_not_called()