Add more tests for ExpiringDict

backend/expiring_dict.py

@@ -4,7 +4,6 @@
 
 from time import time
 from threading import RLock
-import sys
 from typing import Any, Union
 
 from collections import OrderedDict
@@ -22,21 +21,7 @@ def __init__(self, max_len, max_age_seconds, items=None):
         self.max_age = max_age_seconds
         self.lock = RLock()
 
-        if sys.version_info >= (3, 5):
-            self._safe_keys = lambda: list(self.keys())
-        else:
-            self._safe_keys = self.keys
-
-        if items is not None:
-            if self.__is_instance_of_expiring_dict(items):
-                self.__copy_expiring_dict(max_len, max_age_seconds, items)
-            elif self.__is_instance_of_dict(items):
-                self.__copy_dict(items)
-            elif self.__is_reduced_result(items):
-                self.__copy_reduced_result(items)
-
-            else:
-                raise ValueError('can not unpack items')
+        self._safe_keys = lambda: list(self.keys())
 
     def __contains__(self, key):
         """ Return True if the dict has a key, else return False. """
@@ -128,16 +113,6 @@ def items(self):
                 pass
         return r
 
-    def items_with_timestamp(self):
-        """ Return a copy of the dictionary's list of (key, value, timestamp) triples. """
-        r = []
-        for key in self._safe_keys():
-            try:
-                r.append((key, OrderedDict.__getitem__(self, key)))
-            except KeyError:
-                pass
-        return r
-
     def values(self):
         """ Return a copy of the dictionary's list of values.
         See the note for dict.items(). """
@@ -149,30 +124,6 @@ def values(self):
                 pass
         return r
 
-    def fromkeys(self):
-        """ Create a new dictionary with keys from seq and values set to value. """
-        raise NotImplementedError()
-
-    def iteritems(self):
-        """ Return an iterator over the dictionary's (key, value) pairs. """
-        raise NotImplementedError()
-
-    def itervalues(self):
-        """ Return an iterator over the dictionary's values. """
-        raise NotImplementedError()
-
-    def viewitems(self):
-        """ Return a new view of the dictionary's items ((key, value) pairs). """
-        raise NotImplementedError()
-
-    def viewkeys(self):
-        """ Return a new view of the dictionary's keys. """
-        raise NotImplementedError()
-
-    def viewvalues(self):
-        """ Return a new view of the dictionary's values. """
-        raise NotImplementedError()
-
     def __reduce__(self):
         reduced = self.__class__, (self.max_len, self.max_age, ('reduce_result', self.items_with_timestamp()))
         return reduced

tests/test_expiring_dict.py

@@ -2,7 +2,7 @@
 from backend.expiring_dict import ExpiringDict
 
 
-def test_expiring_dict():
+def test_expiring_dict_by_time():
     storage = ExpiringDict(max_len=100, max_age_seconds=1)
 
     # Empty initially
@@ -12,11 +12,29 @@ def test_expiring_dict():
     storage["a"] = "b"
     storage["b"] = "c"
     assert storage.items() == [('a', 'b'), ('b', 'c')]
+    assert storage.values() == ["b", "c"]
     assert "a" in storage
     assert "b" in storage
 
     # Wait so the cache is cleared automatically
     sleep(1.1)
-    assert storage.items() == []
     assert "a" not in storage
+    assert storage.items() == []
+    assert storage.values() == []
     assert "b" not in storage
+
+
+def test_expiring_dict_by_size():
+    storage = ExpiringDict(max_len=2, max_age_seconds=1)
+
+    storage["a"] = "k1"
+    storage["b"] = "k2"
+    assert storage.items() == [('a', 'k1'), ('b', 'k2')]
+
+    # Push new item -> kicks first item
+    storage["c"] = "k3"
+    assert storage.items() == [('b', 'k2'), ("c", "k3")]
+
+    # Reset existing item -> resets time
+    storage["b"] = "k4"
+    assert storage.items() == [('c', 'k3'), ("b", "k4")]