Add cache to /endpoints/publisher/listing

cache/README.md

@@ -0,0 +1,23 @@
+# Local Redis for Development
+
+This setup provides a local Redis instance for development and testing purposes.
+
+## Getting Started
+
+1. Ensure Docker is installed.
+2. ```cd cache``` and run ```docker compose run redis-cli```
+
+3. Then you can interact with the redis server.
+   To check all saved keys use:
+   ```
+   KEYS *
+   ```
+4. To exit the interactive promt, run
+   ```
+   exit
+   ```
+5. To stop the container, run:
+   ```
+   docker compose down
+   ```
+For more Redis CLI commands, check the docs at https://redis.io/learn/howtos/quick-start/cheat-sheet

cache/cache_utility.py

@@ -0,0 +1,8 @@
+from canonicalwebteam.stores_web_redis.utility import RedisCache
+from webapp.config import APP_NAME
+
+redis_cache = RedisCache(
+    namespace=APP_NAME,
+    maxsize=1000,
+    ttl=300,
+)

cache/docker-compose.yaml

@@ -0,0 +1,23 @@
+services:
+  redis:
+    image: redis:7
+    ports:
+      - "6379:6379"
+    volumes:
+      - redis-data:/data
+    command: ["redis-server", "--appendonly", "yes"]
+    healthcheck:
+      test: ["CMD", "redis-cli", "-h", "localhost", "-p", "6379", "ping"]
+      interval: 5s
+      timeout: 2s
+      retries: 5
+
+  redis-cli:
+    image: redis:7
+    entrypoint: ["redis-cli", "-h", "redis", "-p", "6379"]
+    depends_on:
+      redis:
+        condition: service_healthy
+
+volumes:
+  redis-data:

tests/endpoints/endpoint_testing.py

@@ -3,6 +3,7 @@
 
 from webapp.app import create_app
 from webapp.authentication import get_publishergw_authorization_header
+from cache.cache_utility import redis_cache
 
 
 class TestEndpoints(TestCase):
@@ -24,6 +25,15 @@ def _log_in(self, client):
         return get_publishergw_authorization_header(test_macaroon)
 
     def setUp(self):
+        # Clear cache before each test
+        if redis_cache.redis_available:
+            try:
+                redis_cache.client.flushdb()
+            except Exception:
+                pass
+        else:
+            redis_cache.fallback.clear()
+
         self.app = create_app(testing=True)
         self.client = self.app.test_client()
         self._log_in(self.client)

tests/publisher/snaps/tests_listing.py

@@ -1,9 +1,19 @@
 import responses
 from tests.publisher.endpoint_testing import BaseTestCases
+from cache.cache_utility import redis_cache
 
 
 class ListingPageNotAuth(BaseTestCases.EndpointLoggedOut):
     def setUp(self):
+        # Clear cache before each test
+        if redis_cache.redis_available:
+            try:
+                redis_cache.client.flushdb()
+            except Exception:
+                pass
+        else:
+            redis_cache.fallback.clear()
+
         snap_name = "test-snap"
         endpoint_url = "/api/{}/listing".format(snap_name)
 
@@ -12,6 +22,15 @@ def setUp(self):
 
 class GetListingPage(BaseTestCases.EndpointLoggedInErrorHandling):
     def setUp(self):
+        # Clear cache before each test
+        if redis_cache.redis_available:
+            try:
+                redis_cache.client.flushdb()
+            except Exception:
+                pass
+        else:
+            redis_cache.fallback.clear()
+
         snap_name = "test-snap"
 
         api_url = "https://dashboard.snapcraft.io/dev/api/snaps/info/{}"
@@ -275,3 +294,113 @@ def test_failed_categories_api(self):
         self.check_call_by_api_url(responses.calls)
 
         assert response.status_code == 200
+
+    @responses.activate
+    def test_cache_hit_on_second_request(self):
+        """Test that second GET request uses cache instead of API"""
+        payload = {
+            "snap_id": "id",
+            "snap_name": self.snap_name,
+            "title": "Snap title",
+            "summary": "This is a summary",
+            "description": "This is a description",
+            "media": [],
+            "publisher": {
+                "display-name": "The publisher",
+                "username": "toto",
+            },
+            "private": True,
+            "channel_maps_list": [{"map": [{"info": "info"}]}],
+            "contact": "contact adress",
+            "website": "website_url",
+            "public_metrics_enabled": True,
+            "public_metrics_blacklist": False,
+            "license": "License",
+            "video_urls": [],
+            "categories": {"items": []},
+            "status": "published",
+            "update_metadata_on_release": True,
+            "links": {"website": ["https://example.com"]},
+        }
+
+        responses.add(responses.GET, self.api_url, json=payload, status=200)
+        responses.add(
+            responses.GET,
+            "https://api.snapcraft.io/v2/snaps/categories?type=shared",
+            json=[],
+            status=200,
+        )
+
+        # First request should hit API
+        response1 = self.client.get(self.endpoint_url)
+        assert response1.status_code == 200
+
+        snap_info_calls_before = [
+            call
+            for call in responses.calls
+            if self.api_url in call.request.url
+        ]
+        assert len(snap_info_calls_before) == 1
+
+        # Second request should use cache (no additional snap info API calls)
+        response2 = self.client.get(self.endpoint_url)
+        assert response2.status_code == 200
+        snap_info_calls_after = [
+            call
+            for call in responses.calls
+            if self.api_url in call.request.url
+        ]
+        # Should have same number of calls (cache hit)
+        assert len(snap_info_calls_after) == len(snap_info_calls_before)
+
+    @responses.activate
+    def test_cache_stores_data_correctly(self):
+        """Test that cached data matches API response"""
+        payload = {
+            "snap_id": "cached-id",
+            "snap_name": self.snap_name,
+            "title": "Cached Snap Title",
+            "summary": "This is a cached summary",
+            "description": "This is a cached description",
+            "media": [],
+            "publisher": {
+                "display-name": "Cached Publisher",
+                "username": "cached",
+            },
+            "private": False,
+            "channel_maps_list": [{"map": [{"info": "info"}]}],
+            "contact": "cached@example.com",
+            "website": "https://cached.example.com",
+            "public_metrics_enabled": True,
+            "public_metrics_blacklist": False,
+            "license": "MIT",
+            "video_urls": [],
+            "categories": {"items": []},
+            "status": "published",
+            "update_metadata_on_release": False,
+            "links": {"website": ["https://cached.example.com"]},
+        }
+
+        responses.add(responses.GET, self.api_url, json=payload, status=200)
+        responses.add(
+            responses.GET,
+            "https://api.snapcraft.io/v2/snaps/categories?type=shared",
+            json=[],
+            status=200,
+        )
+
+        # First request
+        response1 = self.client.get(self.endpoint_url)
+        assert response1.status_code == 200
+        data1 = response1.get_json()
+
+        # Second request (from cache)
+        response2 = self.client.get(self.endpoint_url)
+        assert response2.status_code == 200
+        data2 = response2.get_json()
+
+        # Verify data matches
+        assert data1["data"]["snap_id"] == data2["data"]["snap_id"]
+        assert data1["data"]["title"] == data2["data"]["title"]
+        assert data1["data"]["summary"] == data2["data"]["summary"]
+        assert data1["data"]["description"] == data2["data"]["description"]