Mercurial > libervia-backend

comparison tests/unit/frontends/test_webrtc.py @ 4140:13dd5660c28f

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
tests (unit/frontends): tests for webrtc implementation: rel 426
author Goffi <goffi@goffi.org>
date Wed, 01 Nov 2023 14:04:25 +0100
parents
children dfccc90cacc6
comparison
equal deleted inserted replaced
4139:6745c6bd4c7a 4140:13dd5660c28f
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 from unittest.mock import AsyncMock, MagicMock
19
20 from gi.repository import Gst
21 import pytest
22
23 from libervia.backend.core import exceptions
24 from libervia.backend.tools.common import data_format
25 from libervia.frontends.tools import webrtc as webrtc_mod
26
27
28
29 @pytest.fixture
30 def host(monkeypatch):
31 host = MagicMock()
32 host.bridge = AsyncMock()
33 host.app.expand = lambda s: s
34 return host
35
36 @pytest.fixture(scope="function")
37 def webrtc(host):
38 """Fixture for WebRTC instantiation."""
39 profile = "test_profile"
40 instance = webrtc_mod.WebRTC(host.bridge, profile)
41
42 instance._set_media_types = MagicMock()
43 instance.start_pipeline = MagicMock()
44 instance.link_element_or_pad = MagicMock()
45 instance.webrtcbin = MagicMock()
46 instance.webrtcbin.emit = MagicMock()
47
48 instance.GstSdp_SDPMessage_new_from_text = MagicMock()
49 instance.GstWebRTC_WebRTCSessionDescription_new = MagicMock()
50 instance.Gst_Promise_new_with_change_func = MagicMock()
51
52 return instance
53
54
55 class TestWebRtc:
56 def test_get_payload_types(self, webrtc):
57 """The method can identify the correct payload types for video and audio."""
58 fake_sdpmsg = MagicMock()
59 fake_media = MagicMock()
60 fake_caps = MagicMock()
61 fake_structure = MagicMock()
62
63 # This side effect will return 'fake_video_encoding' first, then
64 # 'fake_audio_encoding'.
65 fake_structure.__getitem__.side_effect = [
66 "fake_video_encoding",
67 "fake_audio_encoding",
68 ]
69 fake_caps.get_structure.return_value = fake_structure
70 fake_media.get_format.side_effect = ["webrtc-datachannel", "10", "20"]
71 fake_media.get_caps_from_media.return_value = fake_caps
72 fake_sdpmsg.get_media.return_value = fake_media
73 fake_sdpmsg.medias_len.return_value = 1
74 fake_media.formats_len.return_value = 3
75
76 result = webrtc.get_payload_types(
77 fake_sdpmsg, "fake_video_encoding", "fake_audio_encoding"
78 )
79 expected_result = {"fake_video_encoding": 10, "fake_audio_encoding": 20}
80
81 assert result == expected_result
82
83 def test_on_accepted_call(self, webrtc):
84 """The method correctly sets the remote SDP upon acceptance of an outgoing call."""
85 sdp_str = "mock_sdp_string"
86 profile_str = "test_profile"
87
88 webrtc.on_accepted_call(sdp_str, profile_str)
89
90 # remote description must be set
91 assert webrtc.webrtcbin.emit.call_count == 1
92 assert webrtc.webrtcbin.emit.call_args[0][0] == "set-remote-description"
93
94 @pytest.mark.asyncio
95 async def test_answer_call(self, webrtc, monkeypatch):
96 """The method correctly answers an incoming call."""
97 mock_setup_call = AsyncMock()
98
99 def mock_get_payload_types(sdpmsg, video_encoding, audio_encoding):
100 return {"VP8": 96, "OPUS": 97}
101
102 monkeypatch.setattr(webrtc, "setup_call", mock_setup_call)
103 monkeypatch.setattr(webrtc, "get_payload_types", mock_get_payload_types)
104
105 sdp_str = "mock_sdp_string"
106 profile_str = "mock_profile"
107
108 await webrtc.answer_call(sdp_str, profile_str)
109
110 mock_setup_call.assert_called_once_with("responder", audio_pt=97, video_pt=96)
111
112 # remote description must be set
113 assert webrtc.webrtcbin.emit.call_count == 1
114 assert webrtc.webrtcbin.emit.call_args[0][0] == "set-remote-description"
115
116 def test_on_remote_decodebin_stream_video(self, webrtc, monkeypatch):
117 """The method correctly handles video streams from the remote decodebin."""
118 mock_pipeline = MagicMock()
119 monkeypatch.setattr(webrtc, "pipeline", mock_pipeline)
120
121 mock_pad = MagicMock()
122 mock_caps = MagicMock()
123 mock_structure = MagicMock()
124
125 mock_pad.has_current_caps.return_value = True
126 mock_pad.get_current_caps.return_value = mock_caps
127 mock_caps.__len__.return_value = 1
128 mock_caps.__getitem__.return_value = mock_structure
129 mock_structure.get_name.return_value = "video/x-h264"
130 # We use non-standard resolution as example to trigger the workaround
131 mock_structure.get_int.side_effect = lambda x: MagicMock(
132 value=990 if x == "width" else 557
133 )
134
135 webrtc.on_remote_decodebin_stream(None, mock_pad)
136
137 assert webrtc._remote_video_pad == mock_pad
138 mock_pipeline.add.assert_called()
139 mock_pad.link.assert_called()
140
141 def test_on_remote_decodebin_stream_audio(self, webrtc, monkeypatch):
142 """The method correctly handles audio streams from the remote decodebin."""
143 mock_pipeline = MagicMock()
144 monkeypatch.setattr(webrtc, "pipeline", mock_pipeline)
145
146 mock_pad = MagicMock()
147 mock_caps = MagicMock()
148 mock_structure = MagicMock()
149
150 mock_pad.has_current_caps.return_value = True
151 mock_pad.get_current_caps.return_value = mock_caps
152 mock_caps.__len__.return_value = 1
153 mock_caps.__getitem__.return_value = mock_structure
154 mock_structure.get_name.return_value = "audio/x-raw"
155
156 webrtc.on_remote_decodebin_stream(None, mock_pad)
157
158 mock_pipeline.add.assert_called()
159 mock_pad.link.assert_called()
160
161 @pytest.mark.asyncio
162 async def test_setup_call_correct_role(self, host, webrtc, monkeypatch):
163 """Roles are set in setup_call."""
164 monkeypatch.setattr(Gst, "parse_launch", MagicMock())
165 monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))
166
167 await webrtc.setup_call("initiator")
168 assert webrtc.role == "initiator"
169
170 await webrtc.setup_call("responder")
171 assert webrtc.role == "responder"
172
173 with pytest.raises(AssertionError):
174 await webrtc.setup_call("invalid_role")
175
176 @pytest.mark.asyncio
177 async def test_setup_call_test_mode(self, host, webrtc, monkeypatch):
178 """Test mode use fake video and audio in setup_call."""
179 monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))
180 monkeypatch.setattr(webrtc, "sources", webrtc_mod.SINKS_TEST)
181 await webrtc.setup_call("initiator")
182 assert "videotestsrc" in webrtc.gst_pipe_desc
183 assert "audiotestsrc" in webrtc.gst_pipe_desc
184
185 @pytest.mark.asyncio
186 async def test_setup_call_normal_mode(self, host, webrtc, monkeypatch):
187 """Normal mode use real video and audio in setup_call."""
188 monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))
189 monkeypatch.setattr(webrtc, "sources", webrtc_mod.SOURCES_AUTO)
190 await webrtc.setup_call("initiator")
191 assert "v4l2src" in webrtc.gst_pipe_desc
192 assert "pulsesrc" in webrtc.gst_pipe_desc
193
194 @pytest.mark.asyncio
195 async def test_setup_call_with_stun_and_turn(self, host, webrtc, monkeypatch):
196 """STUN and TURN server configurations are done in setup_call."""
197 mock_pipeline = MagicMock()
198 mock_parse_launch = MagicMock()
199 mock_parse_launch.return_value = mock_pipeline
200 monkeypatch.setattr(Gst, "parse_launch", mock_parse_launch)
201
202 mock_pipeline.get_by_name.return_value = webrtc.webrtcbin
203
204 mock_external_disco = [
205 {"type": "stun", "transport": "udp", "host": "stun.host", "port": "3478"},
206 {
207 "type": "turn",
208 "transport": "udp",
209 "host": "turn.host",
210 "port": "3478",
211 "username": "user",
212 "password": "pass",
213 },
214 ]
215
216 monkeypatch.setattr(
217 data_format, "deserialise", MagicMock(return_value=mock_external_disco)
218 )
219
220 mock_emit = AsyncMock()
221 monkeypatch.setattr(webrtc.webrtcbin, "emit", mock_emit)
222
223 mock_set_property = AsyncMock()
224 monkeypatch.setattr(webrtc.webrtcbin, "set_property", mock_set_property)
225
226 await webrtc.setup_call("initiator")
227
228 host.bridge.external_disco_get.assert_called_once_with("", webrtc.profile)
229 mock_set_property.assert_any_call("stun-server", "stun://stun.host:3478")
230 mock_emit.assert_called_once_with(
231 "add-turn-server", "turn://user:pass@turn.host:3478"
232 )
233
234 @pytest.mark.asyncio
235 async def test_setup_call_gstreamer_pipeline_failure(self, webrtc, monkeypatch):
236 """Test setup_call method handling Gstreamer pipeline failure."""
237 monkeypatch.setattr(Gst, "parse_launch", lambda _: None)
238 with pytest.raises(exceptions.InternalError):
239 await webrtc.setup_call("initiator")