diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_disco.py | 155 | ||||
| -rw-r--r-- | tests/test_pubsubstanzas.py | 15 | ||||
| -rw-r--r-- | tests/test_statemachine.py | 261 |
3 files changed, 431 insertions, 0 deletions
diff --git a/tests/test_disco.py b/tests/test_disco.py new file mode 100644 index 00000000..bbe285a6 --- /dev/null +++ b/tests/test_disco.py @@ -0,0 +1,155 @@ +import unittest +from xml.etree import cElementTree as ET +from sleekxmpp.xmlstream.matcher.stanzapath import StanzaPath +from . import xmlcompare + +import sleekxmpp.plugins.xep_0030 as sd + +def stanzaPlugin(stanza, plugin): + stanza.plugin_attrib_map[plugin.plugin_attrib] = plugin + stanza.plugin_tag_map["{%s}%s" % (plugin.namespace, plugin.name)] = plugin + +class testdisco(unittest.TestCase): + + def setUp(self): + self.sd = sd + stanzaPlugin(self.sd.Iq, self.sd.DiscoInfo) + stanzaPlugin(self.sd.Iq, self.sd.DiscoItems) + + def try3Methods(self, xmlstring, iq): + iq2 = self.sd.Iq(None, self.sd.ET.fromstring(xmlstring)) + values = iq2.getValues() + iq3 = self.sd.Iq() + iq3.setValues(values) + self.failUnless(xmlstring == str(iq) == str(iq2) == str(iq3), str(iq)+"3 methods for creating stanza don't match") + + def testCreateInfoQueryNoNode(self): + """Testing disco#info query with no node.""" + iq = self.sd.Iq() + iq['id'] = "0" + iq['disco_info']['node'] = '' + xmlstring = """<iq id="0"><query xmlns="http://jabber.org/protocol/disco#info" /></iq>""" + self.try3Methods(xmlstring, iq) + + def testCreateInfoQueryWithNode(self): + """Testing disco#info query with a node.""" + iq = self.sd.Iq() + iq['id'] = "0" + iq['disco_info']['node'] = 'foo' + xmlstring = """<iq id="0"><query xmlns="http://jabber.org/protocol/disco#info" node="foo" /></iq>""" + self.try3Methods(xmlstring, iq) + + def testCreateInfoQueryNoNode(self): + """Testing disco#items query with no node.""" + iq = self.sd.Iq() + iq['id'] = "0" + iq['disco_items']['node'] = '' + xmlstring = """<iq id="0"><query xmlns="http://jabber.org/protocol/disco#items" /></iq>""" + self.try3Methods(xmlstring, iq) + + def testCreateItemsQueryWithNode(self): + """Testing disco#items query with a node.""" + iq = self.sd.Iq() + iq['id'] = "0" + iq['disco_items']['node'] = 'foo' + xmlstring = """<iq id="0"><query xmlns="http://jabber.org/protocol/disco#items" node="foo" /></iq>""" + self.try3Methods(xmlstring, iq) + + def testInfoIdentities(self): + """Testing adding identities to disco#info.""" + iq = self.sd.Iq() + iq['id'] = "0" + iq['disco_info']['node'] = 'foo' + iq['disco_info'].addIdentity('conference', 'text', 'Chatroom') + xmlstring = """<iq id="0"><query xmlns="http://jabber.org/protocol/disco#info" node="foo"><identity category="conference" type="text" name="Chatroom" /></query></iq>""" + self.try3Methods(xmlstring, iq) + + def testInfoFeatures(self): + """Testing adding features to disco#info.""" + iq = self.sd.Iq() + iq['id'] = "0" + iq['disco_info']['node'] = 'foo' + iq['disco_info'].addFeature('foo') + iq['disco_info'].addFeature('bar') + xmlstring = """<iq id="0"><query xmlns="http://jabber.org/protocol/disco#info" node="foo"><feature var="foo" /><feature var="bar" /></query></iq>""" + self.try3Methods(xmlstring, iq) + + def testItems(self): + """Testing adding features to disco#info.""" + iq = self.sd.Iq() + iq['id'] = "0" + iq['disco_items']['node'] = 'foo' + iq['disco_items'].addItem('user@localhost') + iq['disco_items'].addItem('user@localhost', 'foo') + iq['disco_items'].addItem('user@localhost', 'bar', 'Testing') + xmlstring = """<iq id="0"><query xmlns="http://jabber.org/protocol/disco#items" node="foo"><item jid="user@localhost" /><item node="foo" jid="user@localhost" /><item node="bar" jid="user@localhost" name="Testing" /></query></iq>""" + self.try3Methods(xmlstring, iq) + + def testAddRemoveIdentities(self): + """Test adding and removing identities to disco#info stanza""" + ids = [('automation', 'commands', 'AdHoc'), + ('conference', 'text', 'ChatRoom')] + + info = self.sd.DiscoInfo() + info.addIdentity(*ids[0]) + self.failUnless(info.getIdentities() == [ids[0]]) + + info.delIdentity('automation', 'commands') + self.failUnless(info.getIdentities() == []) + + info.setIdentities(ids) + self.failUnless(info.getIdentities() == ids) + + info.delIdentity('automation', 'commands') + self.failUnless(info.getIdentities() == [ids[1]]) + + info.delIdentities() + self.failUnless(info.getIdentities() == []) + + def testAddRemoveFeatures(self): + """Test adding and removing features to disco#info stanza""" + features = ['foo', 'bar', 'baz'] + + info = self.sd.DiscoInfo() + info.addFeature(features[0]) + self.failUnless(info.getFeatures() == [features[0]]) + + info.delFeature('foo') + self.failUnless(info.getFeatures() == []) + + info.setFeatures(features) + self.failUnless(info.getFeatures() == features) + + info.delFeature('bar') + self.failUnless(info.getFeatures() == ['foo', 'baz']) + + info.delFeatures() + self.failUnless(info.getFeatures() == []) + + def testAddRemoveItems(self): + """Test adding and removing items to disco#items stanza""" + items = [('user@localhost', None, None), + ('user@localhost', 'foo', None), + ('user@localhost', 'bar', 'Test')] + + info = self.sd.DiscoItems() + self.failUnless(True, ""+str(items[0])) + + info.addItem(*(items[0])) + self.failUnless(info.getItems() == [items[0]], info.getItems()) + + info.delItem('user@localhost') + self.failUnless(info.getItems() == []) + + info.setItems(items) + self.failUnless(info.getItems() == items) + + info.delItem('user@localhost', 'foo') + self.failUnless(info.getItems() == [items[0], items[2]]) + + info.delItems() + self.failUnless(info.getItems() == []) + + + +suite = unittest.TestLoader().loadTestsFromTestCase(testdisco) diff --git a/tests/test_pubsubstanzas.py b/tests/test_pubsubstanzas.py index 55407c16..089ee180 100644 --- a/tests/test_pubsubstanzas.py +++ b/tests/test_pubsubstanzas.py @@ -97,6 +97,21 @@ class testpubsubstanzas(unittest.TestCase): iq3.setValues(values) self.failUnless(xmlstring == str(iq) == str(iq2) == str(iq3)) + def testState(self): + "Testing iq/psstate stanzas" + from sleekxmpp.plugins import xep_0004 + iq = self.ps.Iq() + iq['psstate']['node']= 'mynode' + iq['psstate']['item']= 'myitem' + pl = ET.Element('{http://andyet.net/protocol/pubsubqueue}claimed') + iq['psstate']['payload'] = pl + xmlstring = """<iq id="0"><state xmlns="http://jabber.org/protocol/psstate" node="mynode" item="myitem"><claimed xmlns="http://andyet.net/protocol/pubsubqueue" /></state></iq>""" + iq2 = self.ps.Iq(None, self.ps.ET.fromstring(xmlstring)) + iq3 = self.ps.Iq() + values = iq2.getValues() + iq3.setValues(values) + self.failUnless(xmlstring == str(iq) == str(iq2) == str(iq3)) + def testDefault(self): "Testing iq/pubsub_owner/default stanzas" from sleekxmpp.plugins import xep_0004 diff --git a/tests/test_statemachine.py b/tests/test_statemachine.py new file mode 100644 index 00000000..e44b8e48 --- /dev/null +++ b/tests/test_statemachine.py @@ -0,0 +1,261 @@ +import unittest +import time, threading, random, functools + +if __name__ == '__main__': + import sys, os + sys.path.insert(0, os.getcwd()) + import sleekxmpp.xmlstream.statemachine as sm + + +class testStateMachine(unittest.TestCase): + + def setUp(self): pass + + + def testDefaults(self): + "Test ensure transitions occur correctly in a single thread" + s = sm.StateMachine(('one','two','three')) + self.assertTrue(s['one']) + self.failIf(s['two']) + try: + s['booga'] + self.fail('s.booga is an invalid state and should throw an exception!') + except: pass #expected exception + + # just make sure __str__ works, no reason to test its exact value: + print str(s) + + + def testTransitions(self): + "Test ensure transitions occur correctly in a single thread" + s = sm.StateMachine(('one','two','three')) + + self.assertTrue( s.transition('one', 'two') ) + self.assertTrue( s['two'] ) + self.failIf( s['one'] ) + + self.assertTrue( s.transition('two', 'three') ) + self.assertTrue( s['three'] ) + self.failIf( s['two'] ) + + self.assertTrue( s.transition('three', 'one') ) + self.assertTrue( s['one'] ) + self.failIf( s['three'] ) + + # should return False immediately w/ no wait: + self.failIf( s.transition('three', 'one') ) + self.assertTrue( s['one'] ) + self.failIf( s['three'] ) + + # test fail condition w/ a short delay: + self.failIf( s.transition('two', 'three') ) + + # Ensure bad states are weeded out: + try: + s.transition('blah', 'three') + s.fail('Exception expected') + except: pass + + try: + s.transition('one', 'blahblah') + s.fail('Exception expected') + except: pass + + + def testTransitionsBlocking(self): + "Test that transitions block from more than one thread" + + s = sm.StateMachine(('one','two','three')) + self.assertTrue(s['one']) + + now = time.time() + self.failIf( s.transition('two', 'one', wait=5.0) ) + self.assertTrue( time.time() > now + 4 ) + self.assertTrue( time.time() < now + 7 ) + + def testThreadedTransitions(self): + "Test that transitions are atomic in > one thread" + + s = sm.StateMachine(('one','two','three')) + self.assertTrue(s['one']) + + thread_state = {'ready': False, 'transitioned': False} + def t1(): + if s['two']: + print 'thread has already transitioned!' + self.fail() + thread_state['ready'] = True + print 'Thread is ready' + # this will block until the main thread transitions to 'two' + self.assertTrue( s.transition('two','three', wait=20) ) + print 'transitioned to three!' + thread_state['transitioned'] = True + + thread = threading.Thread(target=t1) + thread.daemon = True + thread.start() + start = time.time() + while not thread_state['ready']: + print 'not ready' + if time.time() > start+10: self.fail('Timeout waiting for thread to init!') + time.sleep(0.1) + time.sleep(0.2) # the thread should be blocking on the 'transition' call at this point. + self.failIf( thread_state['transitioned'] ) # ensure it didn't 'go' yet. + print 'transitioning to two!' + self.assertTrue( s.transition('one','two') ) + time.sleep(0.2) # second thread should have transitioned now: + self.assertTrue( thread_state['transitioned'] ) + + + def testForRaceCondition(self): + """Attempt to allow two threads to perform the same transition; + only one should ever make it.""" + + s = sm.StateMachine(('one','two','three')) + + def t1(num): + while True: + if not trigger['go'] or thread_state[num] in (True,False): + time.sleep( random.random()/100 ) # < .01s + if thread_state[num] == 'quit': break + continue + + thread_state[num] = s.transition('one','two' ) +# print '-', + + thread_count = 20 + threads = [] + thread_state = {} + def reset(): + for c in range(thread_count): thread_state[c] = "reset" + trigger = {'go':False} # use of a plain boolean seems to be non-volatile between threads. + + for c in range(thread_count): + thread_state[c] = "reset" + thread = threading.Thread( target= functools.partial(t1,c) ) + threads.append( thread ) + thread.daemon = True + thread.start() + + for x in range(100): # this will take 10s to execute +# print "+", + trigger['go'] = True + time.sleep(.1) + trigger['go'] = False + winners = 0 + for (num, state) in thread_state.items(): + if state == True: winners = winners +1 + elif state != False: raise Exception( "!%d!%s!" % (num,state) ) + + self.assertEqual( 1, winners, "Expected one winner! %d" % winners ) + self.assertTrue( s.ensure('two') ) + self.assertTrue( s.transition('two','one') ) # return to the first state. + reset() + + # now let the threads quit gracefully: + for c in range(thread_count): thread_state[c] = 'quit' + time.sleep(2) + + + def testTransitionFunctions(self): + "test that a `func` argument allows or blocks the transition correctly." + + s = sm.StateMachine(('one','two','three')) + + def alwaysFalse(): return False + def alwaysTrue(): return True + + self.failIf( s.transition('one','two', func=alwaysFalse) ) + self.assertTrue(s['one']) + self.failIf(s['two']) + + self.assertTrue( s.transition('one','two', func=alwaysTrue) ) + self.failIf(s['one']) + self.assertTrue(s['two']) + + + def testTransitionFuncException(self): + "if a transition function throws an exeption, ensure we're in a sane state" + + s = sm.StateMachine(('one','two','three')) + + def alwaysException(): raise Exception('whups!') + + try: + self.failIf( s.transition('one','two', func=alwaysException) ) + self.fail("exception should have been thrown") + except: pass #expected exception + + self.assertTrue(s['one']) + self.failIf(s['two']) + + # ensure a subsequent attempt completes normally: + self.assertTrue( s.transition('one','two') ) + self.failIf(s['one']) + self.assertTrue(s['two']) + + + def testContextManager(self): + + s = sm.StateMachine(('one','two','three')) + + with s.transition_ctx('one','two'): + self.assertTrue( s['one'] ) + self.failIf( s['two'] ) + + #successful transition b/c no exception was thrown + self.assertTrue( s['two'] ) + self.failIf( s['one'] ) + + # failed transition because exception is thrown: + try: + with s.transition_ctx('two','three'): + raise Exception("boom!") + self.fail('exception expected') + except: pass + + self.failIf( s.current_state() in ('one','three') ) + self.assertTrue( s['two'] ) + + def testCtxManagerTransitionFailure(self): + + s = sm.StateMachine(('one','two','three')) + + with s.transition_ctx('two','three') as result: + self.failIf( result ) + self.assertTrue( s['one'] ) + self.failIf( s.current_state in ('two','three') ) + + self.assertTrue( s['one'] ) + + def r1(): + print 'thread 1 started' + self.assertTrue( s.transition('one','two') ) + print 'thread 1 transitioned' + + def r2(): + print 'thread 2 started' + self.failIf( s['two'] ) + with s.transition_ctx('two','three', 10) as result: + self.assertTrue( result ) + self.assertTrue( s['two'] ) + print 'thread 2 will transition on exit from the context manager...' + self.assertTrue( s['three'] ) + print 'transitioned to %s' % s.current_state() + + t1 = threading.Thread(target=r1) + t2 = threading.Thread(target=r2) + + t2.start() # this should block until r1 goes + time.sleep(1) + t1.start() + + t1.join() + t2.join() + + self.assertTrue( s['three'] ) + + +suite = unittest.TestLoader().loadTestsFromTestCase(testStateMachine) + +if __name__ == '__main__': unittest.main() |