mirror of
https://github.com/comfyanonymous/ComfyUI.git
synced 2026-06-21 23:39:35 +08:00
990242a07d
Some checks failed
Python Linting / Run Ruff (push) Has been cancelled
Python Linting / Run Pylint (push) Has been cancelled
Build package / Build Test (3.10) (push) Has been cancelled
Build package / Build Test (3.11) (push) Has been cancelled
Build package / Build Test (3.12) (push) Has been cancelled
Build package / Build Test (3.13) (push) Has been cancelled
Build package / Build Test (3.14) (push) Has been cancelled
Amp-Thread-ID: https://ampcode.com/threads/T-019eddf6-336e-71db-b94b-655afe52505e Co-authored-by: Amp <amp@ampcode.com>
659 lines
25 KiB
Python
659 lines
25 KiB
Python
import json
|
|
import uuid
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
import pytest
|
|
|
|
from app.assets.api.schemas_out import Asset, AssetCreated
|
|
from helpers import get_asset_filename
|
|
|
|
|
|
def test_asset_created_inherits_hash_field():
|
|
"""AssetCreated must inherit `hash` from Asset so POST /api/assets responses emit it.
|
|
|
|
Schema-level guard: integration tests cover the wire shape, but inheritance
|
|
drift (e.g. AssetCreated ever being redefined to no longer extend Asset)
|
|
would silently drop `hash` from a major endpoint without this check.
|
|
"""
|
|
assert "hash" in Asset.model_fields
|
|
assert "hash" in AssetCreated.model_fields
|
|
assert AssetCreated.model_fields["hash"].annotation == Asset.model_fields["hash"].annotation
|
|
|
|
|
|
def test_upload_ok_duplicate_reference(http: requests.Session, api_base: str, make_asset_bytes):
|
|
name = "dup_a.safetensors"
|
|
tags = ["models", "model_type:checkpoints", "unit-tests", "alpha"]
|
|
meta = {"purpose": "dup"}
|
|
data = make_asset_bytes(name)
|
|
files = {"file": (name, data, "application/octet-stream")}
|
|
form = {"tags": json.dumps(tags), "name": name, "user_metadata": json.dumps(meta)}
|
|
r1 = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
a1 = r1.json()
|
|
assert r1.status_code == 201, a1
|
|
assert a1["created_new"] is True
|
|
assert a1["hash"] == a1["asset_hash"]
|
|
|
|
# Second upload with the same data and name creates a new AssetReference (duplicates allowed)
|
|
# Returns 200 because Asset already exists, but a new AssetReference is created
|
|
files = {"file": (name, data, "application/octet-stream")}
|
|
form = {"tags": json.dumps(tags), "name": name, "user_metadata": json.dumps(meta)}
|
|
r2 = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
a2 = r2.json()
|
|
assert r2.status_code in (200, 201), a2
|
|
assert a2["asset_hash"] == a1["asset_hash"]
|
|
assert a2["hash"] == a1["hash"]
|
|
assert a2["id"] != a1["id"] # new reference with same content
|
|
assert a2.get("file_path") is None
|
|
assert a2.get("display_name") is None
|
|
|
|
# Third upload with the same data but different name also creates new AssetReference
|
|
files = {"file": (name, data, "application/octet-stream")}
|
|
form = {"tags": json.dumps(tags), "name": name + "_d", "user_metadata": json.dumps(meta)}
|
|
r3 = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
a3 = r3.json()
|
|
assert r3.status_code in (200, 201), a3
|
|
assert a3["asset_hash"] == a1["asset_hash"]
|
|
assert a3["id"] != a1["id"]
|
|
assert a3["id"] != a2["id"]
|
|
assert a3.get("file_path") is None
|
|
assert a3.get("display_name") is None
|
|
|
|
|
|
def test_upload_fastpath_from_existing_hash_no_file(http: requests.Session, api_base: str):
|
|
# Seed a small file first
|
|
name = "fastpath_seed.safetensors"
|
|
tags = ["input", "unit-tests"]
|
|
meta = {}
|
|
files = {"file": (name, b"B" * 1024, "application/octet-stream")}
|
|
form = {"tags": json.dumps(tags), "name": name, "user_metadata": json.dumps(meta)}
|
|
r1 = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
b1 = r1.json()
|
|
assert r1.status_code == 201, b1
|
|
h = b1["asset_hash"]
|
|
assert b1["hash"] == h
|
|
|
|
# Now POST /api/assets with only hash and no file
|
|
hash_only_tags = ["models", "checkpoints", "unit-tests", "hash-labels"]
|
|
files = [
|
|
("hash", (None, h)),
|
|
("tags", (None, json.dumps(hash_only_tags))),
|
|
("name", (None, "fastpath_copy.safetensors")),
|
|
("user_metadata", (None, json.dumps({"purpose": "copy"}))),
|
|
]
|
|
r2 = http.post(api_base + "/api/assets", files=files, timeout=120)
|
|
b2 = r2.json()
|
|
assert r2.status_code == 200, b2 # fast path returns 200 with created_new == False
|
|
assert b2["created_new"] is False
|
|
assert b2["asset_hash"] == h
|
|
assert b2["hash"] == h
|
|
assert "models" in b2["tags"]
|
|
assert "checkpoints" in b2["tags"]
|
|
assert "uploaded" not in b2["tags"]
|
|
assert not any(tag.startswith("model_type:") for tag in b2["tags"])
|
|
assert b2.get("file_path") is None
|
|
assert b2.get("display_name") is None
|
|
|
|
rg = http.get(f"{api_base}/api/assets/{b2['id']}", timeout=120)
|
|
detail = rg.json()
|
|
assert rg.status_code == 200, detail
|
|
assert detail.get("file_path") is None
|
|
assert detail.get("display_name") is None
|
|
|
|
|
|
def test_create_from_hash_with_model_tags_does_not_synthesize_file_path(
|
|
http: requests.Session, api_base: str
|
|
):
|
|
seed_name = "from_hash_seed.safetensors"
|
|
seed_tags = ["models", "model_type:checkpoints", "unit-tests"]
|
|
files = {"file": (seed_name, b"D" * 1024, "application/octet-stream")}
|
|
form = {
|
|
"tags": json.dumps(seed_tags),
|
|
"name": seed_name,
|
|
"user_metadata": json.dumps({}),
|
|
}
|
|
seed_r = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
seed = seed_r.json()
|
|
assert seed_r.status_code == 201, seed
|
|
|
|
payload = {
|
|
"hash": seed["asset_hash"],
|
|
"name": "from_hash_copy.safetensors",
|
|
"tags": ["models", "model_type:checkpoints", "unit-tests", "spoofed"],
|
|
}
|
|
created_r = http.post(api_base + "/api/assets/from-hash", json=payload, timeout=120)
|
|
created = created_r.json()
|
|
assert created_r.status_code == 201, created
|
|
assert created["created_new"] is False
|
|
assert created["asset_hash"] == seed["asset_hash"]
|
|
assert created.get("file_path") is None
|
|
assert created.get("display_name") is None
|
|
|
|
detail_r = http.get(f"{api_base}/api/assets/{created['id']}", timeout=120)
|
|
detail = detail_r.json()
|
|
assert detail_r.status_code == 200, detail
|
|
assert detail.get("file_path") is None
|
|
assert detail.get("display_name") is None
|
|
|
|
|
|
def test_upload_fastpath_with_known_hash_and_file(
|
|
http: requests.Session, api_base: str
|
|
):
|
|
# Seed
|
|
files = {"file": ("seed.safetensors", b"C" * 128, "application/octet-stream")}
|
|
form = {"tags": json.dumps(["models", "model_type:checkpoints", "unit-tests", "fp"]), "name": "seed.safetensors", "user_metadata": json.dumps({})}
|
|
r1 = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
b1 = r1.json()
|
|
assert r1.status_code == 201, b1
|
|
h = b1["asset_hash"]
|
|
assert b1["hash"] == h
|
|
|
|
# Send both file and hash of existing content -> server must drain file and create from hash (200)
|
|
files = {"file": ("ignored.bin", b"ignored" * 10, "application/octet-stream")}
|
|
form = {"hash": h, "tags": json.dumps(["models", "checkpoints", "unit-tests", "fp"]), "name": "copy_from_hash.safetensors", "user_metadata": json.dumps({})}
|
|
r2 = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
b2 = r2.json()
|
|
assert r2.status_code == 200, b2
|
|
assert b2["created_new"] is False
|
|
assert b2["asset_hash"] == h
|
|
assert b2["hash"] == h
|
|
assert "checkpoints" in b2["tags"]
|
|
assert "uploaded" not in b2["tags"]
|
|
assert not any(tag == "model_type:checkpoints" for tag in b2["tags"])
|
|
|
|
|
|
def test_duplicate_byte_upload_is_reference_only_and_does_not_need_destination(
|
|
http: requests.Session, api_base: str
|
|
):
|
|
data = b"duplicate-reference-only" * 64
|
|
seed_files = {"file": ("duplicate-seed.bin", data, "application/octet-stream")}
|
|
seed_form = {
|
|
"tags": json.dumps(["input", "unit-tests", "duplicate-seed"]),
|
|
"name": "duplicate-seed.bin",
|
|
"user_metadata": json.dumps({}),
|
|
}
|
|
seed_response = http.post(api_base + "/api/assets", data=seed_form, files=seed_files, timeout=120)
|
|
seed = seed_response.json()
|
|
assert seed_response.status_code == 201, seed
|
|
|
|
duplicate_files = {"file": ("duplicate-copy.bin", data, "application/octet-stream")}
|
|
duplicate_form = {
|
|
"tags": json.dumps(["not-a-destination", "unit-tests", "duplicate-copy"]),
|
|
"name": "duplicate-copy.bin",
|
|
"user_metadata": json.dumps({}),
|
|
}
|
|
duplicate_response = http.post(
|
|
api_base + "/api/assets", data=duplicate_form, files=duplicate_files, timeout=120
|
|
)
|
|
duplicate = duplicate_response.json()
|
|
|
|
assert duplicate_response.status_code == 200, duplicate
|
|
assert duplicate["created_new"] is False
|
|
assert duplicate["asset_hash"] == seed["asset_hash"]
|
|
assert "not-a-destination" in duplicate["tags"]
|
|
assert "uploaded" not in duplicate["tags"]
|
|
assert "input" not in duplicate["tags"]
|
|
assert duplicate.get("file_path") is None
|
|
assert duplicate.get("display_name") is None
|
|
|
|
|
|
def test_upload_multiple_tags_fields_are_merged(http: requests.Session, api_base: str):
|
|
data = [
|
|
("tags", "models,model_type:checkpoints"),
|
|
("tags", json.dumps(["unit-tests", "alpha"])),
|
|
("name", "merge.safetensors"),
|
|
("user_metadata", json.dumps({"u": 1})),
|
|
]
|
|
files = {"file": ("merge.safetensors", b"B" * 256, "application/octet-stream")}
|
|
r1 = http.post(api_base + "/api/assets", data=data, files=files, timeout=120)
|
|
created = r1.json()
|
|
assert r1.status_code in (200, 201), created
|
|
aid = created["id"]
|
|
|
|
# Verify all tags are present on the resource
|
|
rg = http.get(f"{api_base}/api/assets/{aid}", timeout=120)
|
|
detail = rg.json()
|
|
assert rg.status_code == 200, detail
|
|
tags = set(detail["tags"])
|
|
assert {"models", "model_type:checkpoints", "unit-tests", "alpha"}.issubset(tags)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("tags", "extension", "subfolder", "expected_prefix", "expected_display_prefix"),
|
|
[
|
|
(["input", "unit-tests"], ".png", "uploads", "input", ""),
|
|
(
|
|
["models", "model_type:checkpoints", "unit-tests"],
|
|
".safetensors",
|
|
"flux",
|
|
"models/checkpoints",
|
|
"checkpoints/",
|
|
),
|
|
],
|
|
)
|
|
def test_upload_response_includes_file_path_and_display_name(
|
|
tags: list[str],
|
|
extension: str,
|
|
subfolder: str,
|
|
expected_prefix: str,
|
|
expected_display_prefix: str,
|
|
http: requests.Session,
|
|
api_base: str,
|
|
make_asset_bytes,
|
|
):
|
|
scope = f"response-paths-{uuid.uuid4().hex[:6]}"
|
|
scoped_tags = [*tags, scope]
|
|
name = f"asset_response_path{extension}"
|
|
|
|
files = {"file": (name, make_asset_bytes(name, 1024), "application/octet-stream")}
|
|
form = {
|
|
"tags": json.dumps(scoped_tags),
|
|
"name": name,
|
|
"user_metadata": json.dumps({}),
|
|
"subfolder": subfolder,
|
|
}
|
|
created_r = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
created = created_r.json()
|
|
assert created_r.status_code in (200, 201), created
|
|
stored_filename = get_asset_filename(created["asset_hash"], extension)
|
|
expected_suffix = f"{subfolder}/{stored_filename}"
|
|
expected_file_path = f"{expected_prefix}/{expected_suffix}"
|
|
expected_display_name = f"{expected_display_prefix}{expected_suffix}"
|
|
|
|
assert created["file_path"] == expected_file_path
|
|
assert created["display_name"] == expected_display_name
|
|
|
|
detail_r = http.get(f"{api_base}/api/assets/{created['id']}", timeout=120)
|
|
detail = detail_r.json()
|
|
assert detail_r.status_code == 200, detail
|
|
assert detail["file_path"] == expected_file_path
|
|
assert detail["display_name"] == expected_display_name
|
|
|
|
list_r = http.get(
|
|
api_base + "/api/assets",
|
|
params={"include_tags": f"unit-tests,{scope}", "limit": "50"},
|
|
timeout=120,
|
|
)
|
|
listed = list_r.json()
|
|
assert list_r.status_code == 200, listed
|
|
match = next(a for a in listed["assets"] if a["id"] == created["id"])
|
|
assert match["file_path"] == expected_file_path
|
|
assert match["display_name"] == expected_display_name
|
|
|
|
|
|
@pytest.mark.parametrize("root", ["input", "output"])
|
|
def test_concurrent_upload_identical_bytes_different_names(
|
|
root: str,
|
|
http: requests.Session,
|
|
api_base: str,
|
|
make_asset_bytes,
|
|
):
|
|
"""
|
|
Two concurrent uploads of identical bytes but different names.
|
|
Expect a single Asset (same hash), two AssetReference rows, and exactly one created_new=True.
|
|
"""
|
|
scope = f"concupload-{uuid.uuid4().hex[:6]}"
|
|
name1, name2 = "cu_a.bin", "cu_b.bin"
|
|
data = make_asset_bytes("concurrent", 4096)
|
|
tags = [root, "unit-tests", scope]
|
|
|
|
def _do_upload(args):
|
|
url, form_data, files_data = args
|
|
with requests.Session() as s:
|
|
return s.post(url, data=form_data, files=files_data, timeout=120)
|
|
|
|
url = api_base + "/api/assets"
|
|
form1 = {"tags": json.dumps(tags), "name": name1, "user_metadata": json.dumps({})}
|
|
files1 = {"file": (name1, data, "application/octet-stream")}
|
|
form2 = {"tags": json.dumps(tags), "name": name2, "user_metadata": json.dumps({})}
|
|
files2 = {"file": (name2, data, "application/octet-stream")}
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as executor:
|
|
futures = list(executor.map(_do_upload, [(url, form1, files1), (url, form2, files2)]))
|
|
r1, r2 = futures
|
|
|
|
b1, b2 = r1.json(), r2.json()
|
|
assert r1.status_code in (200, 201), b1
|
|
assert r2.status_code in (200, 201), b2
|
|
assert b1["asset_hash"] == b2["asset_hash"]
|
|
assert b1["hash"] == b2["hash"]
|
|
assert b1["hash"] == b1["asset_hash"]
|
|
assert b1["id"] != b2["id"]
|
|
|
|
created_flags = sorted([bool(b1.get("created_new")), bool(b2.get("created_new"))])
|
|
assert created_flags == [False, True]
|
|
|
|
rl = http.get(
|
|
api_base + "/api/assets",
|
|
params={"include_tags": f"unit-tests,{scope}", "sort": "name"},
|
|
timeout=120,
|
|
)
|
|
bl = rl.json()
|
|
assert rl.status_code == 200, bl
|
|
names = [a["name"] for a in bl.get("assets", [])]
|
|
assert set([name1, name2]).issubset(names)
|
|
|
|
|
|
def test_create_from_hash_endpoint_404(http: requests.Session, api_base: str):
|
|
payload = {
|
|
"hash": "blake3:" + "0" * 64,
|
|
"name": "nonexistent.bin",
|
|
"tags": ["models", "checkpoints", "unit-tests"],
|
|
}
|
|
r = http.post(api_base + "/api/assets/from-hash", json=payload, timeout=120)
|
|
body = r.json()
|
|
assert r.status_code == 404
|
|
assert body["error"]["code"] == "ASSET_NOT_FOUND"
|
|
|
|
|
|
def test_create_from_hash_accepts_arbitrary_system_looking_tags(
|
|
http: requests.Session, api_base: str
|
|
):
|
|
files = {"file": ("hash-seed.bin", b"hash-seed" * 64, "application/octet-stream")}
|
|
form = {
|
|
"tags": json.dumps(["input", "unit-tests", "hash-seed"]),
|
|
"name": "hash-seed.bin",
|
|
"user_metadata": json.dumps({}),
|
|
}
|
|
seed_response = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
seed = seed_response.json()
|
|
assert seed_response.status_code == 201, seed
|
|
|
|
response = http.post(
|
|
api_base + "/api/assets/from-hash",
|
|
json={
|
|
"hash": seed["asset_hash"],
|
|
"name": "hash-copy.bin",
|
|
"tags": [
|
|
"models",
|
|
"model:true",
|
|
"models:foo",
|
|
"temporary:true",
|
|
"unit-tests",
|
|
"hash-copy",
|
|
],
|
|
},
|
|
timeout=120,
|
|
)
|
|
body = response.json()
|
|
|
|
assert response.status_code == 201, body
|
|
assert "models" in body["tags"]
|
|
assert "model:true" in body["tags"]
|
|
assert "models:foo" in body["tags"]
|
|
assert "temporary:true" in body["tags"]
|
|
assert "uploaded" not in body["tags"]
|
|
|
|
|
|
def test_upload_zero_byte_rejected(http: requests.Session, api_base: str):
|
|
files = {"file": ("empty.safetensors", b"", "application/octet-stream")}
|
|
form = {"tags": json.dumps(["models", "model_type:checkpoints", "unit-tests", "edge"]), "name": "empty.safetensors", "user_metadata": json.dumps({})}
|
|
r = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
body = r.json()
|
|
assert r.status_code == 400
|
|
assert body["error"]["code"] == "EMPTY_UPLOAD"
|
|
|
|
|
|
def test_upload_rejects_arbitrary_labels_without_required_destination_role(http: requests.Session, api_base: str):
|
|
files = {"file": ("badroot.bin", b"A" * 64, "application/octet-stream")}
|
|
form = {"tags": json.dumps(["not-a-root", "whatever"]), "name": "badroot.bin", "user_metadata": json.dumps({})}
|
|
r = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
body = r.json()
|
|
assert r.status_code == 400
|
|
assert body["error"]["code"] == "INVALID_BODY"
|
|
|
|
|
|
def test_upload_user_metadata_must_be_json(http: requests.Session, api_base: str):
|
|
files = {"file": ("badmeta.bin", b"A" * 128, "application/octet-stream")}
|
|
form = {"tags": json.dumps(["models", "model_type:checkpoints", "unit-tests", "edge"]), "name": "badmeta.bin", "user_metadata": "{not json}"}
|
|
r = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
body = r.json()
|
|
assert r.status_code == 400
|
|
assert body["error"]["code"] == "INVALID_BODY"
|
|
|
|
|
|
def test_upload_requires_multipart(http: requests.Session, api_base: str):
|
|
r = http.post(api_base + "/api/assets", json={"foo": "bar"}, timeout=120)
|
|
body = r.json()
|
|
assert r.status_code == 415
|
|
assert body["error"]["code"] == "UNSUPPORTED_MEDIA_TYPE"
|
|
|
|
|
|
def test_upload_missing_file_and_hash(http: requests.Session, api_base: str):
|
|
files = [
|
|
("tags", (None, json.dumps(["models", "model_type:checkpoints", "unit-tests"]))),
|
|
("name", (None, "x.safetensors")),
|
|
]
|
|
r = http.post(api_base + "/api/assets", files=files, timeout=120)
|
|
body = r.json()
|
|
assert r.status_code == 400
|
|
assert body["error"]["code"] == "MISSING_FILE"
|
|
|
|
|
|
def test_upload_models_unknown_model_type(http: requests.Session, api_base: str):
|
|
files = {"file": ("m.safetensors", b"A" * 128, "application/octet-stream")}
|
|
form = {"tags": json.dumps(["models", "model_type:no_such_category", "unit-tests"]), "name": "m.safetensors"}
|
|
r = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
body = r.json()
|
|
assert r.status_code == 400, body
|
|
assert body["error"]["code"] == "INVALID_BODY"
|
|
|
|
|
|
@pytest.mark.parametrize("model_type", ["configs", "custom_nodes"])
|
|
def test_upload_models_rejects_non_model_registered_folder(
|
|
model_type: str, http: requests.Session, api_base: str
|
|
):
|
|
files = {"file": ("not-a-model.py", b"A" * 128, "application/octet-stream")}
|
|
form = {
|
|
"tags": json.dumps(["models", f"model_type:{model_type}", "unit-tests"]),
|
|
"name": "not-a-model.py",
|
|
}
|
|
|
|
response = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
body = response.json()
|
|
|
|
assert response.status_code == 400, body
|
|
assert body["error"]["code"] == "INVALID_BODY"
|
|
|
|
|
|
def test_upload_models_requires_model_type(http: requests.Session, api_base: str):
|
|
files = {"file": ("nocat.safetensors", b"A" * 64, "application/octet-stream")}
|
|
form = {"tags": json.dumps(["models"]), "name": "nocat.safetensors", "user_metadata": json.dumps({})}
|
|
r = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
body = r.json()
|
|
assert r.status_code == 400
|
|
assert body["error"]["code"] == "INVALID_BODY"
|
|
|
|
|
|
def test_upload_extra_tags_are_labels_not_path_components(http: requests.Session, api_base: str):
|
|
files = {"file": ("evil.safetensors", b"A" * 256, "application/octet-stream")}
|
|
form = {"tags": json.dumps(["models", "model_type:checkpoints", "unit-tests", "..", "zzz"]), "name": "evil.safetensors"}
|
|
r = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
body = r.json()
|
|
assert r.status_code == 201, body
|
|
assert ".." in body["tags"]
|
|
assert "zzz" in body["tags"]
|
|
assert "models" in body["tags"]
|
|
assert "model_type:checkpoints" in body["tags"]
|
|
|
|
|
|
def test_upload_subfolder_is_explicit_path_component(
|
|
http: requests.Session, api_base: str, comfy_tmp_base_dir: Path
|
|
):
|
|
files = {"file": ("nested.bin", b"nested" * 64, "application/octet-stream")}
|
|
form = {
|
|
"tags": json.dumps(["input", "unit-tests", "foo"]),
|
|
"subfolder": "foo/bar",
|
|
"name": "nested.bin",
|
|
}
|
|
r = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
body = r.json()
|
|
|
|
assert r.status_code == 201, body
|
|
stored_name = get_asset_filename(body["asset_hash"], ".bin")
|
|
assert (comfy_tmp_base_dir / "input" / "foo" / "bar" / stored_name).exists()
|
|
assert "foo" in body["tags"]
|
|
|
|
|
|
def test_upload_rejects_unsafe_subfolder(http: requests.Session, api_base: str):
|
|
files = {"file": ("escape.bin", b"escape" * 64, "application/octet-stream")}
|
|
form = {
|
|
"tags": json.dumps(["input", "unit-tests"]),
|
|
"subfolder": "../escape",
|
|
"name": "escape.bin",
|
|
}
|
|
r = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
body = r.json()
|
|
|
|
assert r.status_code == 400, body
|
|
assert body["error"]["code"] == "INVALID_BODY"
|
|
|
|
|
|
def test_multipart_upload_accepts_system_looking_extra_labels(
|
|
http: requests.Session, api_base: str
|
|
):
|
|
files = {"file": ("relaxed-labels.bin", b"relaxed" * 64, "application/octet-stream")}
|
|
form = {
|
|
"tags": json.dumps(
|
|
[
|
|
"input",
|
|
"unit-tests",
|
|
"model:true",
|
|
"models:foo",
|
|
"temporary",
|
|
"uploaded:true",
|
|
]
|
|
),
|
|
"name": "relaxed-labels.bin",
|
|
"user_metadata": json.dumps({}),
|
|
}
|
|
response = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
body = response.json()
|
|
|
|
assert response.status_code == 201, body
|
|
assert "input" in body["tags"]
|
|
assert "model:true" in body["tags"]
|
|
assert "models:foo" in body["tags"]
|
|
assert "temporary" in body["tags"]
|
|
assert "uploaded:true" in body["tags"]
|
|
|
|
|
|
def test_multipart_upload_rejects_ambiguous_destination_roles(
|
|
http: requests.Session, api_base: str
|
|
):
|
|
files = {"file": ("ambiguous.bin", b"ambiguous" * 64, "application/octet-stream")}
|
|
form = {
|
|
"tags": json.dumps(["input", "output", "unit-tests"]),
|
|
"name": "ambiguous.bin",
|
|
"user_metadata": json.dumps({}),
|
|
}
|
|
response = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
body = response.json()
|
|
|
|
assert response.status_code == 400, body
|
|
assert body["error"]["code"] == "INVALID_BODY"
|
|
|
|
|
|
def test_multipart_upload_rejects_multiple_model_types_for_models_destination(
|
|
http: requests.Session, api_base: str
|
|
):
|
|
files = {"file": ("ambiguous-model.safetensors", b"ambiguous-model" * 64, "application/octet-stream")}
|
|
form = {
|
|
"tags": json.dumps(
|
|
["models", "model_type:checkpoints", "model_type:loras", "unit-tests"]
|
|
),
|
|
"name": "ambiguous-model.safetensors",
|
|
"user_metadata": json.dumps({}),
|
|
}
|
|
response = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
body = response.json()
|
|
|
|
assert response.status_code == 400, body
|
|
assert body["error"]["code"] == "INVALID_BODY"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("tags", "expected_root", "extension"),
|
|
[
|
|
(["input", "unit-tests", "upload-location-input"], "input", ".bin"),
|
|
(["output", "unit-tests", "upload-location-output"], "output", ".bin"),
|
|
(
|
|
["models", "model_type:checkpoints", "unit-tests", "upload-location-model"],
|
|
"models/checkpoints",
|
|
".safetensors",
|
|
),
|
|
],
|
|
)
|
|
def test_multipart_upload_role_selects_write_location(
|
|
http: requests.Session,
|
|
api_base: str,
|
|
comfy_tmp_base_dir: Path,
|
|
tags: list[str],
|
|
expected_root: str,
|
|
extension: str,
|
|
):
|
|
role = next(tag for tag in tags if tag in {"input", "models", "output"})
|
|
name = f"{role}-role-upload{extension}"
|
|
files = {"file": (name, f"{role}-role-bytes".encode() * 64, "application/octet-stream")}
|
|
form = {
|
|
"tags": json.dumps(tags),
|
|
"name": name,
|
|
"user_metadata": json.dumps({}),
|
|
}
|
|
|
|
response = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
body = response.json()
|
|
|
|
assert response.status_code == 201, body
|
|
stored_name = get_asset_filename(body["asset_hash"], extension)
|
|
expected_disk_path = comfy_tmp_base_dir / expected_root / stored_name
|
|
assert expected_disk_path.exists()
|
|
|
|
|
|
def test_upload_empty_tags_rejected(http: requests.Session, api_base: str):
|
|
files = {"file": ("notags.bin", b"A" * 64, "application/octet-stream")}
|
|
form = {"tags": json.dumps([]), "name": "notags.bin", "user_metadata": json.dumps({})}
|
|
r = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
|
|
body = r.json()
|
|
assert r.status_code == 400
|
|
assert body["error"]["code"] == "INVALID_BODY"
|
|
|
|
|
|
@pytest.mark.parametrize("root", ["input", "output"])
|
|
def test_duplicate_upload_same_display_name_does_not_clobber(
|
|
root: str,
|
|
http: requests.Session,
|
|
api_base: str,
|
|
asset_factory,
|
|
make_asset_bytes,
|
|
):
|
|
"""
|
|
Two uploads use the same tags and the same display name but different bytes.
|
|
With hash-based filenames, they must NOT overwrite each other. Both assets
|
|
remain accessible and serve their original content.
|
|
"""
|
|
scope = f"dup-path-{uuid.uuid4().hex[:6]}"
|
|
display_name = "same_display.bin"
|
|
|
|
d1 = make_asset_bytes(scope + "-v1", 1536)
|
|
d2 = make_asset_bytes(scope + "-v2", 2048)
|
|
tags = [root, "unit-tests", scope]
|
|
|
|
first = asset_factory(display_name, tags, {}, d1)
|
|
second = asset_factory(display_name, tags, {}, d2)
|
|
|
|
assert first["id"] != second["id"]
|
|
assert first["asset_hash"] != second["asset_hash"] # different content
|
|
assert first["name"] == second["name"] == display_name
|
|
|
|
# Both must be independently retrievable
|
|
r1 = http.get(f"{api_base}/api/assets/{first['id']}/content", timeout=120)
|
|
b1 = r1.content
|
|
assert r1.status_code == 200
|
|
assert b1 == d1
|
|
r2 = http.get(f"{api_base}/api/assets/{second['id']}/content", timeout=120)
|
|
b2 = r2.content
|
|
assert r2.status_code == 200
|
|
assert b2 == d2
|