|
500
|
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") |
