SVN: ZODB/branches/tseaver-better_repozo_tests/src/ZODB/scripts/tests/test_repozo.py Add tests for 'do_full_backup' / 'do_incremental_backup'.

Log message for revision 112306:
  Add tests for 'do_full_backup' / 'do_incremental_backup'.

Changed:
  U   ZODB/branches/tseaver-better_repozo_tests/src/ZODB/scripts/tests/test_repozo.py

-=-
Modified: ZODB/branches/tseaver-better_repozo_tests/src/ZODB/scripts/tests/test_repozo.py
===================================================================
--- ZODB/branches/tseaver-better_repozo_tests/src/ZODB/scripts/tests/test_repozo.py	2010-05-14 17:43:39 UTC (rev 112305)
+++ ZODB/branches/tseaver-better_repozo_tests/src/ZODB/scripts/tests/test_repozo.py	2010-05-14 17:44:37 UTC (rev 112306)
@@ -13,12 +13,21 @@
 ##############################################################################
 import unittest
 import os
+try:
+    # the hashlib package is available from Python 2.5
+    from hashlib import md5
+except ImportError:
+    # the md5 package is deprecated in Python 2.6
+    from md5 import new as md5
+
 import ZODB.tests.util  # layer used at class scope
 
 _NOISY = os.environ.get('NOISY_REPOZO_TEST_OUTPUT')
 
 class OurDB:
 
+    _file_name = None
+
     def __init__(self, dir):
         from BTrees.OOBTree import OOBTree
         import transaction
@@ -27,12 +36,13 @@
         conn = self.db.open()
         conn.root()['tree'] = OOBTree()
         transaction.commit()
+        self.pos = self.db.storage._pos
         self.close()
 
     def getdb(self):
         from ZODB import DB
         from ZODB.FileStorage import FileStorage
-        storage_filename = os.path.join(self.dir, 'Data.fs')
+        self._file_name = storage_filename = os.path.join(self.dir, 'Data.fs')
         storage = FileStorage(storage_filename)
         self.db = DB(storage)
 
@@ -63,6 +73,7 @@
                 if keys:
                     del tree[keys[0]]
         transaction.commit()
+        self.pos = self.db.storage._pos
         self.close()
 
 
@@ -96,7 +107,7 @@
         from ZODB.scripts.repozo import main
         main(argv)
 
-    def testRepozo(self):
+    def test_via_monte_carlo(self):
         self.saved_snapshots = []  # list of (name, time) pairs for copies.
 
         for i in range(100):
@@ -168,33 +179,47 @@
             (correctpath, when, ' '.join(argv)))
         self.assertEquals(fguts, gguts, msg)
 
-class Test_delete_old_backups(unittest.TestCase):
 
+class TestBase:
+
     _repository_directory = None
+    _data_directory = None
 
     def tearDown(self):
         if self._repository_directory is not None:
             from shutil import rmtree
             rmtree(self._repository_directory)
+        if self._data_directory is not None:
+            from shutil import rmtree
+            rmtree(self._data_directory)
 
-    def _callFUT(self, options=None, filenames=()):
-        from ZODB.scripts.repozo import delete_old_backups
-        if options is None:
-            options = self._makeOptions(filenames)
-        delete_old_backups(options)
+    def _makeOptions(self, **kw):
+        import tempfile
+        self._repository_directory = tempfile.mkdtemp()
+        class Options(object):
+            repository = self._repository_directory
+            def __init__(self, **kw):
+                self.__dict__.update(kw)
+        return Options(**kw)
 
+
+class Test_delete_old_backups(TestBase, unittest.TestCase):
+
     def _makeOptions(self, filenames=()):
-        import tempfile
-        dir = self._repository_directory = tempfile.mkdtemp()
+        options = super(Test_delete_old_backups, self)._makeOptions()
         for filename in filenames:
-            fqn = os.path.join(dir, filename)
+            fqn = os.path.join(options.repository, filename)
             f = open(fqn, 'wb')
             f.write('testing delete_old_backups')
             f.close()
-        class Options(object):
-            repository = dir
-        return Options()
+        return options
 
+    def _callFUT(self, options=None, filenames=()):
+        from ZODB.scripts.repozo import delete_old_backups
+        if options is None:
+            options = self._makeOptions(filenames)
+        return delete_old_backups(options)
+
     def test_empty_dir_doesnt_raise(self):
         self._callFUT()
         self.assertEqual(len(os.listdir(self._repository_directory)), 0)
@@ -254,8 +279,153 @@
             fqn = os.path.join(self._repository_directory, name)
             self.failUnless(os.path.isfile(fqn))
 
+
+class Test_do_full_backup(TestBase, unittest.TestCase):
+
+    def _callFUT(self, options):
+        from ZODB.scripts.repozo import do_full_backup
+        return do_full_backup(options)
+
+    def _makeDB(self):
+        import tempfile
+        datadir = self._data_directory = tempfile.mkdtemp()
+        return OurDB(self._data_directory)
+
+    def test_dont_overwrite_existing_file(self):
+        from ZODB.scripts.repozo import WouldOverwriteFiles
+        from ZODB.scripts.repozo import gen_filename
+        db = self._makeDB()
+        options = self._makeOptions(full=True,
+                                    file=db._file_name,
+                                    gzip=False,
+                                    test_now = (2010, 5, 14, 10, 51, 22),
+                                   )
+        f = open(os.path.join(self._repository_directory,
+                              gen_filename(options)), 'w')
+        f.write('TESTING')
+        f.flush()
+        f.close()
+        self.assertRaises(WouldOverwriteFiles, self._callFUT, options)
+
+    def test_empty(self):
+        from ZODB.scripts.repozo import gen_filename
+        db = self._makeDB()
+        options = self._makeOptions(file=db._file_name,
+                                    gzip=False,
+                                    killold=False,
+                                    test_now = (2010, 5, 14, 10, 51, 22),
+                                   )
+        self._callFUT(options)
+        target = os.path.join(self._repository_directory,
+                              gen_filename(options))
+        original = open(db._file_name, 'rb').read()
+        self.assertEqual(open(target, 'rb').read(), original)
+        datfile = os.path.join(self._repository_directory,
+                               gen_filename(options, '.dat'))
+        self.assertEqual(open(datfile).read(),
+                         '%s 0 %d %s\n' %
+                            (target, len(original), md5(original).hexdigest()))
+
+
+class Test_do_incremental_backup(TestBase, unittest.TestCase):
+
+    def _callFUT(self, options, reposz, repofiles):
+        from ZODB.scripts.repozo import do_incremental_backup
+        return do_incremental_backup(options, reposz, repofiles)
+
+    def _makeDB(self):
+        import tempfile
+        datadir = self._data_directory = tempfile.mkdtemp()
+        return OurDB(self._data_directory)
+
+    def test_dont_overwrite_existing_file(self):
+        from ZODB.scripts.repozo import WouldOverwriteFiles
+        from ZODB.scripts.repozo import gen_filename
+        from ZODB.scripts.repozo import find_files
+        db = self._makeDB()
+        options = self._makeOptions(full=False,
+                                    file=db._file_name,
+                                    gzip=False,
+                                    test_now = (2010, 5, 14, 10, 51, 22),
+                                    date = None,
+                                   )
+        f = open(os.path.join(self._repository_directory,
+                              gen_filename(options)), 'w')
+        f.write('TESTING')
+        f.flush()
+        f.close()
+        repofiles = find_files(options)
+        self.assertRaises(WouldOverwriteFiles,
+                          self._callFUT, options, 0, repofiles)
+
+    def test_no_changes(self):
+        from ZODB.scripts.repozo import gen_filename
+        db = self._makeDB()
+        oldpos = db.pos
+        options = self._makeOptions(file=db._file_name,
+                                    gzip=False,
+                                    killold=False,
+                                    test_now = (2010, 5, 14, 10, 51, 22),
+                                    date = None,
+                                   )
+        fullfile = os.path.join(self._repository_directory,
+                                '2010-05-14-00-00-00.fs')
+        original = open(db._file_name, 'rb').read()
+        last = len(original)
+        f = open(fullfile, 'wb')
+        f.write(original)
+        f.flush()
+        f.close()
+        datfile = os.path.join(self._repository_directory,
+                                '2010-05-14-00-00-00.dat')
+        repofiles = [fullfile, datfile]
+        self._callFUT(options, oldpos, repofiles)
+        target = os.path.join(self._repository_directory,
+                              gen_filename(options))
+        self.assertEqual(open(target, 'rb').read(), '')
+        self.assertEqual(open(datfile).read(),
+                         '%s %d %d %s\n' %
+                            (target, oldpos, oldpos, md5('').hexdigest()))
+
+    def test_w_changes(self):
+        from ZODB.scripts.repozo import gen_filename
+        db = self._makeDB()
+        oldpos = db.pos
+        options = self._makeOptions(file=db._file_name,
+                                    gzip=False,
+                                    killold=False,
+                                    test_now = (2010, 5, 14, 10, 51, 22),
+                                    date = None,
+                                   )
+        fullfile = os.path.join(self._repository_directory,
+                                '2010-05-14-00-00-00.fs')
+        original = open(db._file_name, 'rb').read()
+        f = open(fullfile, 'wb')
+        f.write(original)
+        f.flush()
+        f.close()
+        datfile = os.path.join(self._repository_directory,
+                                '2010-05-14-00-00-00.dat')
+        repofiles = [fullfile, datfile]
+        db.mutate()
+        newpos = db.pos
+        self._callFUT(options, oldpos, repofiles)
+        target = os.path.join(self._repository_directory,
+                              gen_filename(options))
+        f = open(db._file_name, 'rb')
+        f.seek(oldpos)
+        increment = f.read()
+        self.assertEqual(open(target, 'rb').read(), increment)
+        self.assertEqual(open(datfile).read(),
+                         '%s %d %d %s\n' %
+                            (target, oldpos, newpos,
+                             md5(increment).hexdigest()))
+
+
 def test_suite():
     return unittest.TestSuite([
         unittest.makeSuite(RepozoTests),
         unittest.makeSuite(Test_delete_old_backups),
+        unittest.makeSuite(Test_do_full_backup),
+        unittest.makeSuite(Test_do_incremental_backup),
     ])