persistent.picklecache: 100% branch coverage.

persistent/picklecache.py

@@ -253,7 +253,9 @@ def _invalidate(self, oid):
         if value is not None and value._p_state != GHOST:
             value._p_invalidate()
             node = self.ring.next
-            while node is not self.ring:
+            while True:
+                if node is self.ring:
+                    break  # pragma: no cover belt-and-suspenders
                 if node.object is value:
                     node.prev.next, node.next.prev = node.next, node.prev
                     break

persistent/tests/test_picklecache.py

@@ -89,6 +89,14 @@ def test___setitem___non_string_oid_raises_ValueError(self):
         else:
             self.fail("Didn't raise ValueError with non-string OID.")
 
+    def test___setitem___duplicate_oid_same_obj(self):
+        from persistent._compat import _b
+        KEY = _b('original')
+        cache = self._makeOne()
+        original = self._makePersist()
+        cache[KEY] = original
+        cache[KEY] = original
+
     def test___setitem___duplicate_oid_raises_KeyError(self):
         from persistent._compat import _b
         KEY = _b('original')
@@ -448,6 +456,54 @@ def test_full_sweep(self):
         for oid in oids:
             self.assertTrue(cache.get(oid) is None)
 
+    def test_full_sweep_w_sticky(self):
+        import gc
+        from persistent.interfaces import UPTODATE
+        from persistent.interfaces import STICKY
+        from persistent._compat import _b
+        cache = self._makeOne()
+        oids = []
+        for i in range(100):
+            oid = _b('oid_%04d' % i)
+            oids.append(oid)
+            state = UPTODATE if i > 0 else STICKY
+            cache[oid] = self._makePersist(oid=oid, state=state)
+        self.assertEqual(cache.cache_non_ghost_count, 100)
+
+        cache.full_sweep()
+        gc.collect() # banish the ghosts who are no longer in the ring
+
+        self.assertEqual(cache.cache_non_ghost_count, 1)
+        self.assertTrue(cache.ring.next is not cache.ring)
+
+        self.assertTrue(cache.get(oids[0]) is not None)
+        for oid in oids[1:]:
+            self.assertTrue(cache.get(oid) is None)
+
+    def test_full_sweep_w_changed(self):
+        import gc
+        from persistent.interfaces import UPTODATE
+        from persistent.interfaces import CHANGED
+        from persistent._compat import _b
+        cache = self._makeOne()
+        oids = []
+        for i in range(100):
+            oid = _b('oid_%04d' % i)
+            oids.append(oid)
+            state = UPTODATE if i > 0 else CHANGED
+            cache[oid] = self._makePersist(oid=oid, state=state)
+        self.assertEqual(cache.cache_non_ghost_count, 100)
+
+        cache.full_sweep()
+        gc.collect() # banish the ghosts who are no longer in the ring
+
+        self.assertEqual(cache.cache_non_ghost_count, 1)
+        self.assertTrue(cache.ring.next is not cache.ring)
+
+        self.assertTrue(cache.get(oids[0]) is not None)
+        for oid in oids[1:]:
+            self.assertTrue(cache.get(oid) is None)
+
     def test_minimize(self):
         import gc
         from persistent.interfaces import UPTODATE