# HG changeset patch # User Goffi # Date 1733183526 -3600 # Node ID 699aa8788d982d5c74712dbead2349dfe1bd1b6b # Parent 7c0b7ecb816fc5b13454f8c6162a73e60fa9bbd9 tests (unit/email gateway): add tests for pubsub service: rel 453 diff -r 7c0b7ecb816f -r 699aa8788d98 libervia/backend/plugins/plugin_comp_email_gateway/pubsub_service.py --- a/libervia/backend/plugins/plugin_comp_email_gateway/pubsub_service.py Tue Dec 03 00:13:23 2024 +0100 +++ b/libervia/backend/plugins/plugin_comp_email_gateway/pubsub_service.py Tue Dec 03 00:52:06 2024 +0100 @@ -16,7 +16,6 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . -from pathlib import Path from typing import TYPE_CHECKING from twisted.internet import defer from twisted.words.protocols.jabber import jid, error diff -r 7c0b7ecb816f -r 699aa8788d98 tests/unit/conftest.py --- a/tests/unit/conftest.py Tue Dec 03 00:13:23 2024 +0100 +++ b/tests/unit/conftest.py Tue Dec 03 00:52:06 2024 +0100 @@ -67,6 +67,9 @@ self.initialised = defer.Deferred() self.initialised.callback(None) + def get_local_path(self, *args, **kwargs): + return MagicMock() + def get_test_config(self, section, name, default=None): return self._test_config.get((section or None, name), default) diff -r 7c0b7ecb816f -r 699aa8788d98 tests/unit/test_email_gateway.py --- a/tests/unit/test_email_gateway.py Tue Dec 03 00:13:23 2024 +0100 +++ b/tests/unit/test_email_gateway.py Tue Dec 03 00:52:06 2024 +0100 @@ -24,6 +24,7 @@ import pytest from pytest_twisted import ensureDeferred as ed from twisted.words.protocols.jabber import jid +from wokkel import disco from libervia.backend.plugins.plugin_comp_email_gateway import ( EmailGatewayComponent, @@ -33,16 +34,22 @@ Credentials, UserData, ) +from libervia.backend.plugins.plugin_comp_email_gateway.pubsub_service import ( + EmailGWPubsubService, + NODE_CONFIG, + NODE_CONFIG_VALUES, + NODE_OPTIONS, +) from libervia.backend.plugins.plugin_xep_0131 import HeadersData, Urgency +@pytest.fixture +def email_gw(host): + email_gw = EmailGatewayComponent(host) + email_gw.storage = MagicMock() + email_gw.users_data = {} + return email_gw class TestEmailGatewayComponent: - @pytest.fixture - def email_gw(self, host): - email_gw = EmailGatewayComponent(host) - email_gw.storage = MagicMock() - email_gw.users_data = {} - return email_gw def test_jid_to_email_gateway_jid(self, email_gw): """JID from the gateway is converted to an email address.""" @@ -248,3 +255,84 @@ None, extra={"headers": {"keywords": "test, example", "urgency": "high"}}, ) + + +class TestPubsubService: + + @pytest.fixture + def pubsub_service(self, email_gw, client): + email_gw.client = client + service = EmailGWPubsubService(email_gw) + return service + + @pytest.fixture + def pubsub_resource(self, pubsub_service): + return pubsub_service.resource + + def test_getNodes(self, pubsub_resource, client): + """XEP-0498 well-known node is returned.""" + requestor = client.jid + service = client.pubsub_service + nodeIdentifier = "test_node" + result = pubsub_resource.getNodes(requestor, service, nodeIdentifier) + assert result.result == [pubsub_resource._pfs.namespace] + + @ed + async def test_items(self, pubsub_resource, client, host): + """Items are retrieved from the storage""" + request = MagicMock() + request.sender = client.jid + request.nodeIdentifier = pubsub_resource._pfs.namespace + files = [ + {"id": "1", "name": "file1", "media_type": "application", "media_subtype": + "octet-stream", "size": 123, "hash_algo": "sha-256", "file_hash": "0123456789abcdef", + "created": 123}, + {"id": "2", "name": "file2", "media_type": "application", "media_subtype": + "octet-stream", "size": 456, "hash_algo": "sha-256", "file_hash": "0123456789abcdef", + "created": 123}, + ] + host.memory.get_files = AsyncMock(return_value=files) + result, _ = await pubsub_resource.items(request) + assert len(result) == 2 + + @ed + async def test_retract(self, pubsub_resource, client, host): + """Items are retracted from the storage""" + request = MagicMock() + request.sender = client.jid + request.nodeIdentifier = pubsub_resource._pfs.namespace + request.itemIdentifiers = ["item_1"] + host.memory.file_delete = AsyncMock() + await pubsub_resource.retract(request) + host.memory.file_delete.assert_called_once() + + def test_getConfigurationOptions(self, pubsub_resource): + """Configuration options are returned""" + options = pubsub_resource.getConfigurationOptions() + assert options == NODE_OPTIONS + + def test_getConfiguration(self, pubsub_resource, client): + """Configuration values are returned""" + requestor = client.jid + service = client.pubsub_service + nodeIdentifier = "test_node" + result = pubsub_resource.getConfiguration(requestor, service, nodeIdentifier) + assert result.result == NODE_CONFIG_VALUES + + def test_getNodeInfo(self, pubsub_resource, client): + """Node information is returned""" + requestor = client.jid + service = client.pubsub_service + nodeIdentifier = pubsub_resource._pfs.namespace + info = pubsub_resource.getNodeInfo(requestor, service, nodeIdentifier) + assert info == {"type": "leaf", "meta-data": NODE_CONFIG} + + @ed + async def test_getDiscoInfo(self, pubsub_service, client): + """Disco information is returned""" + requestor = client.jid + target = client.pubsub_service + nodeIdentifier = "" + result = await pubsub_service.getDiscoInfo(requestor, target, nodeIdentifier) + assert len(result) > 0 + assert any(isinstance(info, disco.DiscoFeature) for info in result)