diff options
| -rw-r--r-- | test/mock.py | 465 | ||||
| -rw-r--r-- | test/notify.py | 15 | ||||
| -rw-r--r-- | test/test_sessions.py | 267 |
3 files changed, 747 insertions, 0 deletions
diff --git a/test/mock.py b/test/mock.py new file mode 100644 index 000000000..7c20056c1 --- /dev/null +++ b/test/mock.py @@ -0,0 +1,465 @@ +# +# (c) Dave Kirby 2001 - 2005 +# mock@thedeveloperscoach.com +# +# Original call interceptor and call assertion code by Phil Dawes (pdawes@users.sourceforge.net) +# Call interceptor code enhanced by Bruce Cropley (cropleyb@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 +# +# 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@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: + """ + 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.realClassMethods = None + if realClass: + self.realClassMethods = dict(inspect.getmembers(realClass, inspect.isroutine)) + for retMethod in self.mockReturnValues.keys(): + if not self.realClassMethods.has_key(retMethod): + 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.__class__,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 == None: + return + if not self.realClassMethods.has_key(name): + raise MockInterfaceError("Calling mock method '%s' that was not found in the original class" % name) + + 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 posSeen.has_key(kwp): + 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 = self.kwparams.items() + items.sort() + 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.__class__, self.name) + if not func: + raise NotImplementedError + return func(*allPosParams, **kwparams) + else: + returnVal = self.mock.mockReturnValues.get(self.name) + if isinstance(returnVal, ReturnValuesBase): + returnVal = returnVal.next() + return returnVal + + def checkExpectations(self, thisCall, params, kwparams): + if self.name in self.mock.mockExpectations: + callsMade = len(self.mock.mockCalledMethods[self.name]) + for (expectation, after, until) in self.mock.mockExpectations[self.name]: + if callsMade > after and (until==0 or callsMade < until): + assert expectation(self.mock, thisCall, len(self.mock.mockAllCalledMethods)-1), 'Expectation failed: '+str(thisCall) + + +def _findFunc(cl, name): + """ Depth first search for a method with a given name. """ + if cl.__dict__.has_key(name): + return cl.__dict__[name] + for base in cl.__bases__: + func = _findFunc(base, name) + if func: + return func + return None + + + +class ReturnValuesBase: + def next(self): + try: + return self.iter.next() + except StopIteration: + raise AssertionError("No more return values") + def __iter__(self): + return self + +class ReturnValues(ReturnValuesBase): + def __init__(self, *values): + self.iter = iter(values) + + +class ReturnIterator(ReturnValuesBase): + def __init__(self, iterator): + self.iter = iter(iterator) + + +def expectParams(*params, **keywords): + '''check that the callObj is called with specified params and keywords + ''' + def fn(mockObj, callObj, idx): + return callObj.params == params and callObj.kwparams == keywords + return fn + + +def expectAfter(*methods): + '''check that the function is only called after all the functions in 'methods' + ''' + def fn(mockObj, callObj, idx): + calledMethods = [method.getName() for method in mockObj.mockGetAllCalls()] + #skip last entry, since that is the current call + calledMethods = calledMethods[:-1] + for method in methods: + if method not in calledMethods: + return False + return True + return fn + +def expectException(exception, *args, **kwargs): + ''' raise an exception when the method is called + ''' + def fn(mockObj, callObj, idx): + raise exception(*args, **kwargs) + return fn + + +def expectParam(paramIdx, cond): + '''check that the callObj is called with parameter specified by paramIdx (a position index or keyword) + fulfills the condition specified by cond. + cond is a function that takes a single argument, the value to test. + ''' + def fn(mockObj, callObj, idx): + param = callObj.getParam(paramIdx) + return cond(param) + return fn + +def EQ(value): + def testFn(param): + return param == value + return testFn + +def NE(value): + def testFn(param): + return param != value + return testFn + +def GT(value): + def testFn(param): + return param > value + return testFn + +def LT(value): + def testFn(param): + return param < value + return testFn + +def GE(value): + def testFn(param): + return param >= value + return testFn + +def LE(value): + def testFn(param): + return param <= value + return testFn + +def AND(*condlist): + def testFn(param): + for cond in condlist: + if not cond(param): + return False + return True + return testFn + +def OR(*condlist): + def testFn(param): + for cond in condlist: + if cond(param): + return True + return False + return testFn + +def NOT(cond): + def testFn(param): + return not cond(param) + return testFn + +def MATCHES(regex, *args, **kwargs): + compiled_regex = re.compile(regex, *args, **kwargs) + def testFn(param): + return compiled_regex.match(param) != None + return testFn + +def SEQ(*sequence): + iterator = iter(sequence) + def testFn(param): + try: + cond = iterator.next() + except StopIteration: + raise AssertionError('SEQ exhausted') + return cond(param) + return testFn + +def IS(instance): + def testFn(param): + return param is instance + return testFn + +def ISINSTANCE(class_): + def testFn(param): + return isinstance(param, class_) + return testFn + +def ISSUBCLASS(class_): + def testFn(param): + return issubclass(param, class_) + return testFn + +def CONTAINS(val): + def testFn(param): + return val in param + return testFn + +def IN(container): + def testFn(param): + return param in container + return testFn + +def HASATTR(attr): + def testFn(param): + return hasattr(param, attr) + return testFn + +def HASMETHOD(method): + def testFn(param): + return hasattr(param, method) and callable(getattr(param, method)) + return testFn + +CALLABLE = callable + + + diff --git a/test/notify.py b/test/notify.py new file mode 100644 index 000000000..2e55e6959 --- /dev/null +++ b/test/notify.py @@ -0,0 +1,15 @@ +# mock notify module + +notifications = [] + +def notify(event, jid, account, parameters, advanced_notif_num = None): + notifications.append((event, jid, account, parameters, advanced_notif_num)) + +def get_advanced_notification(event, account, contact): + return None + +def get_show_in_roster(event, account, contact, session = None): + return True + +def get_show_in_systray(event, account, contact, type_ = None): + return True diff --git a/test/test_sessions.py b/test/test_sessions.py new file mode 100644 index 000000000..d09c38ef9 --- /dev/null +++ b/test/test_sessions.py @@ -0,0 +1,267 @@ +import unittest + +import sys +import os.path + +gajim_root = os.path.join(os.path.abspath(os.path.dirname(__file__)), '..') + +sys.path.append(gajim_root + '/src') + +# a temporary version of ~/.gajim for testing +configdir = gajim_root + '/test/tmp' + +import time + +from mock import Mock + +# define _ for i18n +import __builtin__ +__builtin__._ = lambda x: x + +# wipe config directory +import os +if os.path.isdir(configdir): + import shutil + shutil.rmtree(configdir) + +os.mkdir(configdir) + +import common.configpaths +common.configpaths.gajimpaths.init(configdir) +common.configpaths.gajimpaths.init_profile() + +# for some reason common.gajim needs to be imported before xmpppy? +from common import gajim +from common import xmpp + +from common.stanza_session import StanzaSession + +# name to use for the test account +account_name = 'test' + +class MockConnection(Mock): + def __init__(self, *args): + Mock.__init__(self, *args) + self.name = account_name + gajim.connections[self.name] = self + +class TestStanzaSession(unittest.TestCase): + def setUp(self): + self.jid = 'test@example.org/Gajim' + self.conn = MockConnection({'send_stanza': None}) + self.sess = StanzaSession(self.conn, self.jid, None, 'chat') + + def test_generate_thread_id(self): + # thread_id is a string + self.assert_(isinstance(self.sess.thread_id, str)) + + # it should be somewhat long, to avoid clashes + self.assert_(len(self.sess.thread_id) >= 32) + + def test_is_loggable(self): + # by default a session should be loggable + # (unless the no_log_for setting says otherwise) + self.assert_(self.sess.is_loggable()) + + def test_terminate(self): + # termination is sent by default + self.sess.last_send = time.time() + self.sess.terminate() + + self.assertEqual(None, self.sess.status) + + calls = self.conn.mockGetNamedCalls('send_stanza') + msg = calls[0].getParam(0) + + self.assertEqual(msg.getThread(), self.sess.thread_id) + + def test_terminate_without_sendng(self): + # no termination is sent if no messages have been sent in the session + self.sess.terminate() + + self.assertEqual(None, self.sess.status) + + calls = self.conn.mockGetNamedCalls('send_stanza') + self.assertEqual(0, len(calls)) + + def test_terminate_no_remote_xep_201(self): + # no termination is sent if only messages without thread ids have been + # received + self.sess.last_send = time.time() + self.sess.last_receive = time.time() + self.sess.terminate() + + self.assertEqual(None, self.sess.status) + + calls = self.conn.mockGetNamedCalls('send_stanza') + self.assertEqual(0, len(calls)) + +from session import ChatControlSession + +class MockWindow(Mock): + def __init__(self, *args): + Mock.__init__(self, *args) + self.window = Mock() + +class MockChatControl(Mock): + def __init__(self, *args): + Mock.__init__(self, *args) + + self.parent_win = MockWindow({'get_active_control': self}) + self.session = None + + def set_session(self, sess): + self.session = sess + + def __nonzero__(self): + return True + + def __eq__(self, other): + return self is other + +class MockInterface(Mock): + def __init__(self, *args): + Mock.__init__(self, *args) + self.msg_win_mgr = Mock() + self.roster = Mock() + + self.remote_ctrl = None + self.minimized_controls = { account_name: {} } + +class MockLogger(Mock): + def __init__(self): + Mock.__init__(self, {'write': None}) + +class MockContact(Mock): + def __nonzero__(self): + return True + +gajim.interface = MockInterface() +gajim.contacts.add_account(account_name) + +import notify + +class TestChatControlSession(unittest.TestCase): + def setUp(self): + self.jid = 'test@example.org/Gajim' + self.conn = MockConnection({'send_stanza': None}) + self.sess = ChatControlSession(self.conn, self.jid, None) + gajim.logger = MockLogger() + + # initially there are no events + self.assertEqual(0, len(gajim.events.get_events(account_name))) + + # no notifications have been sent + self.assertEqual(0, len(notify.notifications)) + + def tearDown(self): + # remove all events and notifications that were added + gajim.events._events = {} + notify.notifications = [] + + def receive_chat_msg(self, jid, msgtxt): + '''simulate receiving a chat message from jid''' + msg = xmpp.Message() + msg.setBody(msgtxt) + msg.setType('chat') + + tim = time.localtime() + encrypted = False + self.sess.received(jid, msgtxt, tim, encrypted, msg) + + # ----- custom assertions ----- + def assert_new_message_notification(self): + '''a new_message notification has been sent''' + self.assertEqual(1, len(notify.notifications)) + notif = notify.notifications[0] + self.assertEqual('new_message', notif[0]) + + def assert_first_message_notification(self): + '''this message was treated as a first message''' + self.assert_new_message_notification() + notif = notify.notifications[0] + params = notif[3] + first = params[1] + self.assert_(first, 'message should have been treated as a first message') + + def assert_not_first_message_notification(self): + '''this message was not treated as a first message''' + self.assert_new_message_notification() + notif = notify.notifications[0] + params = notif[3] + first = params[1] + self.assert_(not first, 'message was unexpectedly treated as a first message') + + # ----- tests ----- + def test_receive_nocontrol(self): + '''test receiving a message in a blank state''' + jid = 'bct@necronomicorp.com/Gajim' + msgtxt = 'testing one two three' + + self.receive_chat_msg(jid, msgtxt) + + # message was logged + calls = gajim.logger.mockGetNamedCalls('write') + self.assertEqual(1, len(calls)) + + # no ChatControl was open and autopopup was off + # so the message goes into the event queue + self.assertEqual(1, len(gajim.events.get_events(account_name))) + + self.assert_first_message_notification() + + # no control is attached to the session + self.assertEqual(None, self.sess.control) + + def test_receive_already_has_control(self): + '''test receiving a message with a session already attached to a control''' + jid = 'bct@necronomicorp.com/Gajim' + msgtxt = 'testing one two three' + + self.sess.control = MockChatControl() + + self.receive_chat_msg(jid, msgtxt) + + # message was logged + calls = gajim.logger.mockGetNamedCalls('write') + self.assertEqual(1, len(calls)) + + # the message does not go into the event queue + self.assertEqual(0, len(gajim.events.get_events(account_name))) + + self.assert_not_first_message_notification() + + # message was printed to the control + calls = self.sess.control.mockGetNamedCalls('print_conversation') + self.assertEqual(1, len(calls)) + + def test_received_orphaned_control(self): + '''test receiving a message when a control that doesn't have a session attached exists''' + + jid = 'bct@necronomicorp.com/Gajim' + msgtxt = 'testing one two three' + + ctrl = MockChatControl() + gajim.interface.msg_win_mgr = Mock({'get_sessionless_ctrl': ctrl}) + + self.receive_chat_msg(jid, msgtxt) + + # message was logged + calls = gajim.logger.mockGetNamedCalls('write') + self.assertEqual(1, len(calls)) + + # the message does not go into the event queue + self.assertEqual(0, len(gajim.events.get_events(account_name))) + + self.assert_not_first_message_notification() + + # this session is now attached to that control + self.assertEqual(self.sess, ctrl.session) + self.assertEqual(ctrl, self.sess.control, 'foo') + + # message was printed to the control + calls = ctrl.mockGetNamedCalls('print_conversation') + self.assertEqual(1, len(calls)) + +if __name__ == '__main__': + unittest.main() |