libervia-backend/tests/unit/test_plugin_xep_0272.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/>.

import pytest
from unittest.mock import MagicMock, patch
from pytest_twisted import ensureDeferred as ed

from pytest import fixture

from libervia.backend.core import exceptions
from twisted.internet import defer
from twisted.words.protocols.jabber import jid
from twisted.words.xish import domish

from tx_xmpp import muc

from libervia.backend.plugins.plugin_xep_0272 import (
    NS_MUJI,
    PRESENCE_MUJI,
    XEP_0272,
    XEP_0272_handler,
)
from libervia.backend.tools.common import data_format

TEST_ROOM_JID = jid.JID("room@example.com/user")


@fixture
def xep_0272(host) -> XEP_0272:
    """XEP-0272 plugin with mocked dependencies."""
    host.plugins = {
        "XEP-0045": MagicMock(),
        "XEP-0166": MagicMock(),
        "XEP-0167": MagicMock(),
        "XEP-0402": MagicMock(),
        "XEP-0482": MagicMock(),
    }
    return XEP_0272(host)


class TestXEP0272:
    def test_get_presence_user_jid(self, host, client):
        """The real JID is extracted from the muc#user item of the presence."""
        xep_0272 = XEP_0272(host)
        presence_elt = domish.Element((None, "presence"))
        x_elt = presence_elt.addElement((muc.NS_MUC_USER, "x"))
        item_elt = x_elt.addElement((muc.NS_MUC_USER, "item"))
        item_elt["jid"] = "peer@example.com/res"
        assert xep_0272._get_presence_user_jid(presence_elt) == jid.JID(
            "peer@example.com/res"
        )

        # no muc#user item
        assert xep_0272._get_presence_user_jid(domish.Element((None, "presence"))) is None

    def test_get_handler(self, host, client):
        """XEP_0272_handler is instantiated and returned."""
        xep_0272 = XEP_0272(host)
        handler = xep_0272.get_handler(client)
        assert isinstance(handler, XEP_0272_handler)
        assert handler.plugin_parent == xep_0272

    def test_on_muji_request_preparing_state(self, host, client):
        """Our own preparing broadcast finishes the data collection."""
        xep_0272 = XEP_0272(host)
        presence_elt = domish.Element((None, "presence"))
        presence_elt["from"] = TEST_ROOM_JID.full()
        muji_elt = presence_elt.addElement((NS_MUJI, "muji"))
        muji_elt.addElement("preparing")

        muji_data = {"preparing_jids": set(), "to_call": set()}
        with patch.object(xep_0272._muc, "get_room_user_jid", return_value=TEST_ROOM_JID):
            with patch.object(xep_0272, "get_muji_data", return_value=muji_data):
                with patch.object(
                    xep_0272, "try_to_finish_preparation"
                ) as mock_try_to_finish_preparation:
                    xep_0272.on_muji_request(presence_elt, client)
                    assert muji_data["done_collecting"] is True
                    mock_try_to_finish_preparation.assert_called_once()

    def test_on_muji_request_peer_data(self, host, client):
        """The real JID of a data-advertising peer is added to to_call."""
        xep_0272 = XEP_0272(host)
        presence_elt = domish.Element((None, "presence"))
        presence_elt["from"] = "room@example.com/peer"
        presence_elt.addElement((NS_MUJI, "muji"))
        x_elt = presence_elt.addElement((muc.NS_MUC_USER, "x"))
        item_elt = x_elt.addElement((muc.NS_MUC_USER, "item"))
        item_elt["jid"] = "peer@example.com/res"

        room = MagicMock()
        room.roomJID = TEST_ROOM_JID.userhostJID()
        muji_data = {"preparing_jids": set(), "to_call": set()}
        with (
            patch.object(xep_0272._muc, "get_room", return_value=room),
            patch.object(xep_0272._muc, "get_room_user_jid", return_value=TEST_ROOM_JID),
            patch.object(xep_0272, "get_muji_data", return_value=muji_data),
            patch.object(host.bridge, "call_group_setup") as mock_call_group_setup,
        ):
            # the peer's data is collected, but the preparation is not finished yet
            # (our own preparation broadcast has not been received)
            xep_0272.on_muji_request(presence_elt, client)
            assert "done_preparing" not in muji_data
            mock_call_group_setup.assert_not_called()

            # our own preparation broadcast: the data collection is done, finish and
            # ask the frontend to initiate the sessions
            own_presence = domish.Element((None, "presence"))
            own_presence["from"] = TEST_ROOM_JID.full()
            muji_elt = own_presence.addElement((NS_MUJI, "muji"))
            muji_elt.addElement("preparing")
            xep_0272.on_muji_request(own_presence, client)
            assert muji_data["done_preparing"] is True
            mock_call_group_setup.assert_called_once()
            to_call = data_format.deserialise(mock_call_group_setup.call_args.args[1])[
                "to_call"
            ]
            assert to_call == ["peer@example.com/res"]
            # the preparation is already done: a data presence received now is
            # ignored, no new setup is emitted
            mock_call_group_setup.reset_mock()
            xep_0272.on_muji_request(presence_elt, client)
            mock_call_group_setup.assert_not_called()

    @ed
    async def test_on_session_initiate(self, host, client):
        """The MUJI room is only trusted if the sender is a member of the room."""
        xep_0272 = XEP_0272(host)
        jingle_elt = domish.Element((None, "jingle"))
        muji_elt = jingle_elt.addElement((NS_MUJI, "muji"))
        muji_elt["room"] = "room@example.com"

        # sender is a member: the room is stored
        with patch.object(xep_0272._muc, "is_jid_in_room", return_value=True):
            session = {"peer_jid": jid.JID("peer@example.com/res")}
            assert await xep_0272._on_session_initiate(client, session, None, jingle_elt)
            assert session["muji_room"] == "room@example.com"

        # sender is not a member: the room is ignored
        with patch.object(xep_0272._muc, "is_jid_in_room", return_value=False):
            session = {"peer_jid": jid.JID("stranger@example.com")}
            assert await xep_0272._on_session_initiate(client, session, None, jingle_elt)
            assert "muji_room" not in session

    def test_on_muji_request_semi_anonymous_room(self, host, client):
        """A room which does not reveal the real JIDs can't be used for MUJI."""
        xep_0272 = XEP_0272(host)
        presence_elt = domish.Element((None, "presence"))
        presence_elt["from"] = "room@example.com/peer"
        presence_elt.addElement((NS_MUJI, "muji"))
        # no muc#user item: the sender's real JID is not available

        room = MagicMock()
        room.roomJID = TEST_ROOM_JID.userhostJID()
        muji_data = {"preparing_jids": set(), "to_call": set()}
        with (
            patch.object(xep_0272._muc, "get_room", return_value=room),
            patch.object(xep_0272._muc, "get_room_user_jid", return_value=TEST_ROOM_JID),
            patch.object(xep_0272, "get_muji_data", return_value=muji_data),
            pytest.raises(exceptions.ExternalRequestError),
        ):
            xep_0272.on_muji_request(presence_elt, client)

    def test_on_muji_request_own_jid(self, host, client):
        """try_to_finish_preparation is not called for a presence from own JID."""
        xep_0272 = XEP_0272(host)
        presence_elt = domish.Element((None, "presence"))
        presence_elt["from"] = "room@example.com/user"
        presence_elt.addElement((NS_MUJI, "muji"))

        muji_data = {"preparing_jids": set(), "to_call": set()}
        with (
            patch.object(
                xep_0272._muc,
                "get_room_user_jid",
                return_value=jid.JID("room@example.com/user"),
            ),
            patch.object(xep_0272, "get_muji_data", return_value=muji_data),
            patch.object(
                xep_0272, "try_to_finish_preparation"
            ) as mock_try_to_finish_preparation,
        ):
            xep_0272.on_muji_request(presence_elt, client)
            mock_try_to_finish_preparation.assert_not_called()

    def test_try_to_finish_preparation(self, host, client):
        """call_group_setup is only emitted once the data collection is done and no
        other preparation is left to wait."""
        xep_0272 = XEP_0272(host)
        room = MagicMock()
        room.roomJID = TEST_ROOM_JID.userhostJID()
        muji_data = {
            "preparing_jids": set(),
            "to_call": {jid.JID("early@example.com")},
        }

        # signals are synchronous (see _Bridge.send_signal): use a plain mock, not the
        # async one of the bridge fixture
        with (
            patch.object(xep_0272, "get_muji_data", return_value=muji_data),
            patch.object(host.bridge, "call_group_setup") as mock_call_group_setup,
        ):
            # our own preparation broadcast has not been received yet: nothing is done
            xep_0272.try_to_finish_preparation(client, room, muji_data)
            mock_call_group_setup.assert_not_called()
            assert "done_preparing" not in muji_data

            # data collection done, no preparation left: finish and ask the frontend to
            # initiate the sessions
            muji_data["done_collecting"] = True
            xep_0272.try_to_finish_preparation(client, room, muji_data)
            assert muji_data["done_preparing"] is True
            mock_call_group_setup.assert_called_once()
            to_call = data_format.deserialise(mock_call_group_setup.call_args.args[1])[
                "to_call"
            ]
            assert to_call == ["early@example.com"]

            # a preparation is still in progress: nothing more is done
            muji_data["preparing_jids"] = {jid.JID("room@example.com/other")}
            mock_call_group_setup.reset_mock()
            xep_0272.try_to_finish_preparation(client, room, muji_data)
            mock_call_group_setup.assert_not_called()

    @ed
    async def test_call_group_end(self, host, client):
        """call_group_end removes the MUJI presence data and the room bookmark."""
        xep_0272 = XEP_0272(host)
        room_jid = jid.JID("room@example.com")
        room = MagicMock()
        room.roomJID = room_jid
        muji_data = {"to_call": {jid.JID("peer@example.com")}, "done_preparing": True}

        with (
            patch.object(xep_0272._muc, "get_room", return_value=room),
            patch.object(xep_0272, "get_muji_data", return_value=muji_data),
            patch.object(
                client.presence, "send", return_value=defer.succeed(None)
            ) as mock_presence_send,
            patch.object(
                xep_0272._bookmarks, "remove", return_value=defer.succeed(None)
            ) as mock_remove,
        ):
            await xep_0272.call_group_end(client, room_jid)
            mock_presence_send.assert_called_once()
            assert muji_data == {"to_call": set()}
            mock_remove.assert_called_once_with(client, room_jid)

    @ed
    async def test_call_group_data_set(self, host, client):
        """call_group_data_set sends correct presence and muji data."""
        xep_0272 = XEP_0272(host)
        room_jid = jid.JID("room@example.com")
        call_data = {"sdp": "sdp_data"}

        with patch.object(
            xep_0272,
            "generate_presence_and_muji",
            return_value=(MagicMock(), MagicMock()),
        ):
            with patch.object(
                client.presence, "send", return_value=defer.succeed(None)
            ) as mock_presence_send:
                await xep_0272.call_group_data_set(client, room_jid, call_data)
                mock_presence_send.assert_called()

    @ed
    async def test_start_preparation(self, host, client):
        """start_preparation sends correct presence and muji data."""
        xep_0272 = XEP_0272(host)
        room = MagicMock()

        with patch.object(
            xep_0272,
            "generate_presence_and_muji",
            return_value=(MagicMock(), MagicMock()),
        ):
            with patch.object(
                client.presence, "send", return_value=defer.succeed(None)
            ) as mock_presence_send:
                await xep_0272.start_preparation(client, room)
                mock_presence_send.assert_called()


class TestXEP0272Handler:
    def test_connectionInitialized(self, host, client):
        """connectionInitialized adds MUJI presence observer."""
        xep_0272 = XEP_0272(host)
        handler = XEP_0272_handler(xep_0272)
        handler.parent = MagicMock()
        handler.xmlstream = MagicMock()

        with patch.object(handler.xmlstream, "addObserver") as mock_addObserver:
            handler.connectionInitialized()
            mock_addObserver.assert_called_once_with(
                PRESENCE_MUJI, xep_0272.on_muji_request, client=handler.parent
            )

    def test_getDiscoInfo(self):
        """getDiscoInfo returns MUJI feature."""
        handler = XEP_0272_handler(MagicMock())
        info = handler.getDiscoInfo(MagicMock(), MagicMock())
        assert len(info) == 1
        assert info[0] == NS_MUJI

    def test_getDiscoItems(self):
        """getDiscoItems returns empty list."""
        handler = XEP_0272_handler(MagicMock())
        items = handler.getDiscoItems(MagicMock(), MagicMock())
        assert items == []