test: add tests for redis cache for navigation

Author: Yuri Zmytrakov

stac_fastapi/tests/redis/__init__.py

@@ -0,0 +1,80 @@
+import uuid
+
+import pytest
+
+from ..conftest import create_collection, create_item
+
+
+@pytest.mark.asyncio
+async def test_search_pagination_uses_redis_cache(
+    app_client, txn_client, load_test_data
+):
+    """Test Redis caching and navigation for the /search endpoint."""
+
+    collection = load_test_data("test_collection.json")
+    collection_id = f"test-pagination-collection-{uuid.uuid4()}"
+    collection["id"] = collection_id
+    await create_collection(txn_client, collection)
+
+    for i in range(5):
+        item = load_test_data("test_item.json")
+        item["id"] = f"test-pagination-item-{uuid.uuid4()}"
+        item["collection"] = collection_id
+        await create_item(txn_client, item)
+
+    resp = await app_client.post(
+        "/search", json={"collections": [collection_id], "limit": 1}
+    )
+    resp_json = resp.json()
+
+    next_link = next(
+        (link for link in resp_json["links"] if link["rel"] == "next"), None
+    )
+    next_token = next_link["body"]["token"]
+
+    # Expect the previous link on the second page to be retrieved from Redis cache
+    resp2 = await app_client.post(
+        "/search",
+        json={"collections": [collection_id], "limit": 1, "token": next_token},
+    )
+    resp2_json = resp2.json()
+
+    prev_link = next(
+        (link for link in resp2_json["links"] if link["rel"] == "prev"), None
+    )
+    assert prev_link is not None
+
+
+@pytest.mark.asyncio
+async def test_collections_pagination_uses_redis_cache(
+    app_client, txn_client, load_test_data
+):
+    """Test Redis caching and navigation for the /collection endpoint."""
+
+    collection_data = load_test_data("test_collection.json")
+    for i in range(5):
+        collection = collection_data.copy()
+        collection["id"] = f"test-collection-pagination-{uuid.uuid4()}"
+        collection["title"] = f"Test Collection Pagination {i}"
+        await create_collection(txn_client, collection)
+
+    resp = await app_client.get("/collections", params={"limit": 1})
+    assert resp.status_code == 200
+    resp1_json = resp.json()
+
+    next_link = next(
+        (link for link in resp1_json["links"] if link["rel"] == "next"), None
+    )
+    next_token = next_link["href"].split("token=")[1]
+
+    # Expect the previous link on the second page to be retrieved from Redis cache
+    resp2 = await app_client.get(
+        "/collections", params={"limit": 1, "token": next_token}
+    )
+    assert resp2.status_code == 200
+    resp2_json = resp2.json()
+
+    prev_link = next(
+        (link for link in resp2_json["links"] if link["rel"] == "prev"), None
+    )
+    assert prev_link is not None

stac_fastapi/tests/redis/test_redis_pagination.py

@@ -0,0 +1,44 @@
+import pytest
+
+from stac_fastapi.core.redis_utils import connect_redis, get_prev_link, save_self_link
+
+
+@pytest.mark.asyncio
+async def test_redis_connection():
+    """Test Redis connection."""
+    redis = await connect_redis()
+    assert redis is not None
+
+    # Test set/get
+    await redis.set("string_key", "string_value")
+    string_value = await redis.get("string_key")
+    assert string_value == "string_value"
+
+    # Test key retrieval operation
+    exists = await redis.exists("string_key")
+    assert exists == 1
+
+    # Test key deletion
+    await redis.delete("string_key")
+    deleted_value = await redis.get("string_key")
+    assert deleted_value is None
+
+
+@pytest.mark.asyncio
+async def test_redis_utils_functions():
+    redis = await connect_redis()
+    assert redis is not None
+
+    token = "test_token_123"
+    self_link = "http://mywebsite.com/search?token=test_token_123"
+
+    await save_self_link(redis, token, self_link)
+    retrieved_link = await get_prev_link(redis, token)
+    assert retrieved_link == self_link
+
+    await save_self_link(redis, None, "should_not_save")
+    null_result = await get_prev_link(redis, None)
+    assert null_result is None
+
+    non_existent = await get_prev_link(redis, "non_existent_token")
+    assert non_existent is None