plugin XEP-0115: refactoring and complete implementation:

- the XEP-0115 implementation has been completed with validation, caps node handling in disco requests, hash generation fixes, and real support for xml:lang. - the `generate_hash` method has been moved from disco where it's not used anymore to XEP-0115. - Disco doesn't hash systematically, it nos use a per-jid cache for non XEP-0115 supporting client. - code modernisation, with use of `get_plugin_data` for Client data in XEP-0115, type hints, etc. - More comments have been added to link algorithms steps to the relevant part in specification.
2026-04-18 00:51:41 +02:00 · 2026-04-18 00:51:41 +02:00 · dd79a42931
commit dd79a42931
parent f6db575ee4
4 changed files with 634 additions and 156 deletions
--- a/tests/unit/test_plugin_xep_0115.py
+++ b/tests/unit/test_plugin_xep_0115.py
@ -0,0 +1,260 @@
+#!/usr/bin/env python3
+
+# Libervia: an XMPP client
+# Copyright (C) 2009-2026 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/>.
+
+"""Tests for XEP-0115 Entity Capabilities implementation."""
+
+from unittest.mock import MagicMock
+from pytest import fixture
+from twisted.words.protocols.jabber import jid
+from tx_xmpp import data_form, disco
+
+from libervia.backend.plugins.plugin_xep_0115 import (
+    NS_ENTITY_CAPABILITY,
+    NS_CAPS_OPTIMIZE,
+    XEP_0115,
+)
+from libervia.backend.core.constants import Const as C
+from libervia.backend.memory.disco import Discovery
+
+
+@fixture
+def xep_0115(host) -> XEP_0115:
+    """Fixture for initializing XEP-0115 plugin."""
+    return XEP_0115(host)
+
+
+@fixture(autouse=True)
+def setup_memory_disco(host):
+    host.memory.disco = Discovery(host)
+    host.memory.disco.hashes = {}
+    return host
+
+
+class TestXEP0115HashGeneration:
+    """Tests for hash generation per XEP-0115 section 4.1."""
+
+    def test_simple_hash_generation(self, xep_0115):
+        """Test hash generation with simple identity and features (XEP-0115 example)."""
+        disco_infos = disco.DiscoInfo()
+        disco_infos.append(disco.DiscoIdentity("client", "pc", "Exodus 0.9.1"))
+        disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/caps"))
+        disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/disco#info"))
+        disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/disco#items"))
+        disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/muc"))
+
+        cap_hash = xep_0115.generate_hash(disco_infos)
+
+        assert cap_hash == "QgayPKawpkPSDYmwT/WM94uAlu0="
+
+    def test_hash_generation_with_xml_lang(self, xep_0115):
+        """Test hash generation with xml:lang attribute on identities."""
+        disco_infos = disco.DiscoInfo()
+        disco_infos.append(disco.DiscoIdentity("client", "pc", "Psi 0.11", "en"))
+        disco_infos.append(disco.DiscoIdentity("client", "pc", "Ψ 0.11", "el"))
+        disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/caps"))
+        disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/disco#info"))
+        disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/disco#items"))
+        disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/muc"))
+
+        cap_hash = xep_0115.generate_hash(disco_infos)
+        cap_hash2 = xep_0115.generate_hash(disco_infos)
+        # Verify that the hash is deterministic
+        assert cap_hash == cap_hash2
+
+        # Verify identities with lang are properly included in hash
+        # The hash should be different from one without lang
+        disco_infos_no_lang = disco.DiscoInfo()
+        disco_infos_no_lang.append(disco.DiscoIdentity("client", "pc", "Psi 0.11"))
+        disco_infos_no_lang.append(disco.DiscoIdentity("client", "pc", "Ψ 0.11"))
+        disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/caps"))
+        disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/disco#info"))
+        disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/disco#items"))
+        disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/muc"))
+
+        cap_hash_no_lang = xep_0115.generate_hash(disco_infos_no_lang)
+        assert cap_hash != cap_hash_no_lang
+
+    def test_hash_generation_spec_complex_example(self, xep_0115):
+        """Test hash generation against the XEP-0115 §5.3 Complex Generation Example."""
+        disco_infos = disco.DiscoInfo()
+        disco_infos.append(disco.DiscoIdentity("client", "pc", "\u03a8 0.11", "el"))
+        disco_infos.append(disco.DiscoIdentity("client", "pc", "Psi 0.11", "en"))
+        disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/caps"))
+        disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/disco#info"))
+        disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/disco#items"))
+        disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/muc"))
+
+        form = data_form.Form("result")
+        form.formNamespace = "urn:xmpp:dataforms:softwareinfo"
+
+        form_type_field = data_form.Field("hidden", "FORM_TYPE")
+        form_type_field.values = ["urn:xmpp:dataforms:softwareinfo"]
+        form.addField(form_type_field)
+
+        ip_version_field = data_form.Field("text-multi", "ip_version")
+        ip_version_field.values = ["ipv4", "ipv6"]
+        form.addField(ip_version_field)
+
+        os_field = data_form.Field("text", "os")
+        os_field.values = ["Mac"]
+        form.addField(os_field)
+
+        os_version_field = data_form.Field("text", "os_version")
+        os_version_field.values = ["10.5.1"]
+        form.addField(os_version_field)
+
+        software_field = data_form.Field("text", "software")
+        software_field.values = ["Psi"]
+        form.addField(software_field)
+
+        software_version_field = data_form.Field("text", "software_version")
+        software_version_field.values = ["0.11"]
+        form.addField(software_version_field)
+
+        disco_infos.append(form)
+
+        cap_hash = xep_0115.generate_hash(disco_infos)
+
+        # Expected hash from XEP-0115 §5.3
+        assert cap_hash == "q07IKJEyjvHSyhy//CH0CxmKi8w="
+
+
+class TestXEP0115Handler:
+    """Tests for XEP-0115 handler."""
+
+    def test_getDiscoInfo_returns_caps_features(self, xep_0115):
+        """getDiscoInfo returns caps features for non-caps queries."""
+        handler = xep_0115.get_handler(MagicMock())
+
+        result = handler.getDiscoInfo(MagicMock(), MagicMock(), "")
+
+        assert disco.DiscoFeature(NS_ENTITY_CAPABILITY) in result
+        assert disco.DiscoFeature(NS_CAPS_OPTIMIZE) in result
+
+    def test_getDiscoInfo_caps_node_query(self, host, xep_0115):
+        """getDiscoInfo returns cached disco info for caps node queries."""
+        handler = xep_0115.get_handler(MagicMock())
+
+        # Create and cache some disco info
+        disco_infos = disco.DiscoInfo()
+        disco_infos.append(disco.DiscoIdentity("client", "pc", "Test Client"))
+        disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/caps"))
+        disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/disco#info"))
+
+        cap_hash = xep_0115.generate_hash(disco_infos)
+        host.memory.disco.hashes[cap_hash] = disco_infos
+
+        # Query with caps node format
+        node_identifier = f"{C.APP_URL}#{cap_hash}"
+        result = handler.getDiscoInfo(MagicMock(), MagicMock(), node_identifier)
+
+        # Should return the cached disco info items
+        assert len(result) > 0
+        assert any(isinstance(item, disco.DiscoIdentity) for item in result)
+        assert any(isinstance(item, disco.DiscoFeature) for item in result)
+
+        # Verify that the returned items match the cached disco info
+        identities = [item for item in result if isinstance(item, disco.DiscoIdentity)]
+        features = [item for item in result if isinstance(item, disco.DiscoFeature)]
+        assert len(identities) == len(disco_infos.identities)
+        assert len(features) == len(disco_infos.features)
+        assert identities[0].name == "Test Client"
+        assert disco.DiscoFeature("http://jabber.org/protocol/caps") in features
+        assert disco.DiscoFeature("http://jabber.org/protocol/disco#info") in features
+
+    def test_getDiscoInfo_unknown_caps_node(self, xep_0115):
+        """getDiscoInfo returns caps features for unknown caps nodes."""
+        handler = xep_0115.get_handler(MagicMock())
+
+        # Query with unknown hash
+        node_identifier = f"{C.APP_URL}#unknownhash123"
+        result = handler.getDiscoInfo(MagicMock(), MagicMock(), node_identifier)
+
+        # Should return just the caps features, not cached info
+        assert disco.DiscoFeature(NS_ENTITY_CAPABILITY) in result
+        assert disco.DiscoFeature(NS_CAPS_OPTIMIZE) in result
+
+    def test_getDiscoItems_empty(self, xep_0115):
+        """getDiscoItems returns empty list (caps nodes are virtual)."""
+        handler = xep_0115.get_handler(MagicMock())
+
+        result = handler.getDiscoItems(MagicMock(), MagicMock(), "")
+
+        assert result == []
+
+
+class TestXEP0115Validation:
+    """Tests for caps validation."""
+
+    def test_validate_disco_info_well_formed(self, xep_0115):
+        """Validation passes for well-formed disco info."""
+        disco_infos = disco.DiscoInfo()
+        disco_infos.append(disco.DiscoIdentity("client", "pc", "Test"))
+        disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/caps"))
+
+        assert (
+            xep_0115._validate_caps(
+                jid.JID("test@example.com"),
+                "testhash",
+                "http://test.com",
+                disco_infos,
+            )
+            is True
+        )
+
+    def test_validate_disco_info_duplicate_identity(self, xep_0115, capsys):
+        """Validation fails for duplicate identities."""
+        disco_infos = disco.DiscoInfo()
+        disco_infos.append(disco.DiscoIdentity("client", "pc", "Test", "en"))
+        disco_infos.append(
+            disco.DiscoIdentity("client", "pc", "Test2", "en")
+        )  # Duplicate category/type/lang
+
+        assert (
+            xep_0115._validate_caps(
+                jid.JID("test@example.com"),
+                "testhash",
+                "http://test.com",
+                disco_infos,
+            )
+            is False
+        )
+
+        captured = capsys.readouterr()
+        assert "Duplicate identity found" in captured.out
+
+    def test_validate_disco_info_duplicate_feature(self, xep_0115, capsys):
+        """Validation fails for duplicate features."""
+        disco_infos = disco.DiscoInfo()
+        disco_infos.append(disco.DiscoFeature("http://jabber.org/protocol/caps"))
+        disco_infos.append(
+            disco.DiscoFeature("http://jabber.org/protocol/caps")
+        )  # Duplicate
+
+        assert (
+            xep_0115._validate_caps(
+                jid.JID("test@example.com"),
+                "testhash",
+                "http://test.com",
+                disco_infos,
+            )
+            is False
+        )
+
+        captured = capsys.readouterr()
+        assert "Duplicate feature found" in captured.out