From 4651eaecbcfd6f64848819854ac1ab4fe033ddbc Mon Sep 17 00:00:00 2001 From: Chneemann Date: Sat, 29 Mar 2025 07:55:27 +0100 Subject: [PATCH] test: create tests for Redis cache and optimize caching logic --- task_app/caching.py | 62 +++++++++++++++++------------------------- task_app/models.py | 8 ++++-- task_app/tests.py | 66 ++++++++++++++++++++++++++++++++++++++++++++- 3 files changed, 96 insertions(+), 40 deletions(-) diff --git a/task_app/caching.py b/task_app/caching.py index 7e461d6..dcb92dc 100644 --- a/task_app/caching.py +++ b/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" - 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 + return _get_or_set_cache("all_tasks", Task.objects.all()) def get_cached_tasks_by_status(status): - cache_key = f"tasks_by_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 + return _get_or_set_cache(f"tasks_by_status_{status}", Task.objects.filter(status=status)) def get_cached_task_by_id(task_id): - cache_key = f"task_{task_id}" - cached_task_json = cache.get(cache_key) + return _get_or_set_cache(f"task_{task_id}", Task.objects.filter(pk=task_id), single=True) - 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 - except IndexError: - return None + deserialized = list(serializers.deserialize("json", cached_json)) + return deserialized[0].object if single and deserialized else [obj.object for obj in deserialized] + except Exception: + cache.delete(cache_key) - try: - task = Task.objects.get(pk=task_id) - task_json = serializers.serialize('json', [task]) - cache.set(cache_key, task_json, timeout=3600) - return task - except Task.DoesNotExist: - return None \ No newline at end of file + return _fetch_and_cache(cache_key, queryset, single) + +def _fetch_and_cache(cache_key, queryset, single=False): + objects = list(queryset) + + if not objects: + 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 \ No newline at end of file diff --git a/task_app/models.py b/task_app/models.py index fc8fb50..c55c8f8 100644 --- a/task_app/models.py +++ b/task_app/models.py @@ -46,9 +46,11 @@ 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) - + cache_key_task = f"task_{instance.id}" cache.delete(cache_key_task) @@ -60,4 +62,6 @@ def clear_assigned_task_cache(sender, instance, **kwargs): cache.delete(cache_key) cache_key_task = f"task_{instance.task.id}" - cache.delete(cache_key_task) \ No newline at end of file + cache.delete(cache_key_task) + + cache.delete("all_tasks") \ No newline at end of file diff --git a/task_app/tests.py b/task_app/tests.py index 7ce503c..b0c8ffe 100644 --- a/task_app/tests.py +++ b/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}"))