Improve IndividualCache test cases

Author: rayluo

msal/individual_cache.py

@@ -59,6 +59,10 @@ def __init__(self, mapping=None, capacity=None, expires_in=None, lock=None,
         self._expires_in = expires_in
         self._lock = Lock() if lock is None else lock
 
+    def _peek(self):
+        # Returns (sequence, timestamps) without triggering maintenance
+        return self._mapping.get(self._INDEX, ([], {}))
+
     def _validate_key(self, key):
         if key == self._INDEX:
             raise ValueError("key {} is a reserved keyword in {}".format(
@@ -85,7 +89,7 @@ def _set(self, key, value, expires_in):
         # This internal implementation powers both set() and __setitem__(),
         # so that they don't depend on each other.
         self._validate_key(key)
-        sequence, timestamps = self._mapping.get(self._INDEX, ([], {}))
+        sequence, timestamps = self._peek()
         self._maintenance(sequence, timestamps)  # O(logN)
         now = int(time.time())
         expires_at = now + expires_in
@@ -136,7 +140,7 @@ def __getitem__(self, key):  # O(1)
         self._validate_key(key)
         with self._lock:
             # Skip self._maintenance(), because it would need O(logN) time
-            sequence, timestamps = self._mapping.get(self._INDEX, ([], {}))
+            sequence, timestamps = self._peek()
             expires_at, created_at = timestamps[key]  # Would raise KeyError accordingly
             now = int(time.time())
             if not created_at <= now < expires_at:
@@ -155,22 +159,22 @@ def __delitem__(self, key):  # O(1)
         with self._lock:
             # Skip self._maintenance(), because it would need O(logN) time
             self._mapping.pop(key, None)  # O(1)
-            sequence, timestamps = self._mapping.get(self._INDEX, ([], {}))
+            sequence, timestamps = self._peek()
             del timestamps[key]  # O(1)
             self._mapping[self._INDEX] = sequence, timestamps
 
     def __len__(self):  # O(logN)
         """Drop all expired items and return the remaining length"""
         with self._lock:
-            sequence, timestamps = self._mapping.get(self._INDEX, ([], {}))
+            sequence, timestamps = self._peek()
             self._maintenance(sequence, timestamps)  # O(logN)
             self._mapping[self._INDEX] = sequence, timestamps
             return len(timestamps)  # Faster than iter(self._mapping) when it is on disk
 
     def __iter__(self):
         """Drop all expired items and return an iterator of the remaining items"""
         with self._lock:
-            sequence, timestamps = self._mapping.get(self._INDEX, ([], {}))
+            sequence, timestamps = self._peek()
             self._maintenance(sequence, timestamps)  # O(logN)
             self._mapping[self._INDEX] = sequence, timestamps
         return iter(timestamps)  # Faster than iter(self._mapping) when it is on disk

tests/test_individual_cache.py

@@ -10,6 +10,10 @@ def setUp(self):
         self.mapping = {}
         self.m = ExpiringMapping(mapping=self.mapping, capacity=2, expires_in=1)
 
+    def how_many(self):
+        # This helper checks how many items are in the mapping, WITHOUT triggering purge
+        return len(self.m._peek()[1])
+
     def test_should_disallow_accessing_reserved_keyword(self):
         with self.assertRaises(ValueError):
             self.m.get(ExpiringMapping._INDEX)
@@ -40,11 +44,21 @@ def test_iter_should_purge(self):
         sleep(1)
         self.assertEqual([], list(self.m))
 
-    def test_get_should_purge(self):
+    def test_get_should_not_purge_and_should_return_only_when_the_item_is_still_valid(self):
         self.m["thing one"] = "one"
+        self.m["thing two"] = "two"
         sleep(1)
+        self.assertEqual(2, self.how_many(), "We begin with 2 items")
         with self.assertRaises(KeyError):
             self.m["thing one"]
+        self.assertEqual(1, self.how_many(), "get() should not purge the remaining items")
+
+    def test_setitem_should_purge(self):
+        self.m["thing one"] = "one"
+        sleep(1)
+        self.m["thing two"] = "two"
+        self.assertEqual(1, self.how_many(), "setitem() should purge all expired items")
+        self.assertEqual("two", self.m["thing two"], "The remaining item should be thing two")
 
     def test_various_expiring_time(self):
         self.assertEqual(0, len(self.m))