[Zodb-checkins] SVN: ZODB/branches/blob-merge-branch/src/ZODB/Blobs/tests/test_undo.py Add a test for redo of an object's creation.

[Chris McDonough]

Log message for revision 65614:
  Add a test for redo of an object's creation.
  

Changed:
  U   ZODB/branches/blob-merge-branch/src/ZODB/Blobs/tests/test_undo.py

-=-
Modified: ZODB/branches/blob-merge-branch/src/ZODB/Blobs/tests/test_undo.py
===================================================================
--- ZODB/branches/blob-merge-branch/src/ZODB/Blobs/tests/test_undo.py	2006-02-28 20:24:13 UTC (rev 65613)
+++ ZODB/branches/blob-merge-branch/src/ZODB/Blobs/tests/test_undo.py	2006-02-28 20:48:52 UTC (rev 65614)
@@ -23,6 +23,7 @@
 from ZODB.DB import DB
 import transaction
 from ZODB.Blobs.Blob import Blob
+from ZODB import utils
 
 class BlobUndoTests(unittest.TestCase):
 
@@ -44,17 +45,20 @@
         connection = database.open()
         root = connection.root()
         transaction.begin()
-        blob = Blob()
-        blob.open('w').write('this is state 1')
-        root['blob'] = blob
+        root['blob'] = Blob()
         transaction.commit()
 
         serial = base64.encodestring(blob_storage._tid)
 
+        # undo the creation of the previously added blob
         transaction.begin()
-        blob_storage.undo(serial, blob_storage._transaction)
+        database.undo(serial, blob_storage._transaction)
         transaction.commit()
 
+        connection.close()
+        connection = database.open()
+        root = connection.root()
+        # the blob footprint object should exist no longer
         self.assertRaises(KeyError, root.__getitem__, 'blob')
 
     def testUndo(self):
@@ -90,6 +94,79 @@
         self.assertEqual(blob.open('r').read(), 'this is state 1')
         transaction.abort()
 
+    def testRedo(self):
+        base_storage = FileStorage(self.storagefile)
+        blob_storage = BlobStorage(self.blob_dir, base_storage)
+        database = DB(blob_storage)
+        connection = database.open()
+        root = connection.root()
+        blob = Blob()
+
+        transaction.begin()
+        blob.open('w').write('this is state 1')
+        root['blob'] = blob
+        transaction.commit()
+
+        transaction.begin()
+        blob = root['blob']
+        blob.open('w').write('this is state 2')
+        transaction.commit()
+
+        serial = base64.encodestring(blob_storage._tid)
+
+        transaction.begin()
+        database.undo(serial)
+        transaction.commit()
+
+        transaction.begin()
+        blob = root['blob']
+        self.assertEqual(blob.open('r').read(), 'this is state 1')
+        transaction.abort()
+
+        serial = base64.encodestring(blob_storage._tid)
+
+        transaction.begin()
+        database.undo(serial)
+        transaction.commit()
+
+        transaction.begin()
+        blob = root['blob']
+        self.assertEqual(blob.open('r').read(), 'this is state 2')
+        transaction.abort()
+        
+    def testRedoOfCreation(self):
+        base_storage = FileStorage(self.storagefile)
+        blob_storage = BlobStorage(self.blob_dir, base_storage)
+        database = DB(blob_storage)
+        connection = database.open()
+        root = connection.root()
+        blob = Blob()
+
+        transaction.begin()
+        blob.open('w').write('this is state 1')
+        root['blob'] = blob
+        transaction.commit()
+
+        serial = base64.encodestring(blob_storage._tid)
+
+        transaction.begin()
+        database.undo(serial)
+        transaction.commit()
+
+        self.assertRaises(KeyError, root.__getitem__, 'blob')
+
+        serial = base64.encodestring(blob_storage._tid)
+
+        transaction.begin()
+        database.undo(serial)
+        transaction.commit()
+
+        transaction.begin()
+        blob = root['blob']
+        self.assertEqual(blob.open('r').read(), 'this is state 1')
+        transaction.abort()
+
+
 def test_suite():
     suite = unittest.TestSuite()
     suite.addTest(unittest.makeSuite(BlobUndoTests))