fix(wcd): verify imported stickers and remap production assets

backend/tests/test_wcd_import.py

@@ -0,0 +1,169 @@
+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_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)