Mercurial > libervia-backend

comparison tests/unit/test_plugin_xep_0033.py @ 4307:6a0155f410bd

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
test (unit): add test for plugin XEP-0033: those replace the legacy XEP-0033 test from libervia/backend/test/test_plugin_xep_0033.py. rel 450
author Goffi <goffi@goffi.org>
date Thu, 26 Sep 2024 16:12:01 +0200
parents
children
comparison
equal deleted inserted replaced
4306:94e0968987cd 4307:6a0155f410bd
1 #!/usr/bin/env python3
2
3 # Libervia: an XMPP client
4 # Copyright (C) 2009-2024 Jérôme Poisson (goffi@goffi.org)
5
6 # This program is free software: you can redistribute it and/or modify
7 # it under the terms of the GNU Affero General Public License as published by
8 # the Free Software Foundation, either version 3 of the License, or
9 # (at your option) any later version.
10
11 # This program is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 # GNU Affero General Public License for more details.
15
16 # You should have received a copy of the GNU Affero General Public License
17 # along with this program. If not, see <http://www.gnu.org/licenses/>.
18
19 from typing import cast
20 from unittest.mock import AsyncMock, MagicMock, patch
21
22 from pytest_twisted import ensureDeferred as ed
23 from twisted.words.protocols.jabber import jid
24 from twisted.words.xish import domish
25
26 from libervia.backend.models.core import MessageData
27 from libervia.backend.plugins.plugin_xep_0033 import (
28 AddressType,
29 AddressesData,
30 NS_ADDRESS,
31 XEP_0033,
32 )
33
34
35 class TestXEP0033:
36 def test_address_type_from_element(self):
37 """
38 AddressType instance is correctly created from an XML <address> element.
39 """
40 address_elt = domish.Element((NS_ADDRESS, "address"))
41 address_elt["jid"] = "test@example.com"
42 address_elt["desc"] = "Test Description"
43 address_elt["delivered"] = "true"
44 address = AddressType.from_element(address_elt)
45 assert address.jid == jid.JID("test@example.com")
46 assert address.desc == "Test Description"
47 assert address.delivered is True
48
49 def test_address_type_to_element(self):
50 """
51 XML <address> element is correctly built from an AddressType instance.
52 """
53 address = AddressType(
54 jid=jid.JID("test@example.com"),
55 desc="Test Description",
56 delivered=True,
57 )
58 address_elt = address.to_element()
59 assert address_elt.uri == NS_ADDRESS
60 assert address_elt.name == "address"
61 assert address_elt["jid"] == "test@example.com"
62 assert address_elt["desc"] == "Test Description"
63 assert address_elt["delivered"] == "true"
64
65 def test_addresses_data_from_element(self):
66 """
67 AddressesData instance is correctly created from an XML <addresses> element.
68 """
69 addresses_elt = domish.Element((NS_ADDRESS, "addresses"))
70 address_elt1 = addresses_elt.addElement("address")
71 address_elt1["type"] = "to"
72 address_elt1["jid"] = "test1@example.com"
73 address_elt2 = addresses_elt.addElement("address")
74 address_elt2["type"] = "cc"
75 address_elt2["jid"] = "test2@example.com"
76 address_elt3 = addresses_elt.addElement("address")
77 address_elt3["type"] = "bcc"
78 address_elt3["jid"] = "test3@example.com"
79 address_elt4 = addresses_elt.addElement("address")
80 address_elt4["type"] = "noreply"
81 addresses = AddressesData.from_element(addresses_elt)
82 assert addresses.to is not None and len(addresses.to) == 1
83 assert addresses.to[0].jid == jid.JID("test1@example.com")
84 assert addresses.cc is not None and len(addresses.cc) == 1
85 assert addresses.cc[0].jid == jid.JID("test2@example.com")
86 assert addresses.bcc is not None and len(addresses.bcc) == 1
87 assert addresses.bcc[0].jid == jid.JID("test3@example.com")
88 assert addresses.noreply
89
90 def test_addresses_data_to_element(self):
91 """
92 XML <addresses> element is correctly built from an AddressesData instance.
93 """
94 addresses = AddressesData(
95 to=[AddressType(jid=jid.JID("test1@example.com"))],
96 cc=[AddressType(jid=jid.JID("test2@example.com"))],
97 bcc=[AddressType(jid=jid.JID("test3@example.com"))],
98 noreply=True,
99 )
100 addresses_elt = addresses.to_element()
101 assert addresses_elt.uri == NS_ADDRESS
102 assert addresses_elt.name == "addresses"
103 assert len(addresses_elt.children) == 4
104 for elt in addresses_elt.children:
105 assert elt.uri == NS_ADDRESS
106 assert elt.name == "address"
107 assert addresses_elt.children[0]["type"] == "to"
108 assert addresses_elt.children[0]["jid"] == "test1@example.com"
109 assert addresses_elt.children[1]["type"] == "cc"
110 assert addresses_elt.children[1]["jid"] == "test2@example.com"
111 assert addresses_elt.children[2]["type"] == "bcc"
112 assert addresses_elt.children[2]["jid"] == "test3@example.com"
113 assert addresses_elt.children[3]["type"] == "noreply"
114
115 @ed
116 async def test_handle_addresses(self):
117 """
118 Server JID is used, <addresses> element is added and messages are delivered.
119 """
120 xep_0033 = XEP_0033(MagicMock())
121 client = MagicMock()
122 client.server_jid = jid.JID("server.example.com")
123 client.profile = "test_profile"
124
125 mess_data = MessageData(
126 {
127 "to": jid.JID("recipient@example.com"),
128 "extra": {
129 "addresses": {
130 "to": [{"jid": "to@example.com"}],
131 "cc": [{"jid": "cc@example.com"}],
132 }
133 },
134 "xml": domish.Element(("jabber:client", "message")),
135 }
136 )
137
138 xep_0033.host.memory.disco.has_feature = AsyncMock(return_value=True)
139 xep_0033.deliver_messages = AsyncMock()
140
141 with patch.object(xep_0033, "_stop_if_all_delivered", AsyncMock()):
142 result = await xep_0033._handle_addresses(client, mess_data)
143 cast(AsyncMock, xep_0033._stop_if_all_delivered).assert_called_once()
144
145 assert result["to"] == client.server_jid
146 assert "addresses" in result["xml"].children[0].name
147 xep_0033.deliver_messages.assert_called_once()
148
149 @ed
150 async def test_deliver_messages(self):
151 """Delivery is done for all recipients."""
152 xep_0033 = XEP_0033(MagicMock())
153 client = MagicMock()
154 client.server_jid = jid.JID("server.example.com")
155 client.a_send = AsyncMock()
156
157 mess_data = MessageData(
158 {
159 "xml": domish.Element(("jabber:client", "message")),
160 }
161 )
162 to_example_addr = AddressType(jid=jid.JID("to@example.com"))
163 cc_example_addr = AddressType(jid=jid.JID("cc@example.com"))
164 bcc_example_addr = AddressType(jid=jid.JID("bcc@other.com"))
165
166 addr_data = AddressesData(
167 to=[to_example_addr],
168 cc=[cc_example_addr],
169 bcc=[bcc_example_addr],
170 )
171
172 domains = {
173 "example.com": [
174 to_example_addr,
175 cc_example_addr,
176 ],
177 "other.com": [bcc_example_addr],
178 }
179
180 xep_0033.host.memory.disco.has_feature = AsyncMock(return_value=True)
181
182 await xep_0033.deliver_messages(client, mess_data, addr_data, domains)
183
184 # Check that messages were sent to the multicast domain and individual recipients
185 assert client.a_send.call_count == 2
186 calls = client.a_send.call_args_list
187 # First call is to the multicast service.
188 assert calls[0][0][0]["to"] == "example.com"
189 # Second call is the individual BCC.
190 assert calls[1][0][0]["to"] == "bcc@other.com"
191
192 # Everything must have been delivered.
193 assert all(address.delivered for address in addr_data.addresses)
194
195 # And BCC must have been removed.
196 assert addr_data.bcc is None
197
198 @ed
199 async def test_deliver_messages_multicast_only(self):
200 """Delivery is done only to multicast services."""
201 xep_0033 = XEP_0033(MagicMock())
202 client = MagicMock()
203 client.server_jid = jid.JID("server.example.com")
204 client.a_send = AsyncMock()
205
206 mess_data = MessageData(
207 {
208 "xml": domish.Element(("jabber:client", "message")),
209 }
210 )
211 to_example_addr = AddressType(jid=jid.JID("to@example.com"))
212 cc_example_addr = AddressType(jid=jid.JID("cc@example.com"))
213 bcc_example_addr = AddressType(jid=jid.JID("bcc@other.com"))
214
215 addr_data = AddressesData(
216 to=[to_example_addr],
217 cc=[cc_example_addr],
218 bcc=[bcc_example_addr],
219 )
220
221 domains = {
222 "example.com": [
223 to_example_addr,
224 cc_example_addr,
225 ],
226 "other.com": [bcc_example_addr],
227 }
228
229 xep_0033.host.memory.disco.has_feature = AsyncMock(return_value=True)
230
231 await xep_0033.deliver_messages(
232 client, mess_data, addr_data, domains, multicast_only=True
233 )
234
235 # Check that only the multicast message was sent
236 assert client.a_send.call_count == 1
237 assert client.a_send.call_args[0][0]["to"] == "example.com"
238
239 # Check that only addresses from the multicast domain are marked as delivered
240 assert addr_data.to and addr_data.to[0].delivered is True
241 assert addr_data.cc and addr_data.cc[0].delivered is True
242 assert addr_data.bcc and addr_data.bcc[0].delivered is None