J621/backend/apps/library/tests/test_media_cache.py

"""Signed media URLs must be stable, versioned and cacheable.

Regression: signatures embedded the current second, so every API response
re-minted every URL and browsers re-downloaded each image on every poll; the
file responses also carried no cache headers at all.
"""

import base64
import hashlib
import io
import shutil
import tempfile
import time
from pathlib import Path
from unittest import mock

from django.contrib.auth import get_user_model
from django.core import signing
from django.core.files.uploadedfile import SimpleUploadedFile
from django.test import Client, TestCase, override_settings

from PIL import Image

from rest_framework.authtoken.models import Token

from apps.library import services
from apps.library.models import MediaItem, MediaLocation, TempUpload
from apps.library.signing_urls import load_payload, sign_payload

User = get_user_model()

TINY_PNG = base64.b64decode(
    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg=="
)


def png_bytes(width=1200, height=800, color=(20, 120, 200)):
    buffer = io.BytesIO()
    Image.new("RGB", (width, height), color).save(buffer, format="PNG")
    return buffer.getvalue()


class MediaCacheTests(TestCase):
    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        cls._tmp = tempfile.mkdtemp(prefix="j621-cache-")
        cls._media = Path(cls._tmp) / "media"
        cls._watched = cls._media / "library"
        cls._watched.mkdir(parents=True, exist_ok=True)
        cls._settings = override_settings(
            MEDIA_ROOT=str(cls._media), WATCHED_FOLDER=str(cls._watched)
        )
        cls._settings.enable()

    @classmethod
    def tearDownClass(cls):
        cls._settings.disable()
        shutil.rmtree(cls._tmp, ignore_errors=True)
        super().tearDownClass()

    def setUp(self):
        self.user = User.objects.create_user(
            username="cache-uploader", password="cache-pass-123456"
        )
        self.user.role = "uploader"
        self.user.save(update_fields=["role"])
        self.token = Token.objects.create(user=self.user).key
        payload = png_bytes()
        path = self._watched / "cache-image.png"
        path.write_bytes(payload)
        self.item = MediaItem.objects.create(
            md5=hashlib.md5(payload).hexdigest(), size=len(payload)
        )
        MediaLocation.objects.create(
            item=self.item, path=str(path), rel_path=path.name, mtime=time.time()
        )
        self.client = Client()

    def signed(self, action):
        return {
            "sig": sign_payload(
                {"item": self.item.id, "user": self.user.id, "action": action},
                services.MEDIA_FILE_SALT,
            )
        }

    def test_media_response_carries_cache_headers(self):
        response = self.client.get(
            f"/api/files/J-{self.item.id}/raw/", self.signed("raw")
        )
        self.assertEqual(response.status_code, 200)
        self.assertIn("private", response["Cache-Control"])
        self.assertIn(
            f"max-age={services.MEDIA_CACHE_SECONDS}", response["Cache-Control"]
        )
        self.assertIn("immutable", response["Cache-Control"])
        self.assertTrue(response["ETag"])
        self.assertTrue(response["Last-Modified"])

    def test_media_revalidation_returns_304(self):
        url = f"/api/files/J-{self.item.id}/raw/"
        first = self.client.get(url, self.signed("raw"))
        second = self.client.get(
            url, self.signed("raw"), HTTP_IF_NONE_MATCH=first["ETag"]
        )
        self.assertEqual(second.status_code, 304)
        self.assertEqual(second.content, b"")

    def test_image_thumbnail_is_generated_and_reused(self):
        url = f"/api/files/J-{self.item.id}/thumbnail/"
        response = self.client.get(url, self.signed("thumbnail"))
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response["Content-Type"], "image/jpeg")
        thumb = self._media / "thumbs" / f"{self.item.md5}.jpg"
        self.assertTrue(thumb.exists())
        with Image.open(thumb) as image:
            self.assertLessEqual(max(image.size), 480)
        before = thumb.stat().st_mtime_ns
        self.client.get(url, self.signed("thumbnail"))
        self.assertEqual(thumb.stat().st_mtime_ns, before)

    def test_staged_files_cache_briefly(self):
        temp = TempUpload.objects.create(
            user=self.user,
            file=SimpleUploadedFile("staged.png", TINY_PNG, content_type="image/png"),
            original_filename="staged.png",
            md5=hashlib.md5(b"staged").hexdigest(),
            size=len(TINY_PNG),
        )
        signature = sign_payload(
            {"temp": str(temp.id), "user": self.user.id}, services.UPLOAD_FILE_SALT
        )
        response = self.client.get(
            f"/api/uploads/{temp.id}/file/", {"sig": signature}
        )
        self.assertEqual(response.status_code, 200)
        self.assertIn(
            f"max-age={services.TEMP_CACHE_SECONDS}", response["Cache-Control"]
        )
        self.assertNotIn("immutable", response["Cache-Control"])

    def test_legacy_timestamp_signatures_are_accepted(self):
        """URLs minted before the stable scheme must not 500.

        A TimestampSigner HMAC also passes a plain Signer's check, so the
        embedded timestamp used to reach the JSON decoder and blow up.
        """
        payload = {"item": self.item.id, "user": self.user.id, "action": "raw"}
        legacy = signing.dumps(payload, salt=services.MEDIA_FILE_SALT)
        self.assertEqual(
            load_payload(legacy, services.MEDIA_FILE_SALT)["item"], self.item.id
        )
        response = self.client.get(
            f"/api/files/J-{self.item.id}/raw/", {"sig": legacy}
        )
        self.assertEqual(response.status_code, 200)

    def test_expired_and_malformed_signatures_return_none(self):
        payload = {"item": self.item.id, "user": self.user.id, "action": "raw"}
        with mock.patch.object(signing, "time") as clock:
            clock.time.return_value = time.time() - 3 * 86400
            expired = signing.dumps(payload, salt=services.MEDIA_FILE_SALT)
        self.assertIsNone(load_payload(expired, services.MEDIA_FILE_SALT))
        self.assertIsNone(load_payload("bogus", services.MEDIA_FILE_SALT))
        self.assertIsNone(load_payload("a:b", services.MEDIA_FILE_SALT))
        self.assertIsNone(load_payload("", services.MEDIA_FILE_SALT))