[Zodb-checkins] SVN: ZODB/branches/test_repozo/src/ZODB/scripts/ Use a tempdir for backup / restore.
Tres Seaver
tseaver at palladion.com
Wed Dec 2 13:57:21 EST 2009
Log message for revision 106187:
Use a tempdir for backup / restore.
Clean out module-scope imports.
Don't shell out to run repozo: use its main() (passing args).
Suppress noisy output, unless NOISY_REPOZO_TEST_OUTPUT is in the environment.
Changed:
U ZODB/branches/test_repozo/src/ZODB/scripts/repozo.py
U ZODB/branches/test_repozo/src/ZODB/scripts/tests/test_repozo.py
-=-
Modified: ZODB/branches/test_repozo/src/ZODB/scripts/repozo.py
===================================================================
--- ZODB/branches/test_repozo/src/ZODB/scripts/repozo.py 2009-12-02 17:34:25 UTC (rev 106186)
+++ ZODB/branches/test_repozo/src/ZODB/scripts/repozo.py 2009-12-02 18:57:20 UTC (rev 106187)
@@ -106,10 +106,10 @@
print >> sys.stderr, msg % args
-def parseargs():
+def parseargs(argv):
global VERBOSE
try:
- opts, args = getopt.getopt(sys.argv[1:], 'BRvhf:r:FD:o:Qz',
+ opts, args = getopt.getopt(argv, 'BRvhf:r:FD:o:Qz',
['backup', 'recover', 'verbose', 'help',
'file=', 'repository=', 'full', 'date=',
'output=', 'quick', 'gzip'])
@@ -509,8 +509,10 @@
log('Recovered %s bytes, md5: %s', reposz, reposum)
-def main():
- options = parseargs()
+def main(argv=None):
+ if argv is None:
+ argv = sys.argv[1:]
+ options = parseargs(argv)
if options.mode == BACKUP:
do_backup(options)
else:
Modified: ZODB/branches/test_repozo/src/ZODB/scripts/tests/test_repozo.py
===================================================================
--- ZODB/branches/test_repozo/src/ZODB/scripts/tests/test_repozo.py 2009-12-02 17:34:25 UTC (rev 106186)
+++ ZODB/branches/test_repozo/src/ZODB/scripts/tests/test_repozo.py 2009-12-02 18:57:20 UTC (rev 106187)
@@ -15,24 +15,14 @@
import unittest
import os
-import sys
-import random
-import time
-import shutil
-import ZODB
-from ZODB import FileStorage
-import transaction
-from BTrees.OOBTree import OOBTree
+_NOISY = os.environ.get('NOISY_REPOZO_TEST_OUTPUT')
-from ZODB.scripts import tests as tests_module
-
-REPOZO = os.path.join(os.path.dirname(sys.argv[0]), 'repozo')
-
-
class OurDB:
def __init__(self, dir):
+ from BTrees.OOBTree import OOBTree
+ import transaction
self.dir = dir
self.getdb()
conn = self.db.open()
@@ -41,9 +31,11 @@
self.close()
def getdb(self):
+ from ZODB import DB
+ from ZODB.FileStorage import FileStorage
storage_filename = os.path.join(self.dir, 'Data.fs')
- storage = FileStorage.FileStorage(storage_filename)
- self.db = ZODB.DB(storage)
+ storage = FileStorage(storage_filename)
+ self.db = DB(storage)
def gettree(self):
self.getdb()
@@ -61,6 +53,8 @@
def mutate(self):
# Make random mutations to the btree in the database.
+ import random
+ import transaction
tree = self.gettree()
for dummy in range(100):
if random.random() < 0.6:
@@ -83,14 +77,13 @@
def setUp(self):
# compute directory names
- self.basedir = os.path.dirname(tests_module.__file__)
+ import tempfile
+ self.basedir = tempfile.mkdtemp()
self.backupdir = os.path.join(self.basedir, 'backup')
self.datadir = os.path.join(self.basedir, 'data')
self.restoredir = os.path.join(self.basedir, 'restore')
self.copydir = os.path.join(self.basedir, 'copy')
self.currdir = os.getcwd()
- # ensure they have all been deleted
- self.cleanup()
# create empty directories
os.mkdir(self.backupdir)
os.mkdir(self.datadir)
@@ -100,9 +93,14 @@
self.db = OurDB(self.datadir)
def tearDown(self):
- self.cleanup()
+ import shutil
+ shutil.rmtree(self.basedir)
os.chdir(self.currdir)
+ def _callRepozoMain(self, argv):
+ from ZODB.scripts.repozo import main
+ main(argv)
+
def testRepozo(self):
self.saved_snapshots = [] # list of (name, time) pairs for copies.
@@ -111,49 +109,58 @@
# Verify snapshots can be reproduced exactly.
for copyname, copytime in self.saved_snapshots:
- print "Checking that", copyname, "at", copytime, "is reproducible."
+ if _NOISY:
+ print "Checking that", copyname,
+ print "at", copytime, "is reproducible."
self.assertRestored(copyname, copytime)
def mutate_pack_backup(self, i):
+ import random
+ from shutil import copyfile
+ from time import gmtime
+ from time import sleep
self.db.mutate()
# Pack about each tenth time.
if random.random() < 0.1:
- print "packing"
+ if _NOISY:
+ print "packing"
self.db.pack()
self.db.close()
# Make an incremental backup, half the time with gzip (-z).
+ argv = ['-BQr', self.backupdir, '-f', 'Data.fs']
+ if _NOISY:
+ argv.insert(0, '-v')
if random.random() < 0.5:
- cmd = REPOZO + ' -vBQr %s -f Data.fs'
- else:
- cmd = REPOZO + ' -zvBQr %s -f Data.fs'
- os.system(cmd % self.backupdir)
+ argv.insert(0, '-z')
+ self._callRepozoMain(argv)
# Save snapshots to assert that dated restores are possible
if i % 9 == 0:
srcname = os.path.join(self.datadir, 'Data.fs')
- copytime = '%04d-%02d-%02d-%02d-%02d-%02d' % (time.gmtime()[:6])
+ copytime = '%04d-%02d-%02d-%02d-%02d-%02d' % (gmtime()[:6])
copyname = os.path.join(self.copydir, "Data%d.fs" % i)
- shutil.copyfile(srcname, copyname)
+ copyfile(srcname, copyname)
self.saved_snapshots.append((copyname, copytime))
# Make sure the clock moves at least a second.
- time.sleep(1.01)
+ sleep(1.01)
# Verify current Data.fs can be reproduced exactly.
self.assertRestored()
def assertRestored(self, correctpath='Data.fs', when=None):
# Do recovery to time 'when', and check that it's identical to correctpath.
- if when is None:
- extra = ''
- else:
- extra = ' -D ' + when
# restore to Restored.fs
restoredfile = os.path.join(self.restoredir, 'Restored.fs')
- cmd = REPOZO + ' -vRr %s -o ' + restoredfile + extra
- os.system(cmd % self.backupdir)
+ argv = ['-Rr', self.backupdir, '-o', restoredfile]
+ if _NOISY:
+ argv.insert(0, '-v')
+ if when is not None:
+ argv.append('-D')
+ argv.append(when)
+ self._callRepozoMain(argv)
# check restored file content is equal to file that was backed up
f = file(correctpath, 'rb')
@@ -163,18 +170,10 @@
f.close()
g.close()
msg = ("guts don't match\ncorrectpath=%r when=%r\n cmd=%r" %
- (correctpath, when, cmd))
+ (correctpath, when, ' '.join(argv)))
self.assertEquals(fguts, gguts, msg)
- def cleanup(self):
- for dir in [self.datadir, self.backupdir, self.restoredir,
- self.copydir]:
- if os.path.isdir(dir):
- for fname in os.listdir(dir):
- os.remove(os.path.join(dir, fname))
- os.rmdir(dir)
-
def test_suite():
suite = unittest.TestSuite()
suite.addTest(unittest.makeSuite(BasicRepozoTests))
More information about the Zodb-checkins
mailing list