test: create tests for Redis cache and optimize caching logic

task_app/caching.py

@@ -2,48 +2,36 @@ from django.core.cache import cache
 from django.core import serializers
 from .models import Task
 
+CACHE_TIMEOUT = 3600
+
 def get_cached_tasks():
-    cache_key = "all_tasks"
+    return _get_or_set_cache("all_tasks", Task.objects.all())
-    cached_tasks_json = cache.get(cache_key)
-
-    if cached_tasks_json is not None:
-        deserialized_objects = list(serializers.deserialize('json', cached_tasks_json))
-        tasks = [obj.object for obj in deserialized_objects]
-        return tasks
-
-    tasks = Task.objects.all()
-    tasks_json = serializers.serialize('json', tasks)
-    cache.set(cache_key, tasks_json, timeout=3600)
-    return tasks
 
 def get_cached_tasks_by_status(status):
-    cache_key = f"tasks_by_status_{status}"
+    return _get_or_set_cache(f"tasks_by_status_{status}", Task.objects.filter(status=status))
-    cached_tasks_json = cache.get(cache_key)
-
-    if cached_tasks_json is not None:
-        deserialized_objects = list(serializers.deserialize('json', cached_tasks_json))
-        tasks = [obj.object for obj in deserialized_objects]
-        return tasks
-
-    tasks = Task.objects.filter(status=status)
-    tasks_json = serializers.serialize('json', tasks)
-    cache.set(cache_key, tasks_json, timeout=3600)
-    return tasks
 
 def get_cached_task_by_id(task_id):
-    cache_key = f"task_{task_id}"
+    return _get_or_set_cache(f"task_{task_id}", Task.objects.filter(pk=task_id), single=True)
-    cached_task_json = cache.get(cache_key)
 
-    if cached_task_json is not None:
+def _get_or_set_cache(cache_key, queryset, single=False):
+    cached_json = cache.get(cache_key)
+    
+    if cached_json:
         try:
-            return list(serializers.deserialize('json', cached_task_json))[0].object
+            deserialized = list(serializers.deserialize("json", cached_json))
-        except IndexError:
+            return deserialized[0].object if single and deserialized else [obj.object for obj in deserialized]
-            return None
+        except Exception:
+            cache.delete(cache_key)
 
-    try:
+    return _fetch_and_cache(cache_key, queryset, single)
-        task = Task.objects.get(pk=task_id)
+
-        task_json = serializers.serialize('json', [task])
+def _fetch_and_cache(cache_key, queryset, single=False):
-        cache.set(cache_key, task_json, timeout=3600)
+    objects = list(queryset)
-        return task
+    
-    except Task.DoesNotExist:
+    if not objects:
-        return None
+        return None if single else []
+    
+    serialized_json = serializers.serialize("json", objects)
+    cache.set(cache_key, serialized_json, timeout=CACHE_TIMEOUT)
+    
+    return objects[0] if single else objects

task_app/models.py

@@ -46,6 +46,8 @@ class AssignedTask(models.Model):
 @receiver(post_delete, sender=Task)
 def clear_task_cache(sender, instance, **kwargs):
     """Invalidate cache for the affected task status and individual task"""
+    cache.delete("all_tasks")
+
     cache_key = f"tasks_by_status_{instance.status}"
     cache.delete(cache_key)
 
@@ -61,3 +63,5 @@ def clear_assigned_task_cache(sender, instance, **kwargs):
 
     cache_key_task = f"task_{instance.task.id}"
     cache.delete(cache_key_task)
+
+    cache.delete("all_tasks")

task_app/tests.py

@@ -1,3 +1,67 @@
 from django.test import TestCase
+from django.core.cache import cache
+from django.contrib.auth import get_user_model
+from .models import Task
+from .caching import get_cached_tasks, get_cached_tasks_by_status, get_cached_task_by_id
+from .choices import TaskStatus
 
-# Create your tests here.
+User = get_user_model()
+
+class TaskCacheTests(TestCase):
+    def setUp(self):
+        user = User.objects.create(first_name="Test", last_name="User", email="test@example.com")
+        self.user = user
+        self.task1 = Task.objects.create(
+            title="Test Task 1",
+            description="Description 1",
+            category="Work",
+            priority="High",
+            status=TaskStatus.TODO,
+            date="2025-01-01",
+            creator=self.user
+        )
+        self.task2 = Task.objects.create(
+            title="Test Task 2",
+            description="Description 2",
+            category="Personal",
+            priority="Medium",
+            status=TaskStatus.IN_PROGRESS,
+            date="2025-01-02",
+            creator=self.user
+        )
+        cache.clear()
+
+    def test_get_cached_tasks(self):
+        tasks = get_cached_tasks()
+        self.assertEqual(len(tasks), 2)
+        self.assertIn(self.task1, tasks)
+        self.assertIn(self.task2, tasks)
+
+    def test_get_cached_tasks_by_status(self):
+        tasks_todo = get_cached_tasks_by_status(TaskStatus.TODO)
+        self.assertEqual(len(tasks_todo), 1)
+        self.assertEqual(tasks_todo[0], self.task1)
+        
+        tasks_in_progress = get_cached_tasks_by_status(TaskStatus.IN_PROGRESS)
+        self.assertEqual(len(tasks_in_progress), 1)
+        self.assertEqual(tasks_in_progress[0], self.task2)
+
+    def test_get_cached_task_by_id(self):
+        task = get_cached_task_by_id(self.task1.id)
+        self.assertIsNotNone(task)
+        self.assertEqual(task, self.task1)
+
+    def test_cache_invalidation_on_task_save(self):
+        get_cached_tasks()
+        self.task1.status = TaskStatus.DONE
+        self.task1.save()
+        self.assertIsNone(cache.get("all_tasks"))
+        self.assertIsNone(cache.get(f"tasks_by_status_{TaskStatus.TODO}"))
+        self.assertIsNone(cache.get(f"task_{self.task1.id}"))
+
+    def test_cache_invalidation_on_task_delete(self):
+        get_cached_tasks()
+        task_id = self.task1.id
+        self.task1.delete()
+        self.assertIsNone(cache.get("all_tasks"))
+        self.assertIsNone(cache.get(f"task_{task_id}"))