RomanNum3ral
/
ComfyUI
mirror of https://github.com/Comfy-Org/ComfyUI.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add safe local asset file actions

This commit is contained in:
adobeluo 2026-08-06 13:45:01 +08:00
parent 62b3c94bd4
commit 021bead1bc
9 changed files with 749 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

123
app/assets/api/routes.py
View File

@ -31,6 +31,7 @@ from app.assets.services import (
asset_exists,
create_from_hash,
delete_asset_reference,
delete_asset_reference_with_file,
get_asset_detail,
list_assets_page,
list_tags,
@ -40,6 +41,10 @@ from app.assets.services import (
upload_from_temp_path,
)
from app.assets.services.cursor import InvalidCursorError
from app.assets.services.file_location import (
is_loopback_address,
reveal_file_in_file_manager,
)
from app.assets.services.path_utils import compute_display_name
from app.assets.services.tagging import list_tag_histogram
@ -636,20 +641,61 @@ async def update_asset_route(request: web.Request) -> web.Response:
@_require_assets_feature_enabled
async def delete_asset_route(request: web.Request) -> web.Response:
reference_id = str(uuid.UUID(request.match_info["id"]))
owner_id = USER_MANAGER.get_request_user_id(request)
delete_content = request.query.get("delete_content") == "true"
try:
# Deleting an asset is a soft delete of the reference; the underlying
# content is preserved (it may be shared with other references).
deleted = delete_asset_reference(
reference_id=reference_id,
owner_id=USER_MANAGER.get_request_user_id(request),
delete_content_if_orphan=False,
if delete_content:
detail = get_asset_detail(reference_id=reference_id, owner_id=owner_id)
if detail is None:
return _build_error_response(
404,
"ASSET_NOT_FOUND",
f"AssetReference {reference_id} not found.",
)
allowed_roots = []
if "input" in detail.tags:
allowed_roots.append(folder_paths.get_input_directory())
if "output" in detail.tags:
allowed_roots.append(folder_paths.get_output_directory())
if not detail.ref.file_path or not any(
folder_paths.is_within_directory(root, detail.ref.file_path)
for root in allowed_roots
):
return _build_error_response(
403,
"ASSET_DELETE_FORBIDDEN",
"Only tagged files inside the input or output directory can be deleted with their content.",
)
deleted = delete_asset_reference_with_file(
reference_id=reference_id,
owner_id=owner_id,
staging_directory=folder_paths.get_temp_directory(),
expected_file_path=detail.ref.file_path,
allowed_directories=allowed_roots,
allow_ownerless=not user_manager.args.multi_user,
)
else:
deleted = delete_asset_reference(
reference_id=reference_id,
owner_id=owner_id,
delete_content_if_orphan=False,
)
except PermissionError as error:
return _build_error_response(
403,
"ASSET_DELETE_FORBIDDEN",
str(error),
{"id": reference_id},
)
except Exception:
logging.exception(
"delete_asset_reference failed for reference_id=%s, owner_id=%s",
reference_id,
USER_MANAGER.get_request_user_id(request),
owner_id,
)
return _build_error_response(500, "INTERNAL", "Unexpected server error.")
@ -660,6 +706,69 @@ async def delete_asset_route(request: web.Request) -> web.Response:
return web.Response(status=204)
@ROUTES.post(f"/api/assets/{{id:{UUID_RE}}}/open-location")
@_require_assets_feature_enabled
async def open_asset_location_route(request: web.Request) -> web.Response:
reference_id = str(uuid.UUID(request.match_info["id"]))
if not is_loopback_address(request.remote):
return _build_error_response(
403,
"LOCAL_ACCESS_REQUIRED",
"Opening a file location is only available from the ComfyUI host machine.",
)
if request.headers.get("Sec-Fetch-Site") not in (None, "same-origin"):
return _build_error_response(
403,
"CROSS_SITE_REQUEST_FORBIDDEN",
"Cross-site requests cannot open local file locations.",
)
detail = get_asset_detail(
reference_id=reference_id,
owner_id=USER_MANAGER.get_request_user_id(request),
)
if detail is None:
return _build_error_response(
404,
"ASSET_NOT_FOUND",
f"AssetReference {reference_id} not found.",
)
file_path = detail.ref.file_path
output_root = folder_paths.get_output_directory()
if (
"output" not in detail.tags
or not file_path
or not folder_paths.is_within_directory(output_root, file_path)
):
return _build_error_response(
403,
"ASSET_LOCATION_FORBIDDEN",
"Only generated files inside the output directory can be opened.",
)
try:
reveal_file_in_file_manager(file_path)
except FileNotFoundError:
return _build_error_response(
404,
"ASSET_FILE_NOT_FOUND",
"The generated source file no longer exists on disk.",
)
except OSError:
logging.exception(
"Failed to reveal asset file for reference_id=%s", reference_id
)
return _build_error_response(
500,
"FILE_MANAGER_ERROR",
"Unable to open the system file manager.",
)
return web.Response(status=204)
@ROUTES.get("/api/tags")
@_require_assets_feature_enabled
async def get_tags(request: web.Request) -> web.Response:

2
app/assets/database/queries/__init__.py
View File

@ -11,6 +11,7 @@ from app.assets.database.queries.asset import (
from app.assets.database.queries.asset_reference import (
CacheStateRow,
UnenrichedReferenceRow,
any_reference_exists_for_asset_id,
bulk_insert_references_ignore_conflicts,
bulk_update_enrichment_level,
count_active_siblings,
@ -75,6 +76,7 @@ __all__ = [
"RemoveTagsResult",
"SetTagsResult",
"UnenrichedReferenceRow",
"any_reference_exists_for_asset_id",
"add_missing_tag_for_asset_id",
"add_tags_to_reference",
"asset_exists_by_hash",

13
app/assets/database/queries/asset_reference.py
View File

@ -145,6 +145,19 @@ def reference_exists_for_asset_id(
return session.execute(q).first() is not None
def any_reference_exists_for_asset_id(
session: Session,
asset_id: str,
) -> bool:
q = (
select(sa.literal(True))
.select_from(AssetReference)
.where(AssetReference.asset_id == asset_id)
.limit(1)
)
return session.execute(q).first() is not None
def reference_exists(
session: Session,
reference_id: str,

2
app/assets/services/__init__.py
View File

@ -1,6 +1,7 @@
from app.assets.services.asset_management import (
asset_exists,
delete_asset_reference,
delete_asset_reference_with_file,
get_asset_by_hash,
get_asset_detail,
list_assets_page,
@ -72,6 +73,7 @@ __all__ = [
"batch_insert_seed_assets",
"create_from_hash",
"delete_asset_reference",
"delete_asset_reference_with_file",
"get_asset_by_hash",
"get_asset_detail",
"ingest_existing_file",

113
app/assets/services/asset_management.py
View File

@ -1,9 +1,13 @@
import contextlib
import logging
import mimetypes
import os
import uuid
from datetime import timezone
from typing import Sequence
import folder_paths
from app.assets.services.cursor import (
CursorPayload,
InvalidCursorError,
@ -17,6 +21,7 @@ from app.assets.services.cursor import (
from app.assets.database.models import Asset
from app.assets.database.queries import (
any_reference_exists_for_asset_id,
asset_exists_by_hash,
reference_exists_for_asset_id,
delete_reference_by_id,
@ -221,6 +226,114 @@ def delete_asset_reference(
return True
def delete_asset_reference_with_file(
reference_id: str,
owner_id: str,
staging_directory: str,
expected_file_path: str,
allowed_directories: Sequence[str],
allow_ownerless: bool = False,
) -> bool:
with create_session() as session:
ref = get_reference_by_id(session, reference_id=reference_id)
if ref is None or ref.deleted_at is not None:
return False
caller_owner_id = (owner_id or "").strip()
owns_reference = ref.owner_id == caller_owner_id
may_delete_ownerless = allow_ownerless and ref.owner_id == ""
if not (owns_reference or may_delete_ownerless):
raise PermissionError(
"Only the owning user can delete an asset's source file."
)
asset_id = ref.asset_id
file_path = ref.file_path
if not file_path or os.path.realpath(file_path) != os.path.realpath(
expected_file_path
):
raise PermissionError(
"The asset source path changed before it could be deleted."
)
if not any(
folder_paths.is_within_directory(directory, file_path)
for directory in allowed_directories
):
raise PermissionError(
"The asset source file is outside the directories allowed for deletion."
)
deleted = delete_reference_by_id(
session, reference_id=reference_id, owner_id=owner_id
)
if not deleted:
session.rollback()
return False
staged_file_path: str | None = None
if file_path and os.path.isfile(file_path):
source_directory = os.path.dirname(file_path)
os.makedirs(staging_directory, exist_ok=True)
staged_file_path = os.path.join(
staging_directory, f".comfy-delete-{uuid.uuid4().hex}.tmp"
)
if not any(
folder_paths.is_within_directory(directory, file_path)
for directory in allowed_directories
):
raise PermissionError(
"The asset source file moved outside the directories allowed for deletion."
)
try:
os.replace(file_path, staged_file_path)
except OSError:
staged_file_path = os.path.join(
source_directory, f".comfy-delete-{uuid.uuid4().hex}.tmp"
)
if not any(
folder_paths.is_within_directory(directory, file_path)
for directory in allowed_directories
):
raise PermissionError(
"The asset source file moved outside the directories allowed for deletion."
)
os.replace(file_path, staged_file_path)
try:
session.flush()
if not any_reference_exists_for_asset_id(session, asset_id=asset_id):
asset = session.get(Asset, asset_id)
if asset is not None:
session.delete(asset)
session.commit()
except Exception:
session.rollback()
if (
staged_file_path
and file_path
and os.path.isfile(staged_file_path)
and not os.path.exists(file_path)
):
try:
os.replace(staged_file_path, file_path)
except OSError:
logging.exception(
"Failed to restore staged asset file after rollback: %s",
file_path,
)
raise
if staged_file_path:
try:
os.remove(staged_file_path)
except OSError:
logging.exception(
"Failed to remove staged asset file after commit: %s",
staged_file_path,
)
return True
def set_asset_preview(
reference_id: str,
preview_reference_id: str | None = None,

46
app/assets/services/file_location.py Normal file
View File

@ -0,0 +1,46 @@
import ipaddress
import os
import subprocess
import sys
def is_loopback_address(address: str | None) -> bool:
if not address:
return False
try:
parsed = ipaddress.ip_address(address.split("%", 1)[0])
except ValueError:
return False
if parsed.is_loopback:
return True
return bool(
parsed.version == 6 and parsed.ipv4_mapped and parsed.ipv4_mapped.is_loopback
)
def build_file_manager_command(
file_path: str, platform: str | None = None
) -> list[str]:
normalized_path = os.path.abspath(file_path)
current_platform = platform or sys.platform
if current_platform == "win32":
return ["explorer.exe", f"/select,{normalized_path}"]
if current_platform == "darwin":
return ["open", "-R", normalized_path]
return ["xdg-open", os.path.dirname(normalized_path)]
def reveal_file_in_file_manager(file_path: str) -> None:
normalized_path = os.path.abspath(file_path)
if not os.path.isfile(normalized_path):
raise FileNotFoundError(normalized_path)
subprocess.Popen(
build_file_manager_command(normalized_path),
close_fds=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)

205
tests-unit/assets_test/services/test_asset_management.py
View File

@ -1,4 +1,5 @@
"""Tests for asset_management services."""
import folder_paths
import pytest
from sqlalchemy.orm import Session
@ -6,6 +7,7 @@ from app.assets.database.models import Asset, AssetReference
from app.assets.database.queries import ensure_tags_exist, add_tags_to_reference
from app.assets.helpers import get_utc_now
from app.assets.services import (
delete_asset_reference_with_file,
get_asset_detail,
update_asset_metadata,
delete_asset_reference,
@ -215,6 +217,209 @@ class TestDeleteAssetReference:
assert session.get(Asset, asset_id) is None
class TestDeleteAssetReferenceWithFile:
def test_deletes_only_selected_reference_file(
self, mock_create_session, session: Session, temp_dir
):
selected_file = temp_dir / "selected.bin"
retained_file = temp_dir / "retained.bin"
selected_file.write_bytes(b"same-content")
retained_file.write_bytes(b"same-content")
asset = _make_asset(session)
selected_ref = _make_reference(session, asset, name=selected_file.name)
selected_ref.file_path = str(selected_file)
retained_ref = _make_reference(session, asset, name=retained_file.name)
retained_ref.file_path = str(retained_file)
asset_id = asset.id
session.commit()
result = delete_asset_reference_with_file(
reference_id=selected_ref.id,
owner_id="",
staging_directory=str(temp_dir / "staging"),
expected_file_path=str(selected_file),
allowed_directories=[str(temp_dir)],
)
assert result is True
assert not selected_file.exists()
assert retained_file.exists()
assert session.get(Asset, asset_id) is not None
assert session.get(AssetReference, retained_ref.id) is not None
def test_rejects_ownerless_reference_without_explicit_permission(
self, mock_create_session, session: Session, temp_dir
):
selected_file = temp_dir / "selected.bin"
selected_file.write_bytes(b"content")
asset = _make_asset(session)
selected_ref = _make_reference(session, asset, owner_id="")
selected_ref.file_path = str(selected_file)
selected_ref_id = selected_ref.id
session.commit()
with pytest.raises(PermissionError, match="owning user"):
delete_asset_reference_with_file(
reference_id=selected_ref_id,
owner_id="another-user",
staging_directory=str(temp_dir / "staging"),
expected_file_path=str(selected_file),
allowed_directories=[str(temp_dir)],
)
assert selected_file.exists()
assert session.get(AssetReference, selected_ref_id) is not None
def test_allows_ownerless_reference_in_single_user_mode(
self, mock_create_session, session: Session, temp_dir
):
selected_file = temp_dir / "selected.bin"
selected_file.write_bytes(b"content")
asset = _make_asset(session)
selected_ref = _make_reference(session, asset, owner_id="")
selected_ref.file_path = str(selected_file)
selected_ref_id = selected_ref.id
session.commit()
result = delete_asset_reference_with_file(
reference_id=selected_ref_id,
owner_id="default",
staging_directory=str(temp_dir / "staging"),
expected_file_path=str(selected_file),
allowed_directories=[str(temp_dir)],
allow_ownerless=True,
)
assert result is True
assert not selected_file.exists()
assert session.get(AssetReference, selected_ref_id) is None
def test_preserves_soft_deleted_shared_reference(
self, mock_create_session, session: Session, temp_dir
):
selected_file = temp_dir / "selected.bin"
retained_file = temp_dir / "retained.bin"
selected_file.write_bytes(b"same-content")
retained_file.write_bytes(b"same-content")
asset = _make_asset(session)
selected_ref = _make_reference(session, asset, name=selected_file.name)
selected_ref.file_path = str(selected_file)
retained_ref = _make_reference(session, asset, name=retained_file.name)
retained_ref.file_path = str(retained_file)
retained_ref.deleted_at = get_utc_now()
asset_id = asset.id
retained_ref_id = retained_ref.id
session.commit()
result = delete_asset_reference_with_file(
reference_id=selected_ref.id,
owner_id="",
staging_directory=str(temp_dir / "staging"),
expected_file_path=str(selected_file),
allowed_directories=[str(temp_dir)],
)
assert result is True
assert not selected_file.exists()
assert retained_file.exists()
assert session.get(Asset, asset_id) is not None
assert session.get(AssetReference, retained_ref_id) is not None
def test_restores_file_when_commit_fails(
self, mock_create_session, session: Session, temp_dir, monkeypatch
):
selected_file = temp_dir / "selected.bin"
selected_file.write_bytes(b"content")
asset = _make_asset(session)
selected_ref = _make_reference(session, asset, name=selected_file.name)
selected_ref.file_path = str(selected_file)
selected_ref_id = selected_ref.id
session.commit()
def fail_commit(_session):
raise RuntimeError("commit failed")
monkeypatch.setattr(Session, "commit", fail_commit)
with pytest.raises(RuntimeError, match="commit failed"):
delete_asset_reference_with_file(
reference_id=selected_ref_id,
owner_id="",
staging_directory=str(temp_dir / "staging"),
expected_file_path=str(selected_file),
allowed_directories=[str(temp_dir)],
)
session.expire_all()
assert selected_file.read_bytes() == b"content"
assert session.get(AssetReference, selected_ref_id) is not None
def test_rejects_a_source_path_changed_after_authorization(
self, mock_create_session, session: Session, temp_dir
):
selected_file = temp_dir / "selected.bin"
replacement_file = temp_dir / "replacement.bin"
selected_file.write_bytes(b"selected")
replacement_file.write_bytes(b"replacement")
asset = _make_asset(session)
selected_ref = _make_reference(session, asset, name=selected_file.name)
selected_ref.file_path = str(replacement_file)
selected_ref_id = selected_ref.id
session.commit()
with pytest.raises(PermissionError, match="source path changed"):
delete_asset_reference_with_file(
reference_id=selected_ref_id,
owner_id="",
staging_directory=str(temp_dir / "staging"),
expected_file_path=str(selected_file),
allowed_directories=[str(temp_dir)],
)
assert selected_file.exists()
assert replacement_file.exists()
assert session.get(AssetReference, selected_ref_id) is not None
def test_rechecks_containment_immediately_before_staging(
self, mock_create_session, session: Session, temp_dir, monkeypatch
):
selected_file = temp_dir / "managed" / "selected.bin"
selected_file.parent.mkdir()
selected_file.write_bytes(b"content")
asset = _make_asset(session)
selected_ref = _make_reference(session, asset, name=selected_file.name)
selected_ref.file_path = str(selected_file)
selected_ref_id = selected_ref.id
session.commit()
containment_results = iter([True, False])
monkeypatch.setattr(
folder_paths,
"is_within_directory",
lambda *_args: next(containment_results),
)
with pytest.raises(PermissionError, match="moved outside"):
delete_asset_reference_with_file(
reference_id=selected_ref_id,
owner_id="",
staging_directory=str(temp_dir / "staging"),
expected_file_path=str(selected_file),
allowed_directories=[str(temp_dir / "managed")],
)
session.expire_all()
assert selected_file.read_bytes() == b"content"
assert session.get(AssetReference, selected_ref_id) is not None
class TestSetAssetPreview:
def test_sets_preview(self, mock_create_session, session: Session):
asset = _make_asset(session, hash_val="blake3:main")

69
tests-unit/assets_test/services/test_file_location.py Normal file
View File

@ -0,0 +1,69 @@
import subprocess
import pytest
from app.assets.services.file_location import (
build_file_manager_command,
is_loopback_address,
reveal_file_in_file_manager,
)
@pytest.mark.parametrize(
"address",
["127.0.0.1", "127.42.0.9", "::1", "::ffff:127.0.0.1"],
)
def test_accepts_loopback_addresses(address):
assert is_loopback_address(address)
@pytest.mark.parametrize(
"address",
[None, "", "localhost", "0.0.0.0", "192.168.1.20", "::", "fe80::1"],
)
def test_rejects_non_loopback_addresses(address):
assert not is_loopback_address(address)
def test_builds_platform_commands(tmp_path):
source = tmp_path / "folder with spaces" / "render.mp4"
source.parent.mkdir()
source.write_bytes(b"video")
assert build_file_manager_command(str(source), "win32") == [
"explorer.exe",
f"/select,{source}",
]
assert build_file_manager_command(str(source), "darwin") == [
"open",
"-R",
str(source),
]
assert build_file_manager_command(str(source), "linux") == [
"xdg-open",
str(source.parent),
]
def test_reveal_launches_without_a_shell(tmp_path, monkeypatch):
source = tmp_path / "render.png"
source.write_bytes(b"image")
calls = []
def capture_popen(*args, **kwargs):
calls.append((args, kwargs))
monkeypatch.setattr(subprocess, "Popen", capture_popen)
reveal_file_in_file_manager(str(source))
assert len(calls) == 1
args, kwargs = calls[0]
assert args[0]
assert kwargs["close_fds"] is True
assert "shell" not in kwargs
def test_reveal_rejects_missing_files(tmp_path):
with pytest.raises(FileNotFoundError):
reveal_file_in_file_manager(str(tmp_path / "missing.mp4"))

183
tests-unit/assets_test/test_asset_location_route.py Normal file
View File

@ -0,0 +1,183 @@
import asyncio
import json
import uuid
from types import SimpleNamespace
from unittest.mock import Mock
from app.assets.api import routes
def make_request(remote="127.0.0.1", headers=None, query=None):
return SimpleNamespace(
match_info={"id": str(uuid.uuid4())},
remote=remote,
headers=headers or {},
query=query or {},
)
def run_route(request):
return asyncio.run(routes.open_asset_location_route.__wrapped__(request))
def run_delete_route(request):
return asyncio.run(routes.delete_asset_route.__wrapped__(request))
def test_open_location_rejects_lan_clients(monkeypatch):
get_detail = Mock()
monkeypatch.setattr(routes, "get_asset_detail", get_detail)
response = run_route(make_request(remote="192.168.1.50"))
assert response.status == 403
assert json.loads(response.text)["error"]["code"] == "LOCAL_ACCESS_REQUIRED"
get_detail.assert_not_called()
def test_open_location_rejects_cross_site_requests(monkeypatch):
get_detail = Mock()
monkeypatch.setattr(routes, "get_asset_detail", get_detail)
response = run_route(make_request(headers={"Sec-Fetch-Site": "cross-site"}))
assert response.status == 403
assert json.loads(response.text)["error"]["code"] == "CROSS_SITE_REQUEST_FORBIDDEN"
get_detail.assert_not_called()
def test_open_location_reveals_managed_generated_file(tmp_path, monkeypatch):
source = tmp_path / "output" / "video" / "render.mp4"
source.parent.mkdir(parents=True)
source.write_bytes(b"video")
reveal = Mock()
monkeypatch.setattr(
routes,
"USER_MANAGER",
SimpleNamespace(get_request_user_id=lambda _request: "default"),
)
monkeypatch.setattr(
routes,
"get_asset_detail",
lambda **_kwargs: SimpleNamespace(
tags=["output"], ref=SimpleNamespace(file_path=str(source))
),
)
monkeypatch.setattr(
routes.folder_paths,
"get_output_directory",
lambda: str(tmp_path / "output"),
)
monkeypatch.setattr(routes, "reveal_file_in_file_manager", reveal)
response = run_route(make_request())
assert response.status == 204
reveal.assert_called_once_with(str(source))
def test_open_location_rejects_non_output_assets(tmp_path, monkeypatch):
source = tmp_path / "input" / "image.png"
source.parent.mkdir()
source.write_bytes(b"image")
monkeypatch.setattr(
routes,
"USER_MANAGER",
SimpleNamespace(get_request_user_id=lambda _request: "default"),
)
monkeypatch.setattr(
routes,
"get_asset_detail",
lambda **_kwargs: SimpleNamespace(
tags=["input"], ref=SimpleNamespace(file_path=str(source))
),
)
monkeypatch.setattr(
routes.folder_paths,
"get_output_directory",
lambda: str(tmp_path / "output"),
)
response = run_route(make_request())
assert response.status == 403
assert json.loads(response.text)["error"]["code"] == "ASSET_LOCATION_FORBIDDEN"
def test_delete_content_rejects_paths_outside_managed_root(tmp_path, monkeypatch):
source = tmp_path / "outside" / "render.png"
source.parent.mkdir()
source.write_bytes(b"image")
delete_with_file = Mock()
monkeypatch.setattr(
routes,
"USER_MANAGER",
SimpleNamespace(get_request_user_id=lambda _request: "default"),
)
monkeypatch.setattr(
routes,
"get_asset_detail",
lambda **_kwargs: SimpleNamespace(
tags=["output"], ref=SimpleNamespace(file_path=str(source))
),
)
monkeypatch.setattr(
routes.folder_paths,
"get_output_directory",
lambda: str(tmp_path / "output"),
)
monkeypatch.setattr(routes, "delete_asset_reference_with_file", delete_with_file)
response = run_delete_route(make_request(query={"delete_content": "true"}))
assert response.status == 403
assert json.loads(response.text)["error"]["code"] == "ASSET_DELETE_FORBIDDEN"
delete_with_file.assert_not_called()
def test_delete_content_uses_guarded_file_service(tmp_path, monkeypatch):
source = tmp_path / "output" / "render.png"
source.parent.mkdir()
source.write_bytes(b"image")
delete_with_file = Mock(return_value=True)
monkeypatch.setattr(
routes,
"USER_MANAGER",
SimpleNamespace(get_request_user_id=lambda _request: "default"),
)
monkeypatch.setattr(
routes,
"get_asset_detail",
lambda **_kwargs: SimpleNamespace(
tags=["output"], ref=SimpleNamespace(file_path=str(source))
),
)
monkeypatch.setattr(
routes.folder_paths,
"get_output_directory",
lambda: str(tmp_path / "output"),
)
monkeypatch.setattr(
routes.folder_paths,
"get_temp_directory",
lambda: str(tmp_path / "temp"),
)
monkeypatch.setattr(routes.user_manager.args, "multi_user", False)
monkeypatch.setattr(routes, "delete_asset_reference_with_file", delete_with_file)
request = make_request(query={"delete_content": "true"})
response = run_delete_route(request)
assert response.status == 204
delete_with_file.assert_called_once_with(
reference_id=request.match_info["id"],
owner_id="default",
staging_directory=str(tmp_path / "temp"),
expected_file_path=str(source),
allowed_directories=[str(tmp_path / "output")],
allow_ownerless=True,
)