[Zope-CVS] CVS: Packages/pypes/pypes/tests - test_event.py:1.5
test_services.py:1.6
Casey Duncan
casey at zope.com
Mon Feb 9 16:27:36 EST 2004
Update of /cvs-repository/Packages/pypes/pypes/tests
In directory cvs.zope.org:/tmp/cvs-serv14155
Modified Files:
test_event.py test_services.py
Log Message:
dos -> unix line endings
=== Packages/pypes/pypes/tests/test_event.py 1.4 => 1.5 ===
--- Packages/pypes/pypes/tests/test_event.py:1.4 Sun Feb 8 22:56:22 2004
+++ Packages/pypes/pypes/tests/test_event.py Mon Feb 9 16:27:35 2004
@@ -1,334 +1,334 @@
-##############################################################################
-#
-# Copyright (c) 2002 Zope Corporation and Contributors.
-# All Rights Reserved.
-#
-# This software is subject to the provisions of the Zope Public License,
-# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
-# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
-# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
-# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
-# FOR A PARTICULAR PURPOSE.
-#
-##############################################################################
-"""Event service tests
-
-$Id$"""
-
-import unittest
-from zope.interface.verify import verifyObject
-from persistent import Persistent
-from pypes.exceptions import VetoEvent
-from random import choice
-
-class TestListener(Persistent):
- notifies = 0
- fnordifies = 0
- vetoes = 0
-
- def notify(self, msg):
- self.notifies += 1
-
- def fnordify(self, msg):
- self.fnordifies += 1
-
- def vetoify(self, msg):
- self.vetoes += 1
- raise VetoEvent, 'Hah!'
-
- notcallable = 0
-
-class Message:
- pass
-
-class MessageA(Message):
- pass
-
-class MessageB(Message):
- pass
-
-class MessageA2(MessageA):
- pass
-
-class MessageAB(MessageA, MessageB):
- pass
-
-class TestEventService(unittest.TestCase):
-
- def setUp(self):
- from pypes.event import EventService
- self.event = EventService()
-
- def testInterfaces(self):
- from pypes.interfaces import IPersistentService, IEventService
- self.failUnless(verifyObject(IPersistentService, self.event))
- self.failUnless(verifyObject(IEventService, self.event))
-
- def testRegisterListener(self):
- o = TestListener()
- self.failIf(self.event.isListener(o, Message))
- self.failIf(self.event.isListener(o, MessageA))
- self.event.registerListener(o, 'notify', MessageA)
- self.failUnless(self.event.isListener(o, MessageA))
- self.event.registerListener(o, 'notify', Message)
- self.failUnless(self.event.isListener(o, Message))
- self.failUnless(self.event.isListener(o, MessageA))
-
- def testWouldReceive(self):
- o = TestListener()
- oa = TestListener()
- ob = TestListener()
- self.event.registerListener(o, 'notify', Message)
- self.event.registerListener(oa, 'notify', MessageA)
- self.event.registerListener(ob, 'notify', MessageB)
- for msgtype in (Message, MessageA, MessageB, MessageA2, MessageAB):
- self.failUnless(self.event.wouldReceive(o, msgtype))
- self.failIf(self.event.wouldReceive(oa, Message))
- self.failIf(self.event.wouldReceive(oa, MessageB))
- self.failUnless(self.event.wouldReceive(oa, MessageA))
- self.failUnless(self.event.wouldReceive(oa, MessageA2))
- self.failUnless(self.event.wouldReceive(oa, MessageAB))
- self.failIf(self.event.wouldReceive(ob, Message))
- self.failIf(self.event.wouldReceive(ob, MessageA))
- self.failIf(self.event.wouldReceive(ob, MessageA2))
- self.failUnless(self.event.wouldReceive(ob, MessageB))
- self.failUnless(self.event.wouldReceive(ob, MessageAB))
-
- def testRegisterBadMethod(self):
- o = TestListener()
- self.assertRaises(
- AttributeError, self.event.registerListener, o, 'vitrius', Message)
-
- def testRegisterNotCallable(self):
- o = TestListener()
- self.assertRaises(
- TypeError, self.event.registerListener, o, 'notcallable', Message)
-
- def testRegisterBadMessageType(self):
- o = TestListener()
- self.assertRaises(
- TypeError, self.event.registerListener, o, 'notify', [])
-
- def testUnregister(self):
- o = TestListener()
- self.event.registerListener(o, 'notify', MessageA)
- self.event.registerListener(o, 'notify', MessageB)
- self.event.unregisterListener(o, 'notify', MessageA)
- self.failIf(self.event.isListener(o, MessageA))
- self.failUnless(self.event.isListener(o, MessageB))
- self.failUnless(self.event.wouldReceive(o, MessageAB))
-
- def testUnregisterMultipleMethods(self):
- o = TestListener()
- self.event.registerListener(o, 'notify', MessageB)
- self.event.registerListener(o, 'fnordify', MessageB)
- self.event.unregisterListener(o, 'notify', MessageB)
- self.failUnless(self.event.isListener(o, MessageB))
- self.event.unregisterListener(o, 'fnordify', MessageB)
- self.failIf(self.event.isListener(o, MessageB))
-
- def testUnregisterNeverRegistered(self):
- from pypes.exceptions import EventRegistrationError
- self.assertRaises(EventRegistrationError,
- self.event.unregisterListener,
- TestListener(), 'notify', Message)
- self.event.registerListener(TestListener(), 'notify', Message)
- self.assertRaises(EventRegistrationError,
- self.event.unregisterListener,
- TestListener(), 'notify', Message)
-
- def testUnregisterWrongMethodName(self):
- from pypes.exceptions import EventRegistrationError
- o = TestListener()
- self.event.registerListener(o, 'notify', Message)
- self.assertRaises(EventRegistrationError,
- self.event.unregisterListener,
- o, 'bloatify', Message)
-
- def testUnregisterWrongMethodType(self):
- from pypes.exceptions import EventRegistrationError
- o = TestListener()
- self.event.registerListener(o, 'notify', Message)
- self.assertRaises(EventRegistrationError,
- self.event.unregisterListener,
- o, 'notify', MessageA)
-
- def testUnregisterAllListeners(self):
- o1 = TestListener()
- self.event.registerListener(o1, 'notify', MessageA)
- o2 = TestListener()
- self.event.registerListener(o2, 'notify', MessageA)
- o3 = TestListener()
- self.event.registerListener(o3, 'notify', MessageB)
- o4 = TestListener()
- self.event.registerListener(o4, 'notify', MessageAB)
- self.event.unregisterAllListeners(MessageA)
- self.failIf(self.event.isListener(o1, MessageA))
- self.failIf(self.event.isListener(o2, MessageA))
- self.failUnless(self.event.isListener(o3, MessageB))
- self.failUnless(self.event.wouldReceive(o3, MessageAB))
- self.failUnless(self.event.isListener(o4, MessageAB))
- self.event.unregisterAllListeners(MessageA)
-
- def testSend(self):
- o = TestListener()
- self.failIf(self.event.send(MessageA()))
- self.failIf(self.event.send(MessageB()))
- self.failIf(self.event.send(MessageAB()))
- self.event.registerListener(o, 'notify', MessageA)
- self.event.registerListener(o, 'fnordify', MessageB)
- self.failUnless(self.event.send(MessageA()))
- self.failUnless(self.event.send(MessageB()))
- self.failUnless(self.event.send(MessageAB()))
- self.failIf(self.event.send(Message()))
- self.assertEqual(o.notifies, 2)
- self.assertEqual(o.fnordifies, 2)
-
- def testSendMultipleListeners(self):
- nosey = TestListener()
- self.event.registerListener(nosey, 'notify', Message)
- a = TestListener()
- self.event.registerListener(a, 'notify', MessageA)
- a2 = TestListener()
- self.event.registerListener(a2, 'notify', MessageA2)
- b = TestListener()
- self.event.registerListener(b, 'notify', MessageB)
- self.event.registerListener(b, 'fnordify', Message)
- count = {
- MessageA: 0,
- MessageB: 0,
- MessageA2: 0,
- MessageAB: 0,
- }
- types = count.keys()
- for i in xrange(100):
- msgtype = choice(types)
- self.event.send(msgtype())
- count[msgtype] += 1
- self.assertEqual(
- a.notifies, count[MessageA] + count[MessageA2] + count[MessageAB])
- self.assertEqual(a2.notifies, count[MessageA2])
- self.assertEqual(b.notifies, count[MessageB] + count[MessageAB])
- total = reduce(lambda x,y: x+y, count.values())
- self.assertEqual(nosey.notifies, total)
- self.assertEqual(b.fnordifies, total)
-
- def testSendMultipleRegistrations(self):
- o = TestListener()
- self.event.registerListener(o, 'fnordify', MessageA)
- self.event.registerListener(o, 'fnordify', MessageA2)
- self.event.registerListener(o, 'fnordify', MessageB)
- self.event.registerListener(o, 'fnordify', Message)
- self.event.send(MessageA())
- self.event.send(MessageA2())
- self.event.send(MessageB())
- self.event.send(Message())
- self.assertEqual(o.fnordifies, 4)
-
- def testListenerVeto(self):
- o = TestListener()
- self.event.registerListener(o, 'vetoify', MessageA)
- self.assertRaises(VetoEvent, self.event.send, MessageA())
- self.assertRaises(VetoEvent, self.event.send, MessageAB())
- self.failIf(self.event.send(Message()))
-
- def testIterListeners(self):
- listeners = {}
- msgtypes = (MessageA, MessageB, MessageA2)
- for mt in msgtypes:
- listeners[mt] = []
- self.failIf(list(self.event.iterListeners(mt)))
- for i in xrange(100):
- for mt in msgtypes:
- o = TestListener()
- listeners[mt].append((o, 'notify'))
- self.event.registerListener(o, 'notify', mt)
- for mt in msgtypes:
- listeners[mt].sort()
- registered = list(self.event.iterListeners(mt))
- registered.sort()
- self.assertEqual(listeners[mt], registered)
- self.failIf(list(self.event.iterListeners(Message)))
- self.assertRaises(
- StopIteration, self.event.iterListeners(Message).next)
-
- def testIterReceivers(self):
- listeners = {}
- msgtypes = (MessageA, MessageB, MessageA2, MessageAB)
- for mt in msgtypes:
- listeners[mt] = []
- self.failIf(list(self.event.iterReceivers(mt)))
- all = []
- for i in xrange(100):
- for mt in msgtypes:
- o = TestListener()
- listeners[mt].append((o, 'notify'))
- all.append((o, 'notify'))
- self.event.registerListener(o, 'notify', mt)
-
- listeners[MessageA].sort()
- receivers = list(self.event.iterReceivers(MessageA))
- receivers.sort()
- self.assertEqual(listeners[MessageA], receivers)
-
- registered = listeners[MessageA] + listeners[MessageA2]
- registered.sort()
- receivers = list(self.event.iterReceivers(MessageA2))
- receivers.sort()
- self.assertEqual(registered, receivers)
-
- registered = (listeners[MessageA] +
- listeners[MessageB] +
- listeners[MessageAB])
- registered.sort()
- receivers = list(self.event.iterReceivers(MessageAB))
- receivers.sort()
- self.assertEqual(registered, receivers)
-
- self.failIf(list(self.event.iterReceivers(Message)))
- self.assertRaises(
- StopIteration, self.event.iterReceivers(Message).next)
-
- def testIterMessageTypes(self):
- o = TestListener()
- self.event.registerListener(o, 'notify', MessageA)
- self.event.registerListener(o, 'notify', MessageB)
- self.event.registerListener(o, 'fnordify', MessageAB)
- self.event.registerListener(o, 'notify', MessageAB)
- types = [MessageA, MessageB, MessageAB]
- types.sort()
- registered = list(self.event.iterMessageTypes())
- registered.sort()
- self.assertEqual(types, registered)
-
- def testIterMessageTypesNoneRegistered(self):
- self.assertRaises(
- StopIteration, self.event.iterMessageTypes().next)
- o = TestListener()
- self.event.registerListener(o, 'notify', MessageA)
- self.event.unregisterListener(o, 'notify', MessageA)
- self.assertRaises(
- StopIteration, self.event.iterMessageTypes().next)
-
-class TestMessages(unittest.TestCase):
-
- def testIdRegistered(self):
- from pypes.identity import IdRegisteredMessage
- from pypes.interfaces import IIdentityMessage
- ob = TestListener()
- msg = IdRegisteredMessage(ob, 'blah')
- self.failUnless(verifyObject(IIdentityMessage, msg))
- self.failUnless(msg.object is ob)
- self.assertEqual(msg.id, 'blah')
-
- def testIdUnregistered(self):
- from pypes.identity import IdUnregisteredMessage
- from pypes.interfaces import IIdentityMessage
- ob = TestListener()
- msg = IdUnregisteredMessage(ob, 'blah')
- self.failUnless(verifyObject(IIdentityMessage, msg))
- self.failUnless(msg.object is ob)
- self.assertEqual(msg.id, 'blah')
-
-if __name__ == '__main__':
- unittest.main()
+##############################################################################
+#
+# Copyright (c) 2002 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+"""Event service tests
+
+$Id$"""
+
+import unittest
+from zope.interface.verify import verifyObject
+from persistent import Persistent
+from pypes.exceptions import VetoEvent
+from random import choice
+
+class TestListener(Persistent):
+ notifies = 0
+ fnordifies = 0
+ vetoes = 0
+
+ def notify(self, msg):
+ self.notifies += 1
+
+ def fnordify(self, msg):
+ self.fnordifies += 1
+
+ def vetoify(self, msg):
+ self.vetoes += 1
+ raise VetoEvent, 'Hah!'
+
+ notcallable = 0
+
+class Message:
+ pass
+
+class MessageA(Message):
+ pass
+
+class MessageB(Message):
+ pass
+
+class MessageA2(MessageA):
+ pass
+
+class MessageAB(MessageA, MessageB):
+ pass
+
+class TestEventService(unittest.TestCase):
+
+ def setUp(self):
+ from pypes.event import EventService
+ self.event = EventService()
+
+ def testInterfaces(self):
+ from pypes.interfaces import IPersistentService, IEventService
+ self.failUnless(verifyObject(IPersistentService, self.event))
+ self.failUnless(verifyObject(IEventService, self.event))
+
+ def testRegisterListener(self):
+ o = TestListener()
+ self.failIf(self.event.isListener(o, Message))
+ self.failIf(self.event.isListener(o, MessageA))
+ self.event.registerListener(o, 'notify', MessageA)
+ self.failUnless(self.event.isListener(o, MessageA))
+ self.event.registerListener(o, 'notify', Message)
+ self.failUnless(self.event.isListener(o, Message))
+ self.failUnless(self.event.isListener(o, MessageA))
+
+ def testWouldReceive(self):
+ o = TestListener()
+ oa = TestListener()
+ ob = TestListener()
+ self.event.registerListener(o, 'notify', Message)
+ self.event.registerListener(oa, 'notify', MessageA)
+ self.event.registerListener(ob, 'notify', MessageB)
+ for msgtype in (Message, MessageA, MessageB, MessageA2, MessageAB):
+ self.failUnless(self.event.wouldReceive(o, msgtype))
+ self.failIf(self.event.wouldReceive(oa, Message))
+ self.failIf(self.event.wouldReceive(oa, MessageB))
+ self.failUnless(self.event.wouldReceive(oa, MessageA))
+ self.failUnless(self.event.wouldReceive(oa, MessageA2))
+ self.failUnless(self.event.wouldReceive(oa, MessageAB))
+ self.failIf(self.event.wouldReceive(ob, Message))
+ self.failIf(self.event.wouldReceive(ob, MessageA))
+ self.failIf(self.event.wouldReceive(ob, MessageA2))
+ self.failUnless(self.event.wouldReceive(ob, MessageB))
+ self.failUnless(self.event.wouldReceive(ob, MessageAB))
+
+ def testRegisterBadMethod(self):
+ o = TestListener()
+ self.assertRaises(
+ AttributeError, self.event.registerListener, o, 'vitrius', Message)
+
+ def testRegisterNotCallable(self):
+ o = TestListener()
+ self.assertRaises(
+ TypeError, self.event.registerListener, o, 'notcallable', Message)
+
+ def testRegisterBadMessageType(self):
+ o = TestListener()
+ self.assertRaises(
+ TypeError, self.event.registerListener, o, 'notify', [])
+
+ def testUnregister(self):
+ o = TestListener()
+ self.event.registerListener(o, 'notify', MessageA)
+ self.event.registerListener(o, 'notify', MessageB)
+ self.event.unregisterListener(o, 'notify', MessageA)
+ self.failIf(self.event.isListener(o, MessageA))
+ self.failUnless(self.event.isListener(o, MessageB))
+ self.failUnless(self.event.wouldReceive(o, MessageAB))
+
+ def testUnregisterMultipleMethods(self):
+ o = TestListener()
+ self.event.registerListener(o, 'notify', MessageB)
+ self.event.registerListener(o, 'fnordify', MessageB)
+ self.event.unregisterListener(o, 'notify', MessageB)
+ self.failUnless(self.event.isListener(o, MessageB))
+ self.event.unregisterListener(o, 'fnordify', MessageB)
+ self.failIf(self.event.isListener(o, MessageB))
+
+ def testUnregisterNeverRegistered(self):
+ from pypes.exceptions import EventRegistrationError
+ self.assertRaises(EventRegistrationError,
+ self.event.unregisterListener,
+ TestListener(), 'notify', Message)
+ self.event.registerListener(TestListener(), 'notify', Message)
+ self.assertRaises(EventRegistrationError,
+ self.event.unregisterListener,
+ TestListener(), 'notify', Message)
+
+ def testUnregisterWrongMethodName(self):
+ from pypes.exceptions import EventRegistrationError
+ o = TestListener()
+ self.event.registerListener(o, 'notify', Message)
+ self.assertRaises(EventRegistrationError,
+ self.event.unregisterListener,
+ o, 'bloatify', Message)
+
+ def testUnregisterWrongMethodType(self):
+ from pypes.exceptions import EventRegistrationError
+ o = TestListener()
+ self.event.registerListener(o, 'notify', Message)
+ self.assertRaises(EventRegistrationError,
+ self.event.unregisterListener,
+ o, 'notify', MessageA)
+
+ def testUnregisterAllListeners(self):
+ o1 = TestListener()
+ self.event.registerListener(o1, 'notify', MessageA)
+ o2 = TestListener()
+ self.event.registerListener(o2, 'notify', MessageA)
+ o3 = TestListener()
+ self.event.registerListener(o3, 'notify', MessageB)
+ o4 = TestListener()
+ self.event.registerListener(o4, 'notify', MessageAB)
+ self.event.unregisterAllListeners(MessageA)
+ self.failIf(self.event.isListener(o1, MessageA))
+ self.failIf(self.event.isListener(o2, MessageA))
+ self.failUnless(self.event.isListener(o3, MessageB))
+ self.failUnless(self.event.wouldReceive(o3, MessageAB))
+ self.failUnless(self.event.isListener(o4, MessageAB))
+ self.event.unregisterAllListeners(MessageA)
+
+ def testSend(self):
+ o = TestListener()
+ self.failIf(self.event.send(MessageA()))
+ self.failIf(self.event.send(MessageB()))
+ self.failIf(self.event.send(MessageAB()))
+ self.event.registerListener(o, 'notify', MessageA)
+ self.event.registerListener(o, 'fnordify', MessageB)
+ self.failUnless(self.event.send(MessageA()))
+ self.failUnless(self.event.send(MessageB()))
+ self.failUnless(self.event.send(MessageAB()))
+ self.failIf(self.event.send(Message()))
+ self.assertEqual(o.notifies, 2)
+ self.assertEqual(o.fnordifies, 2)
+
+ def testSendMultipleListeners(self):
+ nosey = TestListener()
+ self.event.registerListener(nosey, 'notify', Message)
+ a = TestListener()
+ self.event.registerListener(a, 'notify', MessageA)
+ a2 = TestListener()
+ self.event.registerListener(a2, 'notify', MessageA2)
+ b = TestListener()
+ self.event.registerListener(b, 'notify', MessageB)
+ self.event.registerListener(b, 'fnordify', Message)
+ count = {
+ MessageA: 0,
+ MessageB: 0,
+ MessageA2: 0,
+ MessageAB: 0,
+ }
+ types = count.keys()
+ for i in xrange(100):
+ msgtype = choice(types)
+ self.event.send(msgtype())
+ count[msgtype] += 1
+ self.assertEqual(
+ a.notifies, count[MessageA] + count[MessageA2] + count[MessageAB])
+ self.assertEqual(a2.notifies, count[MessageA2])
+ self.assertEqual(b.notifies, count[MessageB] + count[MessageAB])
+ total = reduce(lambda x,y: x+y, count.values())
+ self.assertEqual(nosey.notifies, total)
+ self.assertEqual(b.fnordifies, total)
+
+ def testSendMultipleRegistrations(self):
+ o = TestListener()
+ self.event.registerListener(o, 'fnordify', MessageA)
+ self.event.registerListener(o, 'fnordify', MessageA2)
+ self.event.registerListener(o, 'fnordify', MessageB)
+ self.event.registerListener(o, 'fnordify', Message)
+ self.event.send(MessageA())
+ self.event.send(MessageA2())
+ self.event.send(MessageB())
+ self.event.send(Message())
+ self.assertEqual(o.fnordifies, 4)
+
+ def testListenerVeto(self):
+ o = TestListener()
+ self.event.registerListener(o, 'vetoify', MessageA)
+ self.assertRaises(VetoEvent, self.event.send, MessageA())
+ self.assertRaises(VetoEvent, self.event.send, MessageAB())
+ self.failIf(self.event.send(Message()))
+
+ def testIterListeners(self):
+ listeners = {}
+ msgtypes = (MessageA, MessageB, MessageA2)
+ for mt in msgtypes:
+ listeners[mt] = []
+ self.failIf(list(self.event.iterListeners(mt)))
+ for i in xrange(100):
+ for mt in msgtypes:
+ o = TestListener()
+ listeners[mt].append((o, 'notify'))
+ self.event.registerListener(o, 'notify', mt)
+ for mt in msgtypes:
+ listeners[mt].sort()
+ registered = list(self.event.iterListeners(mt))
+ registered.sort()
+ self.assertEqual(listeners[mt], registered)
+ self.failIf(list(self.event.iterListeners(Message)))
+ self.assertRaises(
+ StopIteration, self.event.iterListeners(Message).next)
+
+ def testIterReceivers(self):
+ listeners = {}
+ msgtypes = (MessageA, MessageB, MessageA2, MessageAB)
+ for mt in msgtypes:
+ listeners[mt] = []
+ self.failIf(list(self.event.iterReceivers(mt)))
+ all = []
+ for i in xrange(100):
+ for mt in msgtypes:
+ o = TestListener()
+ listeners[mt].append((o, 'notify'))
+ all.append((o, 'notify'))
+ self.event.registerListener(o, 'notify', mt)
+
+ listeners[MessageA].sort()
+ receivers = list(self.event.iterReceivers(MessageA))
+ receivers.sort()
+ self.assertEqual(listeners[MessageA], receivers)
+
+ registered = listeners[MessageA] + listeners[MessageA2]
+ registered.sort()
+ receivers = list(self.event.iterReceivers(MessageA2))
+ receivers.sort()
+ self.assertEqual(registered, receivers)
+
+ registered = (listeners[MessageA] +
+ listeners[MessageB] +
+ listeners[MessageAB])
+ registered.sort()
+ receivers = list(self.event.iterReceivers(MessageAB))
+ receivers.sort()
+ self.assertEqual(registered, receivers)
+
+ self.failIf(list(self.event.iterReceivers(Message)))
+ self.assertRaises(
+ StopIteration, self.event.iterReceivers(Message).next)
+
+ def testIterMessageTypes(self):
+ o = TestListener()
+ self.event.registerListener(o, 'notify', MessageA)
+ self.event.registerListener(o, 'notify', MessageB)
+ self.event.registerListener(o, 'fnordify', MessageAB)
+ self.event.registerListener(o, 'notify', MessageAB)
+ types = [MessageA, MessageB, MessageAB]
+ types.sort()
+ registered = list(self.event.iterMessageTypes())
+ registered.sort()
+ self.assertEqual(types, registered)
+
+ def testIterMessageTypesNoneRegistered(self):
+ self.assertRaises(
+ StopIteration, self.event.iterMessageTypes().next)
+ o = TestListener()
+ self.event.registerListener(o, 'notify', MessageA)
+ self.event.unregisterListener(o, 'notify', MessageA)
+ self.assertRaises(
+ StopIteration, self.event.iterMessageTypes().next)
+
+class TestMessages(unittest.TestCase):
+
+ def testIdRegistered(self):
+ from pypes.identity import IdRegisteredMessage
+ from pypes.interfaces import IIdentityMessage
+ ob = TestListener()
+ msg = IdRegisteredMessage(ob, 'blah')
+ self.failUnless(verifyObject(IIdentityMessage, msg))
+ self.failUnless(msg.object is ob)
+ self.assertEqual(msg.id, 'blah')
+
+ def testIdUnregistered(self):
+ from pypes.identity import IdUnregisteredMessage
+ from pypes.interfaces import IIdentityMessage
+ ob = TestListener()
+ msg = IdUnregisteredMessage(ob, 'blah')
+ self.failUnless(verifyObject(IIdentityMessage, msg))
+ self.failUnless(msg.object is ob)
+ self.assertEqual(msg.id, 'blah')
+
+if __name__ == '__main__':
+ unittest.main()
=== Packages/pypes/pypes/tests/test_services.py 1.5 => 1.6 ===
--- Packages/pypes/pypes/tests/test_services.py:1.5 Fri Jan 30 00:35:39 2004
+++ Packages/pypes/pypes/tests/test_services.py Mon Feb 9 16:27:35 2004
@@ -1,200 +1,200 @@
-##############################################################################
-#
-# Copyright (c) 2002 Zope Corporation and Contributors.
-# All Rights Reserved.
-#
-# This software is subject to the provisions of the Zope Public License,
-# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
-# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
-# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
-# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
-# FOR A PARTICULAR PURPOSE.
-#
-##############################################################################
-"""Service registration and accessor tests"""
-
-import unittest
-import ZODB
-from common import TestConnection, PypesTestCase, PersistenceTest
-from zope.interface.verify import verifyObject
-from persistent import Persistent
-
-class TestService(Persistent):
- pass
-
-
-class TestInitServices(unittest.TestCase):
-
- def setUp(self):
- self.conn = TestConnection()
-
- def testInitServices(self):
- from pypes.serviceconfig import initServices, registry_name
- svcs = initServices(self.conn)
- self.failUnless(svcs is self.conn.root()[registry_name])
-
- def testInitServicesAgain(self):
- from pypes.serviceconfig import initServices
- svcs1 = initServices(self.conn)
- ob = svcs1['foo'] = object()
- svcs2 = initServices(self.conn)
- self.failUnless(svcs2 is svcs1)
- self.failUnless(svcs2['foo'] is ob)
-
- def testInitServicesAgainForce(self):
- from pypes.serviceconfig import initServices
- svcs1 = initServices(self.conn)
- ob = svcs1['foo'] = object()
- svcs2 = initServices(self.conn, force=True)
- self.failUnless(svcs2 is not svcs1)
- self.failIf(svcs2)
-
- def testServicesSpecificToConnection(self):
- from pypes.serviceconfig import initServices, registry_name
- conn1 = TestConnection()
- conn2 = TestConnection()
- svcs1 = initServices(conn1)
- svcs2 = initServices(conn2)
- self.failIf(svcs1 is svcs2)
-
-
-class TestServiceConfig(unittest.TestCase):
-
- def setUp(self):
- from pypes.serviceconfig import initServices
- self.conn = TestConnection()
- initServices(self.conn)
-
- def testAddService(self):
- from pypes.serviceconfig import addService, hasService, registry_name
- svc = TestService()
- addService(self.conn, 'test', svc)
- self.failUnless(hasService(self.conn, 'test'))
- self.failUnless(self.conn.root()[registry_name]['test'] is svc)
-
- def testAddServiceTwice(self):
- from pypes.serviceconfig import addService, registry_name
- from pypes.exceptions import PypesError
- svc = TestService()
- addService(self.conn, 'test', svc)
- svc2 = TestService()
- self.assertRaises(PypesError, addService, self.conn, 'test', svc2)
- self.failUnless(self.conn.root()[registry_name]['test'] is svc)
-
- def testSetService(self):
- from pypes.serviceconfig import setService, hasService, registry_name
- svc = TestService()
- setService(self.conn, 'test', svc)
- self.failUnless(hasService(self.conn, 'test'))
- self.failUnless(self.conn.root()[registry_name]['test'] is svc)
-
- def testSetServiceTwice(self):
- from pypes.serviceconfig import setService, registry_name
- svc = TestService()
- setService(self.conn, 'test', svc)
- svc2 = TestService()
- setService(self.conn, 'test', svc2)
- self.failUnless(self.conn.root()[registry_name]['test'] is svc2)
-
- def testDelService(self):
- from pypes.serviceconfig import addService, delService, registry_name
- svc = TestService()
- addService(self.conn, 'flam', svc)
- svc2 = TestService()
- addService(self.conn, 'bom', svc2)
- delService(self.conn, 'flam')
- self.failIf(self.conn.root()[registry_name].has_key('flam'))
- self.failUnless(self.conn.root()[registry_name]['bom'] is svc2)
- delService(self.conn, 'bom')
- self.failIf(self.conn.root()[registry_name].has_key('bom'))
-
- def testDelServiceNeverRegistered(self):
- from pypes.serviceconfig import delService
- from pypes.exceptions import PypesLookupError
- self.assertRaises(PypesLookupError, delService, self.conn, 'kibi')
-
- def testDelServiceTwice(self):
- from pypes.serviceconfig import addService, delService
- from pypes.exceptions import PypesLookupError
- svc = TestService()
- addService(self.conn, 'yongo', svc)
- delService(self.conn, 'yongo')
- self.assertRaises(PypesLookupError, delService, self.conn, 'yongo')
-
- def testHasService(self):
- from pypes.serviceconfig import addService, hasService
- self.failIf(hasService(self.conn, 'yum'))
- svc = TestService()
- addService(self.conn, 'tum', svc)
- self.failIf(hasService(self.conn, 'yum'))
- self.failUnless(hasService(self.conn, 'tum'))
-
- def testServiceConfigSpecificToConnection(self):
- from pypes.serviceconfig import initServices, addService, hasService
- conn1 = TestConnection()
- conn2 = TestConnection()
- initServices(conn1)
- initServices(conn2)
- addService(conn1, 'flan', TestService())
- self.failIf(hasService(conn2, 'flan'))
-
-
-class TestServiceAccessors(PypesTestCase):
-
- def setUp(self):
- from pypes.serviceconfig import registry_name
- PypesTestCase.setUp(self)
- self.identity = self.root[registry_name]['identity']
-
- def testGetFromConnection(self):
- from pypes import services
- self.failUnless(services.get(self.conn, 'identity') is self.identity)
-
- def testGetFromPersistentOb(self):
- from pypes import services
- ob = Persistent()
- ob._p_jar = self.conn
- self.failUnless(services.get(ob, 'identity') is self.identity)
-
- def testGetUnregisteredService(self):
- from pypes import services
- from pypes.exceptions import PypesLookupError
- self.assertRaises(PypesLookupError, services.get, self.conn, 'flambom')
-
- def testAllFromConnection(self):
- from pypes import services
- self.failUnless(services.all(self.conn)['identity'] is self.identity)
-
- def testAllFromPersistentOb(self):
- from pypes import services
- ob = Persistent()
- ob._p_jar = self.conn
- self.failUnless(services.all(ob)['identity'] is self.identity)
-
- def testIdentity(self):
- from pypes import services
- self.failUnless(services.identity(self.conn) is self.identity)
-
- def testEvent(self):
- from pypes import services
- from pypes.event import EventService
- self.failUnless(isinstance(services.event(self.conn), EventService))
-
-
-class TestInitializeServices(PersistenceTest):
-
- def testInitDefaultServices(self):
- import pypes
- from pypes import services, identity, event
- c1 = self.db.open()
- pypes.initialize(c1)
- get_transaction().commit()
-
- c2 = self.db.open()
- svcs = services.all(c2)
- self.assertEqual(len(svcs), 2)
- self.failUnless(isinstance(svcs['identity'], identity.IdentityService))
- self.failUnless(isinstance(svcs['event'], event.EventService))
-
-if __name__ == '__main__':
- unittest.main()
+##############################################################################
+#
+# Copyright (c) 2002 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+"""Service registration and accessor tests"""
+
+import unittest
+import ZODB
+from common import TestConnection, PypesTestCase, PersistenceTest
+from zope.interface.verify import verifyObject
+from persistent import Persistent
+
+class TestService(Persistent):
+ pass
+
+
+class TestInitServices(unittest.TestCase):
+
+ def setUp(self):
+ self.conn = TestConnection()
+
+ def testInitServices(self):
+ from pypes.serviceconfig import initServices, registry_name
+ svcs = initServices(self.conn)
+ self.failUnless(svcs is self.conn.root()[registry_name])
+
+ def testInitServicesAgain(self):
+ from pypes.serviceconfig import initServices
+ svcs1 = initServices(self.conn)
+ ob = svcs1['foo'] = object()
+ svcs2 = initServices(self.conn)
+ self.failUnless(svcs2 is svcs1)
+ self.failUnless(svcs2['foo'] is ob)
+
+ def testInitServicesAgainForce(self):
+ from pypes.serviceconfig import initServices
+ svcs1 = initServices(self.conn)
+ ob = svcs1['foo'] = object()
+ svcs2 = initServices(self.conn, force=True)
+ self.failUnless(svcs2 is not svcs1)
+ self.failIf(svcs2)
+
+ def testServicesSpecificToConnection(self):
+ from pypes.serviceconfig import initServices, registry_name
+ conn1 = TestConnection()
+ conn2 = TestConnection()
+ svcs1 = initServices(conn1)
+ svcs2 = initServices(conn2)
+ self.failIf(svcs1 is svcs2)
+
+
+class TestServiceConfig(unittest.TestCase):
+
+ def setUp(self):
+ from pypes.serviceconfig import initServices
+ self.conn = TestConnection()
+ initServices(self.conn)
+
+ def testAddService(self):
+ from pypes.serviceconfig import addService, hasService, registry_name
+ svc = TestService()
+ addService(self.conn, 'test', svc)
+ self.failUnless(hasService(self.conn, 'test'))
+ self.failUnless(self.conn.root()[registry_name]['test'] is svc)
+
+ def testAddServiceTwice(self):
+ from pypes.serviceconfig import addService, registry_name
+ from pypes.exceptions import PypesError
+ svc = TestService()
+ addService(self.conn, 'test', svc)
+ svc2 = TestService()
+ self.assertRaises(PypesError, addService, self.conn, 'test', svc2)
+ self.failUnless(self.conn.root()[registry_name]['test'] is svc)
+
+ def testSetService(self):
+ from pypes.serviceconfig import setService, hasService, registry_name
+ svc = TestService()
+ setService(self.conn, 'test', svc)
+ self.failUnless(hasService(self.conn, 'test'))
+ self.failUnless(self.conn.root()[registry_name]['test'] is svc)
+
+ def testSetServiceTwice(self):
+ from pypes.serviceconfig import setService, registry_name
+ svc = TestService()
+ setService(self.conn, 'test', svc)
+ svc2 = TestService()
+ setService(self.conn, 'test', svc2)
+ self.failUnless(self.conn.root()[registry_name]['test'] is svc2)
+
+ def testDelService(self):
+ from pypes.serviceconfig import addService, delService, registry_name
+ svc = TestService()
+ addService(self.conn, 'flam', svc)
+ svc2 = TestService()
+ addService(self.conn, 'bom', svc2)
+ delService(self.conn, 'flam')
+ self.failIf(self.conn.root()[registry_name].has_key('flam'))
+ self.failUnless(self.conn.root()[registry_name]['bom'] is svc2)
+ delService(self.conn, 'bom')
+ self.failIf(self.conn.root()[registry_name].has_key('bom'))
+
+ def testDelServiceNeverRegistered(self):
+ from pypes.serviceconfig import delService
+ from pypes.exceptions import PypesLookupError
+ self.assertRaises(PypesLookupError, delService, self.conn, 'kibi')
+
+ def testDelServiceTwice(self):
+ from pypes.serviceconfig import addService, delService
+ from pypes.exceptions import PypesLookupError
+ svc = TestService()
+ addService(self.conn, 'yongo', svc)
+ delService(self.conn, 'yongo')
+ self.assertRaises(PypesLookupError, delService, self.conn, 'yongo')
+
+ def testHasService(self):
+ from pypes.serviceconfig import addService, hasService
+ self.failIf(hasService(self.conn, 'yum'))
+ svc = TestService()
+ addService(self.conn, 'tum', svc)
+ self.failIf(hasService(self.conn, 'yum'))
+ self.failUnless(hasService(self.conn, 'tum'))
+
+ def testServiceConfigSpecificToConnection(self):
+ from pypes.serviceconfig import initServices, addService, hasService
+ conn1 = TestConnection()
+ conn2 = TestConnection()
+ initServices(conn1)
+ initServices(conn2)
+ addService(conn1, 'flan', TestService())
+ self.failIf(hasService(conn2, 'flan'))
+
+
+class TestServiceAccessors(PypesTestCase):
+
+ def setUp(self):
+ from pypes.serviceconfig import registry_name
+ PypesTestCase.setUp(self)
+ self.identity = self.root[registry_name]['identity']
+
+ def testGetFromConnection(self):
+ from pypes import services
+ self.failUnless(services.get(self.conn, 'identity') is self.identity)
+
+ def testGetFromPersistentOb(self):
+ from pypes import services
+ ob = Persistent()
+ ob._p_jar = self.conn
+ self.failUnless(services.get(ob, 'identity') is self.identity)
+
+ def testGetUnregisteredService(self):
+ from pypes import services
+ from pypes.exceptions import PypesLookupError
+ self.assertRaises(PypesLookupError, services.get, self.conn, 'flambom')
+
+ def testAllFromConnection(self):
+ from pypes import services
+ self.failUnless(services.all(self.conn)['identity'] is self.identity)
+
+ def testAllFromPersistentOb(self):
+ from pypes import services
+ ob = Persistent()
+ ob._p_jar = self.conn
+ self.failUnless(services.all(ob)['identity'] is self.identity)
+
+ def testIdentity(self):
+ from pypes import services
+ self.failUnless(services.identity(self.conn) is self.identity)
+
+ def testEvent(self):
+ from pypes import services
+ from pypes.event import EventService
+ self.failUnless(isinstance(services.event(self.conn), EventService))
+
+
+class TestInitializeServices(PersistenceTest):
+
+ def testInitDefaultServices(self):
+ import pypes
+ from pypes import services, identity, event
+ c1 = self.db.open()
+ pypes.initialize(c1)
+ get_transaction().commit()
+
+ c2 = self.db.open()
+ svcs = services.all(c2)
+ self.assertEqual(len(svcs), 2)
+ self.failUnless(isinstance(svcs['identity'], identity.IdentityService))
+ self.failUnless(isinstance(svcs['event'], event.EventService))
+
+if __name__ == '__main__':
+ unittest.main()
More information about the Zope-CVS
mailing list