[Zope-CVS] CVS: Products/Ape/lib/apelib/tests -
serialtestbase.py:1.3.6.1 testio.py:1.4.2.1
testserialization.py:1.5.2.1 teststorage.py:1.5.2.1
testzope2fs.py:1.4.4.1 testzope2sql.py:1.6.4.1
Shane Hathaway
shane at zope.com
Fri Dec 19 21:53:20 EST 2003
Update of /cvs-repository/Products/Ape/lib/apelib/tests
In directory cvs.zope.org:/tmp/cvs-serv5534/tests
Modified Files:
Tag: ape-0_8-branch
serialtestbase.py testio.py testserialization.py
teststorage.py testzope2fs.py testzope2sql.py
Log Message:
Cleaned up to the point that ApelibImplTests pass.
=== Products/Ape/lib/apelib/tests/serialtestbase.py 1.3 => 1.3.6.1 ===
--- Products/Ape/lib/apelib/tests/serialtestbase.py:1.3 Mon May 26 14:16:02 2003
+++ Products/Ape/lib/apelib/tests/serialtestbase.py Fri Dec 19 21:52:49 2003
@@ -20,111 +20,49 @@
from Persistence import PersistentMapping
from cPickle import dumps, loads
-from apelib.core \
- import classifiers, gateways, keygen, mapper, schemas, serializers
-from apelib.core.interfaces import ISerializer
-
-from apelib.zodb3.serializers \
- import FixedPersistentMapping, RollCall, RemainingState
-
-
-class SimpleItemsSerializer:
-
- __implements__ = ISerializer
-
- schema = schemas.RowSequenceSchema()
- schema.addField('name')
- schema.addField('pickle')
-
- def getSchema(self):
- return self.schema
-
- def serialize(self, object, event):
- res = []
- for k, v in object.items():
- res.append((k, dumps(v)))
- event.notifySerialized(k, v, 0)
- res.sort()
- event.ignoreAttribute('data')
- event.ignoreAttribute('_container')
- return res
-
- def deserialize(self, object, event, state):
- d = {}
- for k, v in state:
- o = loads(v)
- d[k] = o
- event.notifyDeserialized(k, o)
- object.__init__(d)
+from apelib.core import classifiers, gateways, io
+from apelib.core import mapper, oidgen, schemas, serializers
+from apelib.zodb3.serializers import BasicPersistentMapping, RollCall
+from apelib.zodb3.serializers import RemainingState
-class SerialTestBase:
- def setUp(self):
- self.conns = {}
+class TestObject(PersistentMapping):
+ strdata = ""
+
+
+def addMapper(conf, klass, mapper_name):
+ """Adds a simple mapper to the configuration.
+ """
+ serializer = CompositeSerializer(klass.__module__, klass.__name__)
+ gateway = gateways.RAMGateway(serializer.schema)
+ mapper = mapper.Mapper(serializer, gateway)
+ class_name = '%s.%s' % (klass.__module__, klass.__name__)
+ conf.mappers[mapper_name] = mapper
+ conf.classifier.register("class", class_name, mapper_name)
+ return mapper
- cfr = classifiers.FixedClassifier()
- cfr.register('test', 'test_mapper')
- cfr.register('test2', 'test_mapper_2')
-
- ser2 = serializers.CompositeSerializer(
- 'Persistence', 'PersistentMapping')
- fixed_items_serializer = FixedPersistentMapping()
- fixed_items_serializer.add('TestRoot', ('test',), ('test_mapper',))
- fixed_items_serializer.add('TestRoot2', ('test2',), ('test_mapper_2',))
- ser2.addSerializer('fixed_items', fixed_items_serializer)
- ser2.addSerializer('roll_call', RollCall())
-
- root_mapper = mapper.Mapper(
- None, ser2, gateways.CompositeGateway(), cfr)
- self.root_mapper = root_mapper
- # Since all of the keychains used by this mapper are fixed, we
- # don't want any keychains to be generated.
- root_mapper.setKeychainGenerator(keygen.NullKeychainGenerator())
-
- # Create "test_mapper", which allows a "strdata" attribute.
-
- ser1 = serializers.CompositeSerializer(
- 'Persistence', 'PersistentMapping')
- items_serializer = SimpleItemsSerializer()
- ser1.addSerializer('items', items_serializer)
- props_serializer = serializers.StringDataAttribute('strdata')
- ser1.addSerializer('properties', props_serializer)
- ser1.addSerializer('roll_call', RollCall())
-
- gw1 = gateways.CompositeGateway()
- items_gw = gateways.MappingGateway(items_serializer.getSchema())
- self.items_gw = items_gw
- gw1.addGateway('items', items_gw)
- props_gw = gateways.MappingGateway(props_serializer.getSchema())
- self.props_gw = props_gw
- gw1.addGateway('properties', props_gw)
-
- om1 = mapper.Mapper(root_mapper, ser1, gw1)
- self.om1 = om1
- root_mapper.addSubMapper('test_mapper', om1)
-
- # Create "test_mapper_2", which stores a remainder.
-
- ser = serializers.CompositeSerializer(
- 'Persistence', 'PersistentMapping')
- items_serializer = SimpleItemsSerializer()
- ser.addSerializer('items', items_serializer)
- remainder_serializer = RemainingState()
- ser.addSerializer('remainder', remainder_serializer)
-
- gw = gateways.CompositeGateway()
- items_gw = gateways.MappingGateway(items_serializer.getSchema())
- gw.addGateway('items', items_gw)
- remainder_gw = gateways.MappingGateway(
- remainder_serializer.getSchema())
- gw.addGateway('remainder', remainder_gw)
- om = mapper.Mapper(root_mapper, ser, gw)
- root_mapper.addSubMapper('test_mapper_2', om)
+class SerialTestBase:
- root_mapper.checkConfiguration()
+ def setUp(self):
+ schema = schemas.FieldSchema("classification", "classification")
+ cfr = classifiers.SimpleClassifier(gateways.RAMGateway(schema))
+ oid_gen = oidgen.SerialOIDGenerator()
+ self.conf = io.MapperConfiguration({}, cfr, oid_gen)
+
+ m = addMapper(self.conf, PersistentMapping, "pm")
+ m.serializer.add(BasicPersistentMapping())
+ m.serializer.add(RollCall())
+ m.gateway.schema = m.serializer.schema
+
+ m = addMapper(self.conf, TestObject, "tm")
+ m.serializer.add(BasicPersistentMapping())
+ m.serializer.add(RemainingState())
+ m.gateway.schema = m.serializer.schema
+ self.conf.check()
+ self.conns = {}
def tearDown(self):
pass
=== Products/Ape/lib/apelib/tests/testio.py 1.4 => 1.4.2.1 ===
--- Products/Ape/lib/apelib/tests/testio.py:1.4 Thu Aug 14 16:20:05 2003
+++ Products/Ape/lib/apelib/tests/testio.py Fri Dec 19 21:52:49 2003
@@ -22,12 +22,12 @@
from Persistence import PersistentMapping
from apelib.core import io
-from apelib.core.interfaces import IKeyedObjectSystem
+from apelib.core.interfaces import IObjectDatabase
from serialtestbase import SerialTestBase
-class TestKeyedObjectSystem:
- __implements__ = IKeyedObjectSystem
+class TestObjectDatabase:
+ __implements__ = IObjectDatabase
def getObject(self, keychain, hints=None):
raise NotImplementedError
@@ -48,13 +48,13 @@
class ApeIOTests(SerialTestBase, unittest.TestCase):
- def getKeyedObjectSystem(self):
- return TestKeyedObjectSystem()
+ def getObjectDatabase(self):
+ return TestObjectDatabase()
def testImpl(self):
# Test of test :-)
from Interface.Verify import verifyClass
- verifyClass(IKeyedObjectSystem, TestKeyedObjectSystem)
+ verifyClass(IObjectDatabase, TestObjectDatabase)
def testSerializeAndDeserialize(self):
ob = PersistentMapping()
@@ -62,8 +62,8 @@
ob['a'] = 'b'
ob['c'] = 'd'
keychain = ('test',)
- kos = self.getKeyedObjectSystem()
- obsys = io.ObjectSystemIO(self.root_mapper, kos)
+ obj_db = self.getObjectDatabase()
+ obsys = io.ObjectSystemIO(self.conf, obj_db)
event, classified_state = obsys.serialize(keychain, ob)
ob2 = obsys.newObject(classified_state)
@@ -79,9 +79,9 @@
ob['a'] = 'b'
ob['c'] = 'd'
keychain = ('test',)
- kos = self.getKeyedObjectSystem()
- obsys = io.ObjectSystemIO(self.root_mapper, kos)
- gwsys = io.GatewayIO(self.root_mapper, self.conns)
+ obj_db = self.getObjectDatabase()
+ obsys = io.ObjectSystemIO(self.conf, obj_db)
+ gwsys = io.GatewayIO(self.conf, self.conns)
event, classified_state = obsys.serialize(keychain, ob)
gwsys.store(keychain, classified_state)
@@ -106,10 +106,10 @@
root['TestRoot2'] = test2
keychain = ()
- exporter = io.ExportImport(self.root_mapper, self.conns)
+ exporter = io.ExportImport(self.conf, self.conns)
exporter.exportObject(root, keychain)
- importer = io.ExportImport(self.root_mapper, self.conns)
+ importer = io.ExportImport(self.conf, self.conns)
roota = importer.importObject(keychain)
self.assert_(root is not roota)
self.assert_(root['TestRoot'] is not roota['TestRoot'])
=== Products/Ape/lib/apelib/tests/testserialization.py 1.5 => 1.5.2.1 ===
--- Products/Ape/lib/apelib/tests/testserialization.py:1.5 Thu Aug 14 16:20:05 2003
+++ Products/Ape/lib/apelib/tests/testserialization.py Fri Dec 19 21:52:49 2003
@@ -23,93 +23,93 @@
from apelib.core.events \
import LoadEvent, StoreEvent, SerializationEvent, DeserializationEvent
-from apelib.core.exceptions import SerializationError
-from serialtestbase import SerialTestBase
+from apelib.core.interfaces import SerializationError
+from serialtestbase import SerialTestBase, TestObject
-class SimpleInstance:
-
+class SimpleClass:
+ """Represents second-class persistent objects.
+ """
def __init__(self, data):
self.data = data
class SerializationTests(SerialTestBase, unittest.TestCase):
+ """Tests of basic events, serializers, and gateways.
- def getKeyedObjectSystem(self):
- # XXX works for now
- return None
+ No connections or object databases are provided.
+ """
def testSerializeAndDeserialize(self):
- ob = PersistentMapping()
- ob.strdata = '345'
+ ob = TestObject()
ob['a'] = 'b'
ob['c'] = 'd'
- kos = self.getKeyedObjectSystem()
- mapper = self.root_mapper.getSubMapper('test_mapper')
- event = SerializationEvent(kos, mapper, ('',), ob)
- full_state = mapper.getSerializer().serialize(ob, event)
- ob2 = PersistentMapping()
- event = DeserializationEvent(kos, mapper, ('',), ob2)
- mapper.getSerializer().deserialize(ob2, event, full_state)
+ obj_db = None
+ m = self.conf.mappers["tm"]
+ event = SerializationEvent(self.conf, m, '', obj_db, ob)
+ full_state = mapper.serializer.serialize(event)
+ ob2 = TestObject()
+ event = DeserializationEvent(self.conf, m, '', obj_db, ob2)
+ mapper.serializer.deserialize(event, full_state)
self.assertEqual(ob.strdata, ob2.strdata)
self.assertEqual(ob.data, ob2.data)
def testStoreAndLoad(self):
- ob = PersistentMapping()
+ ob = TestObject()
ob.strdata = '345'
ob['a'] = 'b'
ob['c'] = 'd'
- kos = self.getKeyedObjectSystem()
- mapper = self.root_mapper.getSubMapper('test_mapper')
- event = SerializationEvent(kos, mapper, ('',), ob)
- full_state = mapper.getSerializer().serialize(ob, event)
- event = StoreEvent(mapper, ('',), self.conns, None)
- mapper.getGateway().store(event, full_state)
-
- event = LoadEvent(mapper, ('',), self.conns, None)
- full_state, serial = mapper.getGateway().load(event)
- ob2 = PersistentMapping()
- event = DeserializationEvent(kos, mapper, ('',), ob2)
- mapper.getSerializer().deserialize(ob2, event, full_state)
+ obj_db = None
+ m = self.conf.mappers["tm"]
+ event = SerializationEvent(self.conf, m, '', obj_db, ob)
+ full_state = mapper.serializer.serialize(event)
+ event = StoreEvent(self.conf, m, '', self.conns, None, True)
+ mapper.gateway.store(event, full_state)
+
+ event = LoadEvent(self.conf, m, '', self.conns, None)
+ full_state, serial = mapper.gateway.load(event)
+ ob2 = TestObject()
+ event = DeserializationEvent(self.conf, m, '', obj_db, ob2)
+ mapper.serializer.deserialize(event, full_state)
self.assertEqual(ob.strdata, ob2.strdata)
self.assertEqual(ob.data, ob2.data)
def testCatchExtraAttribute(self):
ob = PersistentMapping()
- ob.strdata = '345'
ob.extra = '678'
ob['a'] = 'b'
ob['c'] = 'd'
- kos = self.getKeyedObjectSystem()
- mapper = self.root_mapper.getSubMapper('test_mapper')
- event = SerializationEvent(kos, mapper, ('',), ob)
+ obj_db = None
+ m = self.conf.mappers["pm"]
+ event = SerializationEvent(self.conf, m, '', obj_db, ob)
self.assertRaises(SerializationError,
- mapper.getSerializer().serialize, ob, event)
+ mapper.serializer.serialize, event)
def testSharedAttribute(self):
# Test of an attribute shared between a normal serializer and
# a remainder serializer.
- ob = PersistentMapping()
- data = SimpleInstance('This is a shared piece of data')
+ ob = TestObject()
+ data = SimpleClass('This is a shared piece of data')
ob.extra = data
ob['a'] = data
- kos = self.getKeyedObjectSystem()
- mapper = self.root_mapper.getSubMapper('test_mapper_2')
- event = SerializationEvent(kos, mapper, ('',), ob)
- full_state = mapper.getSerializer().serialize(ob, event)
- event = StoreEvent(mapper, ('',), self.conns, None)
- mapper.getGateway().store(event, full_state)
+ obj_db = None
+ m = self.conf.mappers["tm"]
+ event = SerializationEvent(self.conf, m, '', obj_db, ob)
+ full_state = mapper.serializer.serialize(event)
+ event = StoreEvent(self.conf, m, '', self.conns, None, True)
+ mapper.gateway.store(event, full_state)
# Now load the state into a different object
- event = LoadEvent(mapper, ('',), self.conns, None)
- full_state, serial = mapper.getGateway().load(event)
- ob2 = PersistentMapping()
- event = DeserializationEvent(kos, mapper, ('',), ob2)
- mapper.getSerializer().deserialize(ob2, event, full_state)
+ event = LoadEvent(self.conf, m, '', self.conns, None)
+ full_state, serial = mapper.gateway.load(event)
+ ob2 = TestObject()
+ event = DeserializationEvent(self.conf, m, '', obj_db, ob2)
+ mapper.serializer.deserialize(event, full_state)
self.assertEqual(ob.extra.data, ob2.extra.data)
self.assertEqual(ob.keys(), ob2.keys())
- # Check that both see the *same* object
+ # Check that both ways to access the SimpleClass instance
+ # result in the same object.
self.assert_(ob2['a'] is ob2.extra, (ob2['a'], ob2.extra))
self.assert_(ob2['a'] is not data) # Verify it didn't cheat somehow
=== Products/Ape/lib/apelib/tests/teststorage.py 1.5 => 1.5.2.1 ===
--- Products/Ape/lib/apelib/tests/teststorage.py:1.5 Wed Jul 30 17:33:08 2003
+++ Products/Ape/lib/apelib/tests/teststorage.py Fri Dec 19 21:52:49 2003
@@ -26,7 +26,8 @@
from apelib.zodb3.storage import ApeStorage
from apelib.zodb3.resource import StaticResource
from apelib.zodb3.utils import copyOf
-from serialtestbase import SerialTestBase
+from apelib.core.interfaces import StoreError
+from serialtestbase import SerialTestBase, TestObject
def run_in_thread(f):
@@ -48,7 +49,7 @@
def setUp(self):
SerialTestBase.setUp(self)
- resource = StaticResource(self.root_mapper)
+ resource = StaticResource(self.conf)
storage = ApeStorage(resource, self.conns)
self.storage = storage
db = ApeDB(storage, resource)
@@ -60,13 +61,11 @@
SerialTestBase.tearDown(self)
def testStoreAndLoad(self):
- ob = PersistentMapping()
+ ob = TestObject()
ob.strdata = '345'
ob['a'] = 'b'
ob['c'] = 'd'
- dummy = PersistentMapping()
-
conn1 = self.db.open()
conn2 = None
conn3 = None
@@ -76,16 +75,15 @@
root = conn1.root()
get_transaction().begin()
root['TestRoot'] = ob
- root['TestRoot2'] = dummy
get_transaction().commit()
- ob1 = conn1.getObject(('test',))
+ ob1 = conn1.root()['TestRoot']
self.assertEqual(ob1.strdata, ob.strdata)
self.assertEqual(ob1.items(), ob.items())
# Verify a new object was stored and make a change
get_transaction().begin()
conn2 = self.db.open()
- ob2 = conn2.getObject(('test',))
+ ob2 = conn2.root()['TestRoot']
self.assertEqual(ob2.strdata, ob.strdata)
self.assertEqual(ob2.items(), ob.items())
ob2.strdata = '678'
@@ -93,7 +91,7 @@
# Verify the change was stored and make another change
conn3 = self.db.open()
- ob3 = conn3.getObject(('test',))
+ ob3 = conn3.root()['TestRoot']
self.assertEqual(ob3.strdata, '678')
self.assertEqual(ob3.items(), ob.items())
ob3.strdata = '901'
@@ -101,7 +99,7 @@
conn3.close()
conn3 = None
conn3 = self.db.open()
- ob3 = conn3.getObject(('test',))
+ ob3 = conn3.root()['TestRoot']
self.assertEqual(ob3.strdata, '901')
# Verify we didn't accidentally change the original object
@@ -116,14 +114,11 @@
def testUnmanaged(self):
- ob = PersistentMapping()
+ ob = TestObject()
ob['a'] = 'b'
ob.stowaway = PersistentMapping()
ob.stowaway['c'] = 'd'
- dummy = PersistentMapping()
- dummy.strdata = 'foo'
-
conn1 = self.db.open()
conn2 = None
conn3 = None
@@ -132,10 +127,9 @@
# Load the root and create a new object
root = conn1.root()
get_transaction().begin()
- root['TestRoot'] = dummy
root['TestRoot2'] = ob
get_transaction().commit()
- ob1 = conn1.getObject(('test2',))
+ ob1 = conn1.root()['TestRoot2']
self.assert_(ob1 is ob)
self.assertEqual(ob1.items(), [('a', 'b')])
self.assertEqual(ob1.stowaway.items(), [('c', 'd')])
@@ -143,7 +137,7 @@
# Verify a new object was stored
get_transaction().begin()
conn2 = self.db.open()
- ob2 = conn2.getObject(('test2',))
+ ob2 = conn2.root()['TestRoot2']
self.assertEqual(ob2.items(), [('a', 'b')])
self.assertEqual(ob2.stowaway.items(), [('c', 'd')])
@@ -155,7 +149,7 @@
# Verify the change was stored and make a change to the
# managed persistent object.
conn3 = self.db.open()
- ob3 = conn3.getObject(('test2',))
+ ob3 = conn3.root()['TestRoot2']
self.assertEqual(ob3.items(), [('a', 'b')])
self.assertEqual(ob3.stowaway.items(), [('c', 'e')])
ob3['a'] = 'z'
@@ -163,7 +157,7 @@
conn3.close()
conn3 = None
conn3 = self.db.open()
- ob3 = conn3.getObject(('test2',))
+ ob3 = conn3.root()['TestRoot2']
self.assertEqual(ob3['a'], 'z')
self.assertEqual(ob3.stowaway.items(), [('c', 'e')])
@@ -184,52 +178,47 @@
def testStoreAndLoadBinary(self):
- ob = PersistentMapping()
+ ob = TestObject()
# strdata contains binary characters
ob.strdata = ''.join([chr(n) for n in range(256)]) * 2
- dummy = PersistentMapping()
-
conn1 = self.db.open()
try:
root = conn1.root()
get_transaction().begin()
root['TestRoot'] = ob
- root['TestRoot2'] = dummy
get_transaction().commit()
- ob1 = conn1.getObject(('test',))
+ ob1 = conn1.root()['TestRoot']
self.assertEqual(ob1.strdata, ob.strdata)
self.assertEqual(ob1.items(), ob.items())
finally:
conn1.close()
- def _writeBasicObjects(self, conn):
+ def _writeBasicObject(self, conn):
ob = PersistentMapping()
ob.strdata = 'abc'
- dummy = PersistentMapping()
root = conn.root()
get_transaction().begin()
root['TestRoot'] = ob
- root['TestRoot2'] = dummy
get_transaction().commit()
- return ob, dummy
+ return ob
def _changeTestRoot(self):
- conn2 = self.db.open()
+ conn = self.db.open()
try:
- ob2 = conn2.root()['TestRoot']
- ob2.strdata = 'ghi'
+ ob = conn.root()['TestRoot']
+ ob.strdata = 'ghi'
get_transaction().commit()
finally:
- conn2.close()
+ conn.close()
def testConflictDetection(self):
conn1 = self.db.open()
try:
- ob1, dummy = self._writeBasicObjects(conn1)
+ ob1 = self._writeBasicObject(conn1)
ob1.strdata = 'def'
run_in_thread(self._changeTestRoot)
# Verify that "def" doesn't get written, since it
@@ -245,11 +234,10 @@
# Verify a new object won't overwrite existing objects by accident
conn1 = self.db.open()
try:
- ob1, dummy = self._writeBasicObjects(conn1)
+ ob1 = self._writeBasicObject(conn1)
ob1.strdata = 'def'
conn1.setSerial(ob1, '\0' * 8) # Pretend that it's new
- self.assertRaises(ZODB.POSException.ConflictError,
- get_transaction().commit)
+ self.assertRaises(StoreError, get_transaction().commit)
finally:
conn1.close()
@@ -257,17 +245,13 @@
def testRemainderCyclicReferenceRestoration(self):
# test whether the remainder pickler properly stores cyclic references
# back to the object itself.
- ob1 = PersistentMapping()
+ ob1 = TestObject()
ob1.myself = ob1
- dummy = PersistentMapping()
- dummy.strdata = ''
-
conn1 = self.db.open()
try:
root = conn1.root()
get_transaction().begin()
- root['TestRoot'] = dummy
root['TestRoot2'] = ob1
get_transaction().commit()
@@ -322,9 +306,8 @@
# use _p_serial for hashes.
conn1 = self.db.open()
try:
- ob1, dummy = self._writeBasicObjects(conn1)
+ ob1 = self._writeBasicObject(conn1)
self.assertEqual(ob1._p_serial, "\0" * 8)
- self.assertEqual(dummy._p_serial, "\0" * 8)
finally:
conn1.close()
@@ -335,7 +318,7 @@
try:
new_ob = PersistentMapping()
self.assertEqual(conn1.getSerial(new_ob), '\0' * 8)
- ob1, dummy = self._writeBasicObjects(conn1)
+ ob1 = self._writeBasicObject(conn1)
self.assertNotEqual(conn1.getSerial(ob1), '\0' * 8)
finally:
conn1.close()
@@ -345,8 +328,8 @@
# Verifies the behavior of getSerial() and setSerial().
conn1 = self.db.open()
try:
- ob1, dummy = self._writeBasicObjects(conn1)
- self.assertNotEqual(conn1.getSerial(ob1), '\0' * 8)
+ ob = self._writeBasicObject(conn1)
+ self.assertNotEqual(conn1.getSerial(ob), '\0' * 8)
# Replace the object and verify it gets a new serial.
ob1 = PersistentMapping()
ob1.strdata = 'cba'
@@ -361,8 +344,8 @@
# Verify that setSerial() cleans up.
conn1 = self.db.open()
try:
- conn1.SERIAL_CLEANUP_THRESHOLD = 10
- for n in range(conn1.SERIAL_CLEANUP_THRESHOLD + 1):
+ conn1.serial_cleanup_threshold = 10
+ for n in range(conn1.serial_cleanup_threshold + 1):
new_ob = PersistentMapping()
new_ob._p_oid = 'fake_oid_' + str(n)
old_size = len(conn1._serials or ())
@@ -377,12 +360,12 @@
conn1.close()
- def testGetSources(self):
- sources = self.storage.getSources('\0' * 8)
+ def testGetPollSources(self):
+ sources = self.storage.getPollSources('\0' * 8)
self.assertEqual(sources, {})
# The test passed, but check for a false positive.
- oid = self.storage._oid_encoder.encode(('nonexistent-oid',))
- self.assertRaises(KeyError, self.storage.getSources, oid)
+ oid = 'nonexistent-oid'
+ self.assertRaises(KeyError, self.storage.getPollSources, oid)
if __name__ == '__main__':
=== Products/Ape/lib/apelib/tests/testzope2fs.py 1.4 => 1.4.4.1 ===
--- Products/Ape/lib/apelib/tests/testzope2fs.py:1.4 Wed Jul 9 11:40:08 2003
+++ Products/Ape/lib/apelib/tests/testzope2fs.py Fri Dec 19 21:52:49 2003
@@ -44,7 +44,7 @@
tmpdir = mktemp()
-root_mapper = createMapper('filesystem')
+root_mapper = None
class Zope2FSTests (unittest.TestCase, Zope2TestBase):
@@ -54,6 +54,9 @@
return {'fs': conn}
def setUp(self):
+ global root_mapper
+ if root_mapper is None:
+ root_mapper = createMapper('filesystem')
if not os.path.exists(tmpdir):
os.mkdir(tmpdir)
self.path = tmpdir
=== Products/Ape/lib/apelib/tests/testzope2sql.py 1.6 => 1.6.4.1 ===
--- Products/Ape/lib/apelib/tests/testzope2sql.py:1.6 Wed Jul 9 11:40:08 2003
+++ Products/Ape/lib/apelib/tests/testzope2sql.py Fri Dec 19 21:52:49 2003
@@ -27,7 +27,7 @@
from zope2testbase import Zope2TestBase
-root_mapper = createMapper('sql')
+root_mapper = None
class Zope2SQLTests (Zope2TestBase):
@@ -40,6 +40,9 @@
self.dbapi_kwparams, prefix='test_temp')
def setUp(self):
+ global root_mapper
+ if root_mapper is None:
+ root_mapper = createMapper('sql')
conn = self.getConnector()
self.root_mapper = root_mapper
resource = StaticResource(self.root_mapper)
More information about the Zope-CVS
mailing list