wordcloud/backend/tests/test_wcd_import.py

import asyncio
import json
import hashlib
import sys
import zipfile
from io import BytesIO
from pathlib import Path
from types import SimpleNamespace

from fastapi import UploadFile
from PIL import Image
import pytest


BACKEND_DIR = Path(__file__).resolve().parents[1]
if str(BACKEND_DIR) not in sys.path:
    sys.path.insert(0, str(BACKEND_DIR))

from service import app  # noqa: E402


SVG = (
    '<svg xmlns="http://www.w3.org/2000/svg" width="20" height="10">'
    '<rect width="20" height="10" fill="#3366ff"/></svg>'
)


def png_bytes(color=(51, 102, 255)):
    buffer = BytesIO()
    Image.new("RGB", (1, 1), color).save(buffer, format="PNG")
    return buffer.getvalue()


@pytest.fixture
def isolated_asset_store(tmp_path, monkeypatch):
    assets_dir = tmp_path / "service_assets"
    assets_dir.mkdir()
    monkeypatch.setattr(app, "ASSETS_DIR", assets_dir)
    design_templates_dir = tmp_path / "service_design_templates"
    design_templates_dir.mkdir()
    monkeypatch.setattr(app, "DESIGN_TEMPLATES_DIR", design_templates_dir)
    return assets_dir


def asset_dirs(assets_dir):
    return sorted(path for path in assets_dir.glob("*/*") if path.is_dir())


def test_duplicate_content_is_reused(isolated_asset_store):
    first = app._register_import_asset("first", SVG.encode(), "image/svg+xml")
    second = app._register_import_asset("second", SVG.encode(), "image/svg+xml")

    assert first["asset_id"] == second["asset_id"]
    assert second["name"] == "first"
    assert len(asset_dirs(isolated_asset_store)) == 1


def test_same_name_different_content_gets_unique_assets(isolated_asset_store):
    first = app._register_import_asset("same-name", SVG.encode(), "image/svg+xml")
    second = app._register_import_asset(
        "same-name", SVG.replace("#3366ff", "#ff3366").encode(), "image/svg+xml"
    )

    assert first["asset_id"] != second["asset_id"]
    assert len(asset_dirs(isolated_asset_store)) == 2
    assert first["sha256"] != second["sha256"]


def test_stale_sha256_meta_does_not_reuse_mismatched_file(isolated_asset_store):
    content = SVG.encode()
    existing_id = "asset_stale00000000000000000000000000"
    existing_dir = isolated_asset_store / existing_id[:2] / existing_id
    existing_dir.mkdir(parents=True)
    (existing_dir / "asset.svg").write_text('<svg xmlns="x" width="1" height="1"/>')
    (existing_dir / "meta.json").write_text(
        json.dumps(
            {
                "asset_id": existing_id,
                "name": "old",
                "type": "sticker",
                "mime_type": "image/svg+xml",
                "sha256": hashlib.sha256(content).hexdigest(),
            }
        )
    )

    imported = app._register_import_asset("new", content, "image/svg+xml")

    assert imported["asset_id"] != existing_id
    assert len(asset_dirs(isolated_asset_store)) == 2


def test_reference_with_same_content_creates_sticker_asset(isolated_asset_store):
    content = SVG.encode()
    existing_id = "asset_reference00000000000000000000000"
    existing_dir = isolated_asset_store / existing_id[:2] / existing_id
    existing_dir.mkdir(parents=True)
    (existing_dir / "asset.svg").write_bytes(content)
    (existing_dir / "meta.json").write_text(
        json.dumps(
            {
                "asset_id": existing_id,
                "name": "old",
                "type": "reference",
                "mime_type": "image/svg+xml",
                "sha256": hashlib.sha256(content).hexdigest(),
            }
        )
    )

    imported = app._register_import_asset("new", content, "image/svg+xml")

    assert imported["asset_id"] != existing_id
    assert imported["type"] == "sticker"
    assert len(asset_dirs(isolated_asset_store)) == 2


def test_wcd_production_uses_real_asset_ids(tmp_path, isolated_asset_store, monkeypatch):
    manifest = {
        "format": "wordcloud-canvas",
        "version": 1,
        "name": "test-design",
        "assets": [{"id": "pkg-1", "name": "sticker", "type": "svg", "mimeType": "image/svg+xml"}],
    }
    document_json = {
        "width": 40,
        "height": 20,
        "background": "#ffffff",
        "layers": [],
        "layerFolders": [],
        "elements": [
            {
                "id": "element-1",
                "type": "sticker",
                "name": "sticker",
                "assetId": "pkg-1",
                "x": 5,
                "y": 5,
                "width": 10,
                "height": 10,
                "rotation": 0,
                "opacity": 1,
            }
        ],
    }
    package = BytesIO()
    with zipfile.ZipFile(package, "w") as archive:
        archive.writestr("manifest.json", json.dumps(manifest))
        archive.writestr("document.json", json.dumps(document_json))
        archive.writestr("assets/pkg-1.png", png_bytes())

    class Paths:
        input_dir = tmp_path / "input"
        output_dir = tmp_path / "output"

    def prepare_job_dirs(job_id):
        Paths.input_dir.mkdir(parents=True)
        Paths.output_dir.mkdir(parents=True)
        return Paths

    manager = SimpleNamespace(
        create_job=lambda: "job-test",
        set_artifacts=lambda *args, **kwargs: None,
        set_status=lambda *args, **kwargs: None,
    )
    original_compose = app._compose_design_png
    captured = {}

    def compose(document, file_map, output_path):
        captured["document"] = document
        captured["file_map"] = dict(file_map)
        original_compose(document, file_map, output_path)

    monkeypatch.setattr(app, "manager", manager)
    monkeypatch.setattr(app, "storage", SimpleNamespace(prepare_job_dirs=prepare_job_dirs))
    monkeypatch.setattr(app, "_write_order", lambda *args, **kwargs: None)
    monkeypatch.setattr(app, "_compose_design_png", compose)
    monkeypatch.setattr(app.threading, "Thread", lambda target, **kwargs: SimpleNamespace(start=target))

    upload = UploadFile(filename="test.wcd", file=BytesIO(package.getvalue()))
    asyncio.run(app._create_wcd_job(upload, {}))

    document = captured["document"]
    file_map = captured["file_map"]
    real_ids = [element["assetId"] for element in document["elements"]]
    assert real_ids
    assert all(asset_id in file_map for asset_id in real_ids)
    assert all(Path(asset_id).is_absolute() is False for asset_id in real_ids)

    png_path = isolated_asset_store.parent / "composed.png"
    original_compose(document, file_map, png_path)

    with Image.open(png_path) as image:
        assert image.getpixel((10, 10)) == (51, 102, 255)