Mercurial > libervia-backend
comparison tests/unit/frontends/test_webrtc.py @ 4232:0fbe5c605eb6
tests (unit/webrtc,XEP-0176, XEP-0234): Fix tests and add webrtc file transfer tests:
fix 441
| author | Goffi <goffi@goffi.org> |
|---|---|
| date | Sat, 06 Apr 2024 12:59:50 +0200 |
| parents | dfccc90cacc6 |
| children |
comparison
equal
deleted
inserted
replaced
| 4231:e11b13418ba6 | 4232:0fbe5c605eb6 |
|---|---|
| 160 webrtc.on_remote_decodebin_stream(None, mock_pad) | 160 webrtc.on_remote_decodebin_stream(None, mock_pad) |
| 161 | 161 |
| 162 mock_pipeline.add.assert_called() | 162 mock_pipeline.add.assert_called() |
| 163 mock_pad.link.assert_called() | 163 mock_pad.link.assert_called() |
| 164 | 164 |
| 165 @pytest.mark.skipif(Gst is None) | 165 @pytest.mark.skipif(Gst is None, reason="GStreamer is not available") |
| 166 @pytest.mark.asyncio | 166 @pytest.mark.asyncio |
| 167 async def test_setup_call_correct_role(self, host, webrtc, monkeypatch): | 167 async def test_setup_call_correct_role(self, host, webrtc, monkeypatch): |
| 168 """Roles are set in setup_call.""" | 168 """Roles are set in setup_call.""" |
| 169 monkeypatch.setattr(Gst, "parse_launch", MagicMock()) | 169 monkeypatch.setattr(Gst, "parse_launch", MagicMock()) |
| 170 # we use MagicMock class and not instance on purpose, to pass the "isinstance" | |
| 171 # test of "setup_call". | |
| 172 monkeypatch.setattr(Gst, "Pipeline", MagicMock) | |
| 170 monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[])) | 173 monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[])) |
| 171 | 174 |
| 172 await webrtc.setup_call("initiator") | 175 await webrtc.setup_call("initiator") |
| 173 assert webrtc.role == "initiator" | 176 assert webrtc.role == "initiator" |
| 174 | 177 |
| 194 monkeypatch.setattr(webrtc, "sources", webrtc_mod.SOURCES_AUTO) | 197 monkeypatch.setattr(webrtc, "sources", webrtc_mod.SOURCES_AUTO) |
| 195 await webrtc.setup_call("initiator") | 198 await webrtc.setup_call("initiator") |
| 196 assert "v4l2src" in webrtc.gst_pipe_desc | 199 assert "v4l2src" in webrtc.gst_pipe_desc |
| 197 assert "pulsesrc" in webrtc.gst_pipe_desc | 200 assert "pulsesrc" in webrtc.gst_pipe_desc |
| 198 | 201 |
| 199 @pytest.mark.skipif(Gst is None) | 202 @pytest.mark.skipif(Gst is None, reason="GStreamer is not available") |
| 200 @pytest.mark.asyncio | 203 @pytest.mark.asyncio |
| 201 async def test_setup_call_with_stun_and_turn(self, host, webrtc, monkeypatch): | 204 async def test_setup_call_with_stun_and_turn(self, host, webrtc, monkeypatch): |
| 202 """STUN and TURN server configurations are done in setup_call.""" | 205 """STUN and TURN server configurations are done in setup_call.""" |
| 203 mock_pipeline = MagicMock() | 206 mock_pipeline = MagicMock() |
| 204 mock_parse_launch = MagicMock() | 207 mock_parse_launch = MagicMock() |
| 205 mock_parse_launch.return_value = mock_pipeline | 208 mock_parse_launch.return_value = mock_pipeline |
| 209 # As for "test_setup_call_correct_role" we user MagicMock class and not instance | |
| 210 # on purpose here. | |
| 211 monkeypatch.setattr(Gst, "Pipeline", MagicMock) | |
| 206 monkeypatch.setattr(Gst, "parse_launch", mock_parse_launch) | 212 monkeypatch.setattr(Gst, "parse_launch", mock_parse_launch) |
| 207 | 213 |
| 208 mock_pipeline.get_by_name.return_value = webrtc.webrtcbin | 214 mock_pipeline.get_by_name.return_value = webrtc.webrtcbin |
| 209 | 215 |
| 210 mock_external_disco = [ | 216 mock_external_disco = [ |
| 235 mock_set_property.assert_any_call("stun-server", "stun://stun.host:3478") | 241 mock_set_property.assert_any_call("stun-server", "stun://stun.host:3478") |
| 236 mock_emit.assert_called_once_with( | 242 mock_emit.assert_called_once_with( |
| 237 "add-turn-server", "turn://user:pass@turn.host:3478" | 243 "add-turn-server", "turn://user:pass@turn.host:3478" |
| 238 ) | 244 ) |
| 239 | 245 |
| 240 @pytest.mark.skipif(Gst is None) | 246 @pytest.mark.skipif(Gst is None, reason="GStreamer is not available") |
| 241 @pytest.mark.asyncio | 247 @pytest.mark.asyncio |
| 242 async def test_setup_call_gstreamer_pipeline_failure(self, webrtc, monkeypatch): | 248 async def test_setup_call_gstreamer_pipeline_failure(self, webrtc, monkeypatch): |
| 243 """Test setup_call method handling Gstreamer pipeline failure.""" | 249 """Test setup_call method handling Gstreamer pipeline failure.""" |
| 244 monkeypatch.setattr(Gst, "parse_launch", lambda _: None) | 250 monkeypatch.setattr(Gst, "parse_launch", lambda _: None) |
| 245 with pytest.raises(exceptions.InternalError): | 251 with pytest.raises(exceptions.InternalError): |