Mercurial > libervia-backend

comparison sat/test/test_plugin_xep_0033.py @ 2562:26edcf3a30eb

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
core, setup: huge cleaning: - moved directories from src and frontends/src to sat and sat_frontends, which is the recommanded naming convention - move twisted directory to root - removed all hacks from setup.py, and added missing dependencies, it is now clean - use https URL for website in setup.py - removed "Environment :: X11 Applications :: GTK", as wix is deprecated and removed - renamed sat.sh to sat and fixed its installation - added python_requires to specify Python version needed - replaced glib2reactor which use deprecated code by gtk3reactor sat can now be installed directly from virtualenv without using --system-site-packages anymore \o/
author Goffi <goffi@goffi.org>
date Mon, 02 Apr 2018 19:44:50 +0200
parents src/test/test_plugin_xep_0033.py@0046283a285d
children 56f94936df1e
comparison
equal deleted inserted replaced
2561:bd30dc3ffe5a 2562:26edcf3a30eb
1 #!/usr/bin/env python2
2 # -*- coding: utf-8 -*-
3
4 # SAT: a jabber client
5 # Copyright (C) 2009-2018 Jérôme Poisson (goffi@goffi.org)
6 # Copyright (C) 2013-2016 Adrien Cossa (souliane@mailoo.org)
7
8 # This program is free software: you can redistribute it and/or modify
9 # it under the terms of the GNU Affero General Public License as published by
10 # the Free Software Foundation, either version 3 of the License, or
11 # (at your option) any later version.
12
13 # This program is distributed in the hope that it will be useful,
14 # but WITHOUT ANY WARRANTY; without even the implied warranty of
15 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16 # GNU Affero General Public License for more details.
17
18 # You should have received a copy of the GNU Affero General Public License
19 # along with this program. If not, see <http://www.gnu.org/licenses/>.
20
21 """ Plugin extended addressing stanzas """
22
23 from constants import Const
24 from sat.test import helpers
25 from sat.plugins import plugin_xep_0033 as plugin
26 from sat.core.exceptions import CancelError
27 from twisted.internet import defer
28 from wokkel.generic import parseXml
29 from twisted.words.protocols.jabber.jid import JID
30
31 PROFILE_INDEX = 0
32 PROFILE = Const.PROFILE[PROFILE_INDEX]
33 JID_STR_FROM = Const.JID_STR[1]
34 JID_STR_TO = Const.PROFILE_DICT[PROFILE].host
35 JID_STR_X_TO = Const.JID_STR[0]
36 JID_STR_X_CC = Const.JID_STR[1]
37 JID_STR_X_BCC = Const.JID_STR[2]
38
39 ADDRS = ('to', JID_STR_X_TO, 'cc', JID_STR_X_CC, 'bcc', JID_STR_X_BCC)
40
41
42 class XEP_0033Test(helpers.SatTestCase):
43
44 def setUp(self):
45 self.host = helpers.FakeSAT()
46 self.plugin = plugin.XEP_0033(self.host)
47
48 def test_messageReceived(self):
49 self.host.memory.reinit()
50 xml = u"""
51 <message type="chat" from="%s" to="%s" id="test_1">
52 <body>test</body>
53 <addresses xmlns='http://jabber.org/protocol/address'>
54 <address type='to' jid='%s'/>
55 <address type='cc' jid='%s'/>
56 <address type='bcc' jid='%s'/>
57 </addresses>
58 </message>
59 """ % (JID_STR_FROM, JID_STR_TO, JID_STR_X_TO, JID_STR_X_CC, JID_STR_X_BCC)
60 stanza = parseXml(xml.encode("utf-8"))
61 treatments = defer.Deferred()
62 self.plugin.messageReceivedTrigger(self.host.getClient(PROFILE), stanza, treatments)
63 data = {'extra': {}}
64
65 def cb(data):
66 expected = ('to', JID_STR_X_TO, 'cc', JID_STR_X_CC, 'bcc', JID_STR_X_BCC)
67 msg = 'Expected: %s\nGot: %s' % (expected, data['extra']['addresses'])
68 self.assertEqual(data['extra']['addresses'], '%s:%s\n%s:%s\n%s:%s\n' % expected, msg)
69
70 treatments.addCallback(cb)
71 return treatments.callback(data)
72
73 def _get_mess_data(self):
74 mess_data = {"to": JID(JID_STR_TO),
75 "type": "chat",
76 "message": "content",
77 "extra": {}
78 }
79 mess_data["extra"]["address"] = '%s:%s\n%s:%s\n%s:%s\n' % ADDRS
80 original_stanza = u"""
81 <message type="chat" from="%s" to="%s" id="test_1">
82 <body>content</body>
83 </message>
84 """ % (JID_STR_FROM, JID_STR_TO)
85 mess_data['xml'] = parseXml(original_stanza.encode("utf-8"))
86 return mess_data
87
88 def _assertAddresses(self, mess_data):
89 """The mess_data that we got here has been modified by self.plugin.messageSendTrigger,
90 check that the addresses element has been added to the stanza."""
91 expected = self._get_mess_data()['xml']
92 addresses_extra = """
93 <addresses xmlns='http://jabber.org/protocol/address'>
94 <address type='%s' jid='%s'/>
95 <address type='%s' jid='%s'/>
96 <address type='%s' jid='%s'/>
97 </addresses>""" % ADDRS
98 addresses_element = parseXml(addresses_extra.encode('utf-8'))
99 expected.addChild(addresses_element)
100 self.assertEqualXML(mess_data['xml'].toXml().encode("utf-8"), expected.toXml().encode("utf-8"))
101
102 def _checkSentAndStored(self):
103 """Check that all the recipients got their messages and that the history has been filled.
104 /!\ see the comments in XEP_0033.sendAndStoreMessage"""
105 sent = []
106 stored = []
107 d_list = []
108
109 def cb(entities, to_jid):
110 if host in entities:
111 if host not in sent: # send the message to the entity offering the feature
112 sent.append(host)
113 stored.append(host)
114 stored.append(to_jid) # store in history for each recipient
115 else: # feature not supported, use normal behavior
116 sent.append(to_jid)
117 stored.append(to_jid)
118 helpers.unmuteLogging()
119
120 for to_s in (JID_STR_X_TO, JID_STR_X_CC, JID_STR_X_BCC):
121 to_jid = JID(to_s)
122 host = JID(to_jid.host)
123 helpers.muteLogging()
124 d = self.host.findFeaturesSet([plugin.NS_ADDRESS], jid_=host, profile=PROFILE)
125 d.addCallback(cb, to_jid)
126 d_list.append(d)
127
128 def cb_list(dummy):
129 msg = "/!\ see the comments in XEP_0033.sendAndStoreMessage"
130 sent_recipients = [JID(elt['to']) for elt in self.host.getSentMessages(PROFILE_INDEX)]
131 self.assertEqualUnsortedList(sent_recipients, sent, msg)
132 self.assertEqualUnsortedList(self.host.stored_messages, stored, msg)
133
134 return defer.DeferredList(d_list).addCallback(cb_list)
135
136 def _trigger(self, data):
137 """Execute self.plugin.messageSendTrigger with a different logging
138 level to not pollute the output, then check that the plugin did its
139 job. It should abort sending the message or add the extended
140 addressing information to the stanza.
141 @param data: the data to be processed by self.plugin.messageSendTrigger
142 """
143 pre_treatments = defer.Deferred()
144 post_treatments = defer.Deferred()
145 helpers.muteLogging()
146 self.plugin.messageSendTrigger(self.host.getClient[PROFILE], data, pre_treatments, post_treatments)
147 post_treatments.callback(data)
148 helpers.unmuteLogging()
149 post_treatments.addCallbacks(self._assertAddresses, lambda failure: failure.trap(CancelError))
150 return post_treatments
151
152 def test_messageSendTriggerFeatureNotSupported(self):
153 # feature is not supported, abort the message
154 self.host.memory.reinit()
155 data = self._get_mess_data()
156 return self._trigger(data)
157
158 def test_messageSendTriggerFeatureSupported(self):
159 # feature is supported by the main target server
160 self.host.reinit()
161 self.host.addFeature(JID(JID_STR_TO), plugin.NS_ADDRESS, PROFILE)
162 data = self._get_mess_data()
163 d = self._trigger(data)
164 return d.addCallback(lambda dummy: self._checkSentAndStored())
165
166 def test_messageSendTriggerFeatureFullySupported(self):
167 # feature is supported by all target servers
168 self.host.reinit()
169 self.host.addFeature(JID(JID_STR_TO), plugin.NS_ADDRESS, PROFILE)
170 for dest in (JID_STR_X_TO, JID_STR_X_CC, JID_STR_X_BCC):
171 self.host.addFeature(JID(JID(dest).host), plugin.NS_ADDRESS, PROFILE)
172 data = self._get_mess_data()
173 d = self._trigger(data)
174 return d.addCallback(lambda dummy: self._checkSentAndStored())
175
176 def test_messageSendTriggerFixWrongEntity(self):
177 # check that a wrong recipient entity is fixed by the backend
178 self.host.reinit()
179 self.host.addFeature(JID(JID_STR_TO), plugin.NS_ADDRESS, PROFILE)
180 for dest in (JID_STR_X_TO, JID_STR_X_CC, JID_STR_X_BCC):
181 self.host.addFeature(JID(JID(dest).host), plugin.NS_ADDRESS, PROFILE)
182 data = self._get_mess_data()
183 data["to"] = JID(JID_STR_X_TO)
184 d = self._trigger(data)
185 return d.addCallback(lambda dummy: self._checkSentAndStored())