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: FakeMemory inherits from Memory + more helpers basic support + cleaning

Browse source
This commit is contained in:
souliane 2014-01-05 13:04:54 +01:00
commit 96b1d58eb7
4 changed files with 76 additions and 43 deletions
Download patch file Download diff file Expand all files Collapse all files

3
src/test/constants.py
View file

@ -29,6 +29,9 @@ class Const(object):
TEST_JID_2_STR = u"sender@example.net/house"
TEST_JID_2 = JID(TEST_JID_2_STR)
TEST_JID_3_STR = u"sender@example.net/work"
TEST_JID_3 = JID(TEST_JID_3_STR)
TEST_PROFILE = 'test_profile'
NO_SECURITY_LIMIT = -1

64
src/test/helpers.py
View file

@ -18,15 +18,16 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from sat.core.i18n import _
from sat.core import exceptions
from constants import Const
from wokkel.xmppim import RosterItem
from sat.core.xmpp import SatRosterProtocol
from sat.memory.memory import Params
from sat.memory.memory import Params, Memory
from twisted.trial.unittest import FailTest
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 sat.core import exceptions
import re
@ -54,6 +55,12 @@ class FakeSAT(object):
self.bridge = FakeBridge()
self.memory = FakeMemory(self)
self.trigger = FakeTriggerManager()
self.init()
def init(self):
"""This can be called by tests that check for sent and stored messages"""
self.sent_messages = []
self.stored_messages = []
def delContact(self, to, profile_key):
#TODO
@ -66,8 +73,34 @@ class FakeSAT(object):
pass
def sendMessage(self, to_s, msg, subject=None, mess_type='auto', extra={}, profile_key='@NONE@'):
self.sendAndStoreMessage({"to": JID(to_s)})
def sendAndStoreMessage(self, mess_data, skip_send=False, profile=None):
if not skip_send:
self.sent_messages.append(mess_data["to"])
self.stored_messages.append(mess_data["to"])
pass
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, local server if None
@param cache_only: expect the result to be in cache and don't actually
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)
self.memory.server_features.setdefault(profile, {})
if jid_ is None:
jid_ = self.getClientHostJid(profile_key)
# call FakeMemory.init and FakeMemory.addServerFeature
# in your tests to change the return value of this method
return defer.succeed(jid_ if self.memory.hasServerFeature(feature, jid_, profile_key) else None)
def getClientHostJid(self, profile_key):
return JID(Const.TEST_JID.host)
class FakeBridge(object):
"""Class to simulate and test bridge calls"""
@ -122,20 +155,23 @@ class FakeParams(Params):
return defer.succeed(None)
class FakeMemory(object):
class FakeMemory(Memory):
"""Class to simulate and test memory object"""
def __init__(self, host):
# do not call Memory.__init__, we just want to call the methods that are
# manipulating basic stuff, the others should be overwritten when needed
self.host = host
self.params = FakeParams(host, None)
self.init()
def init(self):
"""Tests that manipulate params and/or entities should
"""Tests that manipulate params, entities, features should
re-initialise the memory first to not fake the result."""
self.params.load_default_params()
self.params.params.clear()
self.entities_data = {} # naive simulation of entities
self.entities_data = {}
self.server_features = {}
def getProfileName(self, profile_key, return_profile_keys=False):
return self.params.getProfileName(profile_key, return_profile_keys)
@ -155,22 +191,6 @@ class FakeMemory(object):
def delWaitingSub(self, contact_jid, profile_key):
pass
def getParams(self, security_limit=Const.NO_SECURITY_LIMIT, app='', profile_key='@NONE@'):
"""profile_key is set to @DEFAULT@ to avoid specifying it always"""
return self.params.getParams(security_limit, app, profile_key='@DEFAULT@')
def updateParams(self, xml, security_limit=Const.SECURITY_LIMIT, app=''):
self.params.updateParams(xml, security_limit, app)
def paramsRegisterApp(self, xml, security_limit=Const.NO_SECURITY_LIMIT, app=''):
return self.params.paramsRegisterApp(xml, security_limit, app)
def setParam(self, name, value, category, security_limit=-1, profile_key='@NONE@'):
self.params.setParam(name, value, category, security_limit, profile_key)
def getParamA(self, name, category, attr="value", profile_key='@NONE@'):
return self.params.getParamA(name, category, attr, profile_key)
def updateEntityData(self, entity_jid, key, value, profile_key):
self.entities_data.setdefault(entity_jid, {})
self.entities_data[entity_jid][key] = value
@ -184,7 +204,7 @@ class FakeMemory(object):
class FakeTriggerManager(object):
def add(self, point_name, callback):
def add(self, point_name, callback, priority=0):
pass
def point(self, point_name, *args, **kwargs):

29
src/test/test_memory.py
View file

@ -92,6 +92,11 @@ class MemoryTest(unittest.TestCase):
"""@param src: a deferred result from Memory.getParams"""
self._assert(src, False)
def _getParams(self, security_limit, app='', profile_key='@NONE@'):
if profile_key == '@NONE@':
profile_key = '@DEFAULT@'
return self.host.memory.getParams(security_limit, app, profile_key)
def test_updateParams(self):
self.host.memory.init()
# check if the update works
@ -107,23 +112,23 @@ class MemoryTest(unittest.TestCase):
params = self._getParamXML()
self.host.memory.init()
self.host.memory.updateParams(params)
self.host.memory.getParams(Const.NO_SECURITY_LIMIT).addCallback(self._assert)
self.host.memory.getParams(0, app='').addCallback(self._assert_not)
self.host.memory.getParams(1, app='').addCallback(self._assert_not)
self._getParams(Const.NO_SECURITY_LIMIT).addCallback(self._assert)
self._getParams(0).addCallback(self._assert_not)
self._getParams(1).addCallback(self._assert_not)
# 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.host.memory.getParams(Const.NO_SECURITY_LIMIT).addCallback(self._assert)
self.host.memory.getParams(0, app='').addCallback(self._assert)
self.host.memory.getParams(1, app='').addCallback(self._assert)
self._getParams(Const.NO_SECURITY_LIMIT).addCallback(self._assert)
self._getParams(0).addCallback(self._assert)
self._getParams(1).addCallback(self._assert)
# 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.host.memory.getParams(Const.NO_SECURITY_LIMIT).addCallback(self._assert)
self.host.memory.getParams(0).addCallback(self._assert_not)
self.host.memory.getParams(1).addCallback(self._assert)
self._getParams(Const.NO_SECURITY_LIMIT).addCallback(self._assert)
self._getParams(0).addCallback(self._assert_not)
self._getParams(1).addCallback(self._assert)
def test_paramsRegisterApp(self):
# tests with no security level on the parameter (most secure)
@ -165,6 +170,6 @@ class MemoryTest(unittest.TestCase):
self.host.memory.init()
params = self._getParamXML(security_level=1)
self.host.memory.paramsRegisterApp(params, 1, Const.APP_NAME)
self.host.memory.getParams(1, '').addCallback(self._assert)
self.host.memory.getParams(1, Const.APP_NAME).addCallback(self._assert)
self.host.memory.getParams(1, 'another_dummy_frontend').addCallback(self._assert_not)
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)

23
src/test/test_plugin_xep_0085.py
View file

@ -40,13 +40,16 @@ class XEP_0085Test(helpers.SatTestCase):
self.host.memory.init()
self.host.memory.setParam(plugin.PARAM_NAME, True, plugin.PARAM_KEY, NO_SECURITY_LIMIT, Const.TEST_PROFILE)
for state in plugin.CHAT_STATES:
xml = """
<message type="chat" from="sender@example.net/house" to="sender@example.net/house" id="test_1">
xml = u"""
<message type="chat" from="%s" to="%s" id="test_1">
%s
<%s xmlns='http://jabber.org/protocol/chatstates'/>
<%s xmlns='%s'/>
</message>
""" % ("<body>test</body>" if state == "active" else "", state)
stanza = parseXml(xml)
""" % (Const.TEST_JID_2_STR,
Const.TEST_JID_STR,
"<body>test</body>" if state == "active" else "",
state, plugin.NS_CHAT_STATES)
stanza = parseXml(xml.encode("utf-8"))
self.host.bridge.expectCall("chatStateReceived", u"sender@example.net/house", state, Const.TEST_PROFILE)
self.plugin.messageReceivedTrigger(stanza, defer.Deferred(), Const.TEST_PROFILE)
@ -54,15 +57,17 @@ class XEP_0085Test(helpers.SatTestCase):
self.host.memory.init()
self.host.memory.setParam(plugin.PARAM_NAME, True, plugin.PARAM_KEY, NO_SECURITY_LIMIT, Const.TEST_PROFILE)
for state in plugin.CHAT_STATES:
mess_data = {"to": "test@example.org/SàT",
mess_data = {"to": Const.TEST_JID,
"type": "chat",
"message": "content",
"extra": {} if state == "active" else {"chat_state": state}}
mess_data['xml'] = parseXml("""
<message type="chat" from="sender@example.net/house" to="test@example.org/SàT" id="test_1">
stanza = u"""
<message type="chat" from="%s" to="%s" id="test_1">
%s
</message>
""" % (("<body>%s</body>" % mess_data['message']) if state == "active" else "",))
""" % (Const.TEST_JID_2_STR, Const.TEST_JID_STR,
("<body>%s</body>" % mess_data['message']) if state == "active" else "")
mess_data['xml'] = parseXml(stanza.encode("utf-8"))
expected = deepcopy(mess_data['xml'])
expected.addElement(state, plugin.NS_CHAT_STATES)
treatments = defer.Deferred()