Mercurial > libervia-backend

comparison tests/unit/test_plugin_xep_0131.py @ 4315:8ee369e6eb99

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
tests (unit): add XEP-0131 tests: rel 451
author Goffi <goffi@goffi.org>
date Sat, 28 Sep 2024 15:57:03 +0200
parents
children
comparison
equal deleted inserted replaced
4314:6a70fcd93a7a 4315:8ee369e6eb99
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 unittest.mock import MagicMock
20
21 from pytest_twisted import ensureDeferred as ed
22 from twisted.internet import defer
23 from twisted.words.xish import domish
24
25 from libervia.backend.models.core import MessageData
26 from libervia.backend.plugins.plugin_xep_0131 import (
27 HeadersData,
28 NS_SHIM,
29 XEP_0131,
30 Urgency,
31 Priority,
32 )
33
34
35 class TestXEP0131:
36
37 def test_headers_data_to_element(self):
38 """HeadersData instance is correctly converted to an XML element"""
39 headers = HeadersData(
40 keywords="test,keyword",
41 urgency=Urgency.high,
42 priority=Priority.urgent,
43 custom_header="custom_value", # type: ignore
44 )
45 headers_elt = headers.to_element()
46
47 assert headers_elt.uri == NS_SHIM
48 assert headers_elt.name == "headers"
49 header_elts = list(headers_elt.elements(NS_SHIM, "header"))
50 assert len(header_elts) == 4
51
52 header_names = set()
53 for header_elt in header_elts:
54 header_names.add(header_elt["name"])
55 if header_elt["name"] == "keywords":
56 assert str(header_elt) == "test,keyword"
57 elif header_elt["name"] == "urgency":
58 assert str(header_elt) == "high"
59 elif header_elt["name"] == "priority":
60 assert str(header_elt) == "urgent"
61 elif header_elt["name"] == "custom_header":
62 assert str(header_elt) == "custom_value"
63
64 assert header_names == {"keywords", "urgency", "priority", "custom_header"}
65
66 def test_headers_data_from_element(self):
67 """HeadersData instance is correctly created from an XML element"""
68 headers_elt = domish.Element((NS_SHIM, "headers"))
69 keywords_elt = headers_elt.addElement("header")
70 keywords_elt["name"] = "keywords"
71 keywords_elt.addContent("test,keyword")
72 urgency_elt = headers_elt.addElement("header")
73 urgency_elt["name"] = "urgency"
74 urgency_elt.addContent("high")
75 custom_elt = headers_elt.addElement("header")
76 custom_elt["name"] = "custom_header"
77 custom_elt.addContent("custom_value")
78
79 headers = HeadersData.from_element(headers_elt)
80
81 assert headers.keywords == "test,keyword"
82 assert headers.urgency == Urgency.high
83 assert headers.priority is None
84 assert headers.custom_header == "custom_value" # type: ignore
85
86 def test_quote_value(self):
87 """Values are correctly quoted when necessary"""
88 xep_0131 = XEP_0131(MagicMock())
89
90 assert xep_0131.quote_value("simple") == "simple"
91 assert xep_0131.quote_value("with space") == '"with space"'
92 assert xep_0131.quote_value('with "quotes"') == '"with \\"quotes\\""'
93 assert xep_0131.quote_value("with,comma") == '"with,comma"'
94
95 def test_unquote_values(self):
96 """Raw header values are correctly unquoted"""
97 xep_0131 = XEP_0131(MagicMock())
98
99 assert xep_0131.unquote_values("simple,another") == ["simple", "another"]
100 assert xep_0131.unquote_values('"quoted value",simple') == [
101 "quoted value",
102 "simple",
103 ]
104 assert xep_0131.unquote_values('"with,comma",simple') == ["with,comma", "simple"]
105 assert xep_0131.unquote_values('"with \\"quotes\\"",simple') == [
106 'with "quotes"',
107 "simple",
108 ]
109
110 @ed
111 async def test_send_message_trigger(self):
112 """Headers are correctly added to the message when sending"""
113 xep_0131 = XEP_0131(MagicMock())
114 client = MagicMock()
115
116 mess_data = MessageData(
117 {
118 "extra": {
119 "keywords": ["test", "keyword"],
120 "headers": {"urgency": "high", "custom_header": "custom_value"},
121 },
122 "xml": domish.Element(("jabber:client", "message")),
123 }
124 )
125
126 pre_xml_treatments = MagicMock()
127 post_xml_treatments = defer.Deferred()
128
129 result = xep_0131.send_message_trigger(
130 client, mess_data, pre_xml_treatments, post_xml_treatments
131 )
132 assert result is True
133
134 post_xml_treatments.callback(mess_data)
135 await post_xml_treatments
136
137 # Check that headers were added to the XML
138 headers_elt = next(mess_data["xml"].elements(NS_SHIM, "headers"))
139
140 header_names = set()
141 for header_elt in headers_elt.elements(NS_SHIM, "header"):
142 header_names.add(header_elt["name"])
143 if header_elt["name"] == "keywords":
144 assert str(header_elt) == "test,keyword"
145 elif header_elt["name"] == "urgency":
146 assert str(header_elt) == "high"
147 elif header_elt["name"] == "custom_header":
148 assert str(header_elt) == "custom_value"
149
150 assert header_names == {"keywords", "urgency", "custom_header"}
151
152 @ed
153 async def test_message_received_trigger(self):
154 """Headers are correctly parsed from received messages"""
155 xep_0131 = XEP_0131(MagicMock())
156 client = MagicMock()
157
158 message_elt = domish.Element(("jabber:client", "message"))
159 headers_elt = message_elt.addElement((NS_SHIM, "headers"))
160 keywords_elt = headers_elt.addElement("header")
161 keywords_elt["name"] = "keywords"
162 keywords_elt.addContent('test,"keyword with space"')
163 urgency_elt = headers_elt.addElement("header")
164 urgency_elt["name"] = "urgency"
165 urgency_elt.addContent("high")
166
167 post_treat = defer.Deferred()
168
169 result = xep_0131.message_received_trigger(client, message_elt, post_treat)
170
171 assert result is True
172
173 mess_data = MessageData({"extra": {}})
174 post_treat.callback(mess_data)
175 await post_treat
176
177 assert mess_data["extra"]["keywords"] == ["test", "keyword with space"]
178 headers = mess_data["extra"]["headers"]
179 assert "keywords" not in headers
180 assert headers["urgency"] == "high"