tests (unit/frontends): tests for webrtc implementation:
rel 426
This commit is contained in:
Goffi
parent
547375c079
commit
6a92ebce37
1 changed files with 239 additions and 0 deletions
239
tests/unit/frontends/test_webrtc.py
Normal file
239
tests/unit/frontends/test_webrtc.py
Normal file
|
|
@ -0,0 +1,239 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
# Libervia: an XMPP client
|
||||
# Copyright (C) 2009-2023 Jérôme Poisson (goffi@goffi.org)
|
||||
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Affero General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Affero General Public License for more details.
|
||||
|
||||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
from gi.repository import Gst
|
||||
import pytest
|
||||
|
||||
from libervia.backend.core import exceptions
|
||||
from libervia.backend.tools.common import data_format
|
||||
from libervia.frontends.tools import webrtc as webrtc_mod
|
||||
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def host(monkeypatch):
|
||||
host = MagicMock()
|
||||
host.bridge = AsyncMock()
|
||||
host.app.expand = lambda s: s
|
||||
return host
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def webrtc(host):
|
||||
"""Fixture for WebRTC instantiation."""
|
||||
profile = "test_profile"
|
||||
instance = webrtc_mod.WebRTC(host.bridge, profile)
|
||||
|
||||
instance._set_media_types = MagicMock()
|
||||
instance.start_pipeline = MagicMock()
|
||||
instance.link_element_or_pad = MagicMock()
|
||||
instance.webrtcbin = MagicMock()
|
||||
instance.webrtcbin.emit = MagicMock()
|
||||
|
||||
instance.GstSdp_SDPMessage_new_from_text = MagicMock()
|
||||
instance.GstWebRTC_WebRTCSessionDescription_new = MagicMock()
|
||||
instance.Gst_Promise_new_with_change_func = MagicMock()
|
||||
|
||||
return instance
|
||||
|
||||
|
||||
class TestWebRtc:
|
||||
def test_get_payload_types(self, webrtc):
|
||||
"""The method can identify the correct payload types for video and audio."""
|
||||
fake_sdpmsg = MagicMock()
|
||||
fake_media = MagicMock()
|
||||
fake_caps = MagicMock()
|
||||
fake_structure = MagicMock()
|
||||
|
||||
# This side effect will return 'fake_video_encoding' first, then
|
||||
# 'fake_audio_encoding'.
|
||||
fake_structure.__getitem__.side_effect = [
|
||||
"fake_video_encoding",
|
||||
"fake_audio_encoding",
|
||||
]
|
||||
fake_caps.get_structure.return_value = fake_structure
|
||||
fake_media.get_format.side_effect = ["webrtc-datachannel", "10", "20"]
|
||||
fake_media.get_caps_from_media.return_value = fake_caps
|
||||
fake_sdpmsg.get_media.return_value = fake_media
|
||||
fake_sdpmsg.medias_len.return_value = 1
|
||||
fake_media.formats_len.return_value = 3
|
||||
|
||||
result = webrtc.get_payload_types(
|
||||
fake_sdpmsg, "fake_video_encoding", "fake_audio_encoding"
|
||||
)
|
||||
expected_result = {"fake_video_encoding": 10, "fake_audio_encoding": 20}
|
||||
|
||||
assert result == expected_result
|
||||
|
||||
def test_on_accepted_call(self, webrtc):
|
||||
"""The method correctly sets the remote SDP upon acceptance of an outgoing call."""
|
||||
sdp_str = "mock_sdp_string"
|
||||
profile_str = "test_profile"
|
||||
|
||||
webrtc.on_accepted_call(sdp_str, profile_str)
|
||||
|
||||
# remote description must be set
|
||||
assert webrtc.webrtcbin.emit.call_count == 1
|
||||
assert webrtc.webrtcbin.emit.call_args[0][0] == "set-remote-description"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_answer_call(self, webrtc, monkeypatch):
|
||||
"""The method correctly answers an incoming call."""
|
||||
mock_setup_call = AsyncMock()
|
||||
|
||||
def mock_get_payload_types(sdpmsg, video_encoding, audio_encoding):
|
||||
return {"VP8": 96, "OPUS": 97}
|
||||
|
||||
monkeypatch.setattr(webrtc, "setup_call", mock_setup_call)
|
||||
monkeypatch.setattr(webrtc, "get_payload_types", mock_get_payload_types)
|
||||
|
||||
sdp_str = "mock_sdp_string"
|
||||
profile_str = "mock_profile"
|
||||
|
||||
await webrtc.answer_call(sdp_str, profile_str)
|
||||
|
||||
mock_setup_call.assert_called_once_with("responder", audio_pt=97, video_pt=96)
|
||||
|
||||
# remote description must be set
|
||||
assert webrtc.webrtcbin.emit.call_count == 1
|
||||
assert webrtc.webrtcbin.emit.call_args[0][0] == "set-remote-description"
|
||||
|
||||
def test_on_remote_decodebin_stream_video(self, webrtc, monkeypatch):
|
||||
"""The method correctly handles video streams from the remote decodebin."""
|
||||
mock_pipeline = MagicMock()
|
||||
monkeypatch.setattr(webrtc, "pipeline", mock_pipeline)
|
||||
|
||||
mock_pad = MagicMock()
|
||||
mock_caps = MagicMock()
|
||||
mock_structure = MagicMock()
|
||||
|
||||
mock_pad.has_current_caps.return_value = True
|
||||
mock_pad.get_current_caps.return_value = mock_caps
|
||||
mock_caps.__len__.return_value = 1
|
||||
mock_caps.__getitem__.return_value = mock_structure
|
||||
mock_structure.get_name.return_value = "video/x-h264"
|
||||
# We use non-standard resolution as example to trigger the workaround
|
||||
mock_structure.get_int.side_effect = lambda x: MagicMock(
|
||||
value=990 if x == "width" else 557
|
||||
)
|
||||
|
||||
webrtc.on_remote_decodebin_stream(None, mock_pad)
|
||||
|
||||
assert webrtc._remote_video_pad == mock_pad
|
||||
mock_pipeline.add.assert_called()
|
||||
mock_pad.link.assert_called()
|
||||
|
||||
def test_on_remote_decodebin_stream_audio(self, webrtc, monkeypatch):
|
||||
"""The method correctly handles audio streams from the remote decodebin."""
|
||||
mock_pipeline = MagicMock()
|
||||
monkeypatch.setattr(webrtc, "pipeline", mock_pipeline)
|
||||
|
||||
mock_pad = MagicMock()
|
||||
mock_caps = MagicMock()
|
||||
mock_structure = MagicMock()
|
||||
|
||||
mock_pad.has_current_caps.return_value = True
|
||||
mock_pad.get_current_caps.return_value = mock_caps
|
||||
mock_caps.__len__.return_value = 1
|
||||
mock_caps.__getitem__.return_value = mock_structure
|
||||
mock_structure.get_name.return_value = "audio/x-raw"
|
||||
|
||||
webrtc.on_remote_decodebin_stream(None, mock_pad)
|
||||
|
||||
mock_pipeline.add.assert_called()
|
||||
mock_pad.link.assert_called()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_setup_call_correct_role(self, host, webrtc, monkeypatch):
|
||||
"""Roles are set in setup_call."""
|
||||
monkeypatch.setattr(Gst, "parse_launch", MagicMock())
|
||||
monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))
|
||||
|
||||
await webrtc.setup_call("initiator")
|
||||
assert webrtc.role == "initiator"
|
||||
|
||||
await webrtc.setup_call("responder")
|
||||
assert webrtc.role == "responder"
|
||||
|
||||
with pytest.raises(AssertionError):
|
||||
await webrtc.setup_call("invalid_role")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_setup_call_test_mode(self, host, webrtc, monkeypatch):
|
||||
"""Test mode use fake video and audio in setup_call."""
|
||||
monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))
|
||||
monkeypatch.setattr(webrtc, "sources", webrtc_mod.SINKS_TEST)
|
||||
await webrtc.setup_call("initiator")
|
||||
assert "videotestsrc" in webrtc.gst_pipe_desc
|
||||
assert "audiotestsrc" in webrtc.gst_pipe_desc
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_setup_call_normal_mode(self, host, webrtc, monkeypatch):
|
||||
"""Normal mode use real video and audio in setup_call."""
|
||||
monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))
|
||||
monkeypatch.setattr(webrtc, "sources", webrtc_mod.SOURCES_AUTO)
|
||||
await webrtc.setup_call("initiator")
|
||||
assert "v4l2src" in webrtc.gst_pipe_desc
|
||||
assert "pulsesrc" in webrtc.gst_pipe_desc
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_setup_call_with_stun_and_turn(self, host, webrtc, monkeypatch):
|
||||
"""STUN and TURN server configurations are done in setup_call."""
|
||||
mock_pipeline = MagicMock()
|
||||
mock_parse_launch = MagicMock()
|
||||
mock_parse_launch.return_value = mock_pipeline
|
||||
monkeypatch.setattr(Gst, "parse_launch", mock_parse_launch)
|
||||
|
||||
mock_pipeline.get_by_name.return_value = webrtc.webrtcbin
|
||||
|
||||
mock_external_disco = [
|
||||
{"type": "stun", "transport": "udp", "host": "stun.host", "port": "3478"},
|
||||
{
|
||||
"type": "turn",
|
||||
"transport": "udp",
|
||||
"host": "turn.host",
|
||||
"port": "3478",
|
||||
"username": "user",
|
||||
"password": "pass",
|
||||
},
|
||||
]
|
||||
|
||||
monkeypatch.setattr(
|
||||
data_format, "deserialise", MagicMock(return_value=mock_external_disco)
|
||||
)
|
||||
|
||||
mock_emit = AsyncMock()
|
||||
monkeypatch.setattr(webrtc.webrtcbin, "emit", mock_emit)
|
||||
|
||||
mock_set_property = AsyncMock()
|
||||
monkeypatch.setattr(webrtc.webrtcbin, "set_property", mock_set_property)
|
||||
|
||||
await webrtc.setup_call("initiator")
|
||||
|
||||
host.bridge.external_disco_get.assert_called_once_with("", webrtc.profile)
|
||||
mock_set_property.assert_any_call("stun-server", "stun://stun.host:3478")
|
||||
mock_emit.assert_called_once_with(
|
||||
"add-turn-server", "turn://user:pass@turn.host:3478"
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_setup_call_gstreamer_pipeline_failure(self, webrtc, monkeypatch):
|
||||
"""Test setup_call method handling Gstreamer pipeline failure."""
|
||||
monkeypatch.setattr(Gst, "parse_launch", lambda _: None)
|
||||
with pytest.raises(exceptions.InternalError):
|
||||
await webrtc.setup_call("initiator")
|
||||
Loading…
Reference in a new issue