libervia-backend/tests/unit/test_plugin_xep_0115.py

#!/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_disco_info_trigger_caps_node_query(self, host, xep_0115):
        """_disco_info_trigger returns cached disco info for caps node queries."""
        # 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}"
        cont, deferred = xep_0115._disco_info_trigger(
            MagicMock(), MagicMock(), node_identifier
        )

        # Should not continue, return cached disco info
        assert cont is False
        assert deferred is not None
        result = deferred.result
        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_disco_info_trigger_unknown_caps_node(self, xep_0115):
        """_disco_info_trigger lets unknown caps nodes pass through."""
        # Query with unknown hash
        node_identifier = f"{C.APP_URL}#unknownhash123"
        cont, deferred = xep_0115._disco_info_trigger(
            MagicMock(), MagicMock(), node_identifier
        )

        # Should continue, let normal flow proceed
        assert cont is True
        assert deferred is None

    def test_disco_info_trigger_non_caps_node(self, xep_0115):
        """_disco_info_trigger ignores non-caps nodes."""
        cont, deferred = xep_0115._disco_info_trigger(
            MagicMock(), MagicMock(), "some:other:node"
        )

        assert cont is True
        assert deferred is None

    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