[Zope-CVS] CVS: Products/Ape/lib/apelib/tests -
serialtestbase.py:1.3.6.2 testio.py:1.4.2.2
testscanner.py:1.2.2.1 testserialization.py:1.5.2.2
testsqlimpl.py:1.1.8.1 teststorage.py:1.5.2.2
testzope2fs.py:1.4.4.2 testzope2sql.py:1.6.4.2
Shane Hathaway
shane at zope.com
Sat Dec 20 02:31:38 EST 2003
Update of /cvs-repository/Products/Ape/lib/apelib/tests
In directory cvs.zope.org:/tmp/cvs-serv30479/tests
Modified Files:
Tag: ape-0_8-branch
serialtestbase.py testio.py testscanner.py
testserialization.py testsqlimpl.py teststorage.py
testzope2fs.py testzope2sql.py
Log Message:
Continued refactoring and renaming.
Over 60 tests now pass.
=== Products/Ape/lib/apelib/tests/serialtestbase.py 1.3.6.1 => 1.3.6.2 ===
--- Products/Ape/lib/apelib/tests/serialtestbase.py:1.3.6.1 Fri Dec 19 21:52:49 2003
+++ Products/Ape/lib/apelib/tests/serialtestbase.py Sat Dec 20 02:31:07 2003
@@ -23,8 +23,8 @@
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
+from apelib.zodb3.serializers import StringToPersistentPM, StringToPicklePM
+from apelib.zodb3.serializers import RemainingState, RollCall
class TestObject(PersistentMapping):
@@ -34,13 +34,14 @@
def addMapper(conf, klass, mapper_name):
"""Adds a simple mapper to the configuration.
"""
- serializer = CompositeSerializer(klass.__module__, klass.__name__)
+ serializer = serializers.CompositeSerializer(
+ klass.__module__, klass.__name__)
gateway = gateways.RAMGateway(serializer.schema)
- mapper = mapper.Mapper(serializer, gateway)
+ m = mapper.Mapper(serializer, gateway)
class_name = '%s.%s' % (klass.__module__, klass.__name__)
- conf.mappers[mapper_name] = mapper
+ conf.mappers[mapper_name] = m
conf.classifier.register("class", class_name, mapper_name)
- return mapper
+ return m
class SerialTestBase:
@@ -52,13 +53,13 @@
self.conf = io.MapperConfiguration({}, cfr, oid_gen)
m = addMapper(self.conf, PersistentMapping, "pm")
- m.serializer.add(BasicPersistentMapping())
- m.serializer.add(RollCall())
+ m.serializer.add("items", StringToPersistentPM())
+ m.serializer.add("rollcall", RollCall())
m.gateway.schema = m.serializer.schema
m = addMapper(self.conf, TestObject, "tm")
- m.serializer.add(BasicPersistentMapping())
- m.serializer.add(RemainingState())
+ m.serializer.add("items", StringToPicklePM())
+ m.serializer.add("remainder", RemainingState())
m.gateway.schema = m.serializer.schema
self.conf.check()
=== Products/Ape/lib/apelib/tests/testio.py 1.4.2.1 => 1.4.2.2 ===
--- Products/Ape/lib/apelib/tests/testio.py:1.4.2.1 Fri Dec 19 21:52:49 2003
+++ Products/Ape/lib/apelib/tests/testio.py Sat Dec 20 02:31:07 2003
@@ -23,21 +23,19 @@
from apelib.core import io
from apelib.core.interfaces import IObjectDatabase
-from serialtestbase import SerialTestBase
+from serialtestbase import SerialTestBase, TestObject
class TestObjectDatabase:
__implements__ = IObjectDatabase
- def getObject(self, keychain, hints=None):
+ def getObject(self, oid, hints=None):
raise NotImplementedError
- loadStub = getObject
-
- def identifyObject(self, obj):
+ def identify(self, obj):
raise NotImplementedError
- def newKey(self):
+ def new_oid(self):
raise NotImplementedError
def getClass(self, module, name):
@@ -57,37 +55,37 @@
verifyClass(IObjectDatabase, TestObjectDatabase)
def testSerializeAndDeserialize(self):
- ob = PersistentMapping()
+ ob = TestObject()
ob.strdata = '345'
ob['a'] = 'b'
ob['c'] = 'd'
- keychain = ('test',)
+ oid = 'test'
obj_db = self.getObjectDatabase()
obsys = io.ObjectSystemIO(self.conf, obj_db)
- event, classified_state = obsys.serialize(keychain, ob)
+ event, classified_state = obsys.serialize(oid, ob)
ob2 = obsys.newObject(classified_state)
- obsys.deserialize(keychain, ob2, classified_state)
+ obsys.deserialize(oid, ob2, classified_state)
self.assertEqual(ob.strdata, ob2.strdata)
self.assertEqual(ob.data, ob2.data)
def testStoreAndLoad(self):
# Tests both serialization and storage
- ob = PersistentMapping()
+ ob = TestObject()
ob.strdata = '345'
ob['a'] = 'b'
ob['c'] = 'd'
- keychain = ('test',)
+ oid = 'test'
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)
+ event, classified_state = obsys.serialize(oid, ob)
+ gwsys.store(oid, classified_state, True)
- event, cs, hash_value = gwsys.load(keychain)
+ event, cs, hash_value = gwsys.load(oid)
ob2 = obsys.newObject(cs)
- obsys.deserialize(keychain, ob2, cs)
+ obsys.deserialize(oid, ob2, cs)
self.assertEqual(ob.strdata, ob2.strdata)
self.assertEqual(ob.data, ob2.data)
@@ -95,22 +93,22 @@
def testExportImport(self):
root = PersistentMapping()
- test1 = PersistentMapping()
+ test1 = TestObject()
test1.strdata = '345'
test1['a'] = 'b'
test1['c'] = 'd'
root['TestRoot'] = test1
- test2 = PersistentMapping()
+ test2 = TestObject()
test2.leftover = 'oops'
test2['true'] = 'undecided'
root['TestRoot2'] = test2
- keychain = ()
+ oid = ''
exporter = io.ExportImport(self.conf, self.conns)
- exporter.exportObject(root, keychain)
+ exporter.exportObject(root, oid)
importer = io.ExportImport(self.conf, self.conns)
- roota = importer.importObject(keychain)
+ roota = importer.importObject(oid)
self.assert_(root is not roota)
self.assert_(root['TestRoot'] is not roota['TestRoot'])
self.assert_(root['TestRoot2'] is not roota['TestRoot2'])
=== Products/Ape/lib/apelib/tests/testscanner.py 1.2 => 1.2.2.1 ===
--- Products/Ape/lib/apelib/tests/testscanner.py:1.2 Wed Jul 30 17:33:08 2003
+++ Products/Ape/lib/apelib/tests/testscanner.py Sat Dec 20 02:31:07 2003
@@ -24,7 +24,7 @@
class FakeRepository:
- def freshen(self, d):
+ def poll(self, d):
res = {}
for source in d.keys():
repo, location = source
@@ -41,7 +41,7 @@
repo = FakeRepository()
- def getSources(self, oid):
+ def getPollSources(self, oid):
return {(self.repo, oid): 10}
@@ -92,7 +92,7 @@
def testAddSource(self):
new_sources = {(self.repo, '5'): 0}
- self.scanner.setSources(5, new_sources)
+ self.scanner.setPollSources(5, new_sources)
self.assertEqual(len(self.scanner.future), 1)
self.assertEqual(self.scanner.future[5][0], new_sources)
@@ -100,13 +100,13 @@
self.conn1.setOIDs([5])
old_sources = {(self.repo, '5'): 0}
- self.scanner.setSources(5, old_sources)
+ self.scanner.setPollSources(5, old_sources)
self.assertEqual(len(self.scanner.future), 0)
self.assertEqual(len(self.scanner.current), 1)
self.assertEqual(self.scanner.current[5], old_sources)
new_sources = {(self.repo, '6'): 0, (self.repo, '7'): 0, }
- self.scanner.setSources(5, new_sources)
+ self.scanner.setPollSources(5, new_sources)
self.assertEqual(len(self.scanner.future), 0)
self.assertEqual(len(self.scanner.current), 1)
self.assertEqual(self.scanner.current[5], new_sources)
@@ -120,7 +120,7 @@
def testScan(self):
self.conn1.setOIDs([5])
new_sources = {(self.repo, '6'): 0, (self.repo, '7'): 0, }
- self.scanner.setSources(5, new_sources)
+ self.scanner.setPollSources(5, new_sources)
to_invalidate = self.scanner.scan()
self.assertEqual(len(to_invalidate), 1)
@@ -137,14 +137,14 @@
storage = FakeStorage()
self.scanner.setStorage(storage)
self.conn1.setOIDs([5])
- expect_sources = storage.getSources(5)
+ expect_sources = storage.getPollSources(5)
self.assertEqual(self.scanner.current[5], expect_sources)
def testUseCachedSources(self):
# Verify the scanner uses previously cached sources when available.
repo = FakeRepository()
sources = {(repo, '999'): -1}
- self.scanner.setSources(5, sources)
+ self.scanner.setPollSources(5, sources)
self.conn1.setOIDs([5])
self.assertEqual(self.scanner.current[5], sources)
@@ -152,7 +152,7 @@
# Verify the scanner sees sources according to transactions.
repo = FakeRepository()
sources = {(repo, '999'): -1}
- self.scanner.setSources(5, sources)
+ self.scanner.setPollSources(5, sources)
self.conn1.setOIDs([5])
sources_2 = {(repo, '999'): -2}
self.scanner.setUncommittedSources(123, 5, sources_2)
=== Products/Ape/lib/apelib/tests/testserialization.py 1.5.2.1 => 1.5.2.2 ===
--- Products/Ape/lib/apelib/tests/testserialization.py:1.5.2.1 Fri Dec 19 21:52:49 2003
+++ Products/Ape/lib/apelib/tests/testserialization.py Sat Dec 20 02:31:07 2003
@@ -47,10 +47,10 @@
obj_db = None
m = self.conf.mappers["tm"]
event = SerializationEvent(self.conf, m, '', obj_db, ob)
- full_state = mapper.serializer.serialize(event)
+ full_state = m.serializer.serialize(event)
ob2 = TestObject()
event = DeserializationEvent(self.conf, m, '', obj_db, ob2)
- mapper.serializer.deserialize(event, full_state)
+ m.serializer.deserialize(event, full_state)
self.assertEqual(ob.strdata, ob2.strdata)
self.assertEqual(ob.data, ob2.data)
@@ -62,15 +62,15 @@
obj_db = None
m = self.conf.mappers["tm"]
event = SerializationEvent(self.conf, m, '', obj_db, ob)
- full_state = mapper.serializer.serialize(event)
+ full_state = m.serializer.serialize(event)
event = StoreEvent(self.conf, m, '', self.conns, None, True)
- mapper.gateway.store(event, full_state)
+ m.gateway.store(event, full_state)
event = LoadEvent(self.conf, m, '', self.conns, None)
- full_state, serial = mapper.gateway.load(event)
+ full_state, serial = m.gateway.load(event)
ob2 = TestObject()
event = DeserializationEvent(self.conf, m, '', obj_db, ob2)
- mapper.serializer.deserialize(event, full_state)
+ m.serializer.deserialize(event, full_state)
self.assertEqual(ob.strdata, ob2.strdata)
self.assertEqual(ob.data, ob2.data)
@@ -82,8 +82,7 @@
obj_db = None
m = self.conf.mappers["pm"]
event = SerializationEvent(self.conf, m, '', obj_db, ob)
- self.assertRaises(SerializationError,
- mapper.serializer.serialize, event)
+ self.assertRaises(SerializationError, m.serializer.serialize, event)
def testSharedAttribute(self):
# Test of an attribute shared between a normal serializer and
@@ -95,16 +94,16 @@
obj_db = None
m = self.conf.mappers["tm"]
event = SerializationEvent(self.conf, m, '', obj_db, ob)
- full_state = mapper.serializer.serialize(event)
+ full_state = m.serializer.serialize(event)
event = StoreEvent(self.conf, m, '', self.conns, None, True)
- mapper.gateway.store(event, full_state)
+ m.gateway.store(event, full_state)
# Now load the state into a different object
event = LoadEvent(self.conf, m, '', self.conns, None)
- full_state, serial = mapper.gateway.load(event)
+ full_state, serial = m.gateway.load(event)
ob2 = TestObject()
event = DeserializationEvent(self.conf, m, '', obj_db, ob2)
- mapper.serializer.deserialize(event, full_state)
+ m.serializer.deserialize(event, full_state)
self.assertEqual(ob.extra.data, ob2.extra.data)
self.assertEqual(ob.keys(), ob2.keys())
=== Products/Ape/lib/apelib/tests/testsqlimpl.py 1.1 => 1.1.8.1 ===
--- Products/Ape/lib/apelib/tests/testsqlimpl.py:1.1 Wed Apr 9 23:09:57 2003
+++ Products/Ape/lib/apelib/tests/testsqlimpl.py Sat Dec 20 02:31:07 2003
@@ -25,6 +25,7 @@
def testSQLImplementations(self):
import apelib.sql
+ import apelib.sql.oidgen
self._testAllInPackage(apelib.sql)
if __name__ == '__main__':
=== Products/Ape/lib/apelib/tests/teststorage.py 1.5.2.1 => 1.5.2.2 ===
--- Products/Ape/lib/apelib/tests/teststorage.py:1.5.2.1 Fri Dec 19 21:52:49 2003
+++ Products/Ape/lib/apelib/tests/teststorage.py Sat Dec 20 02:31:07 2003
@@ -196,7 +196,7 @@
def _writeBasicObject(self, conn):
- ob = PersistentMapping()
+ ob = TestObject()
ob.strdata = 'abc'
root = conn.root()
get_transaction().begin()
@@ -316,7 +316,7 @@
# Verifies the behavior of getSerial().
conn1 = self.db.open()
try:
- new_ob = PersistentMapping()
+ new_ob = TestObject()
self.assertEqual(conn1.getSerial(new_ob), '\0' * 8)
ob1 = self._writeBasicObject(conn1)
self.assertNotEqual(conn1.getSerial(ob1), '\0' * 8)
@@ -362,7 +362,7 @@
def testGetPollSources(self):
sources = self.storage.getPollSources('\0' * 8)
- self.assertEqual(sources, {})
+ self.assert_(not sources)
# The test passed, but check for a false positive.
oid = 'nonexistent-oid'
self.assertRaises(KeyError, self.storage.getPollSources, oid)
=== Products/Ape/lib/apelib/tests/testzope2fs.py 1.4.4.1 => 1.4.4.2 ===
--- Products/Ape/lib/apelib/tests/testzope2fs.py:1.4.4.1 Fri Dec 19 21:52:49 2003
+++ Products/Ape/lib/apelib/tests/testzope2fs.py Sat Dec 20 02:31:07 2003
@@ -31,7 +31,7 @@
from apelib.zodb3.db import ApeDB
from apelib.zodb3.storage import ApeStorage
from apelib.zodb3.resource import StaticResource
-from apelib.zope2.mapper import createMapper
+from apelib.zope2.mapper import loadConf
from apelib.fs.exceptions import FSWriteError
from apelib.fs.connection import FSConnection
from zope2testbase import Zope2TestBase, Folder
@@ -44,7 +44,7 @@
tmpdir = mktemp()
-root_mapper = None
+conf = None
class Zope2FSTests (unittest.TestCase, Zope2TestBase):
@@ -54,17 +54,17 @@
return {'fs': conn}
def setUp(self):
- global root_mapper
- if root_mapper is None:
- root_mapper = createMapper('filesystem')
+ global conf
+ if conf is None:
+ conf = loadConf('filesystem')
if not os.path.exists(tmpdir):
os.mkdir(tmpdir)
self.path = tmpdir
- self.root_mapper = root_mapper
+ self.conf = conf
conns = self._createConnections(tmpdir)
assert len(conns) == 1
self.conn = conns['fs']
- resource = StaticResource(self.root_mapper)
+ resource = StaticResource(self.conf)
storage = ApeStorage(resource, conns)
self.storage = storage
db = ApeDB(storage, resource)
@@ -97,16 +97,14 @@
get_transaction().commit()
for folder in (f, f2, f3):
- keychain = conn.db()._oid_encoder.decode(folder._p_oid)
- key = keychain[-1]
- text = self.conn.readSection(key, 'classification')
+ text = self.conn.readSection(folder._p_oid, 'classification')
self.assert_(text.find('class_name=OFS.Folder.Folder') >= 0)
finally:
conn.close()
def testMismatchedIdDetection(self):
- # FSAutoID should detect when the keychain and ID don't match.
+ # FSAutoID should detect when the OID and ID don't match.
# Normally, the only time they don't match is when an object
# has been moved.
conn = self.db.open()
=== Products/Ape/lib/apelib/tests/testzope2sql.py 1.6.4.1 => 1.6.4.2 ===
--- Products/Ape/lib/apelib/tests/testzope2sql.py:1.6.4.1 Fri Dec 19 21:52:49 2003
+++ Products/Ape/lib/apelib/tests/testzope2sql.py Sat Dec 20 02:31:07 2003
@@ -23,11 +23,11 @@
from apelib.zodb3.db import ApeDB
from apelib.zodb3.storage import ApeStorage
from apelib.zodb3.resource import StaticResource
-from apelib.zope2.mapper import createMapper
+from apelib.zope2.mapper import loadConf
from zope2testbase import Zope2TestBase
-root_mapper = None
+conf = None
class Zope2SQLTests (Zope2TestBase):
@@ -40,12 +40,12 @@
self.dbapi_kwparams, prefix='test_temp')
def setUp(self):
- global root_mapper
- if root_mapper is None:
- root_mapper = createMapper('sql')
+ global conf
+ if conf is None:
+ conf = loadConf('sql')
conn = self.getConnector()
- self.root_mapper = root_mapper
- resource = StaticResource(self.root_mapper)
+ self.conf = conf
+ resource = StaticResource(self.conf)
self.conns = {'db': conn}
storage = ApeStorage(resource, self.conns, clear_all=1)
self.storage = storage
More information about the Zope-CVS
mailing list