Commit 61084543 authored by Tres Seaver's avatar Tres Seaver

DemoStorage was unable to wrap base storages without an '_oid' attribute.

o Most notably, ZEO.ClientStorage (http://www.zope.org/Collectors/Zope/2016).

o Forward-ported from 3.4 branch.
parent e20e5b29
......@@ -33,6 +33,8 @@ from ZODB.tests import StorageTestBase, BasicStorage, VersionStorage, \
PackableStorage, Synchronization, ConflictResolution, RevisionStorage, \
MTStorage, ReadOnlyStorage
from ZODB.tests.testDemoStorage import DemoStorageWrappedBase
from ZEO.ClientStorage import ClientStorage
from ZEO.tests import forker, Cache, CommitLockTests, ThreadTests
......@@ -196,9 +198,42 @@ class MappingStorageTests(GenericTests):
def getConfig(self):
return """<mappingstorage 1/>"""
class DemoStorageWrappedAroundClientStorage(DemoStorageWrappedBase):
def getConfig(self):
return """<mappingstorage 1/>"""
def _makeBaseStorage(self):
logger.info("setUp() %s", self.id())
port = get_port()
zconf = forker.ZEOConfig(('', port))
zport, adminaddr, pid, path = forker.start_zeo_server(self.getConfig(),
zconf, port)
self._pids = [pid]
self._servers = [adminaddr]
self._conf_path = path
_base = ClientStorage(zport, '1', cache_size=20000000,
min_disconnect_poll=0.5, wait=1,
wait_timeout=60)
_base.registerDB(DummyDB(), None)
return _base
def tearDown(self):
DemoStorageWrappedBase.tearDown(self)
os.remove(self._conf_path)
for server in self._servers:
forker.shutdown_zeo_server(server)
if hasattr(os, 'waitpid'):
# Not in Windows Python until 2.3
for pid in self._pids:
os.waitpid(pid, 0)
test_classes = [OneTimeTests,
FileStorageTests,
MappingStorageTests]
MappingStorageTests,
DemoStorageWrappedAroundClientStorage,
]
def test_suite():
suite = unittest.TestSuite()
......
......@@ -102,10 +102,11 @@ class BaseStorage(UndoLogCompatible):
# a reserved oid for the root object). Our new_oid() method
# increments it by 1, and returns the result. It's really a
# 64-bit integer stored as an 8-byte big-endian string.
if base is None:
oid = getattr(base, '_oid', None)
if oid is None:
self._oid = z64
else:
self._oid = base._oid
self._oid = oid
def abortVersion(self, src, transaction):
if transaction is not self._transaction:
......
......@@ -11,7 +11,6 @@
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
import ZODB.DemoStorage
import unittest
from ZODB.tests import StorageTestBase, BasicStorage, \
......@@ -24,6 +23,7 @@ class DemoStorageTests(StorageTestBase.StorageTestBase,
):
def setUp(self):
import ZODB.DemoStorage
self._storage = ZODB.DemoStorage.DemoStorage()
def tearDown(self):
......@@ -55,8 +55,40 @@ class DemoStorageTests(StorageTestBase.StorageTestBase,
pass
class DemoStorageWrappedBase(DemoStorageTests):
def setUp(self):
import ZODB.DemoStorage
self._base = self._makeBaseStorage()
self._storage = ZODB.DemoStorage.DemoStorage(base=self._base)
def tearDown(self):
self._storage.close()
self._base.close()
def _makeBaseStorage(self):
raise NotImplementedError
class DemoStorageWrappedAroundFileStorage(DemoStorageWrappedBase):
def _makeBaseStorage(self):
from ZODB.MappingStorage import MappingStorage
return MappingStorage()
class DemoStorageWrappedAroundMappingStorage(DemoStorageWrappedBase):
def _makeBaseStorage(self):
from ZODB.FileStorage import FileStorage
return FileStorage('FileStorageTests.fs')
def test_suite():
suite = unittest.makeSuite(DemoStorageTests, 'check')
suite = unittest.TestSuite()
suite.addTest(unittest.makeSuite(DemoStorageTests, 'check'))
suite.addTest(unittest.makeSuite(DemoStorageWrappedAroundFileStorage,
'check'))
suite.addTest(unittest.makeSuite(DemoStorageWrappedAroundMappingStorage,
'check'))
return suite
if __name__ == "__main__":
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment