Lerking/Phoenix
1
0
Fork 0
mirror of https://github.com/wxWidgets/Phoenix.git synced 2026-01-07 04:20:07 +01:00
Code Issues Actions Packages Projects Releases Wiki Activity

Pubsub unittest updates from Werner

Browse Source 
git-svn-id: https://svn.wxwidgets.org/svn/wx/wxPython/Phoenix/trunk@73903 c3d73ce0-8a6f-49c7-b76d-6d57e0e08775
This commit is contained in:
Robin Dunn
2013-05-02 01:01:30 +00:00
parent 6d4555b60f
commit 3105b4d95a
13 changed files with 359 additions and 296 deletions
Download Patch File Download Diff File Expand all files Collapse all files

180
unittests/test_lib_pubsub_api3.py
View File

@@ -11,15 +11,11 @@ disabled, as it is more rigorous for testing purposes.
import imp_unittest, unittest
import wtc
from wx.lib.pubsub import pub
from wx.lib.pubsub.utils.notification import IgnoreNotificationsMixin
#---------------------------------------------------------------------------
class lib_pubsub_Except(wtc.WidgetTestCase):
class lib_pubsub_Except(wtc.PubsubTestCase):
def testDOAListenerPubsub(self):
# Verify that a 'temporary' listener (one that will be garbage collected
@@ -34,13 +30,15 @@ class lib_pubsub_Except(wtc.WidgetTestCase):
def __call__(self):
pass
pub.subscribe( Wrapper(listener), 'testDOAListenerPubsub')
assert not pub.getTopic('testDOAListenerPubsub').hasListeners()
assert pub.isValid(listener, 'testDOAListenerPubsub')
self.pub.subscribe( Wrapper(listener), 'testDOAListenerPubsub')
assert not self.pub.getTopic('testDOAListenerPubsub').hasListeners()
assert self.pub.isValid(listener, 'testDOAListenerPubsub')
def testDeadListener(self):
# create a listener for listeners that have died
from wx.lib.pubsub.utils.notification import IgnoreNotificationsMixin
class DeathListener(IgnoreNotificationsMixin):
listenerStr = ''
def notifyDeadListener(self, pubListener, topicObj):
@@ -48,8 +46,8 @@ class lib_pubsub_Except(wtc.WidgetTestCase):
DeathListener.listenerStr = pubListener.name()
dl = DeathListener()
dl.assertEqual = self.assertEqual
pub.addNotificationHandler(dl)
pub.setNotificationFlags(deadListener=True)
self.pub.addNotificationHandler(dl)
self.pub.setNotificationFlags(deadListener=True)
# define a topic, subscribe to it, and kill its listener:
class TempListener:
@@ -58,61 +56,61 @@ class lib_pubsub_Except(wtc.WidgetTestCase):
def __del__(self):
pass #print 'being deleted'
tempListener = TempListener()
expectLisrStr, _ = pub.getListenerID(tempListener)
pub.subscribe(tempListener, 'sadTopic')
expectLisrStr, _ = self.pub.getListenerID(tempListener)
self.pub.subscribe(tempListener, 'sadTopic')
del tempListener
# verify:
assert DeathListener.listenerStr.startswith(expectLisrStr), \
'"%s" !~ "%s"' % (DeathListener.listenerStr, expectLisrStr)
pub.addNotificationHandler(None)
pub.clearNotificationHandlers()
self.pub.addNotificationHandler(None)
self.pub.clearNotificationHandlers()
def testSubscribe(self):
topicName = 'testSubscribe'
def proto(a, b, c=None):
pass
pub.getOrCreateTopic(topicName, proto)
self.pub.getOrCreateTopic(topicName, proto)
def listener(a, b, c=None): pass
# verify that pub.isValid() works too
pub.validate(listener, topicName)
assert pub.isValid(listener, topicName)
# verify that self.pub.isValid() works too
self.pub.validate(listener, topicName)
assert self.pub.isValid(listener, topicName)
self.assertEqual(pub.getTopic(topicName).getNumListeners(), 0)
self.assertEqual(pub.getAssociatedTopics(listener), [])
assert not pub.isSubscribed(listener, topicName)
assert pub.subscribe(listener, topicName)
assert pub.isSubscribed(listener, topicName)
self.assertEqual(self.pub.getTopic(topicName).getNumListeners(), 0)
self.assertEqual(self.pub.getAssociatedTopics(listener), [])
assert not self.pub.isSubscribed(listener, topicName)
assert self.pub.subscribe(listener, topicName)
assert self.pub.isSubscribed(listener, topicName)
def topicNames(listener):
return [t.getName() for t in pub.getAssociatedTopics(listener)]
return [t.getName() for t in self.pub.getAssociatedTopics(listener)]
self.assertEqual(topicNames(listener), [topicName])
# should do nothing if already subscribed:
assert not pub.subscribe(listener, topicName)[1]
self.assertEqual(pub.getTopic(topicName).getNumListeners(), 1)
assert not self.pub.subscribe(listener, topicName)[1]
self.assertEqual(self.pub.getTopic(topicName).getNumListeners(), 1)
# test pub.getAssociatedTopics()
pub.subscribe(listener, 'lt2', )
# test self.pub.getAssociatedTopics()
self.pub.subscribe(listener, 'lt2', )
self.assertEqual(set(topicNames(listener)),
set([topicName,'lt2']))
pub.subscribe(listener, 'lt1.lst1')
self.pub.subscribe(listener, 'lt1.lst1')
self.assertEqual(set(topicNames(listener)),
set([topicName,'lt2','lt1.lst1']))
# test ALL_TOPICS
def listenToAll():
pass
pub.subscribe(listenToAll, pub.ALL_TOPICS)
self.assertEqual(topicNames(listenToAll), [pub.ALL_TOPICS])
self.pub.subscribe(listenToAll, self.pub.ALL_TOPICS)
self.assertEqual(topicNames(listenToAll), [self.pub.ALL_TOPICS])
def testMissingReqdArgs(self):
def proto(a, b, c=None):
pass
pub.getOrCreateTopic('missingReqdArgs', proto)
self.assertRaises(pub.SenderMissingReqdArgs, pub.sendMessage,
self.pub.getOrCreateTopic('missingReqdArgs', proto)
self.assertRaises(self.pub.SenderMissingReqdArgs, self.pub.sendMessage,
'missingReqdArgs', a=1)
@@ -127,23 +125,23 @@ class lib_pubsub_Except(wtc.WidgetTestCase):
def listen1(self, **kwarg):
self.count += 1
self.heardTopic = True
def listen2(self, msgTopic=pub.AUTO_TOPIC, **kwarg):
def listen2(self, msgTopic=self.pub.AUTO_TOPIC, **kwarg):
self.listen2Topics.append(msgTopic.getName())
my = MyListener()
pub.subscribe(my.listen0, 'testSendTopic')
pub.subscribe(my.listen1, 'testSendTopic')
pub.subscribe(my.listen2, 'testSendTopic')
self.pub.subscribe(my.listen0, 'testSendTopic')
self.pub.subscribe(my.listen1, 'testSendTopic')
self.pub.subscribe(my.listen2, 'testSendTopic')
pub.sendMessage('testSendTopic')
self.pub.sendMessage('testSendTopic')
self.assertEqual(my.count, 1)
self.assertEqual(my.heardTopic, True)
pub.subscribe(my.listen0, 'testSendTopic.subtopic')
pub.subscribe(my.listen1, 'testSendTopic.subtopic')
pub.subscribe(my.listen2, 'testSendTopic.subtopic')
self.pub.subscribe(my.listen0, 'testSendTopic.subtopic')
self.pub.subscribe(my.listen1, 'testSendTopic.subtopic')
self.pub.subscribe(my.listen2, 'testSendTopic.subtopic')
pub.sendMessage('testSendTopic.subtopic')
self.pub.sendMessage('testSendTopic.subtopic')
self.assertEqual(my.count, 3)
self.assertEqual([], [topic for topic in my.listen2Topics
if topic not in ('testSendTopic', 'testSendTopic.subtopic')] )
@@ -154,16 +152,16 @@ class lib_pubsub_Except(wtc.WidgetTestCase):
pass
def listenAllArgs(arg1=None, **kwargs):
pass
def listenAllArgs2(arg1=None, msgTopic=pub.AUTO_TOPIC, **kwargs):
def listenAllArgs2(arg1=None, msgTopic=self.pub.AUTO_TOPIC, **kwargs):
pass
pub.subscribe(listen, 'testAcceptAllArgs')
self.pub.subscribe(listen, 'testAcceptAllArgs')
pub.subscribe(listenAllArgs, 'testAcceptAllArgs')
pub.subscribe(listenAllArgs2, 'testAcceptAllArgs')
self.pub.subscribe(listenAllArgs, 'testAcceptAllArgs')
self.pub.subscribe(listenAllArgs2, 'testAcceptAllArgs')
pub.subscribe(listenAllArgs2, 'testAcceptAllArgs.subtopic')
pub.subscribe(listenAllArgs, 'testAcceptAllArgs.subtopic')
self.pub.subscribe(listenAllArgs2, 'testAcceptAllArgs.subtopic')
self.pub.subscribe(listenAllArgs, 'testAcceptAllArgs.subtopic')
def testUnsubAll(self):
@@ -181,15 +179,15 @@ class lib_pubsub_Except(wtc.WidgetTestCase):
lisnr3 = MyListener()
lisnr4 = lisnr3.meth
def lisnrSub(listener=None, topic=None, newSub=None): pass
pub.subscribe(lisnrSub, 'pubsub.subscribe')
self.assertEqual(pub.getTopic('pubsub.subscribe').getNumListeners(), 1)
self.pub.subscribe(lisnrSub, 'pubsub.subscribe')
self.assertEqual(self.pub.getTopic('pubsub.subscribe').getNumListeners(), 1)
def subAll():
pub.subscribe(lisnr1, 'testUnsubAll')
pub.subscribe(lisnr2, 'testUnsubAll')
pub.subscribe(lisnr3, 'testUnsubAll')
pub.subscribe(lisnr4, 'testUnsubAll')
self.assertEqual(pub.getTopic('testUnsubAll').getNumListeners(), 4)
self.pub.subscribe(lisnr1, 'testUnsubAll')
self.pub.subscribe(lisnr2, 'testUnsubAll')
self.pub.subscribe(lisnr3, 'testUnsubAll')
self.pub.subscribe(lisnr4, 'testUnsubAll')
self.assertEqual(self.pub.getTopic('testUnsubAll').getNumListeners(), 4)
def filter(lisnr):
passes = str(lisnr).endswith('meth')
@@ -197,23 +195,23 @@ class lib_pubsub_Except(wtc.WidgetTestCase):
# test unsub many non-pubsub topic listeners
subAll()
pub.unsubAll('testUnsubAll')
self.assertEqual(pub.getTopic('testUnsubAll').getNumListeners(), 0)
self.assertEqual(pub.getTopic('pubsub.subscribe').getNumListeners(), 1)
self.pub.unsubAll('testUnsubAll')
self.assertEqual(self.pub.getTopic('testUnsubAll').getNumListeners(), 0)
self.assertEqual(self.pub.getTopic('pubsub.subscribe').getNumListeners(), 1)
# now same but with filter:
subAll()
unsubed = pub.unsubAll('testUnsubAll', listenerFilter=filter)
self.assertEqual(pub.getTopic('testUnsubAll').getNumListeners(), 3)
self.assertEqual(pub.getTopic('pubsub.subscribe').getNumListeners(), 1)
unsubed = self.pub.unsubAll('testUnsubAll', listenerFilter=filter)
self.assertEqual(self.pub.getTopic('testUnsubAll').getNumListeners(), 3)
self.assertEqual(self.pub.getTopic('pubsub.subscribe').getNumListeners(), 1)
# test unsub all listeners of all topics
subAll()
self.assertEqual(pub.getTopic('testUnsubAll').getNumListeners(), 4)
unsubed = pub.unsubAll(listenerFilter=filter)
self.assertEqual(self.pub.getTopic('testUnsubAll').getNumListeners(), 4)
unsubed = self.pub.unsubAll(listenerFilter=filter)
self.assertEqual(unsubed, [lisnr4])
self.assertEqual(pub.getTopic('testUnsubAll').getNumListeners(), 3)
self.assertEqual(pub.getTopic('pubsub.subscribe').getNumListeners(), 1)
unsubed = set( pub.unsubAll() )
self.assertEqual(self.pub.getTopic('testUnsubAll').getNumListeners(), 3)
self.assertEqual(self.pub.getTopic('pubsub.subscribe').getNumListeners(), 1)
unsubed = set( self.pub.unsubAll() )
expect = set([lisnr1, lisnrSub, lisnr3, lisnr2])
# at least all the 'expected' ones were unsub'd; will be others if this
# test is run after other unit tests in same nosetests run
@@ -221,29 +219,29 @@ class lib_pubsub_Except(wtc.WidgetTestCase):
def testSendForUndefinedTopic(self):
pub.sendMessage('testSendForUndefinedTopic')
assert pub.getTopic('testSendForUndefinedTopic')
self.assertEqual(pub.getTopic('testSendForUndefinedTopic').getArgs(),
self.pub.sendMessage('testSendForUndefinedTopic')
assert self.pub.getTopic('testSendForUndefinedTopic')
self.assertEqual(self.pub.getTopic('testSendForUndefinedTopic').getArgs(),
(None, None))
# must also check for subtopics if parents have listeners since
# filtering of args is affected
def listener():
pass
pub.subscribe(listener, 'testSendForUndefinedTopic')
pub.sendMessage('testSendForUndefinedTopic.subtopic', msg='something')
self.pub.subscribe(listener, 'testSendForUndefinedTopic')
self.pub.sendMessage('testSendForUndefinedTopic.subtopic', msg='something')
def testTopicUnspecifiedError(self):
self.assertRaises(pub.ListenerSpecIncomplete, pub.setTopicUnspecifiedFatal)
pub.setTopicUnspecifiedFatal(checkExisting=False)
self.assertRaises(self.pub.ListenerSpecIncomplete, self.pub.setTopicUnspecifiedFatal)
self.pub.setTopicUnspecifiedFatal(checkExisting=False)
def fn():
pass
LSI = pub.ListenerSpecIncomplete
self.assertRaises(LSI, pub.sendMessage, 'testTopicUnspecifiedError')
self.assertRaises(LSI, pub.subscribe, fn, 'testTopicUnspecifiedError')
pub.setTopicUnspecifiedFatal(False)
pub.sendMessage('testTopicUnspecifiedError')
pub.subscribe(fn, 'testTopicUnspecifiedError')
LSI = self.pub.ListenerSpecIncomplete
self.assertRaises(LSI, self.pub.sendMessage, 'testTopicUnspecifiedError')
self.assertRaises(LSI, self.pub.subscribe, fn, 'testTopicUnspecifiedError')
self.pub.setTopicUnspecifiedFatal(False)
self.pub.sendMessage('testTopicUnspecifiedError')
self.pub.subscribe(fn, 'testTopicUnspecifiedError')
def testArgSpecDerivation(self):
@@ -266,15 +264,15 @@ class lib_pubsub_Except(wtc.WidgetTestCase):
# with getOrCreateTopic(topic, proto), the 'required args' set
# is garanteed to be a subset of 'all args'
pub.getOrCreateTopic('tasd', ok_0)
pub.getOrCreateTopic('tasd.t_1', ok_1)
self.assertRaises(pub.ListenerSpecInvalid, pub.getOrCreateTopic,
self.pub.getOrCreateTopic('tasd', ok_0)
self.pub.getOrCreateTopic('tasd.t_1', ok_1)
self.assertRaises(self.pub.ListenerSpecInvalid, self.pub.getOrCreateTopic,
'tasd.t_1.t_11', err_11)
self.assertRaises(pub.ListenerSpecInvalid, pub.getOrCreateTopic,
self.assertRaises(self.pub.ListenerSpecInvalid, self.pub.getOrCreateTopic,
'tasd.t_1.t_12', err_12)
pub.getOrCreateTopic('tasd.t_2', ok_2)
pub.getOrCreateTopic('tasd.t_2.t_21', ok_21)
self.assertRaises(pub.ListenerSpecInvalid, pub.getOrCreateTopic,
self.pub.getOrCreateTopic('tasd.t_2', ok_2)
self.pub.getOrCreateTopic('tasd.t_2.t_21', ok_21)
self.assertRaises(self.pub.ListenerSpecInvalid, self.pub.getOrCreateTopic,
'tasd.t_2.t_22', err_22)
# with newTopic(), 'required args' specified separately so
@@ -282,19 +280,19 @@ class lib_pubsub_Except(wtc.WidgetTestCase):
def check(subName, required=(), **args):
tName = 'tasd.'+subName
try:
pub.newTopic(tName, 'desc', required, **args)
msg = 'Should have raised pub.ListenerSpecInvalid for %s, %s, %s'
self.pub.newTopic(tName, 'desc', required, **args)
msg = 'Should have raised self.pub.ListenerSpecInvalid for %s, %s, %s'
assert False, msg % (tName, required, args)
except pub.ListenerSpecInvalid, exc:
except self.pub.ListenerSpecInvalid, exc:
#import traceback
#traceback.print_exc()
print 'As expected: ', exc
pub.newTopic('tasd.t_1.t_13', 'desc', ('arg1',), arg1='docs for arg1') # ok_1
self.pub.newTopic('tasd.t_1.t_13', 'desc', ('arg1',), arg1='docs for arg1') # ok_1
check('t_1.t_14', arg1='docs for arg1') # err_11
check('t_1.t_15', ('arg2',), arg2='docs for arg2') # err_12
pub.newTopic('tasd.t_2.t_23', 'desc', ('arg1',), arg1='docs for arg1') # ok_21
self.pub.newTopic('tasd.t_2.t_23', 'desc', ('arg1',), arg1='docs for arg1') # ok_21
check('t_2.t_24', ('arg2',), arg2='docs for arg2') # err_22
# check when no inheritence involved

7
unittests/test_lib_pubsub_defaultlog.py
View File

@@ -9,23 +9,20 @@
import imp_unittest, unittest
import wtc
from wx.lib.pubsub import pub
from wx.lib.pubsub.utils import notification
from StringIO import StringIO
#---------------------------------------------------------------------------
class lib_pubsub_DefaultLog(wtc.WidgetTestCase):
class lib_pubsub_DefaultLog(wtc.PubsubTestCase):
def testNotifications(self):
capture = StringIO()
logger = notification.useNotifyByWriteFile(capture)
def block():
def listener(): pass
pub.subscribe(listener, 'testNotifications')
self.pub.subscribe(listener, 'testNotifications')
block()

54
unittests/test_lib_pubsub_except.py
View File

@@ -9,8 +9,6 @@
import imp_unittest, unittest
import wtc
from wx.lib.pubsub import pub
def throws():
raise RuntimeError('test')
@@ -18,35 +16,35 @@ def throws():
#---------------------------------------------------------------------------
class lib_pubsub_Except(wtc.WidgetTestCase):
class lib_pubsub_Except(wtc.PubsubTestCase):
def testHandleExcept1a(self):
from wx.lib.pubsub.utils.exchandling import ExcPublisher
excPublisher = ExcPublisher( pub.getDefaultTopicMgr() )
pub.setListenerExcHandler(excPublisher)
excPublisher = ExcPublisher(self.pub.getDefaultTopicMgr() )
self.pub.setListenerExcHandler(excPublisher)
# create a listener that raises an exception:
from lib_pubsub_except_raisinglistener import getRaisingListener
raisingListener = getRaisingListener()
pub.setNotificationFlags(all=False)
pub.subscribe(raisingListener, 'testHandleExcept1a')
self.pub.setNotificationFlags(all=False)
self.pub.subscribe(raisingListener, 'testHandleExcept1a')
# first test when a listener raises an exception and exception listener also raises!
class BadUncaughtExcListener:
def __call__(self, listenerStr=None, excTraceback=None):
raise RuntimeError('bad exception listener!')
handler = BadUncaughtExcListener()
pub.subscribe(handler, ExcPublisher.topicUncaughtExc)
self.assertRaises(pub.ExcHandlerError, pub.sendMessage,
self.pub.subscribe(handler, ExcPublisher.topicUncaughtExc)
self.assertRaises(self.pub.ExcHandlerError, self.pub.sendMessage,
'testHandleExcept1a')
def testHandleExcept1b(self):
# create a listener that raises an exception:
from lib_pubsub_except_raisinglistener import getRaisingListener
raisingListener = getRaisingListener()
pub.subscribe(raisingListener, 'testHandleExcept1b')
self.pub.subscribe(raisingListener, 'testHandleExcept1b')
# subscribe a good exception listener and validate
# create the listener for uncaught exceptions in listeners:
@@ -72,25 +70,25 @@ class lib_pubsub_Except(wtc.WidgetTestCase):
assert msg.endswith("global name 'RuntimeError2' is not defined\n")
from wx.lib.pubsub.utils.exchandling import ExcPublisher
topic = pub.getTopic(ExcPublisher.topicUncaughtExc)
topic = self.pub.getTopic(ExcPublisher.topicUncaughtExc)
assert not topic.hasListeners()
handler = UncaughtExcListener()
handler.assertEqual = self.assertEqual
pub.subscribe(handler, ExcPublisher.topicUncaughtExc)
pub.sendMessage('testHandleExcept1b')
self.pub.subscribe(handler, ExcPublisher.topicUncaughtExc)
self.pub.sendMessage('testHandleExcept1b')
# verify that listener isn't stuck in a cyclic reference by sys.exc_info()
del raisingListener
assert not pub.getTopic('testHandleExcept1b').hasListeners()
assert not self.pub.getTopic('testHandleExcept1b').hasListeners()
def testHandleExcept2(self):
#Test sendMessage when one handler, then change handler and verify changed
testTopic = 'testTopics.testHandleExcept2'
pub.subscribe(throws, testTopic)
pub.setListenerExcHandler(None)
self.pub.subscribe(throws, testTopic)
self.pub.setListenerExcHandler(None)
#pubsub.utils.notification.useNotifyByWriteFile()
#assert_equal( pub.getTopic(testTopic).getNumListeners(), 1 )
#assert_equal( self.pub.getTopic(testTopic).getNumListeners(), 1 )
expect = None
@@ -111,33 +109,33 @@ class lib_pubsub_Except(wtc.WidgetTestCase):
global expect
expect = HandlerClass.__name__ #'MyExcHandler'
excHandler = HandlerClass()
pub.setListenerExcHandler(excHandler)
pub.sendMessage(testTopic)
self.pub.setListenerExcHandler(excHandler)
self.pub.sendMessage(testTopic)
assert expect is None
doHandling(MyExcHandler)
doHandling(MyExcHandler2)
# restore to no handling and verify:
pub.setListenerExcHandler(None)
self.assertRaises(RuntimeError, pub.sendMessage, testTopic)
self.pub.setListenerExcHandler(None)
self.assertRaises(RuntimeError, self.pub.sendMessage, testTopic)
def testNoExceptionHandling1(self):
pub.setListenerExcHandler(None)
self.pub.setListenerExcHandler(None)
def raises():
raise RuntimeError('test')
pub.getOrCreateTopic('testNoExceptionTrapping')
pub.subscribe(raises, 'testNoExceptionTrapping')
self.assertRaises(RuntimeError, pub.sendMessage, 'testNoExceptionTrapping')
self.pub.getOrCreateTopic('testNoExceptionTrapping')
self.pub.subscribe(raises, 'testNoExceptionTrapping')
self.assertRaises(RuntimeError, self.pub.sendMessage, 'testNoExceptionTrapping')
def testNoExceptionHandling2(self):
testTopic = 'testTopics.testNoExceptionHandling'
pub.subscribe(throws, testTopic)
assert pub.getListenerExcHandler() is None
self.assertRaises(RuntimeError, pub.sendMessage, testTopic)
self.pub.subscribe(throws, testTopic)
assert self.pub.getListenerExcHandler() is None
self.assertRaises(RuntimeError, self.pub.sendMessage, testTopic)
#---------------------------------------------------------------------------

2
unittests/test_lib_pubsub_listener.py
View File

@@ -26,7 +26,7 @@ class ArgsInfoMock:
#---------------------------------------------------------------------------
class lib_pubsub_ArgsInfo(wtc.WidgetTestCase):
class lib_pubsub_ArgsInfo(wtc.PubsubTestCase):
def test0_ArgsInfo(self):
def listener0(msgTopic = Listener.AUTO_TOPIC):

135
unittests/test_lib_pubsub_notify.py
View File

@@ -11,40 +11,38 @@ import wtc
from difflib import ndiff, unified_diff, context_diff
# setup notification and logging
from wx.lib.pubsub import pub
from wx.lib.pubsub.utils.notification import useNotifyByWriteFile, INotificationHandler
#---------------------------------------------------------------------------
class lib_pubsub_Notify(wtc.WidgetTestCase):
def captureStdout(self):
from StringIO import StringIO
capture = StringIO()
useNotifyByWriteFile( fileObj = capture )
return capture
class lib_pubsub_Notify(wtc.PubsubTestCase):
def testNotifyByPrint(self):
capture = self.captureStdout()
from wx.lib.pubsub.utils.notification import useNotifyByWriteFile
def captureStdout():
from StringIO import StringIO
capture = StringIO()
useNotifyByWriteFile( fileObj = capture )
return capture
capture = captureStdout()
print 'capture before sub: %s' % capture.read()
def listener1(arg1):
pass
pub.subscribe(listener1, 'baz')
pub.sendMessage('baz', arg1=123)
pub.unsubscribe(listener1, 'baz')
self.pub.subscribe(listener1, 'baz')
self.pub.sendMessage('baz', arg1=123)
self.pub.unsubscribe(listener1, 'baz')
def doa():
def listener2():
pass
pub.subscribe(listener2, 'bar')
self.pub.subscribe(listener2, 'bar')
doa()
pub.delTopic('baz')
self.pub.delTopic('baz')
expect = '''\
PUBSUB: New topic "baz" created
@@ -57,102 +55,17 @@ PUBSUB: New topic "bar" created
PUBSUB: Subscribed listener "listener2" to topic "bar"
PUBSUB: Listener "listener2" of Topic "bar" has died
PUBSUB: Topic "baz" destroyed
'''
captured = capture.getvalue()
'''.strip()
captured = capture.getvalue().strip()
print "captured -----------------"
print captured
print "expected -----------------"
print expect
# strip as other wise one has \n, at least on windows
assert captured.strip() == expect.strip(), \
assert captured == expect, \
'\n'.join( unified_diff(expect.splitlines(), captured.splitlines(), n=0) )
def testFlagChanges(self):
savedFlags = pub.getNotificationFlags()
pub.setNotificationFlags(all=True, sendMessage=False, deadListener=False)
flags = pub.getNotificationFlags()
assert not flags['sendMessage']
assert not flags['deadListener']
assert flags['newTopic']
assert flags['delTopic']
assert flags['subscribe']
assert flags['unsubscribe']
pub.setNotificationFlags(subscribe=False, deadListener=True)
flags = pub.getNotificationFlags()
assert not flags['sendMessage']
assert not flags['subscribe']
assert flags['newTopic']
assert flags['delTopic']
assert flags['deadListener']
assert flags['unsubscribe']
pub.setNotificationFlags(all=False, subscribe=True, unsubscribe=True)
flags = pub.getNotificationFlags()
assert not flags['sendMessage']
assert not flags['deadListener']
assert not flags['newTopic']
assert not flags['delTopic']
assert flags['subscribe']
assert flags['unsubscribe']
pub.setNotificationFlags(** savedFlags)
def testNotifications(self):
class Handler(INotificationHandler):
def __init__(self):
self.resetCounts()
def resetCounts(self):
self.counts = dict(send=0, sub=0, unsub=0, delt=0, newt=0, dead=0, all=0)
def notifySubscribe(self, pubListener, topicObj, newSub):
self.counts['sub'] += 1
def notifyUnsubscribe(self, pubListener, topicObj):
self.counts['unsub'] += 1
def notifyDeadListener(self, pubListener, topicObj):
self.counts['dead'] += 1
def notifySend(self, stage, topicObj, pubListener=None):
if stage == 'pre': self.counts['send'] += 1
def notifyNewTopic(self, topicObj, description, required, argsDocs):
self.counts['newt'] += 1
def notifyDelTopic(self, topicName):
self.counts['delt'] += 1
notifiee = Handler()
pub.addNotificationHandler(notifiee)
pub.setNotificationFlags(all=True)
def verify(**ref):
for key, val in notifiee.counts.iteritems():
if key in ref:
self.assertEqual(val, ref[key], "\n%s\n%s" % (notifiee.counts, ref) )
else:
self.assertEqual(val, 0, "%s = %s, expected 0" % (key, val))
notifiee.resetCounts()
verify()
def testListener():
pass
def testListener2():
pass
pub.getOrCreateTopic('newTopic')
verify(newt=1)
pub.subscribe(testListener, 'newTopic')
pub.subscribe(testListener2, 'newTopic')
verify(sub=2)
pub.sendMessage('newTopic')
verify(send=1)
del testListener
verify(dead=1)
pub.unsubscribe(testListener2,'newTopic')
verify(unsub=1)
pub.delTopic('newTopic')
verify(delt=1)
#---------------------------------------------------------------------------

41
unittests/test_lib_pubsub_notify2.py
View File

@@ -10,17 +10,17 @@ import imp_unittest, unittest
import wtc
from wx.lib.pubsub import pub
from wx.lib.pubsub.utils.notification import useNotifyByPubsubMessage
topicMgr = pub.getDefaultTopicMgr()
#---------------------------------------------------------------------------
class lib_pubsub_Notify2(wtc.WidgetTestCase):
class lib_pubsub_Notify2(wtc.PubsubTestCase):
def test0_NotificationTopics(self):
topicMgr = self.pub.getDefaultTopicMgr()
from wx.lib.pubsub.utils.notification import useNotifyByPubsubMessage
assert not topicMgr.getTopic('pubsub', okIfNone=True)
useNotifyByPubsubMessage()
assert topicMgr.getTopic('pubsub')
@@ -40,13 +40,13 @@ class lib_pubsub_Notify2(wtc.WidgetTestCase):
class MyListener:
countSub = 0
countUnsub = 0
def listenerSub(self, msgTopic=pub.AUTO_TOPIC, listener=None,
def listenerSub(self, msgTopic=self.pub.AUTO_TOPIC, listener=None,
topic=None, newSub=None):
self.assertEqual(msgTopic.getName(), 'pubsub.subscribe' )
assert topic.getName() in ('pubsub.unsubscribe', 'testSubscribeNotify')
if newSub:
self.countSub += 1
def listenerUnsub(self, msgTopic=pub.AUTO_TOPIC, topic=None,
def listenerUnsub(self, msgTopic=self.pub.AUTO_TOPIC, topic=None,
listener=None, listenerRaw=None):
assert topic.getName() in ('testSubscribeNotify', 'pubsub.subscribe' )
self.assertEqual(msgTopic.getName(), 'pubsub.unsubscribe' )
@@ -55,28 +55,28 @@ class lib_pubsub_Notify2(wtc.WidgetTestCase):
def listenerTest(self):
raise NotImplementedError # should never get here
pub.setNotificationFlags(subscribe=True, unsubscribe=True)
pub.getOrCreateTopic('testSubscribeNotify')
self.pub.setNotificationFlags(subscribe=True, unsubscribe=True)
self.pub.getOrCreateTopic('testSubscribeNotify')
tmp = MyListener()
tmp.assertEqual = self.assertEqual
pub.subscribe(tmp.listenerSub, 'pubsub.subscribe')
self.pub.subscribe(tmp.listenerSub, 'pubsub.subscribe')
self.assertEqual(tmp.countSub, 0) # don't notify of self subscription
self.assertEqual(tmp.countUnsub, 0)
sl, ok = pub.subscribe(tmp.listenerUnsub, 'pubsub.unsubscribe')
sl, ok = self.pub.subscribe(tmp.listenerUnsub, 'pubsub.unsubscribe')
assert ok
self.assertEqual(tmp.countSub, 1)
self.assertEqual(tmp.countUnsub, 0)
pub.subscribe(tmp.listenerTest, 'testSubscribeNotify')
self.pub.subscribe(tmp.listenerTest, 'testSubscribeNotify')
self.assertEqual(tmp.countUnsub, 0)
pub.unsubscribe(tmp.listenerTest, 'testSubscribeNotify')
self.pub.unsubscribe(tmp.listenerTest, 'testSubscribeNotify')
self.assertEqual(tmp.countUnsub, 1)
pub.unsubscribe(tmp.listenerSub, 'pubsub.subscribe')
self.pub.unsubscribe(tmp.listenerSub, 'pubsub.subscribe')
self.assertEqual(tmp.countSub, 2)
self.assertEqual(tmp.countUnsub, 2)
pub.unsubscribe(tmp.listenerUnsub, 'pubsub.unsubscribe')
self.pub.unsubscribe(tmp.listenerUnsub, 'pubsub.unsubscribe')
self.assertEqual(tmp.countSub, 2)
self.assertEqual(tmp.countUnsub, 2) # don't notify of self unsubscription
@@ -86,7 +86,8 @@ class lib_pubsub_Notify2(wtc.WidgetTestCase):
class SendHandler:
def __init__(self):
self.pre = self.post = 0
def __call__(self, topic=None, stage=None, listener=None, msgTopic=pub.AUTO_TOPIC):
def __call__(self, topic=None, stage=None, listener=None,
msgTopic=self.pub.AUTO_TOPIC):
if stage == 'pre':
self.pre += 1
else:
@@ -96,14 +97,14 @@ class lib_pubsub_Notify2(wtc.WidgetTestCase):
sh = SendHandler()
sh.assertEqual = self.assertEqual
pub.subscribe(sh, 'pubsub.sendMessage')
pub.setNotificationFlags(sendMessage=True)
self.pub.subscribe(sh, 'pubsub.sendMessage')
self.pub.setNotificationFlags(sendMessage=True)
# generate a message that will cause pubsub.sendMessage to be generated too
assert sh.pre == 0
assert sh.post == 0
pub.getOrCreateTopic('testSendNotify')
pub.sendMessage('testSendNotify')
self.pub.getOrCreateTopic('testSendNotify')
self.pub.sendMessage('testSendNotify')
assert sh.pre == 1
assert sh.post == 1

57
unittests/test_lib_pubsub_notify3.py Normal file
View File

@@ -0,0 +1,57 @@
"""
:copyright: Copyright 2006-2009 by Oliver Schoenborn, all rights reserved.
:license: BSD, see LICENSE.txt for details.
"""
import imp_unittest, unittest
import wtc
from difflib import ndiff, unified_diff, context_diff
#---------------------------------------------------------------------------
class lib_pubsub_NotifyFlagChanges(wtc.PubsubTestCase):
def testFlagChanges(self):
savedFlags = self.pub.getNotificationFlags()
self.pub.setNotificationFlags(all=True, sendMessage=False, deadListener=False)
flags = self.pub.getNotificationFlags()
assert not flags['sendMessage']
assert not flags['deadListener']
assert flags['newTopic']
assert flags['delTopic']
assert flags['subscribe']
assert flags['unsubscribe']
self.pub.setNotificationFlags(subscribe=False, deadListener=True)
flags = self.pub.getNotificationFlags()
assert not flags['sendMessage']
assert not flags['subscribe']
assert flags['newTopic']
assert flags['delTopic']
assert flags['deadListener']
assert flags['unsubscribe']
self.pub.setNotificationFlags(all=False, subscribe=True, unsubscribe=True)
flags = self.pub.getNotificationFlags()
assert not flags['sendMessage']
assert not flags['deadListener']
assert not flags['newTopic']
assert not flags['delTopic']
assert flags['subscribe']
assert flags['unsubscribe']
self.pub.setNotificationFlags(** savedFlags)
#---------------------------------------------------------------------------
if __name__ == '__main__':
unittest.main()

84
unittests/test_lib_pubsub_notify4.py Normal file
View File

@@ -0,0 +1,84 @@
"""
:copyright: Copyright 2006-2009 by Oliver Schoenborn, all rights reserved.
:license: BSD, see LICENSE.txt for details.
"""
import imp_unittest, unittest
import wtc
from difflib import ndiff, unified_diff, context_diff
#---------------------------------------------------------------------------
class lib_pubsub_NotifyN(wtc.PubsubTestCase):
def testNotifications(self):
from wx.lib.pubsub.utils.notification import INotificationHandler
class Handler(INotificationHandler):
def __init__(self):
self.resetCounts()
def resetCounts(self):
self.counts = dict(send=0, sub=0, unsub=0, delt=0, newt=0, dead=0, all=0)
def notifySubscribe(self, pubListener, topicObj, newSub):
self.counts['sub'] += 1
def notifyUnsubscribe(self, pubListener, topicObj):
self.counts['unsub'] += 1
def notifyDeadListener(self, pubListener, topicObj):
self.counts['dead'] += 1
def notifySend(self, stage, topicObj, pubListener=None):
if stage == 'pre': self.counts['send'] += 1
def notifyNewTopic(self, topicObj, description, required, argsDocs):
self.counts['newt'] += 1
def notifyDelTopic(self, topicName):
self.counts['delt'] += 1
notifiee = Handler()
self.pub.addNotificationHandler(notifiee)
self.pub.setNotificationFlags(all=True)
def verify(**ref):
for key, val in notifiee.counts.iteritems():
if key in ref:
self.assertEqual(val, ref[key], "\n%s\n%s" % (notifiee.counts, ref) )
else:
self.assertEqual(val, 0, "%s = %s, expected 0" % (key, val))
notifiee.resetCounts()
verify()
def testListener():
pass
def testListener2():
pass
self.pub.getOrCreateTopic('newTopic')
verify(newt=1)
self.pub.subscribe(testListener, 'newTopic')
self.pub.subscribe(testListener2, 'newTopic')
verify(sub=2)
self.pub.sendMessage('newTopic')
verify(send=1)
del testListener
verify(dead=1)
self.pub.unsubscribe(testListener2,'newTopic')
verify(unsub=1)
self.pub.delTopic('newTopic')
verify(delt=1)
#---------------------------------------------------------------------------
if __name__ == '__main__':
unittest.main()

56
unittests/test_lib_pubsub_provider.py
View File

@@ -11,8 +11,6 @@ import wtc
from textwrap import dedent
from wx.lib.pubsub import pub
class my_topics:
class rootTopic1:
@@ -49,11 +47,11 @@ class my_topics:
#---------------------------------------------------------------------------
class lib_pubsub_Except(wtc.WidgetTestCase):
class lib_pubsub_Except(wtc.PubsubTestCase):
def test1(self):
pub.importTopicTree(my_topics)
self.pub.importTopicTree(my_topics)
provString = """
@@ -73,7 +71,7 @@ class lib_pubsub_Except(wtc.WidgetTestCase):
"""
pub.importTopicTree(provString, format=pub.TOPIC_TREE_FROM_STRING)
self.pub.importTopicTree(provString, format=self.pub.TOPIC_TREE_FROM_STRING)
provFile = """
@@ -93,22 +91,22 @@ class lib_pubsub_Except(wtc.WidgetTestCase):
myTopicTree = file('myTopicTree.py', 'w')
myTopicTree.write(dedent(provFile))
myTopicTree.close()
pub.importTopicTree('myTopicTree', format=pub.TOPIC_TREE_FROM_MODULE, lazy=True)
self.pub.importTopicTree('myTopicTree', format=self.pub.TOPIC_TREE_FROM_MODULE, lazy=True)
import os
os.remove('myTopicTree.py')
if os.path.exists('myTopicTree.pyc'):
os.remove('myTopicTree.pyc')
assert not pub.getTopic('rootTopic1.subtopic_2', okIfNone=True)
assert not self.pub.getTopic('rootTopic1.subtopic_2', okIfNone=True)
# the following should create all topic tree since parent
# topics are automatically created
assert pub.getOrCreateTopic('rootTopic1.subtopic_1.subsubtopic_11')
assert pub.getOrCreateTopic('rootTopic1.subtopic_1.subsubtopic_12')
assert pub.getOrCreateTopic('rootTopic1.subtopic_2.subsubtopic_21')
assert self.pub.getOrCreateTopic('rootTopic1.subtopic_1.subsubtopic_11')
assert self.pub.getOrCreateTopic('rootTopic1.subtopic_1.subsubtopic_12')
assert self.pub.getOrCreateTopic('rootTopic1.subtopic_2.subsubtopic_21')
# validate that topic specs were properly parsed
def isValid(topicName, listener):
topic = pub.getTopic(topicName)
topic = self.pub.getTopic(topicName)
assert topic.getDescription()
assert topic.isSendable()
return topic.isValid(listener)
@@ -123,12 +121,12 @@ class lib_pubsub_Except(wtc.WidgetTestCase):
assert isValid('rootTopic1.subtopic_1', sub_1)
assert isValid('rootTopic1.subtopic_1.subsubtopic_11', sub_11)
# no providers have spec for subtopic_2
assert not pub.getTopic('rootTopic1.subtopic_2').isSendable()
assert not self.pub.getTopic('rootTopic1.subtopic_2').isSendable()
#printTreeSpec()
pub.exportTopicTree('newTopicTree')
root2Defn = pub.exportTopicTree(rootTopicName='rootTopic1')
self.pub.exportTopicTree('newTopicTree')
root2Defn = self.pub.exportTopicTree(rootTopicName='rootTopic1')
import os
os.remove('newTopicTree.py')
@@ -163,19 +161,19 @@ class lib_pubsub_Except(wtc.WidgetTestCase):
'''
pass
"""
pub.clearTopicDefnProviders()
treeDoc = pub.importTopicTree(importStr, lazy = True,
format = pub.TOPIC_TREE_FROM_STRING)
self.pub.clearTopicDefnProviders()
treeDoc = self.pub.importTopicTree(importStr, lazy = True,
format = self.pub.TOPIC_TREE_FROM_STRING)
assert treeDoc == '''Tree docs, can be anything you want.'''
root = pub.getOrCreateTopic('test_import_export_no_change.subtopic_1')
root = self.pub.getOrCreateTopic('test_import_export_no_change.subtopic_1')
# few sanity checks
def sub_1(arg1, arg2=None):
pass
assert root.isSendable()
assert pub.isValid(sub_1, 'test_import_export_no_change.subtopic_1')
assert self.pub.isValid(sub_1, 'test_import_export_no_change.subtopic_1')
# export tree
exported = pub.exportTopicTree(rootTopicName='test_import_export_no_change', moduleDoc=treeDoc)
exported = self.pub.exportTopicTree(rootTopicName='test_import_export_no_change', moduleDoc=treeDoc)
#print exported
expectExport = """\
@@ -226,11 +224,11 @@ class lib_pubsub_Except(wtc.WidgetTestCase):
assert diffs == ['- ', '+ ']
# now for module:
modDoc = pub.importTopicTree('lib_pubsub_provider_expect',
format=pub.TOPIC_TREE_FROM_MODULE,
modDoc = self.pub.importTopicTree('lib_pubsub_provider_expect',
format=self.pub.TOPIC_TREE_FROM_MODULE,
lazy=False)
assert modDoc.startswith('\nTree docs, can be anything you')
pub.exportTopicTree('lib_pubsub_provider_actual',
self.pub.exportTopicTree('lib_pubsub_provider_actual',
rootTopicName='test_import_export_no_change2',
moduleDoc=treeDoc)
lines1 = file('lib_pubsub_provider_actual.py', 'r').readlines()
@@ -239,16 +237,16 @@ class lib_pubsub_Except(wtc.WidgetTestCase):
assert not list(diffs)
def test_module_as_class(self):
assert pub.getTopic('root_topic1', True) is None
assert pub.getTopic('root_topic2.sub_topic21', True) is None
assert self.pub.getTopic('root_topic1', True) is None
assert self.pub.getTopic('root_topic2.sub_topic21', True) is None
import lib_pubsub_provider_my_import_topics
pub.importTopicTree(lib_pubsub_provider_my_import_topics)
self.pub.importTopicTree(lib_pubsub_provider_my_import_topics)
assert pub.getTopic('root_topic1') is not None
assert pub.getTopic('root_topic2.sub_topic21') is not None
assert self.pub.getTopic('root_topic1') is not None
assert self.pub.getTopic('root_topic2.sub_topic21') is not None
pub.sendMessage(lib_pubsub_provider_my_import_topics.root_topic1)
self.pub.sendMessage(lib_pubsub_provider_my_import_topics.root_topic1)
#---------------------------------------------------------------------------

2
unittests/test_lib_pubsub_spec.py
View File

@@ -19,7 +19,7 @@ from wx.lib.pubsub.core.topicargspec import \
#---------------------------------------------------------------------------
class lib_pubsub_Specs(wtc.WidgetTestCase):
class lib_pubsub_Specs(wtc.PubsubTestCase):
def test1_create(self):

2
unittests/test_lib_pubsub_topic.py
View File

@@ -21,7 +21,7 @@ from wx.lib.pubsub.core.topicexc import ListenerSpecInvalid
#---------------------------------------------------------------------------
class lib_pubsub_Topic(wtc.WidgetTestCase):
class lib_pubsub_Topic(wtc.PubsubTestCase):
rootTopic = None
treeConfig = TreeConfig()

23
unittests/test_lib_pubsub_topicmgr.py
View File

@@ -27,18 +27,15 @@ from wx.lib.pubsub.core.topicutils import \
TopicNameInvalid, \
validateName
topicMgr = pub.getDefaultTopicMgr()
from wx.lib.pubsub.utils.topictreeprinter import \
printTreeDocs, ITopicTreeVisitor
class lib_pubsub_TopicMgr0_Basic(wtc.WidgetTestCase):
class lib_pubsub_TopicMgr0_Basic(wtc.PubsubTestCase):
"""
Only tests TopicMgr methods. This must use some query methods on
topic objects to validate that TopicMgr did it's job properly.
"""
def failTopicName(self, name):
self.assertRaises(TopicNameInvalid, validateName, name)
@@ -69,7 +66,7 @@ class lib_pubsub_TopicMgr0_Basic(wtc.WidgetTestCase):
self.failTopicName( (ALL_TOPICS,) )
class lib_pubsub_TopicMgr1_GetOrCreate_NoDefnProv(wtc.WidgetTestCase):
class lib_pubsub_TopicMgr1_GetOrCreate_NoDefnProv(wtc.PubsubTestCase):
"""
Only tests TopicMgr methods. This must use some query methods on
topic objects to validate that TopicMgr did it's job properly.
@@ -79,6 +76,7 @@ class lib_pubsub_TopicMgr1_GetOrCreate_NoDefnProv(wtc.WidgetTestCase):
#
# Test the getOrCreateTopic without proto listener
#
topicMgr = self.pub.getDefaultTopicMgr()
def verifyNonSendable(topicObj, nameTuple, parent):
"""Any non-sendable topic will satisfy these conditions:"""
@@ -146,6 +144,7 @@ class lib_pubsub_TopicMgr1_GetOrCreate_NoDefnProv(wtc.WidgetTestCase):
#
# Test the getOrCreateTopic with proto listener
#
topicMgr = self.pub.getDefaultTopicMgr()
rootName = 'GetOrCreate_WithProtoListener'
tName = rootName
@@ -186,7 +185,7 @@ class lib_pubsub_TopicMgr1_GetOrCreate_NoDefnProv(wtc.WidgetTestCase):
assert subsubTopic.isValid(protoListener2)
class lib_pubsub_TopicMgr2_GetOrCreate_DefnProv(wtc.WidgetTestCase):
class lib_pubsub_TopicMgr2_GetOrCreate_DefnProv(wtc.PubsubTestCase):
"""
Test TopicManager when one or more definition providers
can provide for some topic definitions.
@@ -196,6 +195,7 @@ class lib_pubsub_TopicMgr2_GetOrCreate_DefnProv(wtc.WidgetTestCase):
#
# Test the addition and clearing of definition providers
#
topicMgr = self.pub.getDefaultTopicMgr()
class DefnProvider:
pass
@@ -217,6 +217,7 @@ class lib_pubsub_TopicMgr2_GetOrCreate_DefnProv(wtc.WidgetTestCase):
# two so we can check that more than one can work together.
# One provides good definitions, one provides some with errors.
#
topicMgr = self.pub.getDefaultTopicMgr()
class DefnProvider(ITopicDefnProvider):
"""
@@ -317,6 +318,7 @@ class lib_pubsub_TopicMgr2_GetOrCreate_DefnProv(wtc.WidgetTestCase):
#
# Test topic deletion
#
topicMgr = self.pub.getDefaultTopicMgr()
topicMgr.getOrCreateTopic('delTopic.b.c.d.e')
assert topicMgr.getTopic('delTopic.b.c.d.e') is not None
@@ -330,7 +332,7 @@ class lib_pubsub_TopicMgr2_GetOrCreate_DefnProv(wtc.WidgetTestCase):
assert topicMgr.getTopic('delTopic.b.c', okIfNone=True) is None
class lib_pubsub_TopicMgr3_TreeTraverser(wtc.WidgetTestCase):
class lib_pubsub_TopicMgr3_TreeTraverser(wtc.PubsubTestCase):
expectedOutput = '''\
\-- Topic "a2"
\-- Topic "a"
@@ -345,6 +347,7 @@ class lib_pubsub_TopicMgr3_TreeTraverser(wtc.WidgetTestCase):
#
# Test printing of topic tree
#
topicMgr = self.pub.getDefaultTopicMgr()
root = topicMgr.getOrCreateTopic('a2')
topicMgr.getOrCreateTopic('a2.a.a')
@@ -361,8 +364,10 @@ class lib_pubsub_TopicMgr3_TreeTraverser(wtc.WidgetTestCase):
#
# Test traversing with and without filtering, breadth and depth
#
topicMgr = self.pub.getDefaultTopicMgr()
class MyTraverser(ITopicTreeVisitor):
def __init__(self):
def __init__(self, pub):
self.traverser = pub.TopicTreeTraverser(self)
self.calls = ''
self.topics = []
@@ -401,7 +406,7 @@ class lib_pubsub_TopicMgr3_TreeTraverser(wtc.WidgetTestCase):
topicMgr.getOrCreateTopic('traversal.b.D.bar')
def exe(expectCalls, expectTopics, **kwargs):
traverser = MyTraverser()
traverser = MyTraverser(self.pub)
traverser.traverse(root, **kwargs)
self.assertEqual(traverser.topics, expectTopics)
self.assertEqual(traverser.calls, expectCalls)

12
unittests/wtc.py
View File

@@ -104,3 +104,15 @@ def mybytes(text):
#---------------------------------------------------------------------------
class PubsubTestCase(unittest.TestCase):
"""
A testcase that will create ?????
"""
def setUp(self):
from wx.lib.pubsub import pub
self.pub = pub
def tearDown(self):
self.pub.unsubAll()
del self.pub