Add tests for decorators

Author: mojixcoder

qscache/cache/base.py

@@ -29,7 +29,7 @@ def _get_cache_key(self, cache_key: Optional[str] = None) -> str:
     def get_cache_key(self) -> str:
         return self._get_cache_key(self.cache_key)
 
-    def _get_detail_cache_key(self, unique_identifier) -> str:
+    def get_detail_cache_key(self, unique_identifier) -> str:
         detail_cache_key = f"{self.get_cache_key()}_{unique_identifier}"
         return detail_cache_key
 
@@ -165,7 +165,7 @@ def get(
                 - Cached object or None if raise_exception=False and object was not found.
         """
         try:
-            cache_key = self._get_detail_cache_key(unique_identifier)
+            cache_key = self.get_detail_cache_key(unique_identifier)
             obj = cache.get(cache_key)
             if obj is not None:
                 return obj

qscache/cache/decorators.py

@@ -25,7 +25,7 @@ def wrapper(*args, **kwargs):
 
 def clear_cache_detail(
     manager: Type[ManagerType],
-    field_key_identifier: str = "pk",
+    field: str = "pk",
     additional_fields: Optional[List[str]] = None,
 ):
     """
@@ -38,15 +38,13 @@ def clear_cache_detail(
     def clear(func):
         def wrapper(*args, **kwargs):
             obj = func(*args, **kwargs)
-            detail_key = manager._get_detail_cache_key(  # noqa
-                getattr(obj, field_key_identifier)
-            )
+            detail_key = manager.get_detail_cache_key(getattr(obj, field))  # noqa
             if additional_fields is not None:
                 deleted_keys = [detail_key]
                 deleted_keys.append(additional_fields)
                 cache.delete_many(deleted_keys)
             else:
-                cache.delete_many(detail_key)
+                cache.delete(detail_key)
 
         return wrapper
 

qscache/tests/test_cache.py

@@ -8,6 +8,8 @@
 
 from django_redis import get_redis_connection
 
+from qscache import clear_cache_keys, clear_cache_detail
+
 from example_app.models import Example
 from example_app.cache import (
     example_cache_manager,
@@ -65,7 +67,7 @@ def test_get(self) -> None:
             unique_identifier=example_object.pk,
             filter_kwargs={"pk": example_object.pk},
         )
-        detail_cache_key = example_cache_manager._get_detail_cache_key(
+        detail_cache_key = example_cache_manager.get_detail_cache_key(
             unique_identifier=example_object.pk
         )
 
@@ -113,7 +115,7 @@ def test_get_with_filter(self) -> None:
             raise_exception=False,  # if raise_exception is False then we don't raise Http404 instead we return None
         )
 
-        cache_key = example_cache_manager._get_detail_cache_key(
+        cache_key = example_cache_manager.get_detail_cache_key(
             unique_identifier=example_object.pk
         )
 
@@ -139,7 +141,7 @@ def test_get_raise_exception(self) -> None:
                 filter_kwargs=filter_kwargs,
             )
 
-        cache_key = example_cache_manager._get_detail_cache_key(
+        cache_key = example_cache_manager.get_detail_cache_key(
             unique_identifier=example_object.pk
         )
 
@@ -386,17 +388,55 @@ def test_dont_use_prefetch_related_for_list(self) -> None:
 
         # 1 new query is executed because we don't prefetch_related objects in list query when use_prefetch_related_for_list is False
         self.assertEqual(len(connection.queries), 1)
-        
+
         obj = example_cache_user_prefetch_related_manager.get(
-            unique_identifier=example_object.pk, 
+            unique_identifier=example_object.pk,
             filter_kwargs={"pk": example_object.pk},
         )
-        
+
         reset_queries()
-        
-        
+
         for user in obj.users.all():
             first_name = user.first_name
-            
+
         self.assertEqual(len(connection.queries), 0)
-        
+
+    def test_clear_cache_keys_decorator(self) -> None:
+        example_list = example_cache_manager.all()
+
+        self.assertIn(example_cache_manager.cache_key, cache.keys("*"))
+
+        @clear_cache_keys(keys=[example_cache_manager.cache_key])
+        def create_example_object() -> None:
+            example_object = Example(
+                title="MojixCoder", text="Mojix Coder", number=1010
+            )
+            example_object.save()
+
+        create_example_object()
+
+        # Cache key is removed from cache
+        # So next time when we call .all() cache will be updated
+        self.assertNotIn(example_cache_manager.cache_key, cache.keys("*"))
+
+    def test_clear_cache_detail(self) -> None:
+        example_object = Example(title="MojixCoder", text="Mojix Coder", number=1010)
+        example_object.save()
+
+        example_obj = example_cache_manager.get(
+            unique_identifier=example_object.pk, filter_kwargs={"pk": example_object.pk}
+        )
+
+        cache_key = example_cache_manager.get_detail_cache_key(example_object.pk)
+
+        self.assertIn(cache_key, cache.keys("*"))
+
+        @clear_cache_detail(manager=example_cache_manager)
+        def update_example_object() -> Example:
+            example_object.title = "I am updated"
+            example_object.save()
+            return example_object
+        
+        update_example_object()
+
+        self.assertNotIn(cache_key, cache.keys("*"))