Mercurial > libervia-backend
view tests/unit/test_plugin_xep_0234.py @ 4300:7ded09452875
plugin XEP-0077, XEP-0100: Adapt to component, and modernize:
- Plugin XEP-0077 can now be used with component, allowing to register methods to return
registration form, and to (un)register.
- Plugin XEP-0077 now advertises its feature in disco.
- Plugin XEP-0100 has been modernized a bit: it is one of the older plugin in Libervia,
and it has now some type hints, models and async methods.
- Plugin XEP-0100's bridge method `gateways_find` now returns a serialised dict with
relevant data. Former XMLUI version has been moved to `gateways_find_xmlui`.
rel 449
author | Goffi <goffi@goffi.org> |
---|---|
date | Fri, 06 Sep 2024 18:01:31 +0200 |
parents | 0fbe5c605eb6 |
children |
line wrap: on
line source
#!/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.'