tests (unit/XEP-0272): XEP-0272 tests:

fix 429

tests/unit/test_plugin_xep_0272.py

@@ -0,0 +1,183 @@
+#!/usr/bin/env python3
+
+# Libervia: an XMPP client
+# Copyright (C) 2009-2024 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 == []