erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

tests (unit/webrtc,XEP-0176, XEP-0234): Fix tests and add webrtc file transfer tests:

Browse source 
fix 441
This commit is contained in:
Goffi 2024-04-06 12:59:50 +02:00
commit 27b66b3fb2
4 changed files with 197 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

9
tests/unit/conftest.py
View file

@ -39,7 +39,7 @@ def storage():
return AsyncMock()
class MockSAT(LiberviaBackend):
class MockLiberviaBackend(LiberviaBackend):
def __init__(self, bridge, storage):
self._cb_map = {}
@ -76,6 +76,11 @@ class MockSAT(LiberviaBackend):
def clear_test_config(self):
self._test_config.clear()
def register_namespace(self, short_name, namespace):
# Plugin classes may be instantiated several times in tests, so we do nothing in
# this method to avoid ConflictError.
pass
@contextmanager
def use_option_and_reload(self, section, name, value):
self.set_test_config(section, name, value)
@ -104,7 +109,7 @@ class MockSAT(LiberviaBackend):
@fixture(scope="session")
def host(bridge, storage):
host = MockSAT(bridge=bridge, storage=storage)
host = MockLiberviaBackend(bridge=bridge, storage=storage)
return host

12
tests/unit/frontends/test_webrtc.py
View file

@ -162,11 +162,14 @@ class TestWebRtc:
mock_pipeline.add.assert_called()
mock_pad.link.assert_called()
@pytest.mark.skipif(Gst is None)
@pytest.mark.skipif(Gst is None, reason="GStreamer is not available")
@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())
# we use MagicMock class and not instance on purpose, to pass the "isinstance"
# test of "setup_call".
monkeypatch.setattr(Gst, "Pipeline", MagicMock)
monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))
await webrtc.setup_call("initiator")
@ -196,13 +199,16 @@ class TestWebRtc:
assert "v4l2src" in webrtc.gst_pipe_desc
assert "pulsesrc" in webrtc.gst_pipe_desc
@pytest.mark.skipif(Gst is None)
@pytest.mark.skipif(Gst is None, reason="GStreamer is not available")
@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
# As for "test_setup_call_correct_role" we user MagicMock class and not instance
# on purpose here.
monkeypatch.setattr(Gst, "Pipeline", MagicMock)
monkeypatch.setattr(Gst, "parse_launch", mock_parse_launch)
mock_pipeline.get_by_name.return_value = webrtc.webrtcbin
@ -237,7 +243,7 @@ class TestWebRtc:
"add-turn-server", "turn://user:pass@turn.host:3478"
)
@pytest.mark.skipif(Gst is None)
@pytest.mark.skipif(Gst is None, reason="GStreamer is not available")
@pytest.mark.asyncio
async def test_setup_call_gstreamer_pipeline_failure(self, webrtc, monkeypatch):
"""Test setup_call method handling Gstreamer pipeline failure."""

1
tests/unit/test_plugin_xep_0176.py
View file

@ -40,6 +40,7 @@ class TestXEP0176:
"contents": {
content_name: {
"application_data": {"media": "audio"},
"transport": MagicMock(),
"transport_data": {
"local_ice_data": {
"ufrag": "testufrag",

180
tests/unit/test_plugin_xep_0234.py Normal file
View file

@ -0,0 +1,180 @@
#!/usr/bin/env python3
# Libervia: an XMPP client
# Copyright (C) 2009-2024 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
import pytest
from pytest_twisted import ensureDeferred as ed
from twisted.words.protocols.jabber import jid
from libervia.backend.core import exceptions
from libervia.backend.plugins.plugin_xep_0166 import XEP_0166
from libervia.backend.plugins.plugin_xep_0234 import XEP_0234
from libervia.backend.tools import xml_tools
@pytest.fixture(autouse=True)
def no_application_register(monkeypatch):
"""Do not register the application in XEP-0166"""
monkeypatch.setattr(XEP_0166, "register_application", lambda *a, **kw: None)
class TestXEP0234:
@ed
async def test_parse_file_element(self, host, client):
"""``parse_file_element`` updates file_data dictionary correctly."""
xep_0234 = XEP_0234(host)
file_elt = xml_tools.parse(
"""
<file xmlns='urn:xmpp:jingle:apps:file-transfer:5'>
<media-type>text/plain</media-type>
<name>test.txt</name>
<date>2015-07-26T21:46:00+01:00</date>
<size>6144</size>
</file>
"""
)
file_data = {}
result = await xep_0234.parse_file_element(client, file_elt, file_data)
expected = {
"mime_type": "text/plain",
"modified": 1437943560,
"name": "test.txt",
"size": 6144,
}
assert result == expected
@ed
async def test_parse_file_element_no_file_element(self, host, client):
"""Raise NotFound exception if no <file> element is provided."""
xep_0234 = XEP_0234(host)
with pytest.raises(exceptions.DataError):
await xep_0234.parse_file_element(client, None)
@ed
async def test_parse_file_element_invalid_file_element(self, host, client):
"""Raise DataError exception if <file> element is invalid."""
xep_0234 = XEP_0234(host)
file_elt = xml_tools.parse(
"""
<file xmlns='invalid_namespace'>
<name>example.txt</name>
</file>
"""
)
with pytest.raises(exceptions.DataError):
await xep_0234.parse_file_element(client, file_elt)
@ed
async def test_file_receiving_request_conf(self, monkeypatch, host, client):
"""Normal use case call "get_dest_dir"."""
async def mock_defer_confirm(*args, **kwargs):
"""Simulate defer_confirm always confirming."""
return True
monkeypatch.setattr(xml_tools, "defer_confirm", mock_defer_confirm)
xep_0234 = XEP_0234(host)
monkeypatch.setattr(
xep_0234._hash, "parse_hash_elt", MagicMock(return_value=(None, None))
)
mock_get_dest_dir = AsyncMock()
mock_get_dest_dir.return_value = True
monkeypatch.setattr(xep_0234._f, "get_dest_dir", mock_get_dest_dir)
session = {
"peer_jid": jid.JID("peer@example.com"),
"file_accepted": False,
"id": "session_id",
}
content_data = {"application_data": {}, "transport_data": {"webrtc": False}}
content_name = "dummy_content"
file_data = {"progress_id": "123"}
file_elt = xml_tools.parse(
"""
<file xmlns='urn:xmpp:jingle:apps:file-transfer:5'>
<media-type>text/plain</media-type>
<name>test.txt</name>
<date>2015-07-26T21:46:00+01:00</date>
<size>6144</size>
</file>
"""
)
confirmed = await xep_0234._file_receiving_request_conf(
client, session, content_data, content_name, file_data, file_elt
)
assert confirmed is True
assert (
mock_get_dest_dir.called
), '"get_dest_dir" must be called for non WebRTC case.'
@ed
async def test_file_receiving_request_conf_webrtc(self, monkeypatch, host, client):
"""WebRTC use case is handled correctly for received confirmation."""
async def mock_defer_confirm(*args, **kwargs):
"""Simulate defer_confirm always confirming."""
return True
monkeypatch.setattr(xml_tools, "defer_confirm", mock_defer_confirm)
xep_0234 = XEP_0234(host)
monkeypatch.setattr(
xep_0234._hash, "parse_hash_elt", MagicMock(return_value=(None, None))
)
mock_get_dest_dir = AsyncMock()
mock_get_dest_dir.return_value = True
monkeypatch.setattr(xep_0234._f, "get_dest_dir", mock_get_dest_dir)
session = {
"peer_jid": jid.JID("peer@example.com"),
"file_accepted": False,
"id": "session_id",
}
content_data = {"application_data": {}, "transport_data": {"webrtc": True}}
content_name = "dummy_content"
file_data = {"progress_id": "123"}
file_elt = xml_tools.parse(
"""
<file xmlns='urn:xmpp:jingle:apps:file-transfer:5'>
<media-type>text/plain</media-type>
<name>test.txt</name>
<date>2015-07-26T21:46:00+01:00</date>
<size>6144</size>
</file>
"""
)
confirmed = await xep_0234._file_receiving_request_conf(
client, session, content_data, content_name, file_data, file_elt
)
assert confirmed is True
# The file is handled by frontend in WebRTC case, so "get_dest_dir" must not be
# called.
assert (
not mock_get_dest_dir.called
), '"get_dest_dir" must not be called for WebRTC case.'