cmshoppe/tests/test_image_studio_images.py

import io
import os
import socket
import sys
import threading
import time
import unittest
from types import SimpleNamespace
from unittest import mock

sys.path.insert(0, os.path.dirname(__file__))

from _helpers import TempDirMixin

from app import db, image_studio, image_studio_images


class _Response:
    def __init__(
        self,
        content=b"",
        status_code=200,
        headers=None,
        url="https://cdn.example.com/image.png",
        chunk_size=None,
    ):
        self.content = content
        self.status_code = status_code
        self.headers = dict(headers or {})
        self.url = url
        self.chunk_size = chunk_size

    def iter_content(self, chunk_size=65536):
        if self.chunk_size:
            for index in range(0, len(self.content), self.chunk_size):
                yield self.content[index:index + self.chunk_size]
            return
        yield self.content


class ImageStudioImageTests(TempDirMixin, unittest.TestCase):
    def _png_bytes(self, size=(20, 16), color=(80, 120, 200, 255)):
        from PIL import Image

        output = io.BytesIO()
        Image.new("RGBA", size, color).save(output, format="PNG")
        return output.getvalue()

    def _public_dns(self):
        return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443))]

    def _project_and_asset(self, temp_dir, remote_url="https://cdn.example.com/source.png"):
        db_path = os.path.join(temp_dir, "cmshopee.db")
        db.init_db(db_path)
        project = image_studio.create_or_get_project(
            account_alias="alias",
            account_slug="alias_slug",
            item_id="51100639510",
            path=db_path,
        )
        asset = image_studio.sync_original_asset_urls(
            project.id,
            [{"index": 1, "src": remote_url}],
            path=db_path,
        )[0]
        return db_path, project, asset

    def test_download_remote_image_rejects_private_non_image_overlimit_and_private_redirect(self):
        with self.assertRaisesRegex(image_studio_images.ImageStudioImageError, "http/https|本机或内网"):
            image_studio_images.download_remote_image("http://127.0.0.1/a.png")

        non_image_session = SimpleNamespace(
            get=lambda *args, **kwargs: _Response(
                content=b"hello",
                headers={"Content-Type": "text/html"},
            )
        )
        with mock.patch("app.image_studio_images.socket.getaddrinfo", return_value=self._public_dns()):
            with self.assertRaisesRegex(image_studio_images.ImageStudioImageError, "不是图片"):
                image_studio_images.download_remote_image(
                    "https://cdn.example.com/a.png",
                    session=non_image_session,
                )

        oversized_session = SimpleNamespace(
            get=lambda *args, **kwargs: _Response(
                content=b"x",
                headers={"Content-Type": "image/png", "Content-Length": "999"},
            )
        )
        with mock.patch("app.image_studio_images.socket.getaddrinfo", return_value=self._public_dns()):
            with self.assertRaisesRegex(image_studio_images.ImageStudioImageError, "大小上限"):
                image_studio_images.download_remote_image(
                    "https://cdn.example.com/a.png",
                    session=oversized_session,
                    max_bytes=10,
                )

        redirect_session = SimpleNamespace(
            get=lambda *args, **kwargs: _Response(
                content=self._png_bytes(),
                headers={"Content-Type": "image/png"},
                url="http://127.0.0.1/private.png",
            )
        )
        with mock.patch("app.image_studio_images.socket.getaddrinfo", return_value=self._public_dns()):
            with self.assertRaisesRegex(image_studio_images.ImageStudioImageError, "本机或内网"):
                image_studio_images.download_remote_image(
                    "https://cdn.example.com/a.png",
                    session=redirect_session,
                )

    def test_load_thumbnail_decodes_image_without_writing_file(self):
        png = self._png_bytes(size=(640, 480))
        session = SimpleNamespace(
            get=lambda *args, **kwargs: _Response(
                content=png,
                headers={"Content-Type": "image/png"},
            )
        )

        with mock.patch("app.image_studio_images.socket.getaddrinfo", return_value=self._public_dns()):
            result = image_studio_images.load_thumbnail(
                "https://cdn.example.com/a.png",
                key="thumb-1",
                max_size=120,
                session=session,
            )

        self.assertEqual("thumb-1", result.key)
        self.assertLessEqual(result.width, 120)
        self.assertLessEqual(result.height, 120)
        self.assertTrue(result.image_bytes.startswith(b"\x89PNG"))

    def test_thumbnail_loader_uses_memory_cache_and_can_cancel(self):
        calls = []
        successes = []
        errors = []

        def fake_loader(url, key=None):
            calls.append((key, url))
            return image_studio_images.ThumbnailResult(
                key=str(key),
                url=url,
                image_bytes=b"png",
                width=10,
                height=10,
            )

        loader = image_studio_images.ThumbnailLoader(max_workers=1, loader=fake_loader)
        try:
            future = loader.submit("a", "https://cdn.example.com/a.png", successes.append, errors.append)
            future.result(timeout=2)
            for _ in range(20):
                if successes:
                    break
                time.sleep(0.01)
            loader.submit("b", "https://cdn.example.com/a.png", successes.append, errors.append)

            self.assertEqual([("a", "https://cdn.example.com/a.png")], calls)
            self.assertEqual(2, len(successes))
            self.assertFalse(successes[0].from_cache)
            self.assertTrue(successes[1].from_cache)
            self.assertEqual([], errors)

            release = threading.Event()

            def slow_loader(url, key=None):
                release.wait(1)
                return image_studio_images.ThumbnailResult(
                    key=str(key),
                    url=url,
                    image_bytes=b"png",
                    width=10,
                    height=10,
                )

            loader._loader = slow_loader
            future = loader.submit("c", "https://cdn.example.com/c.png", successes.append, errors.append)
            loader.cancel("c")
            release.set()
            future.result(timeout=2)
            time.sleep(0.05)
            self.assertEqual(2, len(successes))
        finally:
            loader.close()

    def test_download_original_asset_writes_atomically_and_is_idempotent(self):
        with self.make_temp_dir() as temp_dir:
            db_path, _, asset = self._project_and_asset(temp_dir)
            png = self._png_bytes()
            session = mock.Mock()
            session.get.return_value = _Response(
                content=png,
                headers={"Content-Type": "image/png"},
            )

            with mock.patch("app.image_studio_images.socket.getaddrinfo", return_value=self._public_dns()):
                downloaded = image_studio_images.download_original_asset(
                    asset.id,
                    path=db_path,
                    config={"data_dir": temp_dir, "image_dir": os.path.join(temp_dir, "images")},
                    session=session,
                )
                second = image_studio_images.download_original_asset(
                    asset.id,
                    path=db_path,
                    config={"data_dir": temp_dir, "image_dir": os.path.join(temp_dir, "images")},
                    session=mock.Mock(),
                )

            self.assertEqual(downloaded.id, second.id)
            self.assertEqual(downloaded.local_path, second.local_path)
            self.assertTrue(os.path.isfile(downloaded.local_path))
            self.assertTrue(downloaded.local_path.endswith(".png"))
            self.assertEqual(image_studio.ASSET_STATUS_AVAILABLE, downloaded.status)
            session.get.assert_called_once()
            originals = os.path.dirname(downloaded.local_path)
            self.assertFalse([name for name in os.listdir(originals) if ".tmp-" in name])

        self.assert_removed(temp_dir)

    def test_download_original_asset_bad_image_leaves_asset_without_local_path(self):
        with self.make_temp_dir() as temp_dir:
            db_path, project, asset = self._project_and_asset(temp_dir)
            session = SimpleNamespace(
                get=lambda *args, **kwargs: _Response(
                    content=b"not an image",
                    headers={"Content-Type": "image/png"},
                )
            )

            with mock.patch("app.image_studio_images.socket.getaddrinfo", return_value=self._public_dns()):
                with self.assertRaisesRegex(image_studio_images.ImageStudioImageError, "解码失败"):
                    image_studio_images.download_original_asset(
                        asset.id,
                        path=db_path,
                        config={"data_dir": temp_dir, "image_dir": os.path.join(temp_dir, "images")},
                        session=session,
                    )

            refreshed = image_studio.get_asset(asset.id, path=db_path)
            self.assertIsNone(refreshed.local_path)
            dirs = image_studio.project_image_dirs(os.path.join(temp_dir, "images"), project)
            self.assertFalse(os.path.exists(dirs["originals"]))

        self.assert_removed(temp_dir)


if __name__ == "__main__":
    unittest.main()