[Zodb-checkins] CVS: ZODB3/bsddb3Storage/bsddb3Storage/tests - test_whitebox.py:1.1.2.2

[Barry Warsaw]

Update of /cvs-repository/ZODB3/bsddb3Storage/bsddb3Storage/tests
In directory cvs.zope.org:/tmp/cvs-serv18011/bsddb3Storage/bsddb3Storage/tests

Modified Files:
      Tag: bdb-nolocks
	test_whitebox.py 
Log Message:
Added a whitebox test for reference counting, using the higher level
zodb api.


=== ZODB3/bsddb3Storage/bsddb3Storage/tests/test_whitebox.py 1.1.2.1 => 1.1.2.2 ===
--- ZODB3/bsddb3Storage/bsddb3Storage/tests/test_whitebox.py:1.1.2.1	Tue Sep 10 16:57:14 2002
+++ ZODB3/bsddb3Storage/bsddb3Storage/tests/test_whitebox.py	Wed Sep 11 14:11:48 2002
@@ -17,13 +17,17 @@
 import unittest
 
 from ZODB.utils import U64
+from ZODB.tests.MinPO import MinPO
 from ZODB.tests.StorageTestBase import zodb_unpickle
 from bsddb3Storage.Minimal import Minimal
 from bsddb3Storage.tests.BerkeleyTestBase import BerkeleyTestBase
+from bsddb3Storage.tests.ZODBTestBase import ZODBTestBase
+
+ZERO = '\0'*8
 
 
 
-class WhiteboxMinimal(BerkeleyTestBase):
+class WhiteboxLowLevel(BerkeleyTestBase):
     ConcreteStorage = Minimal
 
     def checkTableConsistencyAfterCommit(self):
@@ -51,26 +55,15 @@
             c.close()
         eq(len(oids), 1)
         eq(len(oids[oids.keys()[0]]), 1)
-        # Same deal for the pickles table.  There should be only one entry in
-        # this table, and the oid + serial should be consistent with what's in
-        # the serials table.
-        pickles = {}
-        c = self._storage._pickles.cursor()
-        try:
-            rec = c.first()
-            while rec:
-                key, pickle = rec
-                oid = key[:8]
-                serial = key[8:]
-                pickles.setdefault(oid, []).append((serial, pickle))
-                rec = c.next()
-        finally:
-            c.close()
+        # There should now be exactly one entry in the pickles table.
+        pickles = self._storage._pickles.items()
         eq(len(pickles), 1)
-        eq(len(pickles[pickles.keys()[0]]), 1)
-        eq(oids.keys()[0], pickles.keys()[0])
-        eq(oids[oids.keys()[0]][0], pickles[pickles.keys()[0]][0][0])
-        obj = zodb_unpickle(pickles[pickles.keys()[0]][0][1])
+        key, data = pickles[0]
+        poid = key[:8]
+        pserial = key[8:]
+        eq(oid, poid)
+        eq(revid3, pserial)
+        obj = zodb_unpickle(data)
         eq(obj.value, 13)
         # Now verify the refcounts table, which should be empty because the
         # stored object isn't referenced by any other objects.
@@ -78,9 +71,52 @@
 
 
 
+class WhiteboxHighLevel(ZODBTestBase):
+    ConcreteStorage = Minimal
+
+    def checkReferenceCounting(self):
+        eq = self.assertEqual
+        obj = MinPO(11)
+        self._root.obj = obj
+        get_transaction().commit()
+        obj.value = 12
+        get_transaction().commit()
+        obj.value = 13
+        get_transaction().commit()
+        # And now refcount out the object
+        del self._root.obj
+        get_transaction().commit()
+        # Verification stage.  Our serials table should have exactly one
+        # entry, oid == 0
+        keys = self._storage._serials.keys()
+        eq(len(keys), 1)
+        eq(len(self._storage._serials.items()), 1)
+        eq(keys[0], ZERO)
+        # The pickles table now should have exactly one revision of the root
+        # object, and no revisions of the MinPO object, which should have been
+        # collected away.
+        pickles = self._storage._pickles.items()
+        eq(len(pickles), 1)
+        rec = pickles[0]
+        key = rec[0]
+        data = rec[1]
+        eq(key[:8], ZERO)
+        # And that pickle should have no 'obj' attribute.
+        unobj = zodb_unpickle(data)
+        self.failIf(hasattr(unobj, 'obj'))
+        # Our refcounts table should have no entries in it, because the root
+        # object is an island.
+        eq(len(self._storage._refcounts.keys()), 0)
+        # And of course, oids and pendings should be empty too
+        eq(len(self._storage._oids.keys()), 0)
+        eq(len(self._storage._pending.keys()), 0)
+
+
+
 def test_suite():
     suite = unittest.TestSuite()
-    suite.addTest(unittest.makeSuite(WhiteboxMinimal, 'check'))
+    suite.addTest(unittest.makeSuite(WhiteboxLowLevel, 'check'))
+    suite.addTest(unittest.makeSuite(WhiteboxHighLevel, 'check'))
     return suite