erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

test: added tests for XEP-0033

Browse source
This commit is contained in:
souliane 2014-01-05 13:05:31 +01:00
commit 84bfd205dd
3 changed files with 175 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

10
src/core/sat_main.py
View file

@ -68,7 +68,9 @@ class MessageSentAndStored(Exception):
""" Exception to raise if the message has been already sent and stored in the
history by the trigger, so the rest of the process should be stopped. This
should normally be raised by the trigger with the minimal priority """
pass
def __init__(self, reason, mess_data):
Exception(reason)
self.mess_data = mess_data # added for testing purpose
class AbortSendMessage(Exception):
@ -652,10 +654,10 @@ class SAT(service.Service):
def requestServerDisco(self, feature, jid_=None, cache_only=False, profile_key="@NONE"):
"""Discover if a server or its items offer a given feature
@param feature: the feature to check
@param jid_: the jid of the server
@param jid_: the jid of the server, local server if None
@param cache_only: expect the result to be in cache and don't actually
make any request to avoid returning a Deferred. This can be used anytime
for requesting the local server because the data are cached for sure.
make any request. This can be used anytime for requesting a feature on
the local server because the data are cached for sure.
@result: the Deferred entity jid offering the feature, or None
"""
profile = self.memory.getProfileName(profile_key)

7
src/plugins/plugin_xep_0033.py
View file

@ -96,7 +96,7 @@ class XEP_0033(object):
element.addChild(domish.Element((None, 'address'), None, {'type': type_, 'jid': jid_}))
# when the prosody plugin is completed, we can immediately return mess_data from here
self.sendAndStoreMessage(mess_data, entries, profile)
return Failure(MessageSentAndStored("XEP-0033 took over"))
return Failure(MessageSentAndStored("XEP-0033 took over", mess_data))
d = self.host.requestServerDisco(NS_ADDRESS, profile_key=profile)
d.addCallbacks(discoCallback, lambda dummy: discoCallback(None))
return d
@ -151,7 +151,10 @@ class XEP_0033(object):
def post_treat_addr(data, addresses):
data['extra']['addresses'] = ""
for address in addresses:
data['extra']['addresses'] += '%s:%s\n' % (address['type'], address['jid'])
# Depending how message has been constructed, we could get here
# some noise like "\n " instead of an address element.
if isinstance(address, domish.Element):
data['extra']['addresses'] += '%s:%s\n' % (address['type'], address['jid'])
return data
try:

164
src/test/test_plugin_xep_0033.py Normal file
View file

@ -0,0 +1,164 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# SAT: a jabber client
# Copyright (C) 2009, 2010, 2011, 2012, 2013 Jérôme Poisson (goffi@goffi.org)
# Copyright (C) 2013 Adrien Cossa (souliane@mailoo.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/>.
""" Plugin extended addressing stanzas """
from constants import Const
from sat.test import helpers
from sat.plugins import plugin_xep_0033 as plugin
from sat.memory.memory import NO_SECURITY_LIMIT
from sat.core.sat_main import AbortSendMessage, MessageSentAndStored
from copy import deepcopy
from twisted.internet import defer
from wokkel.generic import parseXml
from twisted.words.protocols.jabber.jid import JID
class XEP_0033Test(helpers.SatTestCase):
def setUp(self):
self.host = helpers.FakeSAT()
self.plugin = plugin.XEP_0033(self.host)
def test_messageReceived(self):
self.host.memory.init()
xml = u"""
<message type="chat" from="%s" to="%s" id="test_1">
<body>test</body>
<addresses xmlns='http://jabber.org/protocol/address'>
<address type='to' jid='%s'/>
<address type='cc' jid='%s'/>
<address type='bcc' jid='%s'/>
</addresses>
</message>
""" % (Const.TEST_JID_2_STR, self.host.getClientHostJid(Const.TEST_PROFILE),
Const.TEST_JID_STR, Const.TEST_JID_2_STR, Const.TEST_JID_3_STR)
stanza = parseXml(xml.encode("utf-8"))
treatments = defer.Deferred()
self.plugin.messageReceivedTrigger(stanza, treatments, Const.TEST_PROFILE)
data = {'extra': {}}
def assert_(data):
expected = ('to', Const.TEST_JID_STR, 'cc', Const.TEST_JID_2_STR, 'bcc', Const.TEST_JID_3_STR)
assert(data['extra']['addresses'] == '%s:%s\n%s:%s\n%s:%s\n' % expected)
treatments.addCallback(assert_)
treatments.callback(data)
def test_sendMessageTrigger(self):
mess_data = {"to": self.host.getClientHostJid(Const.TEST_PROFILE),
"type": "chat",
"message": "content",
"extra": {}
}
addresses = ('to', Const.TEST_JID_STR, 'cc', Const.TEST_JID_2_STR, 'bcc', Const.TEST_JID_3_STR)
mess_data["extra"]["address"] = '%s:%s\n%s:%s\n%s:%s\n' % addresses
original_stanza = u"""
<message type="chat" from="%s" to="%s" id="test_1">
<body>content</body>
</message>
""" % (Const.TEST_JID_2_STR, self.host.getClientHostJid(Const.TEST_PROFILE))
mess_data['xml'] = parseXml(original_stanza.encode("utf-8"))
expected = deepcopy(mess_data['xml'])
addresses_extra = """
<addresses xmlns='http://jabber.org/protocol/address'>
<address type='%s' jid='%s'/>
<address type='%s' jid='%s'/>
<address type='%s' jid='%s'/>
</addresses>""" % addresses
addresses_element = parseXml(addresses_extra.encode('utf-8'))
expected.addChild(addresses_element)
def assert_(mess_data):
"""The mess_data that we got here as been modified by self.plugin.sendMessageTrigger,
check that the addresses element has been added to the stanza."""
self.assertEqualXML(mess_data['xml'].toXml().encode("utf-8"), expected.toXml().encode("utf-8"))
def fail_(failure, exception_class):
"""If the failure does encapsulate the expected exception, it will be silently
trapped, otherwise it will be re-raised and will make the test fail"""
if exception_class == MessageSentAndStored:
assert_(failure.value.mess_data)
failure.trap(exception_class)
def checkSentAndStored():
"""Check that all the recipients got their messages and that the history has been filled.
/!\ see the comments in XEP_0033.sendAndStoreMessage"""
sent = []
stored = []
cache = set()
for to_s in [addresses[1], addresses[3], addresses[5]]:
to_jid = JID(to_s)
host = JID(to_jid.host)
if self.host.memory.hasServerFeature(plugin.NS_ADDRESS, host, Const.TEST_PROFILE):
if host not in cache:
sent.append(host)
stored.append(host)
cache.add(host)
stored.append(to_jid)
else:
sent.append(to_jid)
stored.append(to_jid)
try:
assert(len(self.host.sent_messages) == len(sent))
assert(len(self.host.stored_messages) == len(stored))
assert(set(self.host.sent_messages) == set(sent))
assert(set(self.host.stored_messages) == set(stored))
except AssertionError as e:
print "----------------------------------------------------"
print "Comparing sent and stored messages failed!"
print self.host.sent_messages
print sent
print self.host.stored_messages
print stored
print "/!\ see the comments in XEP_0033.sendAndStoreMessage"
print "----------------------------------------------------"
raise e
# feature is not supported, abort the message
self.host.memory.init()
treatments = defer.Deferred()
data = deepcopy(mess_data)
self.plugin.sendMessageTrigger(data, treatments, Const.TEST_PROFILE)
treatments.addCallbacks(assert_, lambda failure: fail_(failure, AbortSendMessage))
treatments.callback(data)
# feature is supported
self.host.init()
self.host.memory.init()
self.host.memory.addServerFeature(plugin.NS_ADDRESS, self.host.getClientHostJid(Const.TEST_PROFILE), Const.TEST_PROFILE)
treatments = defer.Deferred()
data = deepcopy(mess_data)
self.plugin.sendMessageTrigger(data, treatments, Const.TEST_PROFILE)
treatments.addCallbacks(assert_, lambda failure: fail_(failure, MessageSentAndStored))
treatments.callback(data)
checkSentAndStored()
# check that a wrong recipient entity is fixed by the backend
self.host.init()
self.host.memory.init()
self.host.memory.addServerFeature(plugin.NS_ADDRESS, self.host.getClientHostJid(Const.TEST_PROFILE), Const.TEST_PROFILE)
treatments = defer.Deferred()
data = deepcopy(mess_data)
data["to"] = Const.TEST_JID
self.plugin.sendMessageTrigger(data, treatments, Const.TEST_PROFILE)
treatments.addCallbacks(assert_, lambda failure: fail_(failure, MessageSentAndStored))
treatments.callback(mess_data)
checkSentAndStored()