Mercurial > libervia-desktop

comparison tests/unit/test_plugin_calls.py @ 507:f6b8300e8234

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
tests (calls): add tests for the "Calls" class: rel 425
author Goffi <goffi@goffi.org>
date Wed, 25 Oct 2023 15:29:12 +0200
parents b0f70be331c5
children f0ce49b360c8
comparison
equal deleted inserted replaced
506:0480f883f0a6 507:f6b8300e8234
29 29
30 @pytest.fixture 30 @pytest.fixture
31 def host(monkeypatch): 31 def host(monkeypatch):
32 host = MagicMock() 32 host = MagicMock()
33 host.a_bridge = AsyncMock() 33 host.a_bridge = AsyncMock()
34 host.app.expand = lambda s: s
34 monkeypatch.setattr(G, "_host", host, raising=False) 35 monkeypatch.setattr(G, "_host", host, raising=False)
35 return host 36 return host
36 37
37 38
38 @pytest.fixture(scope="function") 39 @pytest.fixture(scope="function")
50 instance.GstSdp_SDPMessage_new_from_text = MagicMock() 51 instance.GstSdp_SDPMessage_new_from_text = MagicMock()
51 instance.GstWebRTC_WebRTCSessionDescription_new = MagicMock() 52 instance.GstWebRTC_WebRTCSessionDescription_new = MagicMock()
52 instance.Gst_Promise_new_with_change_func = MagicMock() 53 instance.Gst_Promise_new_with_change_func = MagicMock()
53 54
54 return instance 55 return instance
56
57
58 @pytest.fixture
59 def calls(monkeypatch, host):
60 """Fixture for Call UI instantiation."""
61 for attr in ("header_box", "local_video", "remote_video", "screen_manager"):
62 monkeypatch.setattr(
63 plugin_wid_calls.Calls,
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()
73 calls.header_input = MagicMock()
74 calls.header_input.text = "fake_jid@domain"
75 calls.webrtc = MagicMock()
76 calls.webrtc.setup_call = AsyncMock()
77 calls.webrtc.start_pipeline = MagicMock()
78 calls.end_call = AsyncMock()
79
80 return calls
55 81
56 82
57 class TestWebRtc: 83 class TestWebRtc:
58 def test_get_payload_types(self, webrtc): 84 def test_get_payload_types(self, webrtc):
59 """The method can identify the correct payload types for video and audio.""" 85 """The method can identify the correct payload types for video and audio."""
239 async def test_setup_call_gstreamer_pipeline_failure(self, webrtc, monkeypatch): 265 async def test_setup_call_gstreamer_pipeline_failure(self, webrtc, monkeypatch):
240 """Test setup_call method handling Gstreamer pipeline failure.""" 266 """Test setup_call method handling Gstreamer pipeline failure."""
241 monkeypatch.setattr(Gst, "parse_launch", lambda _: None) 267 monkeypatch.setattr(Gst, "parse_launch", lambda _: None)
242 with pytest.raises(exceptions.InternalError): 268 with pytest.raises(exceptions.InternalError):
243 await webrtc.setup_call("initiator") 269 await webrtc.setup_call("initiator")
270
271
272 class TestCalls:
273
274 @pytest.mark.asyncio
275 async def test_toggle_call_sid_none(self, monkeypatch, calls):
276 """Call is started when there is not sid set."""
277 monkeypatch.setattr(calls.webrtc, "sid", None)
278
279 await calls.toggle_call()
280
281 calls.webrtc.setup_call.assert_called_once_with("initiator")
282 calls.webrtc.start_pipeline.assert_called_once()
283 assert calls.in_call == True
284
285 @pytest.mark.asyncio
286 async def test_toggle_call_sid_set(self, monkeypatch, host, calls):
287 """Call is ended when a sid is set"""
288 monkeypatch.setattr(calls.webrtc, "sid", "test_sid")
289
290 await calls.toggle_call()
291
292 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)
294 assert calls.in_call == False
295
296
297 @pytest.mark.asyncio
298 async def test_on_incoming_call_sid_none(self, monkeypatch, host, calls):
299 """Incoming call is accepted if no ongoing call."""
300 monkeypatch.setattr(calls.webrtc, "sid", None)
301 fake_action_id = "fake_action_id"
302 fake_action_data = {"session_id": "test_sid"}
303 fake_profile = "fake_profile"
304
305 await calls.on_incoming_call(fake_action_data, fake_action_id, fake_profile)
306
307 assert calls.in_call == True
308 assert calls.webrtc.sid == "test_sid"
309 host.a_bridge.action_launch.assert_called_once_with(
310 fake_action_id,
311 data_format.serialise({"cancelled": False}),
312 fake_profile
313 )
314
315 @pytest.mark.asyncio
316 async def test_on_incoming_call_sid_set(self, monkeypatch, host, calls):
317 """Incoming call is ignored if there's an ongoing call."""
318 monkeypatch.setattr(calls.webrtc, "sid", "fake_old_sid")
319 fake_action_id = "fake_action_id"
320 fake_action_data = {"session_id": "test_sid_new"}
321 fake_profile = "fake_profile"
322
323 await calls.on_incoming_call(fake_action_data, fake_action_id, fake_profile)
324
325 # Ensuring the state hasn't been changed to True
326 assert calls.in_call == False
327 host.a_bridge.action_launch.assert_not_called()
328
329 @pytest.mark.asyncio
330 async def test_on_call_setup_initiator(self, calls):
331 """Correct method called if role is 'initiator'."""
332 setup_data = {
333 "role": "initiator",
334 "sdp": "fake_sdp"
335 }
336 profile = "fake_profile"
337
338 await calls.on_call_setup(setup_data, profile)
339
340 calls.webrtc.on_accepted_call.assert_called_once_with(setup_data["sdp"], profile)
341
342 @pytest.mark.asyncio
343 async def test_on_call_setup_responder(self, monkeypatch, calls):
344 """Correct method called if role is 'responder'."""
345 monkeypatch.setattr(
346 calls.webrtc,
347 "answer_call",
348 AsyncMock()
349 )
350 setup_data = {
351 "role": "responder",
352 "sdp": "fake_sdp"
353 }
354 profile = "fake_profile"
355
356 await calls.on_call_setup(setup_data, profile)
357
358 calls.webrtc.answer_call.assert_called_once_with(setup_data["sdp"], profile)
359 calls.webrtc.on_accepted_call.assert_not_called()
360
361 @pytest.mark.asyncio
362 async def test_on_call_setup_invalid_role(self, calls):
363 """Nothing is called if role is neither 'initiator' nor 'responder'."""
364 setup_data = {
365 "role": "invalid_role",
366 "sdp": "fake_sdp"
367 }
368 profile = "fake_profile"
369
370 await calls.on_call_setup(setup_data, profile)
371
372 calls.webrtc.answer_call.assert_not_called()
373 calls.webrtc.on_accepted_call.assert_not_called()