tests (unit/email gateway): add tests for pubsub service:

rel 453

tests/unit/conftest.py

@@ -67,6 +67,9 @@ class MockLiberviaBackend(LiberviaBackend):
         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)
 

tests/unit/test_email_gateway.py

@@ -24,6 +24,7 @@ from unittest.mock import AsyncMock, MagicMock
 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 @@ from libervia.backend.plugins.plugin_comp_email_gateway.models import (
     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 @@ class TestEmailGatewayComponent:
             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)