[Zope-Checkins] CVS: Zope/lib/python/ZODB/tests - BasicStorage.py:1.5.14.1 ConflictResolution.py:1.3.12.1 PackableStorage.py:1.4.24.1 testTransaction.py:1.1.12.1

[Shane Hathaway]

Update of /cvs-repository/Zope/lib/python/ZODB/tests
In directory cvs.zope.org:/tmp/cvs-serv29115/lib/python/ZODB/tests

Modified Files:
      Tag: NR-branch
	BasicStorage.py ConflictResolution.py PackableStorage.py 
	testTransaction.py 
Log Message:
Sync NR-branch with trunk.  Sorry about so many checkin messages...


=== Zope/lib/python/ZODB/tests/BasicStorage.py 1.5 => 1.5.14.1 ===
 
 from ZODB.tests.MinPO import MinPO
-from ZODB.tests.StorageTestBase import zodb_unpickle
+from ZODB.tests.StorageTestBase import zodb_unpickle, zodb_pickle
 
 ZERO = '\0'*8
 
@@ -100,7 +100,6 @@
             data = self._storage.loadSerial(oid, revid)
             assert zodb_unpickle(data) == value
     
-
     def checkConflicts(self):
         oid = self._storage.new_oid()
         revid1 = self._dostore(oid, data=MinPO(11))
@@ -108,3 +107,15 @@
         self.assertRaises(POSException.ConflictError,
                           self._dostore,
                           oid, revid=revid1, data=MinPO(13))
+
+    def checkWriteAfterAbort(self):
+        oid = self._storage.new_oid()
+        self._storage.tpc_begin(self._transaction)
+        revid = self._storage.store(oid, ZERO, zodb_pickle(MinPO(5)),
+                                    '', self._transaction)
+        # Now abort this transaction
+        self._storage.tpc_abort(self._transaction)
+        # Now start all over again
+        self._transaction = Transaction()
+        oid = self._storage.new_oid()
+        revid = self._dostore(oid=oid, data=MinPO(6))


=== Zope/lib/python/ZODB/tests/ConflictResolution.py 1.3 => 1.3.12.1 ===
 class PCounter3(PCounter):
     def _p_resolveConflict(self, oldState, savedState, newState):
-        raise AttributeError, "no attribute"
+        raise AttributeError, "no attribute (testing conflict resolution)"
 
 class PCounter4(PCounter):
     def _p_resolveConflict(self, oldState, savedState):


=== Zope/lib/python/ZODB/tests/PackableStorage.py 1.4 => 1.4.24.1 ===
         pobj = pickle.loads(data)
         assert pobj.getoid() == oid and pobj.value == 3
+
+    def checkPackOnlyOneObject(self):
+        loads = self._makeloader()
+        # Create a root object.  This can't be an instance of Object,
+        # otherwise the pickling machinery will serialize it as a persistent
+        # id and not as an object that contains references (persistent ids) to
+        # other objects.
+        root = Root()
+        # Create a persistent object, with some initial state
+        obj1 = self._newobj()
+        oid1 = obj1.getoid()
+        # Create another persistent object, with some initial state.  Make
+        # sure it's oid is greater than the first object's oid.
+        obj2 = self._newobj()
+        oid2 = obj2.getoid()
+        assert oid2 > oid1
+        # Link the root object to the persistent objects, in order to keep
+        # them alive.  Store the root object.
+        root.obj1 = obj1
+        root.obj2 = obj2
+        root.value = 0
+        revid0 = self._dostoreNP(ZERO, data=dumps(root))
+        # Make sure the root can be retrieved
+        data, revid = self._storage.load(ZERO, '')
+        assert revid == revid0 and loads(data).value == 0
+        # Commit three different revisions of the first object
+        obj1.value = 1
+        revid1 = self._dostoreNP(oid1, data=pickle.dumps(obj1))
+        obj1.value = 2
+        revid2 = self._dostoreNP(oid1, revid=revid1, data=pickle.dumps(obj1))
+        obj1.value = 3
+        revid3 = self._dostoreNP(oid1, revid=revid2, data=pickle.dumps(obj1))
+        # Now make sure all three revisions can be extracted
+        data = self._storage.loadSerial(oid1, revid1)
+        pobj = pickle.loads(data)
+        assert pobj.getoid() == oid1 and pobj.value == 1
+        data = self._storage.loadSerial(oid1, revid2)
+        pobj = pickle.loads(data)
+        assert pobj.getoid() == oid1 and pobj.value == 2
+        data = self._storage.loadSerial(oid1, revid3)
+        pobj = pickle.loads(data)
+        assert pobj.getoid() == oid1 and pobj.value == 3
+        # Now commit a revision of the second object
+        obj2.value = 11
+        revid4 = self._dostoreNP(oid2, data=pickle.dumps(obj2))
+        # And make sure the revision can be extracted
+        data = self._storage.loadSerial(oid2, revid4)
+        pobj = pickle.loads(data)
+        assert pobj.getoid() == oid2 and pobj.value == 11
+        # Now pack just revisions 1 and 2 of object1.  Object1's current
+        # revision should stay alive because it's pointed to by the root, as
+        # should Object2's current revision.
+        self._storage.pack(time.time(), referencesf)
+        # Make sure the revisions are gone, but that object zero, object2, and
+        # revision 3 of object1 are still there and correct.
+        data, revid = self._storage.load(ZERO, '')
+        assert revid == revid0 and loads(data).value == 0
+        self.assertRaises(KeyError,
+                          self._storage.loadSerial, oid1, revid1)
+        self.assertRaises(KeyError,
+                          self._storage.loadSerial, oid1, revid2)
+        data = self._storage.loadSerial(oid1, revid3)
+        pobj = pickle.loads(data)
+        assert pobj.getoid() == oid1 and pobj.value == 3
+        data, revid = self._storage.load(oid1, '')
+        assert revid == revid3
+        pobj = pickle.loads(data)
+        assert pobj.getoid() == oid1 and pobj.value == 3
+        data, revid = self._storage.load(oid2, '')
+        assert revid == revid4 and loads(data).value == 11
+        data = self._storage.loadSerial(oid2, revid4)
+        pobj = pickle.loads(data)
+        assert pobj.getoid() == oid2 and pobj.value == 11


=== Zope/lib/python/ZODB/tests/testTransaction.py 1.1 => 1.1.12.1 ===
     pass
 
+def test_suite():
+    return unittest.makeSuite(AllTests, 'check')
+
 def main():
-    tests = unittest.makeSuite(AllTests, 'check')
+    tests = test_suite()
     runner = unittest.TextTestRunner()
     runner.run(tests)