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

"""Tests for the random image endpoint (`/api/random/`, `/random`).

Used by the SPA's Random page and by shell greeting scripts (fish_greeting
with fastfetch), so the response contract matters:
  * JSON with an absolute, directly fetchable URL,
  * signed for authenticated callers (image viewers send no headers),
  * fastfetch mode restricted to png/jpg/gif,
  * rating filters and guest visibility applied server-side.
"""

import hashlib
import shutil
import tempfile
import time
from pathlib import Path

from django.contrib.auth import get_user_model
from django.test import Client, TestCase, override_settings

from rest_framework.authtoken.models import Token

from apps.library.models import MediaItem, MediaLocation

User = get_user_model()

IMAGE_EXTENSIONS = {"png", "jpg", "jpeg", "gif", "webp", "apng"}
FASTFETCH_EXTENSIONS = {"png", "jpg", "jpeg", "gif"}


class RandomItemTests(TestCase):
    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        cls._tmp = tempfile.mkdtemp(prefix="j621-random-")
        cls._watched = Path(cls._tmp) / "library"
        cls._watched.mkdir(parents=True, exist_ok=True)
        cls._settings = override_settings(
            MEDIA_ROOT=cls._tmp, 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="random-user", password="random-pass-123456"
        )
        token = Token.objects.create(user=self.user)
        self.authed = Client()
        self.authed.defaults["HTTP_AUTHORIZATION"] = f"Token {token.key}"
        self.guest = Client()

    def make_item(self, label, extension, rating, *, hidden=False):
        path = self._watched / f"{label}.{extension}"
        path.write_bytes(b"random-" + label.encode())
        item = MediaItem.objects.create(
            md5=hashlib.md5(label.encode()).hexdigest(),
            size=path.stat().st_size,
            rating=rating,
            uploaded_by=self.user,
        )
        MediaLocation.objects.create(
            item=item, path=str(path), rel_path=path.name, mtime=time.time()
        )
        if hidden:
            MediaItem.objects.filter(pk=item.pk).update(hidden_from_guests=True)
        return item

    def test_returns_image_with_signed_absolute_url(self):
        item = self.make_item("plain", "png", "s")
        response = self.authed.get("/api/random/")
        self.assertEqual(response.status_code, 200)
        data = response.json()
        self.assertEqual(data["j_id"], f"J-{item.id}")
        self.assertEqual(data["extension"], "png")
        self.assertEqual(data["kind"], "image")
        self.assertEqual(data["rating"], "s")
        self.assertTrue(data["url"].startswith("http"))
        self.assertIn("sig=", data["url"])
        self.assertIn("download=1", data["download_url"])
        self.assertFalse(data["fastfetch"])

    def test_guest_url_is_unsigned_and_still_serves(self):
        self.make_item("guest", "jpg", "s")
        data = self.guest.get("/api/random/").json()
        self.assertNotIn("sig=", data["url"])
        path = data["url"].replace("http://testserver", "")
        self.assertEqual(self.guest.get(path).status_code, 200)

    def test_default_mode_returns_images_only(self):
        self.make_item("movie", "mp4", "s")
        self.make_item("picture", "webp", "s")
        for _ in range(10):
            extension = self.authed.get("/api/random/").json()["extension"]
            self.assertIn(extension, IMAGE_EXTENSIONS)

    def test_fastfetch_mode_flag_and_user_agent_restrict_formats(self):
        self.make_item("movie", "mp4", "s")
        self.make_item("modern", "webp", "s")
        self.make_item("picture", "png", "s")
        self.make_item("animation", "gif", "s")

        attempts = [("flag", {"fastfetch": "1"}, {}), ("ua", {}, {"HTTP_USER_AGENT": "fastfetch/2.18.1"})]
        for label, params, headers in attempts:
            for _ in range(15):
                response = self.authed.get("/api/random/", params, **headers)
                self.assertEqual(response.status_code, 200, label)
                data = response.json()
                self.assertIn(data["extension"], FASTFETCH_EXTENSIONS, label)
                self.assertTrue(data["fastfetch"], label)

    def test_rating_filter(self):
        self.make_item("safe", "png", "s")
        explicit = self.make_item("explicit", "png", "e")
        for _ in range(10):
            data = self.authed.get("/api/random/", {"rating": "e"}).json()
            self.assertEqual(data["j_id"], f"J-{explicit.id}")
            self.assertEqual(data["rating"], "e")
        self.assertEqual(self.authed.get("/api/random/", {"rating": "q"}).status_code, 404)

    def test_guests_never_receive_hidden_items(self):
        self.make_item("hidden", "png", "s", hidden=True)
        self.assertEqual(self.guest.get("/api/random/").status_code, 404)
        self.assertEqual(self.authed.get("/api/random/").status_code, 200)

    def test_no_match_returns_404(self):
        self.make_item("movie", "mp4", "s")  # images only
        self.assertEqual(self.authed.get("/api/random/").status_code, 404)

    def test_short_top_level_alias(self):
        self.make_item("alias", "gif", "s")
        self.assertEqual(self.guest.get("/random/").status_code, 200)
        self.assertEqual(self.guest.get("/random").status_code, 200)