[Zope-CVS] CVS: Packages/pypes/pypes/tests - test_extent.py:1.2
Casey Duncan
casey at zope.com
Mon Feb 9 01:20:12 EST 2004
Update of /cvs-repository/Packages/pypes/pypes/tests
In directory cvs.zope.org:/tmp/cvs-serv352/tests
Modified Files:
test_extent.py
Log Message:
Sketch some tests for extent maps
=== Packages/pypes/pypes/tests/test_extent.py 1.1 => 1.2 ===
--- Packages/pypes/pypes/tests/test_extent.py:1.1 Mon Feb 9 00:24:51 2004
+++ Packages/pypes/pypes/tests/test_extent.py Mon Feb 9 01:20:12 2004
@@ -16,6 +16,7 @@
$Id$"""
import unittest
+from sets import Set
from zope.interface import Interface, implements
from zope.interface.verify import verifyObject
from pypes.interfaces import IExtentMap, IExtentService
@@ -30,7 +31,7 @@
class BazClass(BarClass):
pass
-class SplatClass(object):
+class SplatClass(BazClass, BarClass):
pass
class IMoe(Interface):
@@ -42,7 +43,7 @@
class ICurly(ILarry):
pass
-class IJoe(Interface):
+class IJoe(ICurly, IMoe):
pass
class TestExtentMap:
@@ -140,6 +141,9 @@
self.extent = self.makeExtentService(interface_extents=False)
self.testGetClass()
self.testContainsClass()
+ all = list(iter(self.extent))
+ expected = [FooClass, BarClass, BazClass]
+ self.assertEqual(sort(all), sort(expected))
self.assertRaises(TypeError, self.testGetInterface)
self.assertRaises(TypeError, self.testContainsInterface)
@@ -147,13 +151,98 @@
self.extent = self.makeExtentService(class_extents=False)
self.testGetInterface()
self.testContainsInterface()
+ all = list(iter(self.extent))
+ expected = [IMoe, ILarry, ICurly]
+ self.assertEqual(sort(all), sort(expected))
self.assertRaises(TypeError, self.testGetClass)
self.assertRaises(TypeError, self.testContainsClass)
def testMustEnableClassesOrInterfaces(self):
self.assertRaises(AssertionError, self.makeExtentService,
interface_extents=False, class_extents=False)
+
+class DummyEventService:
+ """Test Event Service Scaffold"""
+
+ def __init__(self):
+ self._listeners = Set()
+
+ def registerListener(obj, method_name, message_type):
+ self._listeners.add((obj, method_name, message_type))
+
+ def unregisterListener(obj, method_name, message_type):
+ self._listeners.remove((obj, method_name, message_type))
+
+ def send(message):
+ for obj, method, msg_type in self._listeners:
+ if isinstance(message, msg_type):
+ getattr(obj. method)(message)
+
+ def hasReg(obj, method_name, message_type):
+ return (obj, method_name, message_type) in self._listeners
+
+class DummyIdentityService:
+
+ def __init__(self, conn):
+ self._nextid = 1
+ self._objs = {}
+ self._conn = conn
+
+ def register(obj):
+ from pypes import services
+ from pypes.identity import IdRegisteredEvent
+ obj._pypesid_ = self._nextid
+ self._objs[self._nextid] = obj
+ self._nextid += 1
+ services.event(obj).send(IdRegisteredEvent(obj, obj._pypesid_))
+
+ def remove(obj):
+ from pypes import services
+ from pypes.identity import IdUnregisteredEvent
+ id = obj._pypesid_
+ del self._objs[id]
+ del obj._pypesid_
+ services.event(obj).send(IdUnregisteredEvent(obj, id))
+
+class TestClassExtentMap(unittest.TestCase):
+
+ def setUp(self):
+ from pypes.extent import ClassExtentMap
+ from pypes.serviceconfig import initServices, addService
+ self.conn = TestConnection()
+ initServices(self.conn)
+ self.event = DummyEventService()
+ addService(self.conn, 'event', self.event)
+ self.identity = DummyIdentityService(self.conn)
+ addService(self.conn, 'identity', self.identity)
+ orig_extent_factory = ClassExtentMap._extent_factory
+ ClassExtentMap._extent_factory = TestExtent
+ self.cemap = ClassExtentMap(self.conn)
+ ClassExtentMap._extent_factory = orig_extent_factory
+
+ def _makeObj(self, cls, *args, **kw):
+ obj = cls(*args, **kw)
+ self.identity.register(obj)
+ return obj
+
+ def xtestInterface(self):
+ from pypes.interfaces import IExtentMap
+ self.failUnless(verifyObject(IExtentMap, self.cemap))
+
+ def xtestIsListeningForIdEvents(self):
+ from pypes.identity import IdRegisteredEvent, IdUnregisteredEvent
+ self.failUnless(self.event.hasReg(
+ self.cemap, '_notifyIdRegistered', IdRegisteredEvent))
+ self.failUnless(self.event.hasReg(
+ self.cemap, '_notifyIdUnregistered', IdUnregisteredEvent))
+
+ def xtestAddSimpleObject(self):
+ obj = BarClass()
+ self.cemap.add(obj)
+ extent = self.cemap[BarClass]
+ self.assertEqual(len(extent), 1)
+ self.failUnless(obj in extent)
if __name__ == '__main__':
unittest.main()
More information about the Zope-CVS
mailing list