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/>.

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

from pytest import fixture
from twisted.internet import defer
from twisted.words.protocols.jabber import jid
from twisted.words.xish import domish

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:
    """Fixture for initializing XEP-0272 plugin."""
    host.plugins = {
        "XEP-0045": MagicMock(),
        "XEP-0166": MagicMock(),
        "XEP-0167": MagicMock(),
        "XEP-0249": MagicMock(),
    }
    return XEP_0272(host)


class TestXEP0272:
    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):
        """try_to_finish_preparation is called when preparing_state is True."""
        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()}
        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)
                    mock_try_to_finish_preparation.assert_called_once()

    def test_on_muji_request_not_preparing_state(self, host, client):
        """try_to_finish_preparation is not called when preparing_state is False."""
        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 = {"done_preparing": True, "preparing_jids": set(), "to_call": set()}
        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)
                mock_try_to_finish_preparation.assert_not_called()

    def test_on_muji_request_own_jid(self, host, client):
        """try_to_finish_preparation is not called when the presence is 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"))

        client.jid = jid.JID("room@example.com/user")
        with 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):
        """try_to_finish_preparation sets done_preparing to True."""
        xep_0272 = XEP_0272(host)
        room = MagicMock()
        muji_data = {"preparing_jids": set(), "to_call": set()}

        with patch.object(xep_0272, "get_muji_data", return_value=muji_data):
            xep_0272.try_to_finish_preparation(client, room, muji_data)
            assert muji_data["done_preparing"] is True

    @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, "a_send", return_value=defer.succeed(None)
            ) as mock_a_send:
                await xep_0272.call_group_data_set(client, room_jid, call_data)
                mock_a_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, "a_send", return_value=defer.succeed(None)
            ) as mock_a_send:
                await xep_0272.start_preparation(client, room)
                mock_a_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 == []