# HG changeset patch # User Goffi # Date 1715787316 -7200 # Node ID 5eb13251fd750df63caf20cdcdc5e733af3980d4 # Parent a7d4007a8fa5afda56289e0093e528877683c855 tests (unit/XEP-0272): XEP-0272 tests: fix 429 diff -r a7d4007a8fa5 -r 5eb13251fd75 tests/unit/test_plugin_xep_0272.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/tests/unit/test_plugin_xep_0272.py Wed May 15 17:35:16 2024 +0200 @@ -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 . + +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 == []