[Zodb-checkins] SVN: ZODB/trunk/src/Z Updated most tests to use a commob test framework to try to make sure
Jim Fulton
jim at zope.com
Tue Oct 28 17:33:23 EDT 2008
Log message for revision 92661:
Updated most tests to use a commob test framework to try to make sure
tests have a clean transaction state and to avoid leaving files behind.
Changed:
U ZODB/trunk/src/ZEO/scripts/manual_tests/testzeopack.py
U ZODB/trunk/src/ZEO/tests/ConnectionTests.py
U ZODB/trunk/src/ZEO/tests/testAuth.py
U ZODB/trunk/src/ZEO/tests/testZEO.py
U ZODB/trunk/src/ZEO/tests/test_cache.py
U ZODB/trunk/src/ZODB/scripts/manual_tests/testfstest.py
U ZODB/trunk/src/ZODB/scripts/manual_tests/testzeopack.py
U ZODB/trunk/src/ZODB/tests/Corruption.py
U ZODB/trunk/src/ZODB/tests/RecoveryStorage.py
U ZODB/trunk/src/ZODB/tests/StorageTestBase.py
U ZODB/trunk/src/ZODB/tests/blob_connection.txt
U ZODB/trunk/src/ZODB/tests/blob_importexport.txt
U ZODB/trunk/src/ZODB/tests/blob_layout.txt
U ZODB/trunk/src/ZODB/tests/blob_packing.txt
U ZODB/trunk/src/ZODB/tests/blob_tempdir.txt
U ZODB/trunk/src/ZODB/tests/blob_transaction.txt
U ZODB/trunk/src/ZODB/tests/testCache.py
U ZODB/trunk/src/ZODB/tests/testConfig.py
U ZODB/trunk/src/ZODB/tests/testConnection.py
U ZODB/trunk/src/ZODB/tests/testDB.py
U ZODB/trunk/src/ZODB/tests/testDemoStorage.py
U ZODB/trunk/src/ZODB/tests/testFileStorage.py
U ZODB/trunk/src/ZODB/tests/testMappingStorage.py
U ZODB/trunk/src/ZODB/tests/testRecover.py
U ZODB/trunk/src/ZODB/tests/testZODB.py
U ZODB/trunk/src/ZODB/tests/test_fsdump.py
U ZODB/trunk/src/ZODB/tests/test_storage.py
U ZODB/trunk/src/ZODB/tests/testblob.py
U ZODB/trunk/src/ZODB/tests/testconflictresolution.py
U ZODB/trunk/src/ZODB/tests/testfsoids.py
U ZODB/trunk/src/ZODB/tests/testhistoricalconnections.py
U ZODB/trunk/src/ZODB/tests/testpersistentclass.py
-=-
Modified: ZODB/trunk/src/ZEO/scripts/manual_tests/testzeopack.py
===================================================================
--- ZODB/trunk/src/ZEO/scripts/manual_tests/testzeopack.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZEO/scripts/manual_tests/testzeopack.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -8,7 +8,6 @@
import os
import socket
-import tempfile
import threading
import time
import unittest
@@ -22,11 +21,12 @@
class PackerTests(StorageTestBase):
def setUp(self):
+ StorageTestBase.setUp(self)
self.started = 0
def start(self):
self.started =1
- self.path = tempfile.mktemp(suffix=".fs")
+ self.path = 'Data.fs'
self._storage = FileStorage(self.path)
self.db = ZODB.DB(self._storage)
self.do_updates()
@@ -37,27 +37,21 @@
self._dostore()
def tearDown(self):
- if not self.started:
- return
- self.db.close()
- self._storage.close()
- self.exit.close()
- try:
- os.kill(self.pid, 9)
- except os.error:
- pass
- try:
- os.waitpid(self.pid, 0)
- except os.error, err:
- ##print "waitpid failed", err
- pass
- for ext in '', '.old', '.lock', '.index', '.tmp':
- path = self.path + ext
+ if self.started:
+ self.db.close()
+ self.exit.close()
try:
- os.remove(path)
+ os.kill(self.pid, 9)
except os.error:
pass
+ try:
+ os.waitpid(self.pid, 0)
+ except os.error, err:
+ ##print "waitpid failed", err
+ pass
+ StorageTestBase.tearDown(self)
+
def set_inet_addr(self):
self.host = socket.gethostname()
self.port = forker.get_port()
@@ -80,7 +74,7 @@
assert not os.path.exists(self.path + ".old")
def testAF_UNIXPack(self):
- self.addr = tempfile.mktemp(suffix=".zeo-socket")
+ self.addr = "zeo-socket"
self.start()
status = os.system("zeopack.py -U %s" % self.addr)
assert status == 0
Modified: ZODB/trunk/src/ZEO/tests/ConnectionTests.py
===================================================================
--- ZODB/trunk/src/ZEO/tests/ConnectionTests.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZEO/tests/ConnectionTests.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -17,7 +17,6 @@
import time
import random
import asyncore
-import tempfile
import threading
import logging
@@ -33,6 +32,7 @@
from ZODB.tests.MinPO import MinPO
from ZODB.tests.StorageTestBase \
import zodb_pickle, zodb_unpickle, handle_all_serials, handle_serials
+import ZODB.tests.util
import transaction
from transaction import Transaction
@@ -93,7 +93,7 @@
monitor = 0
db_class = DummyDB
- def setUp(self):
+ def setUp(self, before=None):
"""Test setup for connection tests.
This starts only one server; a test may start more servers by
@@ -102,8 +102,7 @@
"""
self.__super_setUp()
logging.info("setUp() %s", self.id())
- fd, self.file = tempfile.mkstemp()
- os.close(fd)
+ self.file = 'storage_conf'
self.addr = []
self._pids = []
self._servers = []
@@ -141,8 +140,7 @@
for c in self.caches:
for i in 0, 1:
for ext in "", ".trace", ".lock":
- base = "%s-%s.zec%s" % (c, "1", ext)
- path = os.path.join(tempfile.tempdir, base)
+ path = "%s-%s.zec%s" % (c, "1", ext)
# On Windows before 2.3, we don't have a way to wait for
# the spawned server(s) to close, and they inherited
# file descriptors for our open files. So long as those
@@ -183,7 +181,7 @@
self.caches.append(cache)
storage = TestClientStorage(self.addr,
client=cache,
- var=tempfile.tempdir,
+ var='.',
cache_size=cache_size,
wait=wait,
min_disconnect_poll=0.1,
Modified: ZODB/trunk/src/ZEO/tests/testAuth.py
===================================================================
--- ZODB/trunk/src/ZEO/tests/testAuth.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZEO/tests/testAuth.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -30,7 +30,9 @@
realm = None
def setUp(self):
- self.pwfile = tempfile.mktemp()
+ fd, self.pwfile = tempfile.mkstemp('pwfile')
+ os.close(fd)
+
if self.realm:
self.pwdb = self.dbclass(self.pwfile, self.realm)
else:
@@ -38,6 +40,7 @@
self.pwdb.add_user("foo", "bar")
self.pwdb.save()
self._checkZEOpasswd()
+
self.__super_setUp()
def _checkZEOpasswd(self):
@@ -51,8 +54,8 @@
zeopasswd.main(args + ["foo", "bar"])
def tearDown(self):
+ os.remove(self.pwfile)
self.__super_tearDown()
- os.remove(self.pwfile)
def getConfig(self, path, create, read_only):
return "<mappingstorage 1/>"
Modified: ZODB/trunk/src/ZEO/tests/testZEO.py
===================================================================
--- ZODB/trunk/src/ZEO/tests/testZEO.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZEO/tests/testZEO.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -222,6 +222,7 @@
blob_cache_dir = None
def setUp(self):
+ StorageTestBase.StorageTestBase.setUp(self)
logger.info("setUp() %s", self.id())
port = get_port()
zconf = forker.ZEOConfig(('', port))
@@ -232,7 +233,9 @@
self._conf_path = path
if not self.blob_cache_dir:
# This is the blob cache for ClientStorage
- self.blob_cache_dir = tempfile.mkdtemp()
+ self.blob_cache_dir = tempfile.mkdtemp(
+ 'blob_cache',
+ dir=os.path.abspath(os.getcwd()))
self._storage = ClientStorage(
zport, '1', cache_size=20000000,
min_disconnect_poll=0.5, wait=1,
@@ -242,14 +245,13 @@
def tearDown(self):
self._storage.close()
- os.remove(self._conf_path)
- ZODB.blob.remove_committed_dir(self.blob_cache_dir)
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)
+ StorageTestBase.StorageTestBase.tearDown(self)
def runTest(self):
try:
@@ -300,30 +302,23 @@
level = 2
- def setUp(self):
- self._storage = ZODB.FileStorage.FileStorage("Source.fs", create=True)
- self._dst = ZODB.FileStorage.FileStorage("Dest.fs", create=True)
-
def getConfig(self):
- filename = self.__fs_base = tempfile.mktemp()
return """\
<filestorage 1>
path %s
</filestorage>
- """ % filename
+ """ % tempfile.mktemp(dir='.')
def _new_storage(self):
port = get_port()
zconf = forker.ZEOConfig(('', port))
zport, adminaddr, pid, path = forker.start_zeo_server(self.getConfig(),
zconf, port)
- blob_cache_dir = tempfile.mkdtemp()
-
self._pids.append(pid)
self._servers.append(adminaddr)
- self._conf_paths.append(path)
- self.blob_cache_dirs.append(blob_cache_dir)
+ blob_cache_dir = tempfile.mkdtemp(dir='.')
+
storage = ClientStorage(
zport, '1', cache_size=20000000,
min_disconnect_poll=0.5, wait=1,
@@ -332,10 +327,9 @@
return storage
def setUp(self):
+ StorageTestBase.StorageTestBase.setUp(self)
self._pids = []
self._servers = []
- self._conf_paths = []
- self.blob_cache_dirs = []
self._storage = self._new_storage()
self._dst = self._new_storage()
@@ -344,16 +338,13 @@
self._storage.close()
self._dst.close()
- for p in self._conf_paths:
- os.remove(p)
- for p in self.blob_cache_dirs:
- ZODB.blob.remove_committed_dir(p)
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)
+ StorageTestBase.StorageTestBase.tearDown(self)
def new_dest(self):
return self._new_storage()
@@ -364,12 +355,11 @@
level = 2
def getConfig(self):
- filename = self.__fs_base = tempfile.mktemp()
return """\
<filestorage 1>
- path %s
+ path Data.fs
</filestorage>
- """ % filename
+ """
def checkInterfaceFromRemoteStorage(self):
# ClientStorage itself doesn't implement IStorageIteration, but the
@@ -416,10 +406,10 @@
return """
<demostorage 1>
<filestorage 1>
- path %s
+ path Data.fs
</filestorage>
</demostorage>
- """ % tempfile.mktemp()
+ """
def checkUndoZombie(self):
# The test base class IteratorStorage assumes that we keep undo data
@@ -615,23 +605,19 @@
class CommonBlobTests:
- def tearDown(self):
- super(BlobAdaptedFileStorageTests, self).tearDown()
- if os.path.exists(self.blobdir):
- # Might be gone already if the super() method deleted
- # the shared directory. Don't worry.
- shutil.rmtree(self.blobdir)
-
def getConfig(self):
return """
<blobstorage 1>
- blob-dir %s
+ blob-dir blobs
<filestorage 2>
- path %s
+ path Data.fs
</filestorage>
</blobstorage>
- """ % (self.blobdir, self.filestorage)
+ """
+ blobdir = 'blobs'
+ blob_cache_dir = 'blob_cache'
+
def checkStoreBlob(self):
from ZODB.utils import oid_repr, tid_repr
from ZODB.blob import Blob, BLOB_SUFFIX
@@ -713,24 +699,17 @@
def checkTransactionBufferCleanup(self):
oid = self._storage.new_oid()
- handle, blob_file_name = tempfile.mkstemp()
- os.close(handle)
- open(blob_file_name, 'w').write('I am a happy blob.')
+ open('blob_file', 'w').write('I am a happy blob.')
t = transaction.Transaction()
self._storage.tpc_begin(t)
self._storage.storeBlob(
- oid, ZODB.utils.z64, 'foo', blob_file_name, '', t)
+ oid, ZODB.utils.z64, 'foo', 'blob_file', '', t)
self._storage.close()
class BlobAdaptedFileStorageTests(FullGenericTests, CommonBlobTests):
"""ZEO backed by a BlobStorage-adapted FileStorage."""
- def setUp(self):
- self.blobdir = tempfile.mkdtemp() # blob directory on ZEO server
- self.filestorage = tempfile.mktemp()
- super(BlobAdaptedFileStorageTests, self).setUp()
-
def checkStoreAndLoadBlob(self):
from ZODB.utils import oid_repr, tid_repr
from ZODB.blob import Blob, BLOB_SUFFIX
@@ -814,13 +793,9 @@
class BlobWritableCacheTests(FullGenericTests, CommonBlobTests):
- def setUp(self):
- self.blobdir = self.blob_cache_dir = tempfile.mkdtemp()
- self.filestorage = tempfile.mktemp()
- self.shared_blob_dir = True
- super(BlobWritableCacheTests, self).setUp()
+ blob_cache_dir = 'blobs'
+ shared_blob_dir = True
-
class StorageServerClientWrapper:
def __init__(self):
Modified: ZODB/trunk/src/ZEO/tests/test_cache.py
===================================================================
--- ZODB/trunk/src/ZEO/tests/test_cache.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZEO/tests/test_cache.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -43,7 +43,8 @@
printable = ""
hex = ""
for character in line:
- if character in string.printable and not ord(character) in [12,13,9]:
+ if (character in string.printable
+ and not ord(character) in [12,13,9]):
printable += character
else:
printable += '.'
Modified: ZODB/trunk/src/ZODB/scripts/manual_tests/testfstest.py
===================================================================
--- ZODB/trunk/src/ZODB/scripts/manual_tests/testfstest.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/scripts/manual_tests/testfstest.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -9,11 +9,12 @@
import struct
import tempfile
import unittest
+import ZODB.tests.util
import fstest
from fstest import FormatError, U64
-class TestCorruptedFS(unittest.TestCase):
+class TestCorruptedFS(ZODB.tests.util.TestCase):
f = open('test-checker.fs', 'rb')
datafs = f.read()
@@ -21,17 +22,14 @@
del f
def setUp(self):
- self._temp = tempfile.mktemp()
+ ZODB.tests.util.TestCase.setUp(self)
+ self._temp = 'Data.fs'
self._file = open(self._temp, 'wb')
def tearDown(self):
if not self._file.closed:
self._file.close()
- if os.path.exists(self._temp):
- try:
- os.remove(self._temp)
- except os.error:
- pass
+ ZODB.tests.util.TestCase.tearDown(self)
def noError(self):
if not self._file.closed:
Modified: ZODB/trunk/src/ZODB/scripts/manual_tests/testzeopack.py
===================================================================
--- ZODB/trunk/src/ZODB/scripts/manual_tests/testzeopack.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/scripts/manual_tests/testzeopack.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -22,11 +22,12 @@
class PackerTests(StorageTestBase):
def setUp(self):
+ StorageTestBase.setUp(self)
self.started = 0
def start(self):
self.started =1
- self.path = tempfile.mktemp(suffix=".fs")
+ self.path = 'Data.fs'
self._storage = FileStorage(self.path)
self.db = ZODB.DB(self._storage)
self.do_updates()
@@ -37,26 +38,19 @@
self._dostore()
def tearDown(self):
- if not self.started:
- return
- self.db.close()
- self._storage.close()
- self.exit.close()
- try:
- os.kill(self.pid, 9)
- except os.error:
- pass
- try:
- os.waitpid(self.pid, 0)
- except os.error, err:
- ##print "waitpid failed", err
- pass
- for ext in '', '.old', '.lock', '.index', '.tmp':
- path = self.path + ext
+ if self.started:
+ self.db.close()
+ self.exit.close()
try:
- os.remove(path)
+ os.kill(self.pid, 9)
except os.error:
pass
+ try:
+ os.waitpid(self.pid, 0)
+ except os.error, err:
+ ##print "waitpid failed", err
+ pass
+ StorageTestBase.tearDown(self)
def set_inet_addr(self):
self.host = socket.gethostname()
Modified: ZODB/trunk/src/ZODB/tests/Corruption.py
===================================================================
--- ZODB/trunk/src/ZODB/tests/Corruption.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/Corruption.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -16,7 +16,6 @@
import os
import random
import stat
-import tempfile
import ZODB, ZODB.FileStorage
from StorageTestBase import StorageTestBase
@@ -24,13 +23,9 @@
class FileStorageCorruptTests(StorageTestBase):
def setUp(self):
- self.path = tempfile.mktemp()
- self._storage = ZODB.FileStorage.FileStorage(self.path, create=1)
+ StorageTestBase.setUp(self)
+ self._storage = ZODB.FileStorage.FileStorage('Data.fs', create=1)
- def tearDown(self):
- self._storage.close()
- self._storage.cleanup()
-
def _do_stores(self):
oids = []
for i in range(5):
@@ -49,16 +44,15 @@
self._close()
# truncation the index file
- path = self.path + '.index'
- self.failUnless(os.path.exists(path))
- f = open(path, 'r+')
+ self.failUnless(os.path.exists('Data.fs.index'))
+ f = open('Data.fs.index', 'r+')
f.seek(0, 2)
size = f.tell()
f.seek(size / 2)
f.truncate()
f.close()
- self._storage = ZODB.FileStorage.FileStorage(self.path)
+ self._storage = ZODB.FileStorage.FileStorage('Data.fs')
self._check_stores(oids)
def checkCorruptedIndex(self):
@@ -66,14 +60,13 @@
self._close()
# truncation the index file
- path = self.path + '.index'
- self.failUnless(os.path.exists(path))
- size = os.stat(path)[stat.ST_SIZE]
- f = open(path, 'r+')
+ self.failUnless(os.path.exists('Data.fs.index'))
+ size = os.stat('Data.fs.index')[stat.ST_SIZE]
+ f = open('Data.fs.index', 'r+')
while f.tell() < size:
f.seek(random.randrange(1, size / 10), 1)
f.write('\000')
f.close()
- self._storage = ZODB.FileStorage.FileStorage(self.path)
+ self._storage = ZODB.FileStorage.FileStorage('Data.fs')
self._check_stores(oids)
Modified: ZODB/trunk/src/ZODB/tests/RecoveryStorage.py
===================================================================
--- ZODB/trunk/src/ZODB/tests/RecoveryStorage.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/RecoveryStorage.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -24,6 +24,7 @@
class RecoveryStorage(IteratorDeepCompare):
+
# Requires a setUp() that creates a self._dst destination storage
def checkSimpleRecovery(self):
oid = self._storage.new_oid()
Modified: ZODB/trunk/src/ZODB/tests/StorageTestBase.py
===================================================================
--- ZODB/trunk/src/ZODB/tests/StorageTestBase.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/StorageTestBase.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -21,8 +21,6 @@
import sys
import time
-import types
-import unittest
from cPickle import Pickler, Unpickler
from cStringIO import StringIO
@@ -30,6 +28,7 @@
from ZODB.utils import u64
from ZODB.tests.MinPO import MinPO
+import ZODB.tests.util
ZERO = '\0'*8
@@ -78,7 +77,7 @@
u = Unpickler(f)
u.persistent_load = persistent_load
klass_info = u.load()
- if isinstance(klass_info, types.TupleType):
+ if isinstance(klass_info, tuple):
if isinstance(klass_info[0], type):
# Unclear: what is the second part of klass_info?
klass, xxx = klass_info
@@ -119,13 +118,13 @@
"""
d = {}
for arg in args:
- if isinstance(arg, types.StringType):
+ if isinstance(arg, str):
d[oid] = arg
elif arg is None:
pass
else:
for oid, serial in arg:
- if not isinstance(serial, types.StringType):
+ if not isinstance(serial, str):
raise serial # error from ZEO server
d[oid] = serial
return d
@@ -142,14 +141,12 @@
return sys.modules[name]
-class StorageTestBase(unittest.TestCase):
+class StorageTestBase(ZODB.tests.util.TestCase):
# It would be simpler if concrete tests didn't need to extend
# setUp() and tearDown().
- def setUp(self):
- # You need to override this with a setUp that creates self._storage
- self._storage = None
+ _storage = None
def _close(self):
# You should override this if closing your storage requires additional
@@ -159,6 +156,7 @@
def tearDown(self):
self._close()
+ ZODB.tests.util.TestCase.tearDown(self)
def _dostore(self, oid=None, revid=None, data=None,
already_pickled=0, user=None, description=None):
@@ -176,7 +174,7 @@
revid = ZERO
if data is None:
data = MinPO(7)
- if type(data) == types.IntType:
+ if type(data) == int:
data = MinPO(data)
if not already_pickled:
data = zodb_pickle(data)
Modified: ZODB/trunk/src/ZODB/tests/blob_connection.txt
===================================================================
--- ZODB/trunk/src/ZODB/tests/blob_connection.txt 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/blob_connection.txt 2008-10-28 21:33:22 UTC (rev 92661)
@@ -33,10 +33,9 @@
>>> import ZODB.FileStorage
>>> from ZODB.blob import BlobStorage
>>> from ZODB.DB import DB
- >>> from tempfile import mkdtemp
>>> base_storage = ZODB.FileStorage.FileStorage(
... 'BlobTests.fs', create=True)
- >>> blob_dir = mkdtemp()
+ >>> blob_dir = 'blobs'
>>> blob_storage = BlobStorage(blob_dir, base_storage)
>>> database = DB(blob_storage)
@@ -106,4 +105,3 @@
>>> database.close()
>>> database2.close()
>>> blob_storage.close()
- >>> base_storage.cleanup()
Modified: ZODB/trunk/src/ZODB/tests/blob_importexport.txt
===================================================================
--- ZODB/trunk/src/ZODB/tests/blob_importexport.txt 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/blob_importexport.txt 2008-10-28 21:33:22 UTC (rev 92661)
@@ -23,11 +23,10 @@
>>> from persistent.mapping import PersistentMapping
>>> import shutil
>>> import transaction
- >>> from tempfile import mkdtemp, mktemp
- >>> storagefile1 = mktemp()
- >>> blob_dir1 = mkdtemp()
- >>> storagefile2 = mktemp()
- >>> blob_dir2 = mkdtemp()
+ >>> storagefile1 = 'Data.fs.1'
+ >>> blob_dir1 = 'blobs1'
+ >>> storagefile2 = 'Data.fs.2'
+ >>> blob_dir2 = 'blobs2'
We need an database with an undoing blob supporting storage:
@@ -62,7 +61,7 @@
>>> conn = root1['blobdata']._p_jar
>>> oid = root1['blobdata']._p_oid
- >>> exportfile = mktemp()
+ >>> exportfile = 'export'
>>> nothing = connection1.exportFile(oid, exportfile)
Import our exported data into database2:
Modified: ZODB/trunk/src/ZODB/tests/blob_layout.txt
===================================================================
--- ZODB/trunk/src/ZODB/tests/blob_layout.txt 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/blob_layout.txt 2008-10-28 21:33:22 UTC (rev 92661)
@@ -84,40 +84,37 @@
1. Non-existing directories will trigger a bushy layout:
->>> import tempfile
->>> import shutil
->>> d = tempfile.mkdtemp()
->>> shutil.rmtree(d)
->>> auto_layout_select(d)
+>>> import os, shutil
+>>> auto_layout_select('blobs')
'bushy'
2. Empty directories will trigger a bushy layout too:
->>> d = tempfile.mkdtemp()
->>> auto_layout_select(d)
+>>> os.mkdir('blobs')
+>>> auto_layout_select('blobs')
'bushy'
3. If the directory contains a marker for the strategy it will be used:
>>> from ZODB.blob import LAYOUT_MARKER
>>> import os.path
->>> open(os.path.join(d, LAYOUT_MARKER), 'wb').write('bushy')
->>> auto_layout_select(d)
+>>> open(os.path.join('blobs', LAYOUT_MARKER), 'wb').write('bushy')
+>>> auto_layout_select('blobs')
'bushy'
->>> open(os.path.join(d, LAYOUT_MARKER), 'wb').write('lawn')
->>> auto_layout_select(d)
+>>> open(os.path.join('blobs', LAYOUT_MARKER), 'wb').write('lawn')
+>>> auto_layout_select('blobs')
'lawn'
->>> shutil.rmtree(d)
+>>> shutil.rmtree('blobs')
4. If the directory does not contain a marker but other files, we assume that
it was created with an earlier version of the blob implementation and uses our
`lawn` layout:
->>> d = tempfile.mkdtemp()
->>> open(os.path.join(d, '0x0101'), 'wb').write('foo')
->>> auto_layout_select(d)
+>>> os.mkdir('blobs')
+>>> open(os.path.join('blobs', '0x0101'), 'wb').write('foo')
+>>> auto_layout_select('blobs')
'lawn'
->>> shutil.rmtree(d)
+>>> shutil.rmtree('blobs')
Directory layout markers
@@ -127,8 +124,7 @@
it will leave a marker with the choosen layout if no marker exists yet:
>>> from ZODB.blob import FilesystemHelper
->>> d = tempfile.mkdtemp()
->>> blobs = os.path.join(d, 'blobs')
+>>> blobs = 'blobs'
>>> fsh = FilesystemHelper(blobs)
>>> fsh.layout_name
'bushy'
@@ -154,7 +150,7 @@
>>> import ZODB.FileStorage
>>> from ZODB.blob import BlobStorage
->>> datafs = os.path.join(d, 'data.fs')
+>>> datafs = 'data.fs'
>>> base_storage = ZODB.FileStorage.FileStorage(datafs)
>>> os.mkdir(blobs)
@@ -164,13 +160,14 @@
'lawn'
>>> open(os.path.join(blobs, LAYOUT_MARKER), 'rb').read()
'lawn'
->>> blob_storage = BlobStorage(blobs, base_storage, layout='bushy') # doctest: +ELLIPSIS
+>>> blob_storage = BlobStorage('blobs', base_storage, layout='bushy')
+... # doctest: +ELLIPSIS
Traceback (most recent call last):
ValueError: Directory layout `bushy` selected for blob directory .../blobs/, but marker found for layout `lawn`
>>> base_storage.close()
->>> rmtree(d)
+>>> rmtree('blobs')
Migrating between directory layouts
@@ -189,7 +186,8 @@
Create a `lawn` directory structure and migrate it to the new `bushy` one:
>>> from ZODB.blob import FilesystemHelper
->>> d = tempfile.mkdtemp()
+>>> d = 'd'
+>>> os.mkdir(d)
>>> old = os.path.join(d, 'old')
>>> old_fsh = FilesystemHelper(old, 'lawn')
>>> old_fsh.create()
Modified: ZODB/trunk/src/ZODB/tests/blob_packing.txt
===================================================================
--- ZODB/trunk/src/ZODB/tests/blob_packing.txt 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/blob_packing.txt 2008-10-28 21:33:22 UTC (rev 92661)
@@ -23,11 +23,9 @@
>>> from ZODB.blob import Blob, BlobStorage
>>> from ZODB import utils
>>> from ZODB.DB import DB
- >>> import shutil
>>> import transaction
- >>> from tempfile import mkdtemp, mktemp
- >>> storagefile = mktemp()
- >>> blob_dir = mkdtemp()
+ >>> storagefile = 'Data.fs'
+ >>> blob_dir = 'blobs'
A helper method to assure a unique timestamp across multiple platforms:
@@ -261,7 +259,8 @@
knowledge that the underlying storage's pack method is also called:
>>> def dummy_pack(time, ref):
- ... print "_blobs_pack_is_in_progress =", blob_storage._blobs_pack_is_in_progress
+ ... print "_blobs_pack_is_in_progress =",
+ ... print blob_storage._blobs_pack_is_in_progress
... return base_pack(time, ref)
>>> base_pack = base_storage.pack
>>> base_storage.pack = dummy_pack
@@ -270,7 +269,3 @@
>>> blob_storage._blobs_pack_is_in_progress
False
>>> base_storage.pack = base_pack
-
-Clean up our blob directory:
-
- >>> rmtree(blob_dir)
Modified: ZODB/trunk/src/ZODB/tests/blob_tempdir.txt
===================================================================
--- ZODB/trunk/src/ZODB/tests/blob_tempdir.txt 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/blob_tempdir.txt 2008-10-28 21:33:22 UTC (rev 92661)
@@ -30,10 +30,9 @@
>>> from ZODB.MappingStorage import MappingStorage
>>> from ZODB.blob import BlobStorage
>>> from ZODB.DB import DB
- >>> from tempfile import mkdtemp
>>> import os.path
>>> base_storage = MappingStorage('test')
- >>> blob_dir = mkdtemp()
+ >>> blob_dir = os.path.abspath('blobs')
>>> blob_storage = BlobStorage(blob_dir, base_storage)
>>> database = DB(blob_storage)
Modified: ZODB/trunk/src/ZODB/tests/blob_transaction.txt
===================================================================
--- ZODB/trunk/src/ZODB/tests/blob_transaction.txt 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/blob_transaction.txt 2008-10-28 21:33:22 UTC (rev 92661)
@@ -21,10 +21,8 @@
>>> from ZODB.blob import Blob, BlobStorage
>>> from ZODB.DB import DB
>>> import transaction
- >>> import tempfile
- >>> from tempfile import mkdtemp
>>> base_storage = MappingStorage("test")
- >>> blob_dir = mkdtemp()
+ >>> blob_dir = 'blobs'
>>> blob_storage = BlobStorage(blob_dir, base_storage)
>>> database = DB(blob_storage)
>>> connection1 = database.open()
@@ -351,7 +349,7 @@
... def tpc_abort(self):
... pass
>>> base_storage = DummyBaseStorage()
- >>> blob_dir2 = mkdtemp()
+ >>> blob_dir2 = 'blobs2'
>>> blob_storage2 = BlobStorage(blob_dir2, base_storage)
>>> committed_blob_dir = blob_storage2.fshelper.getPathForOID(0)
>>> os.makedirs(committed_blob_dir)
Modified: ZODB/trunk/src/ZODB/tests/testCache.py
===================================================================
--- ZODB/trunk/src/ZODB/tests/testCache.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/testCache.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -29,22 +29,23 @@
import ZODB
import ZODB.MappingStorage
from ZODB.tests.MinPO import MinPO
+import ZODB.tests.util
from ZODB.utils import p64
from persistent import Persistent
-class CacheTestBase(unittest.TestCase):
+class CacheTestBase(ZODB.tests.util.TestCase):
def setUp(self):
+ ZODB.tests.util.TestCase.setUp(self)
store = ZODB.MappingStorage.MappingStorage()
self.db = ZODB.DB(store,
cache_size = self.CACHE_SIZE)
self.conns = []
def tearDown(self):
- for conn in self.conns:
- conn.close()
self.db.close()
+ ZODB.tests.util.TestCase.tearDown(self)
CACHE_SIZE = 20
@@ -89,10 +90,8 @@
class DBMethods(CacheTestBase):
- __super_setUp = CacheTestBase.setUp
-
def setUp(self):
- self.__super_setUp()
+ CacheTestBase.setUp(self)
for i in range(4):
self.noodle_new_connection()
Modified: ZODB/trunk/src/ZODB/tests/testConfig.py
===================================================================
--- ZODB/trunk/src/ZODB/tests/testConfig.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/testConfig.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -12,22 +12,18 @@
#
##############################################################################
-import tempfile
-import unittest
-
import transaction
+import unittest
import ZODB.config
-from ZODB.POSException import ReadOnlyError
+import ZODB.POSException
+import ZODB.tests.util
-class ConfigTestBase(unittest.TestCase):
+class ConfigTestBase(ZODB.tests.util.TestCase):
+
def _opendb(self, s):
return ZODB.config.databaseFromString(s)
- def tearDown(self):
- if getattr(self, "storage", None) is not None:
- self.storage.cleanup()
-
def _test(self, s):
db = self._opendb(s)
self.storage = db._storage
@@ -58,28 +54,26 @@
""")
def test_file_config1(self):
- path = tempfile.mktemp()
self._test(
"""
<zodb>
<filestorage>
- path %s
+ path Data.fs
</filestorage>
</zodb>
- """ % path)
+ """)
def test_file_config2(self):
- path = tempfile.mktemp()
cfg = """
<zodb>
<filestorage>
- path %s
+ path Data.fs
create false
read-only true
</filestorage>
</zodb>
- """ % path
- self.assertRaises(ReadOnlyError, self._test, cfg)
+ """
+ self.assertRaises(ZODB.POSException.ReadOnlyError, self._test, cfg)
def test_demo_config(self):
cfg = """
Modified: ZODB/trunk/src/ZODB/tests/testConnection.py
===================================================================
--- ZODB/trunk/src/ZODB/tests/testConnection.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/testConnection.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -23,19 +23,18 @@
from ZODB.utils import p64, u64
from ZODB.tests.warnhook import WarningsHook
from zope.interface.verify import verifyObject
+import ZODB.tests.util
-class ConnectionDotAdd(unittest.TestCase):
+class ConnectionDotAdd(ZODB.tests.util.TestCase):
def setUp(self):
+ ZODB.tests.util.TestCase.setUp(self)
from ZODB.Connection import Connection
self.db = StubDatabase()
self.datamgr = Connection(self.db)
self.datamgr.open()
self.transaction = StubTransaction()
- def tearDown(self):
- transaction.abort()
-
def check_add(self):
from ZODB.POSException import InvalidObjectReference
obj = StubObject()
@@ -524,10 +523,11 @@
def setValueWithSize(self, size=0): self.value = size*' '
__init__ = setValueWithSize
-class EstimatedSizeTests(unittest.TestCase):
+class EstimatedSizeTests(ZODB.tests.util.TestCase):
"""check that size estimations are handled correctly."""
def setUp(self):
+ ZODB.tests.util.TestCase.setUp(self)
self.db = db = databaseFromString("<zodb>\n<mappingstorage/>\n</zodb>")
self.conn = c = db.open()
self.obj = obj = _PlayPersistent()
@@ -545,7 +545,8 @@
transaction.commit()
new_size = obj._p_estimated_size
self.assert_(new_size > size)
- self.assertEqual(cache.total_estimated_size, cache_size + new_size - size)
+ self.assertEqual(cache.total_estimated_size,
+ cache_size + new_size - size)
def test_size_set_on_write_savepoint(self):
obj, cache = self.obj, self.conn._cache
Modified: ZODB/trunk/src/ZODB/tests/testDB.py
===================================================================
--- ZODB/trunk/src/ZODB/tests/testDB.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/testDB.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -11,36 +11,30 @@
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
+
+from ZODB.tests.MinPO import MinPO
+from zope.testing import doctest
+import datetime
import os
import time
-import unittest
-import datetime
-
import transaction
-
-from zope.testing import doctest
-
+import unittest
import ZODB
-import ZODB.FileStorage
+import ZODB.tests.util
-from ZODB.tests.MinPO import MinPO
-
# Return total number of connections across all pools in a db._pools.
def nconn(pools):
return sum([len(pool.all) for pool in pools.values()])
-class DBTests(unittest.TestCase):
+class DBTests(ZODB.tests.util.TestCase):
def setUp(self):
- self.__path = os.path.abspath('test.fs')
- store = ZODB.FileStorage.FileStorage(self.__path)
- self.db = ZODB.DB(store)
+ ZODB.tests.util.TestCase.setUp(self)
+ self.db = ZODB.DB('test.fs')
def tearDown(self):
self.db.close()
- for s in ('', '.index', '.lock', '.tmp'):
- if os.path.exists(self.__path+s):
- os.remove(self.__path+s)
+ ZODB.tests.util.TestCase.tearDown(self)
def dowork(self):
c = self.db.open()
Modified: ZODB/trunk/src/ZODB/tests/testDemoStorage.py
===================================================================
--- ZODB/trunk/src/ZODB/tests/testDemoStorage.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/testDemoStorage.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -16,7 +16,7 @@
import transaction
from ZODB.DB import DB
from zope.testing import doctest
-import zope.testing.setupstack
+import ZODB.tests.util
import ZODB.utils
import ZODB.DemoStorage
from ZODB.tests import (
@@ -44,11 +44,9 @@
):
def setUp(self):
+ StorageTestBase.StorageTestBase.setUp(self)
self._storage = ZODB.DemoStorage.DemoStorage()
- def tearDown(self):
- self._storage.close()
-
def checkOversizeNote(self):
# This base class test checks for the common case where a storage
# doesnt support huge transaction metadata. This storage doesnt
@@ -85,13 +83,13 @@
class DemoStorageWrappedBase(DemoStorageTests):
def setUp(self):
- import ZODB.DemoStorage
+ StorageTestBase.StorageTestBase.setUp(self)
self._base = self._makeBaseStorage()
self._storage = ZODB.DemoStorage.DemoStorage(base=self._base)
def tearDown(self):
- self._storage.close()
self._base.close()
+ StorageTestBase.StorageTestBase.tearDown(self)
def _makeBaseStorage(self):
raise NotImplementedError
@@ -112,8 +110,7 @@
def setUp(test):
random.seed(0)
- zope.testing.setupstack.setUpDirectory(test)
- zope.testing.setupstack.register(test, transaction.abort)
+ ZODB.tests.util.setUp(test)
def testSomeDelegation():
r"""
@@ -158,11 +155,11 @@
return unittest.TestSuite((
doctest.DocFileSuite('synchronized.txt'),
doctest.DocTestSuite(
- setUp=setUp, tearDown=zope.testing.setupstack.tearDown,
+ setUp=setUp, tearDown=ZODB.tests.util.tearDown,
),
doctest.DocFileSuite(
'README.txt',
- setUp=setUp, tearDown=zope.testing.setupstack.tearDown,
+ setUp=setUp, tearDown=ZODB.tests.util.tearDown,
),
))
@@ -172,11 +169,11 @@
def test_suite():
suite = unittest.TestSuite((
doctest.DocTestSuite(
- setUp=setUp, tearDown=zope.testing.setupstack.tearDown,
+ setUp=setUp, tearDown=ZODB.tests.util.tearDown,
),
doctest.DocFileSuite(
'../DemoStorage.test',
- setUp=setUp, tearDown=zope.testing.setupstack.tearDown,
+ setUp=setUp, tearDown=ZODB.tests.util.tearDown,
),
))
suite.addTest(unittest.makeSuite(DemoStorageTests, 'check'))
Modified: ZODB/trunk/src/ZODB/tests/testFileStorage.py
===================================================================
--- ZODB/trunk/src/ZODB/tests/testFileStorage.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/testFileStorage.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -33,12 +33,9 @@
**kwargs)
def setUp(self):
+ StorageTestBase.StorageTestBase.setUp(self)
self.open(create=1)
- def tearDown(self):
- self._storage.close()
- self._storage.cleanup()
-
class FileStorageTests(
BaseFileStorageTests,
BasicStorage.BasicStorage,
@@ -296,14 +293,14 @@
):
def setUp(self):
+ StorageTestBase.StorageTestBase.setUp(self)
self._storage = ZODB.FileStorage.FileStorage("Source.fs", create=True)
self._dst = ZODB.FileStorage.FileStorage("Dest.fs", create=True)
def tearDown(self):
- self._storage.close()
self._dst.close()
- self._storage.cleanup()
- self._dst.cleanup()
+ StorageTestBase.StorageTestBase.tearDown(self)
+
def new_dest(self):
return ZODB.FileStorage.FileStorage('Dest.fs')
@@ -316,24 +313,16 @@
raise Exception
-class FileStorageNoRestoreRecoveryTest(
- StorageTestBase.StorageTestBase,
- RecoveryStorage.RecoveryStorage,
- ):
+class FileStorageNoRestoreRecoveryTest(FileStorageRecoveryTest):
# This test actually verifies a code path of
# BaseStorage.copyTransactionsFrom. For simplicity of implementation, we
# use a FileStorage deprived of its restore method.
def setUp(self):
+ StorageTestBase.StorageTestBase.setUp(self)
self._storage = FileStorageNoRestore("Source.fs", create=True)
self._dst = FileStorageNoRestore("Dest.fs", create=True)
- def tearDown(self):
- self._storage.close()
- self._dst.close()
- self._storage.cleanup()
- self._dst.cleanup()
-
def new_dest(self):
return FileStorageNoRestore('Dest.fs')
Modified: ZODB/trunk/src/ZODB/tests/testMappingStorage.py
===================================================================
--- ZODB/trunk/src/ZODB/tests/testMappingStorage.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/testMappingStorage.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -40,11 +40,9 @@
):
def setUp(self):
+ StorageTestBase.StorageTestBase.setUp(self, )
self._storage = ZODB.MappingStorage.MappingStorage()
- def tearDown(self):
- self._storage.close()
-
def checkOversizeNote(self):
# This base class test checks for the common case where a storage
# doesnt support huge transaction metadata. This storage doesnt
Modified: ZODB/trunk/src/ZODB/tests/testRecover.py
===================================================================
--- ZODB/trunk/src/ZODB/tests/testRecover.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/testRecover.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -17,38 +17,38 @@
import os
import random
import sys
-import tempfile
import unittest
import StringIO
import ZODB
+import ZODB.tests.util
from ZODB.FileStorage import FileStorage
import ZODB.fsrecover
from persistent.mapping import PersistentMapping
import transaction
-class RecoverTest(unittest.TestCase):
+class RecoverTest(ZODB.tests.util.TestCase):
level = 2
path = None
def setUp(self):
- self.path = tempfile.mktemp(suffix=".fs")
+ ZODB.tests.util.TestCase.setUp(self)
+ self.path = 'source.fs'
self.storage = FileStorage(self.path)
self.populate()
- self.dest = tempfile.mktemp(suffix=".fs")
+ self.dest = 'dest.fs'
self.recovered = None
def tearDown(self):
self.storage.close()
if self.recovered is not None:
self.recovered.close()
- self.storage.cleanup()
temp = FileStorage(self.dest)
temp.close()
- temp.cleanup()
+ ZODB.tests.util.TestCase.tearDown(self)
def populate(self):
db = ZODB.DB(self.storage)
Modified: ZODB/trunk/src/ZODB/tests/testZODB.py
===================================================================
--- ZODB/trunk/src/ZODB/tests/testZODB.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/testZODB.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -20,6 +20,7 @@
from ZODB.POSException import ReadConflictError, ConflictError
from ZODB.POSException import TransactionFailedError
from ZODB.tests.warnhook import WarningsHook
+import ZODB.tests.util
from persistent import Persistent
from persistent.mapping import PersistentMapping
@@ -38,13 +39,18 @@
def _p_independent(self):
return 0
-class ZODBTests(unittest.TestCase):
+class ZODBTests(ZODB.tests.util.TestCase):
def setUp(self):
+ ZODB.tests.util.TestCase.setUp(self)
self._storage = ZODB.FileStorage.FileStorage(
'ZODBTests.fs', create=1)
self._db = ZODB.DB(self._storage)
+ def tearDown(self):
+ self._db.close()
+ ZODB.tests.util.TestCase.tearDown(self)
+
def populate(self):
transaction.begin()
conn = self._db.open()
@@ -56,10 +62,6 @@
transaction.commit()
conn.close()
- def tearDown(self):
- self._db.close()
- self._storage.cleanup()
-
def checkExportImport(self, abort_it=False):
self.populate()
conn = self._db.open()
@@ -431,9 +433,10 @@
transaction.abort()
conn.close()
-class ReadConflictTests(unittest.TestCase):
+class ReadConflictTests(ZODB.tests.util.TestCase):
def setUp(self):
+ ZODB.tests.utils.TestCase.setUp(self)
self._storage = ZODB.MappingStorage.MappingStorage()
def readConflict(self, shouldFail=True):
Modified: ZODB/trunk/src/ZODB/tests/test_fsdump.py
===================================================================
--- ZODB/trunk/src/ZODB/tests/test_fsdump.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/test_fsdump.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -11,17 +11,13 @@
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
-
r"""
fsdump test
===========
-Let's get a temp file path to work with first.
+Let's get a path to work with first.
->>> import tempfile
->>> path = tempfile.mktemp('.fs', 'Data')
->>> print 'path:', path #doctest: +ELLIPSIS
-path: ...Data...fs
+>>> path = 'Data.fs'
More imports.
@@ -69,10 +65,12 @@
Clean up.
>>> st.close()
->>> st.cleanup() # remove .fs, .index, etc
"""
from zope.testing import doctest
+import zope.testing.setupstack
def test_suite():
- return doctest.DocTestSuite()
+ return doctest.DocTestSuite(
+ setUp=zope.testing.setupstack.setUpDirectory,
+ tearDown=zope.testing.setupstack.tearDown)
Modified: ZODB/trunk/src/ZODB/tests/test_storage.py
===================================================================
--- ZODB/trunk/src/ZODB/tests/test_storage.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/test_storage.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -154,6 +154,7 @@
):
def setUp(self):
+ StorageTestBase.StorageTestBase.setUp(self)
self._storage = MinimalMemoryStorage()
# we don't implement undo
Modified: ZODB/trunk/src/ZODB/tests/testblob.py
===================================================================
--- ZODB/trunk/src/ZODB/tests/testblob.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/testblob.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -32,7 +32,6 @@
import struct
import sys
import sys
-import tempfile
import time
import transaction
import unittest
@@ -42,7 +41,6 @@
import ZODB.tests.IteratorStorage
import ZODB.tests.util
import zope.testing.renormalizing
-import zope.testing.setupstack
def new_time():
@@ -60,48 +58,31 @@
return new_time
-class BlobConfigTestBase(ConfigTestBase):
+class ZODBBlobConfigTest(ConfigTestBase):
- def setUp(self):
- super(BlobConfigTestBase, self).setUp()
-
- self.blob_dir = tempfile.mkdtemp()
-
- def tearDown(self):
- super(BlobConfigTestBase, self).tearDown()
-
- shutil.rmtree(self.blob_dir)
-
-
-class ZODBBlobConfigTest(BlobConfigTestBase):
-
def test_map_config1(self):
self._test(
"""
<zodb>
<blobstorage>
- blob-dir %s
+ blob-dir blobs
<mappingstorage/>
</blobstorage>
</zodb>
- """ % self.blob_dir)
+ """)
def test_file_config1(self):
- path = tempfile.mktemp()
self._test(
"""
<zodb>
<blobstorage>
- blob-dir %s
+ blob-dir blobs
<filestorage>
- path %s
+ path Data.fs
</filestorage>
</blobstorage>
</zodb>
- """ %(self.blob_dir, path))
- os.unlink(path)
- os.unlink(path+".index")
- os.unlink(path+".tmp")
+ """)
def test_blob_dir_needed(self):
self.assertRaises(ZConfig.ConfigurationSyntaxError,
@@ -115,29 +96,16 @@
""")
-class BlobTests(unittest.TestCase):
+class BlobCloneTests(ZODB.tests.util.TestCase):
- def setUp(self):
- self.test_dir = tempfile.mkdtemp()
- self.here = os.getcwd()
- os.chdir(self.test_dir)
- self.storagefile = 'Data.fs'
- self.blob_dir = 'blobs'
-
- def tearDown(self):
- os.chdir(self.here)
- ZODB.blob.remove_committed_dir(self.test_dir)
-
-class BlobCloneTests(BlobTests):
-
def testDeepCopyCanInvalidate(self):
"""
Tests regression for invalidation problems related to missing
readers and writers values in cloned objects (see
http://mail.zope.org/pipermail/zodb-dev/2008-August/012054.html)
"""
- base_storage = FileStorage(self.storagefile)
- blob_storage = BlobStorage(self.blob_dir, base_storage)
+ base_storage = FileStorage('Data.fs')
+ blob_storage = BlobStorage('blobs', base_storage)
database = DB(blob_storage)
connection = database.open()
root = connection.root()
@@ -161,11 +129,11 @@
database.close()
-class BlobUndoTests(BlobTests):
+class BlobUndoTests(ZODB.tests.util.TestCase):
def testUndoWithoutPreviousVersion(self):
- base_storage = FileStorage(self.storagefile)
- blob_storage = BlobStorage(self.blob_dir, base_storage)
+ base_storage = FileStorage('Data.fs')
+ blob_storage = BlobStorage('blobs', base_storage)
database = DB(blob_storage)
connection = database.open()
root = connection.root()
@@ -181,8 +149,8 @@
database.close()
def testUndo(self):
- base_storage = FileStorage(self.storagefile)
- blob_storage = BlobStorage(self.blob_dir, base_storage)
+ base_storage = FileStorage('Data.fs')
+ blob_storage = BlobStorage('blobs', base_storage)
database = DB(blob_storage)
connection = database.open()
root = connection.root()
@@ -205,8 +173,8 @@
database.close()
def testUndoAfterConsumption(self):
- base_storage = FileStorage(self.storagefile)
- blob_storage = BlobStorage(self.blob_dir, base_storage)
+ base_storage = FileStorage('Data.fs')
+ blob_storage = BlobStorage('blobs', base_storage)
database = DB(blob_storage)
connection = database.open()
root = connection.root()
@@ -231,8 +199,8 @@
database.close()
def testRedo(self):
- base_storage = FileStorage(self.storagefile)
- blob_storage = BlobStorage(self.blob_dir, base_storage)
+ base_storage = FileStorage('Data.fs')
+ blob_storage = BlobStorage('bobs', base_storage)
database = DB(blob_storage)
connection = database.open()
root = connection.root()
@@ -263,8 +231,8 @@
database.close()
def testRedoOfCreation(self):
- base_storage = FileStorage(self.storagefile)
- blob_storage = BlobStorage(self.blob_dir, base_storage)
+ base_storage = FileStorage('Data.fs')
+ blob_storage = BlobStorage('blobs', base_storage)
database = DB(blob_storage)
connection = database.open()
root = connection.root()
@@ -288,12 +256,11 @@
database.close()
-class RecoveryBlobStorage(unittest.TestCase,
+class RecoveryBlobStorage(ZODB.tests.util.TestCase,
ZODB.tests.IteratorStorage.IteratorDeepCompare):
def setUp(self):
- self.globs = {}
- zope.testing.setupstack.setUpDirectory(self)
+ ZODB.tests.util.TestCase.setUp(self)
self._storage = BlobStorage(
'src_blobs', ZODB.FileStorage.FileStorage("Source.fs", create=True))
self._dst = BlobStorage(
@@ -302,7 +269,7 @@
def tearDown(self):
self._storage.close()
self._dst.close()
- zope.testing.setupstack.tearDown(self)
+ ZODB.tests.util.TestCase.tearDown(self)
# Requires a setUp() that creates a self._dst destination storage
def testSimpleBlobRecovery(self):
@@ -367,10 +334,8 @@
>>> from ZODB.MappingStorage import MappingStorage
>>> from ZODB.blob import BlobStorage
>>> from ZODB.DB import DB
- >>> from tempfile import mkdtemp
>>> base_storage = MappingStorage("test")
- >>> blob_dir = mkdtemp()
- >>> blob_storage = BlobStorage(blob_dir, base_storage)
+ >>> blob_storage = BlobStorage('blobs', base_storage)
>>> database = DB(blob_storage)
>>> connection = database.open()
>>> root = connection.root()
@@ -417,11 +382,9 @@
>>> from ZODB.blob import BlobStorage
>>> from ZODB.DB import DB
>>> from ZODB.serialize import referencesf
- >>> from tempfile import mkdtemp
>>> base_storage = MappingStorage("test")
- >>> blob_dir = mkdtemp()
- >>> blob_storage = BlobStorage(blob_dir, base_storage)
+ >>> blob_storage = BlobStorage('blobs', base_storage)
>>> database = DB(blob_storage)
>>> connection = database.open()
>>> root = connection.root()
@@ -435,8 +398,6 @@
Clean up:
>>> database.close()
- >>> import shutil
- >>> shutil.rmtree(blob_dir)
"""
@@ -453,12 +414,9 @@
>>> from ZODB.blob import BlobStorage
>>> from ZODB.DB import DB
>>> from ZODB.serialize import referencesf
- >>> from tempfile import mkdtemp, mktemp
- >>> storagefile = mktemp()
- >>> base_storage = FileStorage(storagefile)
- >>> blob_dir = mkdtemp()
- >>> blob_storage = BlobStorage(blob_dir, base_storage)
+ >>> base_storage = FileStorage('Data.fs')
+ >>> blob_storage = BlobStorage('blobs', base_storage)
>>> database = DB(blob_storage)
>>> connection = database.open()
>>> root = connection.root()
@@ -472,14 +430,6 @@
Clean up:
>>> database.close()
- >>> import shutil
- >>> shutil.rmtree(blob_dir)
-
- >>> os.unlink(storagefile)
- >>> os.unlink(storagefile+".index")
- >>> os.unlink(storagefile+".tmp")
-
-
"""
@@ -490,33 +440,29 @@
>>> from ZODB.FileStorage.FileStorage import FileStorage
>>> from ZODB.blob import BlobStorage
- >>> from tempfile import mkdtemp
>>> import os.path
- >>> working_directory = mkdtemp()
- >>> base_storage = FileStorage(os.path.join(working_directory, 'Data.fs'))
- >>> blob_storage = BlobStorage(os.path.join(working_directory, 'blobs'),
- ... base_storage)
+ >>> base_storage = FileStorage('Data.fs')
+ >>> blob_storage = BlobStorage('blobs', base_storage)
Two directories are created:
- >>> blob_dir = os.path.join(working_directory, 'blobs')
- >>> os.path.isdir(blob_dir)
+ >>> os.path.isdir('blobs')
True
- >>> tmp_dir = os.path.join(blob_dir, 'tmp')
+ >>> tmp_dir = os.path.join('blobs', 'tmp')
>>> os.path.isdir(tmp_dir)
True
They are only accessible by the owner:
- >>> oct(os.stat(blob_dir).st_mode)
+ >>> oct(os.stat('blobs').st_mode)
'040700'
>>> oct(os.stat(tmp_dir).st_mode)
'040700'
These settings are recognized as secure:
- >>> blob_storage.fshelper.isSecure(blob_dir)
+ >>> blob_storage.fshelper.isSecure('blobs')
True
>>> blob_storage.fshelper.isSecure(tmp_dir)
True
@@ -531,8 +477,6 @@
Clean up:
>>> blob_storage.close()
- >>> import shutil
- >>> shutil.rmtree(working_directory)
"""
@@ -554,12 +498,9 @@
>>> from ZODB.blob import BlobStorage
>>> from ZODB.DB import DB
>>> from ZODB.serialize import referencesf
- >>> from tempfile import mkdtemp, mktemp
- >>> storagefile = mktemp()
- >>> base_storage = FileStorage(storagefile)
- >>> blob_dir = mkdtemp()
- >>> blob_storage = BlobStorage(blob_dir, base_storage)
+ >>> base_storage = FileStorage('Data.fs')
+ >>> blob_storage = BlobStorage('blobs', base_storage)
>>> database = DB(blob_storage)
>>> connection = database.open()
>>> root = connection.root()
@@ -587,12 +528,6 @@
Clean up:
>>> database.close()
- >>> import shutil
- >>> rmtree(blob_dir)
-
- >>> os.unlink(storagefile)
- >>> os.unlink(storagefile+".index")
- >>> os.unlink(storagefile+".tmp")
"""
def is_blob_record():
@@ -639,7 +574,7 @@
"""
def setUp(test):
- zope.testing.setupstack.setUpDirectory(test)
+ ZODB.tests.util.setUp(test)
test.globs['rmtree'] = zope.testing.setupstack.rmtree
def test_suite():
Modified: ZODB/trunk/src/ZODB/tests/testconflictresolution.py
===================================================================
--- ZODB/trunk/src/ZODB/tests/testconflictresolution.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/testconflictresolution.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -15,18 +15,19 @@
$Id$
"""
import unittest
-from zope.testing import doctest, module, setupstack
+from zope.testing import doctest, module
+import ZODB.tests.util
def setUp(test):
+ ZODB.tests.util.setUp(test)
module.setUp(test, 'ConflictResolution_txt')
- setupstack.setUpDirectory(test)
def tearDown(test):
test.globs['db'].close()
test.globs['db1'].close()
test.globs['db2'].close()
module.tearDown(test)
- setupstack.tearDown(test)
+ ZODB.tests.util.tearDown(test)
def test_suite():
return unittest.TestSuite((
Modified: ZODB/trunk/src/ZODB/tests/testfsoids.py
===================================================================
--- ZODB/trunk/src/ZODB/tests/testfsoids.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/testfsoids.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -16,12 +16,9 @@
fsoids test, of the workhorse fsoids.Trace class
================================================
-Let's get a temp file path to work with first.
+Let's get a path to work with first.
->>> import tempfile
->>> path = tempfile.mktemp('.fs', 'Data')
->>> print 'path:', path #doctest: +ELLIPSIS
-path: ...Data...fs
+>>> path = 'Data.fs'
More imports.
Modified: ZODB/trunk/src/ZODB/tests/testhistoricalconnections.py
===================================================================
--- ZODB/trunk/src/ZODB/tests/testhistoricalconnections.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/testhistoricalconnections.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -16,19 +16,21 @@
"""
import unittest
from zope.testing import doctest, module
+import ZODB.tests.util
def setUp(test):
+ ZODB.tests.util.setUp(test)
module.setUp(test, 'historical_connections_txt')
def tearDown(test):
test.globs['db'].close()
test.globs['db2'].close()
test.globs['storage'].close()
- test.globs['storage'].cleanup()
# the DB class masks the module because of __init__ shenanigans
DB_module = __import__('ZODB.DB', globals(), locals(), ['chicken'])
DB_module.time = test.globs['original_time']
module.tearDown(test)
+ ZODB.tests.util.tearDown(test)
def test_suite():
return unittest.TestSuite((
Modified: ZODB/trunk/src/ZODB/tests/testpersistentclass.py
===================================================================
--- ZODB/trunk/src/ZODB/tests/testpersistentclass.py 2008-10-28 21:33:18 UTC (rev 92660)
+++ ZODB/trunk/src/ZODB/tests/testpersistentclass.py 2008-10-28 21:33:22 UTC (rev 92661)
@@ -52,14 +52,15 @@
def setUp(test):
+ ZODB.tests.util.setUp(test)
test.globs['some_database'] = ZODB.tests.util.DB()
module = FakeModule('ZODB.persistentclass_txt', test.globs)
sys.modules[module.__name__] = module
def tearDown(test):
- transaction.abort()
test.globs['some_database'].close()
del sys.modules['ZODB.persistentclass_txt']
+ ZODB.tests.util.tearDown(test)
def test_suite():
return unittest.TestSuite((
More information about the Zodb-checkins
mailing list