zlei6/ComfyUI
1
0
Fork 0
mirror of https://github.com/Comfy-Org/ComfyUI.git synced 2026-08-26 19:02:36 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge master into qa/video-edit-combined

Browse Source 
# Conflicts:
#	comfy_api/latest/_input_impl/video_types.py
This commit is contained in:
Claude
2026-08-20 23:33:45 +00:00
parent ea441aae23 dcbcf8c2e1
commit c148b26568
126 changed files with 10668 additions and 1117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
tests-unit/assets_test/services/test_asset_response_loader_path.py
View File

@@ -43,7 +43,7 @@ def test_uses_persisted_loader_path_without_recomputing():
loader_path="SENTINEL/stored.safetensors",
)
resp = _build_asset_response(result)
resp = _build_asset_response(result, {})
assert resp.loader_path == "SENTINEL/stored.safetensors"
@@ -67,7 +67,7 @@ def test_null_stored_loader_path_is_served_as_null(tmp_path: Path):
mock_fp.models_dir = str(models)
result = _make_result(file_path=str(f), loader_path=None)
resp = _build_asset_response(result)
resp = _build_asset_response(result, {})
assert resp.loader_path is None
assert resp.display_name == "checkpoints/bar.safetensors"
@@ -77,7 +77,7 @@ def test_all_path_fields_null_without_file_path():
"""API-created / hash-only references (no file_path) expose no paths."""
result = _make_result(file_path=None, loader_path=None)
resp = _build_asset_response(result)
resp = _build_asset_response(result, {})
assert resp.loader_path is None
assert resp.display_name is None

300
tests-unit/assets_test/services/test_asset_response_preview_url.py Normal file
View File

@@ -0,0 +1,300 @@
from datetime import datetime
from pathlib import Path
from unittest.mock import patch
import pytest
from app.assets.api.routes import _build_asset_response
from app.assets.services.schemas import AssetData, AssetDetailResult, ReferenceData
_TS = datetime(2024, 1, 1, 0, 0, 0)
@pytest.fixture
def sandboxed_comfy_roots(tmp_path: Path):
with patch("app.assets.services.path_utils.folder_paths") as fp:
fp.get_input_directory.return_value = str(tmp_path / "input")
fp.get_output_directory.return_value = str(tmp_path / "output")
fp.get_temp_directory.return_value = str(tmp_path / "temp")
fp.models_dir = str(tmp_path / "models")
yield tmp_path
def _make_result(
*,
ref_id: str = "ref-1",
name: str = "ComfyUI_temp_abcde_00001_.png",
file_path: str | None = None,
mime_type: str | None = "image/png",
preview_id: str | None = None,
tags: list[str] | None = None,
user_metadata: dict | None = None,
with_asset: bool = True,
) -> AssetDetailResult:
ref = ReferenceData(
id=ref_id,
name=name,
file_path=file_path,
loader_path=None,
user_metadata=user_metadata,
preview_id=preview_id,
created_at=_TS,
updated_at=_TS,
last_access_time=_TS,
)
asset = (
AssetData(hash="blake3:abc", size_bytes=1024, mime_type=mime_type)
if with_asset
else None
)
return AssetDetailResult(ref=ref, asset=asset, tags=tags or [])
@pytest.mark.parametrize(
("root", "relative"),
[
("temp", "ComfyUI_temp_abcde_00001_.png"),
("output", "ComfyUI_00001_.png"),
("input", "example.png"),
],
)
def test_every_view_root_gets_a_preview_url(
sandboxed_comfy_roots: Path, root: str, relative: str
):
resp = _build_asset_response(
_make_result(name=relative, file_path=str(sandboxed_comfy_roots / root / relative)),
{},
)
assert resp.preview_url == f"/api/view?type={root}&filename={relative}", (
f"a file in {root} must get a preview URL; temp is the one the old "
f"tag chain fell off the end of"
)
@pytest.mark.parametrize(
"tags",
[[], ["input"], ["output"], ["models", "model_type:checkpoints"]],
)
def test_preview_url_does_not_depend_on_tags(
sandboxed_comfy_roots: Path, tags: list[str]
):
resp = _build_asset_response(
_make_result(file_path=str(sandboxed_comfy_roots / "temp" / "a.png"), tags=tags), {}
)
assert resp.preview_url == "/api/view?type=temp&filename=a.png", (
"tags are user-editable; removing one must not destroy the preview"
)
def test_preview_url_does_not_depend_on_the_metadata_filename(
sandboxed_comfy_roots: Path,
):
resp = _build_asset_response(
_make_result(
file_path=str(sandboxed_comfy_roots / "output" / "a.png"), user_metadata=None
),
{},
)
assert resp.preview_url == "/api/view?type=output&filename=a.png", (
"a reference carrying no metadata filename must still get a preview"
)
def test_subfolder_is_split_out_and_both_halves_encoded(sandboxed_comfy_roots: Path):
resp = _build_asset_response(
_make_result(
name="my shot.png",
file_path=str(sandboxed_comfy_roots / "output" / "runs & takes" / "my shot.png"),
),
{},
)
assert resp.preview_url == (
"/api/view?type=output&filename=my%20shot.png&subfolder=runs%20%26%20takes"
), "an unencoded & or space in the path would break the query string"
def test_preview_id_resolves_through_the_page_lookup(sandboxed_comfy_roots: Path):
result = _make_result(
file_path=str(sandboxed_comfy_roots / "models" / "checkpoints" / "m.safetensors"),
mime_type="application/safetensors",
preview_id="preview-ref",
)
resp = _build_asset_response(
result, {"preview-ref": str(sandboxed_comfy_roots / "output" / "thumb.png")}
)
assert resp.preview_url == "/api/view?type=output&filename=thumb.png", (
"a nominated preview stands in for content with no visual form"
)
assert resp.preview_id == "preview-ref"
def test_unresolvable_preview_id_yields_no_url(sandboxed_comfy_roots: Path):
result = _make_result(
file_path=str(sandboxed_comfy_roots / "output" / "a.png"), preview_id="gone"
)
resp = _build_asset_response(result, {})
assert resp.preview_url is None, (
"a preview absent from the lookup is soft-deleted, invisible or "
"path-less, so advertising it would promise a URL that 404s - and the "
"asset's own bytes are a different picture, not a degraded one"
)
@pytest.mark.parametrize(
("name", "mime_type"),
[
("notes.txt", "text/plain"),
("notes.md", "text/markdown"),
("rows.csv", "text/csv"),
("page.html", "text/html"),
],
)
def test_text_is_previewable(sandboxed_comfy_roots: Path, name: str, mime_type: str):
resp = _build_asset_response(
_make_result(
name=name,
file_path=str(sandboxed_comfy_roots / "output" / name),
mime_type=mime_type,
),
{},
)
assert resp.preview_url == f"/api/view?type=output&filename={name}", (
"text assets are rendered as a snippet from preview_url, so withholding "
"it leaves that with nothing to fetch; the dangerous members stay safe "
"because /api/view forces them to download, not because they get no URL"
)
@pytest.mark.parametrize(
"mime_type",
["application/safetensors", "application/gguf", "application/octet-stream"],
)
def test_no_preview_url_for_content_a_browser_cannot_render(
sandboxed_comfy_roots: Path, mime_type: str
):
resp = _build_asset_response(
_make_result(
name="model.safetensors",
file_path=str(sandboxed_comfy_roots / "input" / "model.safetensors"),
mime_type=mime_type,
),
{},
)
assert resp.preview_url is None, (
"content a browser cannot render must not advertise itself as a preview"
)
@pytest.mark.parametrize(
("name", "expected"),
[
("shot.png", "/api/view?type=temp&filename=shot.png"),
("clip.mp4", "/api/view?type=temp&filename=clip.mp4"),
("model.safetensors", None),
],
)
def test_missing_mime_type_falls_back_to_the_path(
sandboxed_comfy_roots: Path, name: str, expected: str | None
):
resp = _build_asset_response(
_make_result(
name=name, file_path=str(sandboxed_comfy_roots / "temp" / name), mime_type=None
),
{},
)
assert resp.preview_url == expected, (
"a previewable file must not lose its preview just because the scan "
"that found it recorded no mime type"
)
@pytest.mark.parametrize(
("name", "stored_filename", "expected"),
[
("untitled", "shot.png", "/api/view?type=temp&filename=shot.png"),
("shot.png", "weights.safetensors", None),
],
)
def test_previewability_follows_the_path_not_the_editable_name(
sandboxed_comfy_roots: Path, name: str, stored_filename: str, expected: str | None
):
resp = _build_asset_response(
_make_result(
name=name,
file_path=str(sandboxed_comfy_roots / "temp" / stored_filename),
mime_type=None,
),
{},
)
assert resp.preview_url == expected, (
"name is editable through PUT /api/assets/{id}, so deriving "
"previewability from it would let a rename create or destroy a preview "
"without the bytes changing"
)
def test_mime_type_parameters_do_not_defeat_the_media_check(
sandboxed_comfy_roots: Path,
):
resp = _build_asset_response(
_make_result(
file_path=str(sandboxed_comfy_roots / "temp" / "a.png"),
mime_type="IMAGE/PNG; charset=binary",
),
{},
)
assert resp.preview_url == "/api/view?type=temp&filename=a.png"
def test_no_preview_url_for_a_model(sandboxed_comfy_roots: Path):
resp = _build_asset_response(
_make_result(
name="m.png",
file_path=str(sandboxed_comfy_roots / "models" / "checkpoints" / "m.png"),
),
{},
)
assert resp.preview_url is None, (
"models is not a root /api/view can address, whatever the file is"
)
def test_no_preview_url_for_a_path_outside_every_root(sandboxed_comfy_roots: Path):
resp = _build_asset_response(_make_result(file_path="/elsewhere/a.png"), {})
assert resp.preview_url is None, (
"/api/view cannot address a file outside the roots it serves"
)
def test_no_preview_url_without_a_file_path(sandboxed_comfy_roots: Path):
resp = _build_asset_response(_make_result(file_path=None), {})
assert resp.preview_url is None, (
"an API-created reference has no path, so no view URL can be derived"
)
def test_no_preview_url_without_content(sandboxed_comfy_roots: Path):
resp = _build_asset_response(
_make_result(
file_path=str(sandboxed_comfy_roots / "temp" / "a.png"), with_asset=False
),
{},
)
assert resp.preview_url is None, "no asset row means there is nothing to preview"

419
tests-unit/assets_test/test_list_filter.py
View File

@@ -1,10 +1,14 @@
import time
import uuid
import warnings
import pytest
import requests
from helpers import assert_hash_fields_consistent
from app.assets.api import routes as assets_routes
from app.assets.api import schemas_in
def test_list_assets_paging_and_sort(http: requests.Session, api_base: str, asset_factory, make_asset_bytes):
names = ["a1_u.safetensors", "a2_u.safetensors", "a3_u.safetensors"]
@@ -337,3 +341,418 @@ def test_list_assets_name_contains_literal_underscore(
assert b["name"] not in names, "Underscore must be escaped — should not match 'fooxbar'"
assert c["name"] not in names, "Underscore must be escaped — should not match 'foobar'"
assert body["total"] == 1
def test_list_assets_tags_any_alone(http, api_base, asset_factory, make_asset_bytes):
scope = f"lf-any-{uuid.uuid4().hex[:6]}"
t = ["models", "model_type:checkpoints", "unit-tests", scope]
a = asset_factory("any_a.safetensors", [*t, f"{scope}-alpha"], {}, make_asset_bytes("any_a"))
b = asset_factory("any_b.safetensors", [*t, f"{scope}-beta"], {}, make_asset_bytes("any_b"))
c = asset_factory("any_c.safetensors", [*t, f"{scope}-gamma"], {}, make_asset_bytes("any_c"))
r = http.get(
api_base + "/api/assets",
params={"tags_any": f"{scope}-alpha,{scope}-beta", "limit": "50"},
timeout=120,
)
body = r.json()
assert r.status_code == 200, body
names = [x["name"] for x in body["assets"]]
assert a["name"] in names
assert b["name"] in names
assert c["name"] not in names
def test_list_assets_tags_any_with_tags_all(http, api_base, asset_factory, make_asset_bytes):
scope = f"lf-anyall-{uuid.uuid4().hex[:6]}"
t = ["models", "model_type:checkpoints", "unit-tests", scope]
alpha, beta = f"{scope}-alpha", f"{scope}-beta"
x = asset_factory("aa_x.safetensors", [*t, alpha], {}, make_asset_bytes("aa_x"))
y = asset_factory("aa_y.safetensors", [*t, beta], {}, make_asset_bytes("aa_y"))
w = asset_factory("aa_w.safetensors", t, {}, make_asset_bytes("aa_w"))
d = asset_factory(
"aa_d.safetensors",
["models", "model_type:checkpoints", "unit-tests", f"{scope}-other", alpha],
{},
make_asset_bytes("aa_d"),
)
r = http.get(
api_base + "/api/assets",
params={"tags_all": f"unit-tests,{scope}", "tags_any": f"{alpha},{beta}", "limit": "50"},
timeout=120,
)
body = r.json()
assert r.status_code == 200, body
names = [a["name"] for a in body["assets"]]
assert x["name"] in names
assert y["name"] in names
assert w["name"] not in names, "asset matching tags_all but not tags_any must be excluded"
assert d["name"] not in names, "asset matching tags_any but not tags_all must be excluded"
def test_list_assets_tags_none_wins_over_tags_any(http, api_base, asset_factory, make_asset_bytes):
scope = f"lf-nonewins-{uuid.uuid4().hex[:6]}"
t = ["models", "model_type:checkpoints", "unit-tests", scope]
alpha, beta = f"{scope}-alpha", f"{scope}-beta"
x = asset_factory("nw_x.safetensors", [*t, alpha], {}, make_asset_bytes("nw_x"))
y = asset_factory("nw_y.safetensors", [*t, alpha, beta], {}, make_asset_bytes("nw_y"))
r = http.get(
api_base + "/api/assets",
params={"tags_any": alpha, "tags_none": beta, "limit": "50"},
timeout=120,
)
body = r.json()
assert r.status_code == 200, body
names = [a["name"] for a in body["assets"]]
assert x["name"] in names
assert y["name"] not in names, "tags_none must exclude an asset even when it matches tags_any"
def test_list_assets_empty_tag_filter_lists_behave_as_absent(http, api_base, asset_factory, make_asset_bytes):
scope = f"lf-empty-{uuid.uuid4().hex[:6]}"
t = ["models", "model_type:checkpoints", "unit-tests", scope]
a = asset_factory("em_a.safetensors", t, {}, make_asset_bytes("em_a"))
b = asset_factory("em_b.safetensors", t, {}, make_asset_bytes("em_b"))
expected = {a["name"], b["name"]}
# Empty new-name lists impose no constraint.
r1 = http.get(
api_base + "/api/assets",
params={"tags_all": f"unit-tests,{scope}", "tags_any": "", "tags_none": ""},
timeout=120,
)
b1 = r1.json()
assert r1.status_code == 200, b1
assert {x["name"] for x in b1["assets"]} == expected
# An empty new-name param alongside old names must not trigger validation.
r2 = http.get(
api_base + "/api/assets",
params={"include_tags": f"unit-tests,{scope}", "tags_any": ""},
timeout=120,
)
b2 = r2.json()
assert r2.status_code == 200, b2
assert {x["name"] for x in b2["assets"]} == expected
# An empty tags_all next to include_tags is not a mixed-spelling conflict.
r3 = http.get(
api_base + "/api/assets",
params={"include_tags": f"unit-tests,{scope}", "tags_all": ""},
timeout=120,
)
b3 = r3.json()
assert r3.status_code == 200, b3
assert {x["name"] for x in b3["assets"]} == expected
def test_list_assets_old_names_match_new_names(http, api_base, asset_factory, make_asset_bytes):
scope = f"lf-alias-{uuid.uuid4().hex[:6]}"
t = ["models", "model_type:checkpoints", "unit-tests", scope]
alpha, beta = f"{scope}-alpha", f"{scope}-beta"
asset_factory("al_a.safetensors", [*t, alpha], {}, make_asset_bytes("al_a"))
asset_factory("al_b.safetensors", [*t, beta], {}, make_asset_bytes("al_b"))
def names_for(params: dict) -> tuple[list, int]:
r = http.get(api_base + "/api/assets", params={**params, "sort": "name", "order": "asc"}, timeout=120)
body = r.json()
assert r.status_code == 200, body
return [x["name"] for x in body["assets"]], body["total"]
# include_tags ≡ tags_all
old_names, old_total = names_for({"include_tags": f"unit-tests,{scope}"})
new_names, new_total = names_for({"tags_all": f"unit-tests,{scope}"})
assert old_names == new_names
assert old_total == new_total
# exclude_tags ≡ tags_none (and old/new spellings mix across slots)
old_names, old_total = names_for({"include_tags": f"unit-tests,{scope}", "exclude_tags": alpha})
new_names, new_total = names_for({"tags_all": f"unit-tests,{scope}", "tags_none": alpha})
mixed_names, mixed_total = names_for({"include_tags": f"unit-tests,{scope}", "tags_none": alpha})
assert old_names == new_names == mixed_names == ["al_b.safetensors"]
assert old_total == new_total == mixed_total == 1
@pytest.mark.parametrize(
"params,expected_parameters",
[
({"include_tags": "mx-x", "tags_all": "mx-y"}, ["include_tags", "tags_all"]),
({"exclude_tags": "mx-x", "tags_none": "mx-y"}, ["exclude_tags", "tags_none"]),
],
ids=["include_tags_with_tags_all", "exclude_tags_with_tags_none"],
)
def test_list_assets_mixed_tag_spellings_rejected(http, api_base, params, expected_parameters):
r = http.get(api_base + "/api/assets", params=params, timeout=120)
body = r.json()
assert r.status_code == 400, body
assert body["error"]["code"] == "INVALID_TAG_FILTER"
assert body["error"]["details"]["parameters"] == expected_parameters
@pytest.mark.parametrize(
"params,conflicting,parameters",
[
(
{"tags_all": "cf-x", "tags_none": "cf-x"},
["cf-x"],
["tags_all", "tags_none"],
),
(
{"include_tags": "cf-x", "tags_none": "cf-x"},
["cf-x"],
["include_tags", "tags_none"],
),
(
{"tags_all": "cf-a,cf-b", "tags_none": "cf-b,cf-c"},
["cf-b"],
["tags_all", "tags_none"],
),
],
ids=["new_names", "include_tags_remapped", "partial_overlap"],
)
def test_list_assets_all_none_conflict_rejected(http, api_base, params, conflicting, parameters):
r = http.get(api_base + "/api/assets", params=params, timeout=120)
body = r.json()
assert r.status_code == 400, body
assert body["error"]["code"] == "INVALID_TAG_FILTER"
assert body["error"]["details"]["conflicting_tags"] == conflicting
assert body["error"]["details"]["parameters"] == parameters
def test_list_assets_any_none_overlap_accepted(http, api_base, asset_factory, make_asset_bytes):
scope = f"lf-deadterm-{uuid.uuid4().hex[:6]}"
t = ["models", "model_type:checkpoints", "unit-tests", scope]
alpha, beta = f"{scope}-alpha", f"{scope}-beta"
x = asset_factory("dt_x.safetensors", [*t, alpha], {}, make_asset_bytes("dt_x"))
y = asset_factory("dt_y.safetensors", [*t, beta], {}, make_asset_bytes("dt_y"))
# alpha is a dead term (in both tags_any and tags_none) but the query is valid.
r = http.get(
api_base + "/api/assets",
params={"tags_any": f"{alpha},{beta}", "tags_none": alpha, "limit": "50"},
timeout=120,
)
body = r.json()
assert r.status_code == 200, body
names = [a["name"] for a in body["assets"]]
assert y["name"] in names
assert x["name"] not in names
def test_list_assets_legacy_include_exclude_conflict_still_200(http, api_base, asset_factory, make_asset_bytes):
scope = f"lf-legacy-{uuid.uuid4().hex[:6]}"
t = ["models", "model_type:checkpoints", "unit-tests", scope]
asset_factory("lg_a.safetensors", t, {}, make_asset_bytes("lg_a"))
# Old names only: the self-contradictory query stays an empty 200, never a 400.
r = http.get(
api_base + "/api/assets",
params={"include_tags": scope, "exclude_tags": scope},
timeout=120,
)
body = r.json()
assert r.status_code == 200, body
assert body["assets"] == []
def test_tags_refine_new_tag_filters(http, api_base, asset_factory, make_asset_bytes):
scope = f"rf-{uuid.uuid4().hex[:6]}"
t = ["models", "model_type:checkpoints", "unit-tests", scope]
alpha, beta = f"{scope}-alpha", f"{scope}-beta"
asset_factory("rf_a.safetensors", [*t, alpha], {}, make_asset_bytes("rf_a"))
asset_factory("rf_b.safetensors", [*t, beta], {}, make_asset_bytes("rf_b"))
r = http.get(
api_base + "/api/assets/tags/refine",
params={"tags_any": f"{alpha},{beta}", "tags_none": alpha},
timeout=120,
)
body = r.json()
assert r.status_code == 200, body
counts = body["tag_counts"]
assert counts.get(beta) == 1
assert alpha not in counts
r2 = http.get(
api_base + "/api/assets/tags/refine",
params={"tags_all": "rf-x", "tags_none": "rf-x"},
timeout=120,
)
body2 = r2.json()
assert r2.status_code == 400, body2
assert body2["error"]["code"] == "INVALID_TAG_FILTER"
assert body2["error"]["details"]["conflicting_tags"] == ["rf-x"]
def test_list_assets_cross_slot_old_new_combinations(http, api_base, asset_factory, make_asset_bytes):
"""Old and new spellings of *different* slots combine freely; only
same-slot mixing is rejected."""
scope = f"lf-cross-{uuid.uuid4().hex[:6]}"
t = ["models", "model_type:checkpoints", "unit-tests", scope]
alpha, beta = f"{scope}-alpha", f"{scope}-beta"
a = asset_factory("cs_a.safetensors", [*t, alpha], {}, make_asset_bytes("cs_a"))
b = asset_factory("cs_b.safetensors", [*t, beta], {}, make_asset_bytes("cs_b"))
def names_for(params: dict) -> set:
r = http.get(api_base + "/api/assets", params=params, timeout=120)
body = r.json()
assert r.status_code == 200, body
return {x["name"] for x in body["assets"]}
assert names_for(
{"include_tags": f"unit-tests,{scope}", "tags_any": alpha}
) == {a["name"]}
assert names_for(
{"tags_all": f"unit-tests,{scope}", "exclude_tags": alpha}
) == {b["name"]}
assert names_for(
{"tags_any": f"{alpha},{beta}", "exclude_tags": alpha}
) == {b["name"]}
def test_list_assets_repeated_query_keys_concatenate(http, api_base, asset_factory, make_asset_bytes):
"""Repeated occurrences of a tag param concatenate before the CSV split
(Core-local behavior, not a cross-platform guarantee)."""
scope = f"lf-repeat-{uuid.uuid4().hex[:6]}"
t = ["models", "model_type:checkpoints", "unit-tests", scope]
alpha, beta = f"{scope}-alpha", f"{scope}-beta"
a = asset_factory("rp_a.safetensors", [*t, alpha], {}, make_asset_bytes("rp_a"))
b = asset_factory("rp_b.safetensors", [*t, beta], {}, make_asset_bytes("rp_b"))
# requests encodes a list value as repeated keys: tags_any=<alpha>&tags_any=<beta>
r = http.get(
api_base + "/api/assets",
params={"tags_any": [alpha, beta], "limit": "50"},
timeout=120,
)
body = r.json()
assert r.status_code == 200, body
names = {x["name"] for x in body["assets"]}
assert {a["name"], b["name"]} <= names
def test_list_assets_tags_any_cursor_pagination_consistent(http, api_base, asset_factory, make_asset_bytes):
scope = f"lf-anypage-{uuid.uuid4().hex[:6]}"
t = ["models", "model_type:checkpoints", "unit-tests", scope]
alpha = f"{scope}-alpha"
expected = set()
for i in range(3):
made = asset_factory(f"pg_{i}.safetensors", [*t, alpha], {}, make_asset_bytes(f"pg_{i}"))
expected.add(made["name"])
r1 = http.get(
api_base + "/api/assets",
params={"tags_any": alpha, "limit": "2", "sort": "name", "order": "asc"},
timeout=120,
)
b1 = r1.json()
assert r1.status_code == 200, b1
assert b1["total"] == 3
assert b1["has_more"] is True
assert b1.get("next_cursor"), "expected a keyset cursor on the first page"
r2 = http.get(
api_base + "/api/assets",
params={
"tags_any": alpha,
"limit": "2",
"sort": "name",
"order": "asc",
"after": b1["next_cursor"],
},
timeout=120,
)
b2 = r2.json()
assert r2.status_code == 200, b2
assert b2["has_more"] is False
page1 = {x["name"] for x in b1["assets"]}
page2 = {x["name"] for x in b2["assets"]}
assert not page1 & page2, "cursor pages must not overlap"
assert page1 | page2 == expected
def test_tags_refine_mixed_spellings_rejected_and_legacy_conflict_kept(http, api_base):
r = http.get(
api_base + "/api/assets/tags/refine",
params={"include_tags": "rfmx-x", "tags_all": "rfmx-y"},
timeout=120,
)
body = r.json()
assert r.status_code == 400, body
assert body["error"]["code"] == "INVALID_TAG_FILTER"
assert body["error"]["details"]["parameters"] == ["include_tags", "tags_all"]
# Old names only: the refine route keeps legacy behaviour too.
r2 = http.get(
api_base + "/api/assets/tags/refine",
params={"include_tags": "rfmx-z", "exclude_tags": "rfmx-z"},
timeout=120,
)
body2 = r2.json()
assert r2.status_code == 200, body2
assert body2["tag_counts"] == {}
def test_list_assets_tag_values_case_sensitive(http, api_base, asset_factory, make_asset_bytes):
"""Case-distinct tags are distinct; the all/none conflict check is byte-exact."""
scope = f"lf-case-{uuid.uuid4().hex[:6]}"
t = ["models", "model_type:checkpoints", "unit-tests", scope]
upper, lower = f"{scope}-ALPHA", f"{scope}-alpha"
a = asset_factory("cx_a.safetensors", [*t, upper], {}, make_asset_bytes("cx_a"))
b = asset_factory("cx_b.safetensors", [*t, lower], {}, make_asset_bytes("cx_b"))
def names_for(params: dict) -> set:
r = http.get(api_base + "/api/assets", params=params, timeout=120)
body = r.json()
assert r.status_code == 200, body
return {x["name"] for x in body["assets"]}
assert names_for({"tags_all": f"unit-tests,{scope},{upper}"}) == {a["name"]}
assert names_for({"tags_any": lower, "limit": "50"}) == {b["name"]}
# Case-distinct all/none pair is NOT a conflict — byte-exact comparison.
assert names_for({"tags_all": f"unit-tests,{scope},{upper}", "tags_none": lower}) == {a["name"]}
def test_tag_list_cap_applies_to_all_spellings(http, api_base):
"""The cap covers the legacy spellings too."""
big = ",".join(f"cap-{i}" for i in range(101))
for param in ("tags_any", "include_tags"):
r = http.get(api_base + "/api/assets", params={param: big}, timeout=120)
body = r.json()
assert r.status_code == 400, body
assert body["error"]["code"] == "INVALID_TAG_FILTER"
assert body["error"]["details"]["parameter"] == param
assert body["error"]["details"]["max"] == 100
exact = ",".join(f"cap-{i}" for i in range(100))
r = http.get(api_base + "/api/assets", params={"tags_any": exact}, timeout=120)
assert r.status_code == 200, r.json()
# The cap counts normalized (deduped) tags, not raw CSV items.
dups = ",".join("cap-dup" for _ in range(150))
r = http.get(api_base + "/api/assets", params={"tags_any": dups}, timeout=120)
assert r.status_code == 200, r.json()
def test_resolve_tag_filters_no_deprecation_warning():
"""The deprecated-field warning is for API clients; the server's own remap
shim must not fire it on every request."""
for q in (
schemas_in.ListAssetsQuery(tags_all="a", tags_none="b"),
schemas_in.TagsRefineQuery(tags_any="c"),
):
with warnings.catch_warnings():
warnings.simplefilter("error", DeprecationWarning)
assets_routes._resolve_tag_filters(q)
def test_tag_filter_alias_fields_marked_deprecated():
for model in (schemas_in.ListAssetsQuery, schemas_in.TagsRefineQuery):
props = model.model_json_schema()["properties"]
for field in ("include_tags", "exclude_tags"):
assert props[field].get("deprecated") is True, (model.__name__, field)
for field in ("tags_all", "tags_any", "tags_none"):
assert "deprecated" not in props[field], (model.__name__, field)

216
tests-unit/assets_test/test_preview_url.py Normal file
View File

@@ -0,0 +1,216 @@
import contextlib
import json
import re
import uuid
import requests
def test_preview_url_serves_the_asset(
http: requests.Session, api_base: str, asset_factory, make_asset_bytes
):
scope = f"preview-url-{uuid.uuid4().hex[:6]}"
name = f"{scope}.png"
data = make_asset_bytes(name, 2048)
body = asset_factory(name, ["output", "unit-tests", scope], {}, data)
assert re.fullmatch(r"/api/view\?type=output&filename=[^&]+", body["preview_url"]), (
f"unexpected preview URL shape: {body['preview_url']!r}"
)
r = http.get(api_base + body["preview_url"], timeout=120)
assert r.status_code == 200, r.text
assert r.content == data, "the preview URL must serve the asset's own bytes"
def test_preview_url_honours_range_requests(
http: requests.Session, api_base: str, asset_factory, make_asset_bytes
):
scope = f"preview-range-{uuid.uuid4().hex[:6]}"
name = f"{scope}.png"
data = make_asset_bytes(name, 2048)
body = asset_factory(name, ["output", "unit-tests", scope], {}, data)
r = http.get(
api_base + body["preview_url"], headers={"Range": "bytes=10-109"}, timeout=120
)
assert r.status_code == 206, (
f"expected a partial response, got {r.status_code}: native <video>/<audio> "
f"seek by byte range and will not play a source that ignores it"
)
assert r.content == data[10:110], "the served range must be the requested one"
def test_preview_url_needs_no_user_header(
api_base: str, asset_factory, make_asset_bytes
):
scope = f"preview-anon-{uuid.uuid4().hex[:6]}"
name = f"{scope}.png"
data = make_asset_bytes(name, 1024)
body = asset_factory(name, ["input", "unit-tests", scope], {}, data)
with requests.Session() as bare:
r = bare.get(api_base + body["preview_url"], timeout=120)
assert r.status_code == 200, (
f"a browser fetching <img src> cannot attach a Comfy-User header, so the "
f"preview URL must resolve without one; got {r.status_code}: {r.text}"
)
assert r.content == data
def test_preview_url_survives_tag_removal(
http: requests.Session, api_base: str, asset_factory, make_asset_bytes
):
scope = f"preview-tags-{uuid.uuid4().hex[:6]}"
name = f"{scope}.png"
data = make_asset_bytes(name, 2048)
body = asset_factory(name, ["input", "unit-tests", scope], {}, data)
aid = body["id"]
preview_url = body["preview_url"]
assert preview_url, "an uploaded image starts out with a preview"
r = http.delete(
f"{api_base}/api/assets/{aid}/tags", json={"tags": ["input"]}, timeout=120
)
assert r.status_code == 200, r.text
after = http.get(f"{api_base}/api/assets/{aid}", timeout=120).json()
assert "input" not in after["tags"]
assert after["preview_url"] == preview_url, (
"dropping the tag that used to select the view type must not take the "
"preview with it"
)
assert http.get(api_base + preview_url, timeout=120).status_code == 200
def test_preview_url_is_the_nominated_preview_when_one_is_set(
http: requests.Session, api_base: str, asset_factory, make_asset_bytes
):
scope = f"preview-id-{uuid.uuid4().hex[:6]}"
thumb_name = f"{scope}_thumb.png"
thumb_data = make_asset_bytes(thumb_name, 1024)
thumb = asset_factory(thumb_name, ["input", "unit-tests", scope], {}, thumb_data)
model_name = f"{scope}.safetensors"
files = {"file": (model_name, make_asset_bytes(model_name, 2048), "application/octet-stream")}
form_data = {
"tags": json.dumps(["models", "model_type:checkpoints", "unit-tests", scope]),
"name": model_name,
"preview_id": thumb["id"],
}
r = http.post(api_base + "/api/assets", files=files, data=form_data, timeout=120)
model = r.json()
assert r.status_code in (200, 201), model
try:
assert model["preview_id"] == thumb["id"]
assert model["preview_url"] == thumb["preview_url"], (
"a nominated preview stands in for content with no visual form"
)
got = http.get(api_base + model["preview_url"], timeout=120)
assert got.status_code == 200, got.text
assert got.content == thumb_data
finally:
with contextlib.suppress(Exception):
http.delete(f"{api_base}/api/assets/{model['id']}", timeout=30)
def test_soft_deleted_preview_is_not_advertised(
http: requests.Session, api_base: str, asset_factory, make_asset_bytes
):
scope = f"preview-gone-{uuid.uuid4().hex[:6]}"
thumb_name = f"{scope}_thumb.png"
thumb = asset_factory(
thumb_name, ["input", "unit-tests", scope], {}, make_asset_bytes(thumb_name, 1024)
)
parent_name = f"{scope}_parent.png"
parent = asset_factory(
parent_name, ["input", "unit-tests", scope], {}, make_asset_bytes(parent_name, 1024)
)
r = http.put(
f"{api_base}/api/assets/{parent['id']}",
json={"preview_id": thumb["id"]},
timeout=120,
)
assert r.status_code == 200, r.text
assert r.json()["preview_url"] == thumb["preview_url"]
assert http.delete(f"{api_base}/api/assets/{thumb['id']}", timeout=30).status_code in (200, 204)
after = http.get(f"{api_base}/api/assets/{parent['id']}", timeout=120).json()
assert after["preview_id"] == thumb["id"]
assert after.get("preview_url") is None, (
"soft delete leaves the parent's preview_id pointing at it, so the "
"response has to drop the URL rather than promise one that 404s"
)
def test_no_preview_url_for_content_a_browser_cannot_render(
http: requests.Session, api_base: str, asset_factory, make_asset_bytes
):
scope = f"preview-model-{uuid.uuid4().hex[:6]}"
name = f"{scope}.safetensors"
body = asset_factory(
name,
["models", "model_type:checkpoints", "unit-tests", scope],
{},
make_asset_bytes(name, 2048),
)
assert body.get("preview_url") is None, (
"model weights have no preview of their own"
)
listed = http.get(
api_base + "/api/assets", params={"include_tags": scope}, timeout=120
).json()["assets"]
assert [a.get("preview_url") for a in listed] == [None], (
"the list route must withhold it too, not just the detail route"
)
def test_text_asset_gets_a_preview_url_that_serves_its_content(
http: requests.Session, api_base: str, asset_factory, make_asset_bytes
):
scope = f"preview-text-{uuid.uuid4().hex[:6]}"
name = f"{scope}.txt"
data = b"line one\nline two\n" + make_asset_bytes(name, 256)
body = asset_factory(name, ["output", "unit-tests", scope], {}, data)
assert body.get("preview_url"), (
"text assets are rendered as a snippet fetched from preview_url"
)
r = http.get(api_base + body["preview_url"], timeout=120)
assert r.status_code == 200, r.text
assert r.content == data
def test_dangerous_text_preview_is_still_forced_to_download(
http: requests.Session, api_base: str, asset_factory
):
scope = f"preview-html-{uuid.uuid4().hex[:6]}"
name = f"{scope}.html"
data = f"<html><script>alert('{scope}')</script></html>".encode()
body = asset_factory(name, ["output", "unit-tests", scope], {}, data)
assert body.get("preview_url"), "text/html matches the previewable prefix"
r = http.get(api_base + body["preview_url"], timeout=120)
r.content
assert r.status_code == 200
ct = r.headers.get("Content-Type", "").lower()
cd = r.headers.get("Content-Disposition", "").lower()
assert ct.startswith("application/octet-stream"), (
f"admitting text/ to the previewable set must not let HTML render "
f"inline in the app origin; got {ct!r}"
)
assert "attachment" in cd, f"expected a forced download, got {cd!r}"

28
tests-unit/assets_test/test_sync_references.py
View File

@@ -150,7 +150,7 @@ def test_needs_verify_toggling(session, temp_dir, case):
)
session.commit()
with patch("app.assets.scanner.get_prefixes_for_root", return_value=[str(temp_dir)]):
with patch("app.assets.scanner.get_scan_prefixes_for_root", return_value=[str(temp_dir)]):
sync_references_with_filesystem(session, "models")
session.commit()
@@ -185,7 +185,7 @@ def test_is_missing_flag(session, temp_dir, case):
_make_asset(session, "a1", fp, "r1", asset_hash="blake3:abc", mtime_ns=mtime)
session.commit()
with patch("app.assets.scanner.get_prefixes_for_root", return_value=[str(temp_dir)]):
with patch("app.assets.scanner.get_scan_prefixes_for_root", return_value=[str(temp_dir)]):
sync_references_with_filesystem(session, "models")
session.commit()
@@ -200,7 +200,7 @@ def test_seed_asset_all_missing_deletes_asset(session, temp_dir):
_make_asset(session, "seed1", fp, "r1", asset_hash=None, mtime_ns=999)
session.commit()
with patch("app.assets.scanner.get_prefixes_for_root", return_value=[str(temp_dir)]):
with patch("app.assets.scanner.get_scan_prefixes_for_root", return_value=[str(temp_dir)]):
sync_references_with_filesystem(session, "models")
session.commit()
@@ -215,7 +215,7 @@ def test_seed_asset_some_exist_returns_survivors(session, temp_dir):
_make_asset(session, "seed1", fp, "r1", asset_hash=None, mtime_ns=mtime)
session.commit()
with patch("app.assets.scanner.get_prefixes_for_root", return_value=[str(temp_dir)]):
with patch("app.assets.scanner.get_scan_prefixes_for_root", return_value=[str(temp_dir)]):
survivors = sync_references_with_filesystem(
session, "models", collect_existing_paths=True,
)
@@ -240,7 +240,7 @@ def test_hashed_asset_prunes_missing_refs_when_one_is_ok(session, temp_dir):
session.add(ref_gone)
session.commit()
with patch("app.assets.scanner.get_prefixes_for_root", return_value=[str(temp_dir)]):
with patch("app.assets.scanner.get_scan_prefixes_for_root", return_value=[str(temp_dir)]):
sync_references_with_filesystem(session, "models")
session.commit()
@@ -255,7 +255,7 @@ def test_hashed_asset_all_missing_keeps_refs(session, temp_dir):
_make_asset(session, "h1", fp, "r1", asset_hash="blake3:aaa", mtime_ns=999)
session.commit()
with patch("app.assets.scanner.get_prefixes_for_root", return_value=[str(temp_dir)]):
with patch("app.assets.scanner.get_scan_prefixes_for_root", return_value=[str(temp_dir)]):
sync_references_with_filesystem(session, "models")
session.commit()
@@ -272,7 +272,7 @@ def test_missing_tag_added_when_all_refs_gone(session, temp_dir):
_make_asset(session, "h1", fp, "r1", asset_hash="blake3:aaa", mtime_ns=999)
session.commit()
with patch("app.assets.scanner.get_prefixes_for_root", return_value=[str(temp_dir)]):
with patch("app.assets.scanner.get_scan_prefixes_for_root", return_value=[str(temp_dir)]):
sync_references_with_filesystem(
session, "models", update_missing_tags=True,
)
@@ -295,7 +295,7 @@ def test_missing_tag_removed_when_ref_ok(session, temp_dir):
))
session.commit()
with patch("app.assets.scanner.get_prefixes_for_root", return_value=[str(temp_dir)]):
with patch("app.assets.scanner.get_scan_prefixes_for_root", return_value=[str(temp_dir)]):
sync_references_with_filesystem(
session, "models", update_missing_tags=True,
)
@@ -313,7 +313,7 @@ def test_missing_tags_not_touched_when_flag_false(session, temp_dir):
_make_asset(session, "h1", fp, "r1", asset_hash="blake3:aaa", mtime_ns=999)
session.commit()
with patch("app.assets.scanner.get_prefixes_for_root", return_value=[str(temp_dir)]):
with patch("app.assets.scanner.get_scan_prefixes_for_root", return_value=[str(temp_dir)]):
sync_references_with_filesystem(
session, "models", update_missing_tags=False,
)
@@ -329,7 +329,7 @@ def test_returns_none_when_collect_false(session, temp_dir):
_make_asset(session, "a1", fp, "r1", asset_hash="blake3:abc", mtime_ns=mtime)
session.commit()
with patch("app.assets.scanner.get_prefixes_for_root", return_value=[str(temp_dir)]):
with patch("app.assets.scanner.get_scan_prefixes_for_root", return_value=[str(temp_dir)]):
result = sync_references_with_filesystem(
session, "models", collect_existing_paths=False,
)
@@ -338,7 +338,7 @@ def test_returns_none_when_collect_false(session, temp_dir):
def test_returns_empty_set_for_no_prefixes(session):
with patch("app.assets.scanner.get_prefixes_for_root", return_value=[]):
with patch("app.assets.scanner.get_scan_prefixes_for_root", return_value=[]):
result = sync_references_with_filesystem(
session, "models", collect_existing_paths=True,
)
@@ -348,7 +348,7 @@ def test_returns_empty_set_for_no_prefixes(session):
def test_no_references_is_noop(session, temp_dir):
"""No crash and no side effects when there are no references."""
with patch("app.assets.scanner.get_prefixes_for_root", return_value=[str(temp_dir)]):
with patch("app.assets.scanner.get_scan_prefixes_for_root", return_value=[str(temp_dir)]):
survivors = sync_references_with_filesystem(
session, "models", collect_existing_paths=True,
)
@@ -388,7 +388,7 @@ def test_sync_does_not_resurrect_soft_deleted_ref(session, temp_dir):
_soft_delete_ref(session, "r1")
session.commit()
with patch("app.assets.scanner.get_prefixes_for_root", return_value=[str(temp_dir)]):
with patch("app.assets.scanner.get_scan_prefixes_for_root", return_value=[str(temp_dir)]):
sync_references_with_filesystem(session, "models")
session.commit()
@@ -472,7 +472,7 @@ def test_sync_ignores_soft_deleted_seed_asset(session, temp_dir):
_soft_delete_ref(session, "r1")
session.commit()
with patch("app.assets.scanner.get_prefixes_for_root", return_value=[str(temp_dir)]):
with patch("app.assets.scanner.get_scan_prefixes_for_root", return_value=[str(temp_dir)]):
sync_references_with_filesystem(session, "models")
session.commit()

180
tests-unit/assets_test/test_temp_assets.py Normal file
View File

@@ -0,0 +1,180 @@
import os
import tempfile
from pathlib import Path
from unittest.mock import patch
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from app.assets.database.models import Asset, AssetReference, Base
from app.assets.database.queries.asset_reference import (
mark_references_missing_outside_prefixes,
)
from app.assets.scanner import (
collect_paths_for_roots,
get_owned_prefixes,
get_temp_prefixes,
sync_prefixes_with_filesystem,
)
from app.assets.services.file_utils import get_mtime_ns
@pytest.fixture(autouse=True)
def autoclean_unit_test_assets():
"""Override parent autouse fixture - temp asset tests don't need server cleanup."""
yield
@pytest.fixture
def session():
engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)
with Session(engine) as sess:
yield sess
@pytest.fixture
def comfy_dirs():
with tempfile.TemporaryDirectory() as base:
dirs = {
name: Path(base) / name
for name in ("models", "input", "output", "temp", "elsewhere")
}
for d in dirs.values():
d.mkdir()
with (
patch("folder_paths.get_input_directory", return_value=str(dirs["input"])),
patch("folder_paths.get_output_directory", return_value=str(dirs["output"])),
patch("folder_paths.get_temp_directory", return_value=str(dirs["temp"])),
patch(
"app.assets.scanner.get_comfy_models_folders",
return_value=[("checkpoints", [str(dirs["models"])], set())],
),
):
yield dirs
def _write(directory: Path, name: str) -> str:
p = directory / name
p.write_bytes(b"\x00" * 100)
return str(p)
def _register(
session: Session,
file_path: str,
ref_id: str,
*,
mtime_ns: int,
asset_hash: str | None = "",
) -> None:
if asset_hash == "":
asset_hash = f"blake3:{ref_id}"
session.add(Asset(id=f"asset-{ref_id}", hash=asset_hash, size_bytes=100))
session.flush()
session.add(
AssetReference(
id=ref_id,
asset_id=f"asset-{ref_id}",
name=os.path.basename(file_path),
owner_id="",
file_path=file_path,
mtime_ns=mtime_ns,
)
)
session.flush()
def _mtime(path: str) -> int:
return get_mtime_ns(os.stat(path, follow_symlinks=True))
def test_owned_prefixes_include_temp(comfy_dirs):
owned = get_owned_prefixes()
assert str(comfy_dirs["temp"]) in owned, (
"temp must be owned, or the prune disowns assets whose files are present"
)
for name in ("models", "input", "output"):
assert str(comfy_dirs[name]) in owned, f"{name} must stay owned"
def test_discovery_does_not_walk_temp(comfy_dirs):
temp_file = _write(comfy_dirs["temp"], "preview.png")
output_file = _write(comfy_dirs["output"], "render.png")
with patch("app.assets.scanner.collect_models_files", return_value=[]):
paths = collect_paths_for_roots(("models", "input", "output"))
assert output_file in paths, "scan roots must still be walked"
assert temp_file not in paths, (
"temp is wiped before every scan, so walking it only ever finds nothing"
)
def test_prune_keeps_live_temp_reference(session, comfy_dirs):
temp_file = _write(comfy_dirs["temp"], "preview.png")
stray_file = _write(comfy_dirs["elsewhere"], "stray.png")
_register(session, temp_file, "temp-ref", mtime_ns=_mtime(temp_file))
_register(session, stray_file, "stray-ref", mtime_ns=_mtime(stray_file))
session.commit()
marked = mark_references_missing_outside_prefixes(session, get_owned_prefixes())
session.commit()
session.expire_all()
assert marked == 1, "only the reference outside every owned directory is disowned"
assert session.get(AssetReference, "temp-ref").is_missing is False, (
"a temp file on disk is not missing, however often the prune runs"
)
assert session.get(AssetReference, "stray-ref").is_missing is True, (
"owning temp must not stop the prune disowning files elsewhere"
)
def test_temp_sync_marks_deleted_file_missing(session, comfy_dirs):
temp_file = _write(comfy_dirs["temp"], "preview.png")
_register(session, temp_file, "temp-ref", mtime_ns=_mtime(temp_file))
session.commit()
os.remove(temp_file)
sync_prefixes_with_filesystem(session, get_temp_prefixes())
session.commit()
session.expire_all()
assert session.get(AssetReference, "temp-ref").is_missing is True, (
"nothing else stats temp, so this pass is what retires a wiped file"
)
def test_temp_sync_drops_unhashed_asset_whose_file_is_gone(session, comfy_dirs):
temp_file = _write(comfy_dirs["temp"], "preview.png")
_register(session, temp_file, "temp-ref", mtime_ns=_mtime(temp_file), asset_hash=None)
session.commit()
os.remove(temp_file)
sync_prefixes_with_filesystem(session, get_temp_prefixes())
session.commit()
session.expire_all()
assert session.get(AssetReference, "temp-ref") is None, (
"an unhashed asset with no surviving reference is retired, not kept as missing"
)
assert session.get(Asset, "asset-temp-ref") is None, (
"the orphaned asset row goes with its last reference"
)
def test_temp_sync_keeps_live_file(session, comfy_dirs):
temp_file = _write(comfy_dirs["temp"], "preview.png")
_register(session, temp_file, "temp-ref", mtime_ns=_mtime(temp_file))
session.commit()
sync_prefixes_with_filesystem(session, get_temp_prefixes())
session.commit()
session.expire_all()
ref = session.get(AssetReference, "temp-ref")
assert ref.is_missing is False, "the file is still there"
assert ref.needs_verify is False, "an unchanged file needs no re-verification"