changeset c4e5f8d381e0 in /home/hg/repos/python-nbxmpp
details:http://hg.gajim.org/python-nbxmpp?cmd=changeset;node=c4e5f8d381e0
description: add test suite
diffstat:
test/lib/__init__.py | 13 +
test/lib/mock.py | 463 ++++++++++++++++++++++++++++++++++
test/lib/xmpp_mocks.py | 95 ++++++
test/runtests.py | 51 +++
test/unit/__init__.py | 5 +
test/unit/test_xmpp_client_nb.py | 163 +++++++++++
test/unit/test_xmpp_dispatcher_nb.py | 97 +++++++
test/unit/test_xmpp_smacks.py | 133 +++++++++
test/unit/test_xmpp_transports_nb.py | 78 +++++
test/unit/test_xmpp_transports_nb2.py | 276 ++++++++++++++++++++
10 files changed, 1374 insertions(+), 0 deletions(-)
diffs (truncated from 1414 to 300 lines):
diff -r 9046fb3554d1 -r c4e5f8d381e0 test/lib/__init__.py
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/test/lib/__init__.py Sun Apr 07 22:07:13 2013 +0200
@@ -0,0 +1,13 @@
+import sys
+import os
+import getopt
+
+root = os.path.join(os.path.abspath(os.path.dirname(__file__)), '../..')
+
+# look for modules in the CWD, then gajim/test/lib, then gajim/src,
+# then everywhere else
+sys.path.insert(1, root)
+sys.path.insert(1, root + '/test/lib')
+
+def setup_env():
+ pass
diff -r 9046fb3554d1 -r c4e5f8d381e0 test/lib/mock.py
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/test/lib/mock.py Sun Apr 07 22:07:13 2013 +0200
@@ -0,0 +1,463 @@
+#
+# (c) Dave Kirby 2001 - 2005
+# mock(a)thedeveloperscoach.com
+#
+# Original call interceptor and call assertion code by Phil Dawes (pdawes(a)users.sourceforge.net)
+# Call interceptor code enhanced by Bruce Cropley (cropleyb(a)yahoo.com.au)
+#
+# This Python module and associated files are released under the FreeBSD
+# license. Essentially, you can do what you like with it except pretend you wrote
+# it yourself.
+#
+#
+# Copyright (c) 2005, Dave Kirby
+# Copyright (c) 2009, Yann Leboulanger
+#
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+#
+# * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+#
+# * Redistributions in binary form must reproduce the above copyright
+# notice, this list of conditions and the following disclaimer in the
+# documentation and/or other materials provided with the distribution.
+#
+# * Neither the name of this library nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+# ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
+# ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+#
+# mock(a)thedeveloperscoach.com
+
+
+"""
+Mock object library for Python. Mock objects can be used when unit testing
+to remove a dependency on another production class. They are typically used
+when the dependency would either pull in lots of other classes, or
+significantly slow down the execution of the test.
+They are also used to create exceptional conditions that cannot otherwise
+be easily triggered in the class under test.
+"""
+
+__version__ = "0.1.0"
+
+# Added in Python 2.1
+import inspect
+import re
+
+class MockInterfaceError(Exception):
+ pass
+
+class Mock(object):
+ """
+ The Mock class emulates any other class for testing purposes.
+ All method calls are stored for later examination.
+ """
+
+ def __init__(self, returnValues=None, realClass=None):
+ """
+ The Mock class constructor takes a dictionary of method names and
+ the values they return. Methods that are not in the returnValues
+ dictionary will return None.
+ You may also supply a class whose interface is being mocked.
+ All calls will be checked to see if they appear in the original
+ interface. Any calls to methods not appearing in the real class
+ will raise a MockInterfaceError. Any calls that would fail due to
+ non-matching parameter lists will also raise a MockInterfaceError.
+ Both of these help to prevent the Mock class getting out of sync
+ with the class it is Mocking.
+ """
+ self.mockCalledMethods = {}
+ self.mockAllCalledMethods = []
+ self.mockReturnValues = returnValues or {}
+ self.mockExpectations = {}
+ self.realClass = realClass
+ self.realClassMethods = None
+ if realClass:
+ self.realClassMethods = dict(inspect.getmembers(realClass, inspect.isroutine))
+ for retMethod in self.mockReturnValues.keys():
+ if retMethod not in self.realClassMethods:
+ raise MockInterfaceError("Return value supplied for method '%s' that was not in the original class" % retMethod)
+ self._setupSubclassMethodInterceptors()
+
+ def _setupSubclassMethodInterceptors(self):
+ methods = inspect.getmembers(self.realClass, inspect.isroutine)
+ baseMethods = dict(inspect.getmembers(Mock, inspect.ismethod))
+ for m in methods:
+ name = m[0]
+ # Don't record calls to methods of Mock base class.
+ if not name in baseMethods:
+ self.__dict__[name] = MockCallable(name, self, handcrafted=True)
+
+ def __getattr__(self, name):
+ return MockCallable(name, self)
+
+ def mockAddReturnValues(self, **methodReturnValues ):
+ self.mockReturnValues.update(methodReturnValues)
+
+ def mockSetExpectation(self, name, testFn, after=0, until=0):
+ self.mockExpectations.setdefault(name, []).append((testFn, after, until))
+
+ def _checkInterfaceCall(self, name, callParams, callKwParams):
+ """
+ Check that a call to a method of the given name to the original
+ class with the given parameters would not fail. If it would fail,
+ raise a MockInterfaceError.
+ Based on the Python 2.3.3 Reference Manual section 5.3.4: Calls.
+ """
+ if self.realClassMethods is None:
+ return
+ if name not in self.realClassMethods:
+ return
+
+ func = self.realClassMethods[name]
+ try:
+ args, varargs, varkw, defaults = inspect.getargspec(func)
+ except TypeError:
+ # func is not a Python function. It is probably a builtin,
+ # such as __repr__ or __coerce__. TODO: Checking?
+ # For now assume params are OK.
+ return
+
+ # callParams doesn't include self; args does include self.
+ numPosCallParams = 1 + len(callParams)
+
+ if numPosCallParams > len(args) and not varargs:
+ raise MockInterfaceError("Original %s() takes at most %s arguments (%s given)" %
+ (name, len(args), numPosCallParams))
+
+ # Get the number of positional arguments that appear in the call,
+ # also check for duplicate parameters and unknown parameters
+ numPosSeen = _getNumPosSeenAndCheck(numPosCallParams, callKwParams, args, varkw)
+
+ lenArgsNoDefaults = len(args) - len(defaults or [])
+ if numPosSeen < lenArgsNoDefaults:
+ raise MockInterfaceError("Original %s() takes at least %s arguments (%s given)" % (name, lenArgsNoDefaults, numPosSeen))
+
+ def mockGetAllCalls(self):
+ """
+ Return a list of MockCall objects,
+ representing all the methods in the order they were called.
+ """
+ return self.mockAllCalledMethods
+ getAllCalls = mockGetAllCalls # deprecated - kept for backward compatibility
+
+ def mockGetNamedCalls(self, methodName):
+ """
+ Return a list of MockCall objects,
+ representing all the calls to the named method in the order they were called.
+ """
+ return self.mockCalledMethods.get(methodName, [])
+ getNamedCalls = mockGetNamedCalls # deprecated - kept for backward compatibility
+
+ def mockCheckCall(self, index, name, *args, **kwargs):
+ '''test that the index-th call had the specified name and parameters'''
+ call = self.mockAllCalledMethods[index]
+ assert name == call.getName(), "%r != %r" % (name, call.getName())
+ call.checkArgs(*args, **kwargs)
+
+
+def _getNumPosSeenAndCheck(numPosCallParams, callKwParams, args, varkw):
+ """
+ Positional arguments can appear as call parameters either named as
+ a named (keyword) parameter, or just as a value to be matched by
+ position. Count the positional arguments that are given by either
+ keyword or position, and check for duplicate specifications.
+ Also check for arguments specified by keyword that do not appear
+ in the method's parameter list.
+ """
+ posSeen = {}
+ for arg in args[:numPosCallParams]:
+ posSeen[arg] = True
+ for kwp in callKwParams:
+ if kwp in posSeen:
+ raise MockInterfaceError("%s appears as both a positional and named parameter." % kwp)
+ if kwp in args:
+ posSeen[kwp] = True
+ elif not varkw:
+ raise MockInterfaceError("Original method does not have a parameter '%s'" % kwp)
+ return len(posSeen)
+
+class MockCall:
+ """
+ MockCall records the name and parameters of a call to an instance
+ of a Mock class. Instances of MockCall are created by the Mock class,
+ but can be inspected later as part of the test.
+ """
+ def __init__(self, name, params, kwparams ):
+ self.name = name
+ self.params = params
+ self.kwparams = kwparams
+
+ def checkArgs(self, *args, **kwargs):
+ assert args == self.params, "%r != %r" % (args, self.params)
+ assert kwargs == self.kwparams, "%r != %r" % (kwargs, self.kwparams)
+
+ def getParam( self, n ):
+ if isinstance(n, int):
+ return self.params[n]
+ elif isinstance(n, str):
+ return self.kwparams[n]
+ else:
+ raise IndexError, 'illegal index type for getParam'
+
+ def getNumParams(self):
+ return len(self.params)
+
+ def getNumKwParams(self):
+ return len(self.kwparams)
+
+ def getName(self):
+ return self.name
+
+ #pretty-print the method call
+ def __str__(self):
+ s = self.name + "("
+ sep = ''
+ for p in self.params:
+ s = s + sep + repr(p)
+ sep = ', '
+ items = sorted(self.kwparams.items())
+ for k, v in items:
+ s = s + sep + k + '=' + repr(v)
+ sep = ', '
+ s = s + ')'
+ return s
+ def __repr__(self):
+ return self.__str__()
+
+class MockCallable:
+ """
+ Intercepts the call and records it, then delegates to either the mock's
+ dictionary of mock return values that was passed in to the constructor,
+ or a handcrafted method of a Mock subclass.
+ """
+ def __init__(self, name, mock, handcrafted=False):
+ self.name = name
+ self.mock = mock
+ self.handcrafted = handcrafted
+
+ def __call__(self, *params, **kwparams):
+ self.mock._checkInterfaceCall(self.name, params, kwparams)
+ thisCall = self.recordCall(params, kwparams)
+ self.checkExpectations(thisCall, params, kwparams)
+ return self.makeCall(params, kwparams)
+
+ def recordCall(self, params, kwparams):
+ """
+ Record the MockCall in an ordered list of all calls, and an ordered
+ list of calls for that method name.
+ """
+ thisCall = MockCall(self.name, params, kwparams)
+ calls = self.mock.mockCalledMethods.setdefault(self.name, [])
+ calls.append(thisCall)
+ self.mock.mockAllCalledMethods.append(thisCall)
+ return thisCall
+
+ def makeCall(self, params, kwparams):
+ if self.handcrafted:
+ allPosParams = (self.mock,) + params
+ func = _findFunc(self.mock.realClass, self.name)
+ if not func:
+ raise NotImplementedError
+ return func(*allPosParams, **kwparams)
+ else:
+ returnVal = self.mock.mockReturnValues.get(self.name)
+ if isinstance(returnVal, ReturnValuesBase):
changeset d9b0db7645cd in /home/hg/repos/gajim
branches:
details:http://hg.gajim.org/gajim?cmd=changeset;node=d9b0db7645cd
description: move ConnectionPEP class to connection_handlers to prevent import loop
diffstat:
src/common/connection_handlers.py | 167 +++++++++++++++++++-
src/common/pep.py | 169 +-------------------
src/common/zeroconf/connection_handlers_zeroconf.py | 6 +-
3 files changed, 170 insertions(+), 172 deletions(-)
diffs (truncated from 384 to 300 lines):
diff -r ee6b8e032b5d -r d9b0db7645cd src/common/connection_handlers.py
--- a/src/common/connection_handlers.py Fri Apr 05 21:35:39 2013 +0200
+++ b/src/common/connection_handlers.py Sat Apr 06 10:14:28 2013 +0200
@@ -48,7 +48,6 @@
from common import jingle_xtls
from common.commands import ConnectionCommands
from common.pubsub import ConnectionPubSub
-from common.pep import ConnectionPEP
from common.protocol.caps import ConnectionCaps
from common.protocol.bytestream import ConnectionSocks5Bytestream
from common.protocol.bytestream import ConnectionIBBytestream
@@ -767,6 +766,172 @@
p = self.add_sha(p)
self.connection.send(p)
+
+class ConnectionPEP(object):
+
+ def __init__(self, account, dispatcher, pubsub_connection):
+ self._account = account
+ self._dispatcher = dispatcher
+ self._pubsub_connection = pubsub_connection
+ self.reset_awaiting_pep()
+
+ def pep_change_account_name(self, new_name):
+ self._account = new_name
+
+ def reset_awaiting_pep(self):
+ self.to_be_sent_activity = None
+ self.to_be_sent_mood = None
+ self.to_be_sent_tune = None
+ self.to_be_sent_nick = None
+ self.to_be_sent_location = None
+
+ def send_awaiting_pep(self):
+ """
+ Send pep info that were waiting for connection
+ """
+ if self.to_be_sent_activity:
+ self.send_activity(*self.to_be_sent_activity)
+ if self.to_be_sent_mood:
+ self.send_mood(*self.to_be_sent_mood)
+ if self.to_be_sent_tune:
+ self.send_tune(*self.to_be_sent_tune)
+ if self.to_be_sent_nick:
+ self.send_nick(self.to_be_sent_nick)
+ if self.to_be_sent_location:
+ self.send_location(self.to_be_sent_location)
+ self.reset_awaiting_pep()
+
+ def _pubsubEventCB(self, xmpp_dispatcher, msg):
+ ''' Called when we receive <message /> with pubsub event. '''
+ gajim.nec.push_incoming_event(PEPReceivedEvent(None, conn=self,
+ stanza=msg))
+
+ def send_activity(self, activity, subactivity=None, message=None):
+ if self.connected == 1:
+ # We are connecting, keep activity in mem and send it when we'll be
+ # connected
+ self.to_be_sent_activity = (activity, subactivity, message)
+ return
+ if not self.pep_supported:
+ return
+ item = nbxmpp.Node('activity', {'xmlns': nbxmpp.NS_ACTIVITY})
+ if activity:
+ i = item.addChild(activity)
+ if subactivity:
+ i.addChild(subactivity)
+ if message:
+ i = item.addChild('text')
+ i.addData(message)
+ self._pubsub_connection.send_pb_publish('', nbxmpp.NS_ACTIVITY, item,
+ '0')
+
+ def retract_activity(self):
+ if not self.pep_supported:
+ return
+ self.send_activity(None)
+ # not all client support new XEP, so we still retract
+ self._pubsub_connection.send_pb_retract('', nbxmpp.NS_ACTIVITY, '0')
+
+ def send_mood(self, mood, message=None):
+ if self.connected == 1:
+ # We are connecting, keep mood in mem and send it when we'll be
+ # connected
+ self.to_be_sent_mood = (mood, message)
+ return
+ if not self.pep_supported:
+ return
+ item = nbxmpp.Node('mood', {'xmlns': nbxmpp.NS_MOOD})
+ if mood:
+ item.addChild(mood)
+ if message:
+ i = item.addChild('text')
+ i.addData(message)
+ self._pubsub_connection.send_pb_publish('', nbxmpp.NS_MOOD, item, '0')
+
+ def retract_mood(self):
+ if not self.pep_supported:
+ return
+ self.send_mood(None)
+ # not all client support new XEP, so we still retract
+ self._pubsub_connection.send_pb_retract('', nbxmpp.NS_MOOD, '0')
+
+ def send_tune(self, artist='', title='', source='', track=0, length=0,
+ items=None):
+ if self.connected == 1:
+ # We are connecting, keep tune in mem and send it when we'll be
+ # connected
+ self.to_be_sent_tune = (artist, title, source, track, length, items)
+ return
+ if not self.pep_supported:
+ return
+ item = nbxmpp.Node('tune', {'xmlns': nbxmpp.NS_TUNE})
+ if artist:
+ i = item.addChild('artist')
+ i.addData(artist)
+ if title:
+ i = item.addChild('title')
+ i.addData(title)
+ if source:
+ i = item.addChild('source')
+ i.addData(source)
+ if track:
+ i = item.addChild('track')
+ i.addData(track)
+ if length:
+ i = item.addChild('length')
+ i.addData(length)
+ if items:
+ item.addChild(payload=items)
+ self._pubsub_connection.send_pb_publish('', nbxmpp.NS_TUNE, item, '0')
+
+ def retract_tune(self):
+ if not self.pep_supported:
+ return
+ self.send_tune(None)
+ # not all client support new XEP, so we still retract
+ self._pubsub_connection.send_pb_retract('', nbxmpp.NS_TUNE, '0')
+
+ def send_nickname(self, nick):
+ if self.connected == 1:
+ # We are connecting, keep nick in mem and send it when we'll be
+ # connected
+ self.to_be_sent_nick = nick
+ return
+ if not self.pep_supported:
+ return
+ item = nbxmpp.Node('nick', {'xmlns': nbxmpp.NS_NICK})
+ item.addData(nick)
+ self._pubsub_connection.send_pb_publish('', nbxmpp.NS_NICK, item, '0')
+
+ def retract_nickname(self):
+ if not self.pep_supported:
+ return
+ self.send_nickname(None)
+ # not all client support new XEP, so we still retract
+ self._pubsub_connection.send_pb_retract('', nbxmpp.NS_NICK, '0')
+
+ def send_location(self, info):
+ if self.connected == 1:
+ # We are connecting, keep location in mem and send it when we'll be
+ # connected
+ self.to_be_sent_location = info
+ return
+ if not self.pep_supported:
+ return
+ item = nbxmpp.Node('geoloc', {'xmlns': nbxmpp.NS_LOCATION})
+ for field in LOCATION_DATA:
+ if info.get(field, None):
+ i = item.addChild(field)
+ i.addData(info[field])
+ self._pubsub_connection.send_pb_publish('', nbxmpp.NS_LOCATION, item, '0')
+
+ def retract_location(self):
+ if not self.pep_supported:
+ return
+ self.send_location({})
+ # not all client support new XEP, so we still retract
+ self._pubsub_connection.send_pb_retract('', nbxmpp.NS_LOCATION, '0')
+
# basic connection handlers used here and in zeroconf
class ConnectionHandlersBase:
def __init__(self):
diff -r ee6b8e032b5d -r d9b0db7645cd src/common/pep.py
--- a/src/common/pep.py Fri Apr 05 21:35:39 2013 +0200
+++ b/src/common/pep.py Sat Apr 06 10:14:28 2013 +0200
@@ -512,171 +512,4 @@
SUPPORTED_PERSONAL_USER_EVENTS = [UserMoodPEP, UserTunePEP, UserActivityPEP,
- UserNicknamePEP, UserLocationPEP]
-
-from common.connection_handlers_events import PEPReceivedEvent
-
-class ConnectionPEP(object):
-
- def __init__(self, account, dispatcher, pubsub_connection):
- self._account = account
- self._dispatcher = dispatcher
- self._pubsub_connection = pubsub_connection
- self.reset_awaiting_pep()
-
- def pep_change_account_name(self, new_name):
- self._account = new_name
-
- def reset_awaiting_pep(self):
- self.to_be_sent_activity = None
- self.to_be_sent_mood = None
- self.to_be_sent_tune = None
- self.to_be_sent_nick = None
- self.to_be_sent_location = None
-
- def send_awaiting_pep(self):
- """
- Send pep info that were waiting for connection
- """
- if self.to_be_sent_activity:
- self.send_activity(*self.to_be_sent_activity)
- if self.to_be_sent_mood:
- self.send_mood(*self.to_be_sent_mood)
- if self.to_be_sent_tune:
- self.send_tune(*self.to_be_sent_tune)
- if self.to_be_sent_nick:
- self.send_nick(self.to_be_sent_nick)
- if self.to_be_sent_location:
- self.send_location(self.to_be_sent_location)
- self.reset_awaiting_pep()
-
- def _pubsubEventCB(self, xmpp_dispatcher, msg):
- ''' Called when we receive <message /> with pubsub event. '''
- gajim.nec.push_incoming_event(PEPReceivedEvent(None, conn=self,
- stanza=msg))
-
- def send_activity(self, activity, subactivity=None, message=None):
- if self.connected == 1:
- # We are connecting, keep activity in mem and send it when we'll be
- # connected
- self.to_be_sent_activity = (activity, subactivity, message)
- return
- if not self.pep_supported:
- return
- item = nbxmpp.Node('activity', {'xmlns': nbxmpp.NS_ACTIVITY})
- if activity:
- i = item.addChild(activity)
- if subactivity:
- i.addChild(subactivity)
- if message:
- i = item.addChild('text')
- i.addData(message)
- self._pubsub_connection.send_pb_publish('', nbxmpp.NS_ACTIVITY, item,
- '0')
-
- def retract_activity(self):
- if not self.pep_supported:
- return
- self.send_activity(None)
- # not all client support new XEP, so we still retract
- self._pubsub_connection.send_pb_retract('', nbxmpp.NS_ACTIVITY, '0')
-
- def send_mood(self, mood, message=None):
- if self.connected == 1:
- # We are connecting, keep mood in mem and send it when we'll be
- # connected
- self.to_be_sent_mood = (mood, message)
- return
- if not self.pep_supported:
- return
- item = nbxmpp.Node('mood', {'xmlns': nbxmpp.NS_MOOD})
- if mood:
- item.addChild(mood)
- if message:
- i = item.addChild('text')
- i.addData(message)
- self._pubsub_connection.send_pb_publish('', nbxmpp.NS_MOOD, item, '0')
-
- def retract_mood(self):
- if not self.pep_supported:
- return
- self.send_mood(None)
- # not all client support new XEP, so we still retract
- self._pubsub_connection.send_pb_retract('', nbxmpp.NS_MOOD, '0')
-
- def send_tune(self, artist='', title='', source='', track=0, length=0,
- items=None):
- if self.connected == 1:
- # We are connecting, keep tune in mem and send it when we'll be
- # connected
- self.to_be_sent_tune = (artist, title, source, track, length, items)
- return
- if not self.pep_supported:
- return
- item = nbxmpp.Node('tune', {'xmlns': nbxmpp.NS_TUNE})
- if artist:
- i = item.addChild('artist')
- i.addData(artist)
- if title:
- i = item.addChild('title')
- i.addData(title)
- if source: