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: use the SatTestCase methods instead of builtin "assert" in tests for memory, plugin xep-0033

Browse source
This commit is contained in:
souliane 2014-01-07 09:27:53 +01:00
commit 6be767ae3b
3 changed files with 73 additions and 62 deletions
Download patch file Download diff file Expand all files Collapse all files

18
src/test/helpers.py
View file

@ -28,6 +28,7 @@ from twisted.trial import unittest
from twisted.internet import defer
from twisted.words.protocols.jabber.jid import JID
from xml.etree import cElementTree as etree
from collections import Counter
import re
@ -48,6 +49,10 @@ class DifferentXMLException(FailTest):
pass
class DifferentListException(FailTest):
pass
class FakeSAT(object):
"""Class to simulate a SAT instance"""
@ -292,3 +297,16 @@ class SatTestCase(unittest.TestCase):
print "was expecting:\n-\n%s\n-\n\n" % etree.tostring(expected_elt, encoding='utf-8')
print "---"
raise DifferentXMLException
def assertEqualUnsortedList(self, a, b, msg):
counter_a = Counter(a)
counter_b = Counter(b)
if counter_a != counter_b:
print "---"
print "Unsorted lists are not equals:"
print "got : %s" % counter_a
print "was expecting: %s" % counter_b
if msg:
print msg
print "---"
raise DifferentListException

78
src/test/test_memory.py
View file

@ -73,24 +73,28 @@ class MemoryTest(unittest.TestCase):
return True
return False
def _assert(self, src, true=True):
"""@param src: a deferred result from Memory.getParams"""
assertion = self._paramExists(src=minidom.parseString(src.encode("utf-8")))
if not true:
assertion = not assertion
try:
assert(assertion)
except AssertionError as e:
def assertParam_generic(self, src=None, exists=True, deferred=False):
msg = "Expected parameter not found!\n" if exists else "Unexpected parameter found!\n"
if deferred == False:
# in this stack we can see the line where the error came from,
# if limit=5 is not enough you can increase the value
print "---------------------- stack start ----------------------"
traceback.print_stack(limit=5 if true else 6)
print "----------------------- stack end -----------------------"
raise e
# if limit=5, 6 is not enough you can increase the value
msg += "\n".join(traceback.format_stack(limit=5 if exists else 6))
assertion = self._paramExists(src=src)
getattr(self, "assert%s" % exists)(assertion, msg)
def _assert_not(self, src):
def assertParamExists(self, src=None):
self.assertParam_generic(src, True)
def assertParamNotExists(self, src=None):
self.assertParam_generic(src, False)
def assertParamExists_async(self, src):
"""@param src: a deferred result from Memory.getParams"""
self._assert(src, False)
self.assertParam_generic(minidom.parseString(src.encode("utf-8")), True, True)
def assertParamNotExists_async(self, src):
"""@param src: a deferred result from Memory.getParams"""
self.assertParam_generic(minidom.parseString(src.encode("utf-8")), False, True)
def _getParams(self, security_limit, app='', profile_key='@NONE@'):
if profile_key == '@NONE@':
@ -101,7 +105,7 @@ class MemoryTest(unittest.TestCase):
self.host.memory.init()
# check if the update works
self.host.memory.updateParams(self._getParamXML())
assert(self._paramExists())
self.assertParamExists()
previous = self.host.memory.params.dom.cloneNode(True)
# now check if it is really updated and not duplicated
self.host.memory.updateParams(self._getParamXML())
@ -112,64 +116,64 @@ class MemoryTest(unittest.TestCase):
params = self._getParamXML()
self.host.memory.init()
self.host.memory.updateParams(params)
self._getParams(Const.NO_SECURITY_LIMIT).addCallback(self._assert)
self._getParams(0).addCallback(self._assert_not)
self._getParams(1).addCallback(self._assert_not)
self._getParams(Const.NO_SECURITY_LIMIT).addCallback(self.assertParamExists_async)
self._getParams(0).addCallback(self.assertParamNotExists_async)
self._getParams(1).addCallback(self.assertParamNotExists_async)
# tests with security level 0 on the parameter (not secure)
params = self._getParamXML(security_level=0)
self.host.memory.init()
self.host.memory.updateParams(params)
self._getParams(Const.NO_SECURITY_LIMIT).addCallback(self._assert)
self._getParams(0).addCallback(self._assert)
self._getParams(1).addCallback(self._assert)
self._getParams(Const.NO_SECURITY_LIMIT).addCallback(self.assertParamExists_async)
self._getParams(0).addCallback(self.assertParamExists_async)
self._getParams(1).addCallback(self.assertParamExists_async)
# tests with security level 1 on the parameter (more secure)
params = self._getParamXML(security_level=1)
self.host.memory.init()
self.host.memory.updateParams(params)
self._getParams(Const.NO_SECURITY_LIMIT).addCallback(self._assert)
self._getParams(0).addCallback(self._assert_not)
self._getParams(1).addCallback(self._assert)
self._getParams(Const.NO_SECURITY_LIMIT).addCallback(self.assertParamExists_async)
self._getParams(0).addCallback(self.assertParamNotExists_async)
self._getParams(1).addCallback(self.assertParamExists_async)
def test_paramsRegisterApp(self):
# tests with no security level on the parameter (most secure)
params = self._getParamXML()
self.host.memory.init()
self.host.memory.paramsRegisterApp(params, Const.NO_SECURITY_LIMIT, Const.APP_NAME)
assert(self._paramExists())
self.assertParamExists()
self.host.memory.init()
self.host.memory.paramsRegisterApp(params, 0, Const.APP_NAME)
assert(not self._paramExists())
self.assertParamNotExists()
self.host.memory.init()
self.host.memory.paramsRegisterApp(params, 1, Const.APP_NAME)
assert(not self._paramExists())
self.assertParamNotExists()
# tests with security level 0 on the parameter (not secure)
params = self._getParamXML(security_level=0)
self.host.memory.init()
self.host.memory.paramsRegisterApp(params, Const.NO_SECURITY_LIMIT, Const.APP_NAME)
assert(self._paramExists())
self.assertParamExists()
self.host.memory.init()
self.host.memory.paramsRegisterApp(params, 0, Const.APP_NAME)
assert(self._paramExists())
self.assertParamExists()
self.host.memory.init()
self.host.memory.paramsRegisterApp(params, 1, Const.APP_NAME)
assert(self._paramExists())
self.assertParamExists()
# tests with security level 1 on the parameter (more secure)
params = self._getParamXML(security_level=1)
self.host.memory.init()
self.host.memory.paramsRegisterApp(params, Const.NO_SECURITY_LIMIT, Const.APP_NAME)
assert(self._paramExists())
self.assertParamExists()
self.host.memory.init()
self.host.memory.paramsRegisterApp(params, 0, Const.APP_NAME)
assert(not self._paramExists())
self.assertParamNotExists()
self.host.memory.init()
self.host.memory.paramsRegisterApp(params, 1, Const.APP_NAME)
assert(self._paramExists())
self.assertParamExists()
def test_paramsRegisterApp_getParams(self):
# test retrieving the parameter for a specific frontend
self.host.memory.init()
params = self._getParamXML(security_level=1)
self.host.memory.paramsRegisterApp(params, 1, Const.APP_NAME)
self._getParams(1, '').addCallback(self._assert)
self._getParams(1, Const.APP_NAME).addCallback(self._assert)
self._getParams(1, 'another_dummy_frontend').addCallback(self._assert_not)
self._getParams(1, '').addCallback(self.assertParamExists_async)
self._getParams(1, Const.APP_NAME).addCallback(self.assertParamExists_async)
self._getParams(1, 'another_dummy_frontend').addCallback(self.assertParamNotExists_async)

39
src/test/test_plugin_xep_0033.py
View file

@ -55,11 +55,12 @@ class XEP_0033Test(helpers.SatTestCase):
self.plugin.messageReceivedTrigger(stanza, treatments, Const.TEST_PROFILE)
data = {'extra': {}}
def assert_(data):
def cb(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)
msg = 'Expected: %s\nGot: %s' % (expected, data['extra']['addresses'])
self.assertEqual(data['extra']['addresses'], '%s:%s\n%s:%s\n%s:%s\n' % expected, msg)
treatments.addCallback(assert_)
treatments.addCallback(cb)
treatments.callback(data)
def test_sendMessageTrigger(self):
@ -86,16 +87,16 @@ class XEP_0033Test(helpers.SatTestCase):
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,
def assertAddresses(mess_data):
"""The mess_data that we got here has 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):
def sendMessageErrback(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)
assertAddresses(failure.value.mess_data)
failure.trap(exception_class)
def checkSentAndStored():
@ -116,28 +117,16 @@ class XEP_0033Test(helpers.SatTestCase):
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
msg = "/!\ see the comments in XEP_0033.sendAndStoreMessage"
self.assertEqualUnsortedList(self.host.sent_messages, sent, msg)
self.assertEqualUnsortedList(self.host.stored_messages, stored, msg)
# 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.addCallbacks(assertAddresses, lambda failure: sendMessageErrback(failure, AbortSendMessage))
treatments.callback(data)
# feature is supported
@ -147,7 +136,7 @@ class XEP_0033Test(helpers.SatTestCase):
treatments = defer.Deferred()
data = deepcopy(mess_data)
self.plugin.sendMessageTrigger(data, treatments, Const.TEST_PROFILE)
treatments.addCallbacks(assert_, lambda failure: fail_(failure, MessageSentAndStored))
treatments.addCallbacks(assertAddresses, lambda failure: sendMessageErrback(failure, MessageSentAndStored))
treatments.callback(data)
checkSentAndStored()
@ -159,6 +148,6 @@ class XEP_0033Test(helpers.SatTestCase):
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.addCallbacks(assertAddresses, lambda failure: sendMessageErrback(failure, MessageSentAndStored))
treatments.callback(mess_data)
checkSentAndStored()