feat(assets): protect referenced stickers from deletion
This commit is contained in:
@@ -1399,6 +1399,26 @@ def _zip_asset_entry(zf: zipfile.ZipFile, pkg_id: str) -> str:
|
||||
return sorted(candidates)[0]
|
||||
|
||||
|
||||
def _asset_reference_names(asset_id: str) -> list[str]:
|
||||
usages: list[str] = []
|
||||
data_sources = (
|
||||
(DESIGN_TEMPLATES_DIR, "template.json"),
|
||||
(PROJECTS_DIR, "project.json"),
|
||||
)
|
||||
for root, filename in data_sources:
|
||||
for item in _list_dirs(root):
|
||||
path = item / filename
|
||||
if not path.exists():
|
||||
continue
|
||||
try:
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
except Exception:
|
||||
continue
|
||||
if asset_id in json.dumps(data, ensure_ascii=False):
|
||||
usages.append(str(data.get("name") or item.name))
|
||||
return usages
|
||||
|
||||
|
||||
def _remap_import_asset_ids(document: dict, asset_map: dict[str, str]) -> None:
|
||||
elements = document.get("elements")
|
||||
if not isinstance(elements, list):
|
||||
@@ -1796,6 +1816,10 @@ def delete_asset(asset_id: str) -> None:
|
||||
meta = _read_asset_meta(d)
|
||||
if not meta:
|
||||
raise HTTPException(status_code=404, detail="asset not found")
|
||||
usages = _asset_reference_names(asset_id)
|
||||
if usages:
|
||||
names = "、".join(dict.fromkeys(usages))
|
||||
raise HTTPException(status_code=409, detail=f"素材仍被以下设计引用,无法删除:{names}")
|
||||
shutil.rmtree(d, ignore_errors=True)
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,102 @@
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
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 fastapi import HTTPException # noqa: E402
|
||||
from service import app # noqa: E402
|
||||
|
||||
|
||||
SVG = '<svg xmlns="http://www.w3.org/2000/svg" width="20" height="10"/>'
|
||||
ASSET_ID = "asset_deleteprotected000000000000000000"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def isolated_storage(tmp_path, monkeypatch):
|
||||
assets_dir = tmp_path / "service_assets"
|
||||
templates_dir = tmp_path / "service_design_templates"
|
||||
projects_dir = tmp_path / "service_projects"
|
||||
for directory in (assets_dir, templates_dir, projects_dir):
|
||||
directory.mkdir()
|
||||
monkeypatch.setattr(app, "ASSETS_DIR", assets_dir)
|
||||
monkeypatch.setattr(app, "DESIGN_TEMPLATES_DIR", templates_dir)
|
||||
monkeypatch.setattr(app, "PROJECTS_DIR", projects_dir)
|
||||
return assets_dir
|
||||
|
||||
|
||||
def create_sticker_asset(assets_dir, asset_id=ASSET_ID):
|
||||
asset_dir = assets_dir / asset_id[:2] / asset_id
|
||||
asset_dir.mkdir(parents=True)
|
||||
(asset_dir / "asset.svg").write_text(SVG, encoding="utf-8")
|
||||
(asset_dir / "meta.json").write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"asset_id": asset_id,
|
||||
"name": "protected sticker",
|
||||
"type": "sticker",
|
||||
"mime_type": "image/svg+xml",
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
|
||||
def test_delete_asset_blocks_design_template_reference(isolated_storage):
|
||||
create_sticker_asset(isolated_storage)
|
||||
templates_dir = isolated_storage.parent / "service_design_templates"
|
||||
template_dir = templates_dir / "tm" / "tmpl_test"
|
||||
template_dir.mkdir(parents=True)
|
||||
(template_dir / "template.json").write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"template_id": "tmpl_test",
|
||||
"name": "TEST模板",
|
||||
"document": {"elements": [{"type": "sticker", "assetId": ASSET_ID}]},
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
with pytest.raises(HTTPException) as excinfo:
|
||||
app.delete_asset(ASSET_ID)
|
||||
|
||||
assert excinfo.value.status_code == 409
|
||||
assert "TEST模板" in str(excinfo.value.detail)
|
||||
assert (isolated_storage / ASSET_ID[:2] / ASSET_ID / "meta.json").exists()
|
||||
|
||||
|
||||
def test_delete_asset_blocks_project_reference(isolated_storage):
|
||||
create_sticker_asset(isolated_storage)
|
||||
projects_dir = isolated_storage.parent / "service_projects"
|
||||
project_dir = projects_dir / "pr" / "proj_test"
|
||||
project_dir.mkdir(parents=True)
|
||||
(project_dir / "project.json").write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"project_id": "proj_test",
|
||||
"name": "生产项目",
|
||||
"stickers": [{"assetId": ASSET_ID}],
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
with pytest.raises(HTTPException) as excinfo:
|
||||
app.delete_asset(ASSET_ID)
|
||||
|
||||
assert excinfo.value.status_code == 409
|
||||
assert "生产项目" in str(excinfo.value.detail)
|
||||
|
||||
|
||||
def test_delete_asset_allows_unreferenced_sticker(isolated_storage):
|
||||
create_sticker_asset(isolated_storage)
|
||||
|
||||
app.delete_asset(ASSET_ID)
|
||||
|
||||
assert not (isolated_storage / ASSET_ID[:2] / ASSET_ID).exists()
|
||||
Reference in New Issue
Block a user