diff --git a/app/assets/api/routes.py b/app/assets/api/routes.py index 68126b6a5..76c81032c 100644 --- a/app/assets/api/routes.py +++ b/app/assets/api/routes.py @@ -10,7 +10,6 @@ from typing import Any from aiohttp import web from pydantic import ValidationError -import folder_paths from app import user_manager from app.assets.api import schemas_in, schemas_out from app.assets.services import schemas @@ -39,6 +38,10 @@ from app.assets.services import ( update_asset_metadata, upload_from_temp_path, ) +from app.assets.services.path_utils import ( + compute_asset_response_paths, + is_comfy_model_folder_name, +) from app.assets.services.tagging import list_tag_histogram ROUTES = web.RouteTableDef() @@ -160,9 +163,16 @@ def _build_asset_response(result: schemas.AssetDetailResult | schemas.UploadResu preview_url = None else: preview_url = _build_preview_url_from_view(result.tags, result.ref.user_metadata) + if result.ref.file_path: + paths = compute_asset_response_paths(result.ref.file_path) + file_path, display_name = paths if paths else (None, None) + else: + file_path, display_name = None, None return schemas_out.Asset( id=result.ref.id, name=result.ref.name, + file_path=file_path, + display_name=display_name, asset_hash=result.asset.hash if result.asset else None, size=int(result.asset.size_bytes) if result.asset else None, mime_type=result.asset.mime_type if result.asset else None, @@ -401,10 +411,7 @@ async def upload_asset(request: web.Request) -> web.Response: ) if spec.tags and spec.tags[0] == "models": - if ( - len(spec.tags) < 2 - or spec.tags[1] not in folder_paths.folder_names_and_paths - ): + if len(spec.tags) < 2 or not is_comfy_model_folder_name(spec.tags[1]): delete_temp_file_if_exists(parsed.tmp_path) category = spec.tags[1] if len(spec.tags) >= 2 else "" return _build_error_response( diff --git a/app/assets/api/schemas_out.py b/app/assets/api/schemas_out.py index d99b1098d..79c5bd236 100644 --- a/app/assets/api/schemas_out.py +++ b/app/assets/api/schemas_out.py @@ -9,7 +9,19 @@ class Asset(BaseModel): ``id`` here is the AssetReference id, not the content-addressed Asset id.""" id: str - name: str + name: str = Field( + ..., + deprecated=True, + description="Deprecated: use `file_path` or `display_name`. Free user-chosen label with no path semantics.", + ) + file_path: str | None = Field( + default=None, + description="Logical namespace key for the asset. For files under ComfyUI registered model-folder paths, uses `models//`; not a physical path or unique identity.", + ) + display_name: str | None = Field( + default=None, + description="Human-facing display label for the asset, usually the filename relative to its root or registered model-folder name.", + ) asset_hash: str | None = None size: int | None = None mime_type: str | None = None diff --git a/app/assets/services/path_utils.py b/app/assets/services/path_utils.py index 892140ffb..c5dac295c 100644 --- a/app/assets/services/path_utils.py +++ b/app/assets/services/path_utils.py @@ -6,15 +6,21 @@ import folder_paths from app.assets.helpers import normalize_tags -_NON_MODEL_FOLDER_NAMES = frozenset({"custom_nodes"}) +# These names are bootstrapped into folder_names_and_paths by core but are not +# model folders (matching /api/experiment/models' exclusion). Intentionally +# duplicated here so the assets layer stays decoupled from the legacy +# model-manager code it will eventually replace. +_NON_MODEL_FOLDER_NAMES = frozenset({"configs", "custom_nodes"}) + +AssetRoot = Literal["input", "output", "temp", "models"] def get_comfy_models_folders() -> list[tuple[str, list[str]]]: """Build list of (folder_name, base_paths[]) for all model locations. - Includes every category registered in folder_names_and_paths, + Includes every folder name registered in folder_names_and_paths, regardless of whether its paths are under the main models_dir, - but excludes non-model entries like custom_nodes. + but excludes non-model entries like configs and custom_nodes. """ targets: list[tuple[str, list[str]]] = [] for name, values in folder_paths.folder_names_and_paths.items(): @@ -26,6 +32,16 @@ def get_comfy_models_folders() -> list[tuple[str, list[str]]]: return targets +def get_comfy_model_folder_names() -> set[str]: + """Return valid model folder names for public asset model paths.""" + return {name for name, _paths in get_comfy_models_folders()} + + +def is_comfy_model_folder_name(folder_name: str) -> bool: + """Return whether a folder name resolves to a public model folder name.""" + return folder_paths.map_legacy(folder_name) in get_comfy_model_folder_names() + + def resolve_destination_from_tags(tags: list[str]) -> tuple[str, list[str]]: """Validates and maps tags -> (base_dir, subdirs_for_fs)""" if not tags: @@ -34,8 +50,11 @@ def resolve_destination_from_tags(tags: list[str]) -> tuple[str, list[str]]: if root == "models": if len(tags) < 2: raise ValueError("at least two tags required for model asset") + folder_name = folder_paths.map_legacy(tags[1]) + if not is_comfy_model_folder_name(tags[1]): + raise ValueError(f"unknown model category '{tags[1]}'") try: - bases = folder_paths.folder_names_and_paths[tags[1]][0] + bases = folder_paths.folder_names_and_paths[folder_name][0] except KeyError: raise ValueError(f"unknown model category '{tags[1]}'") if not bases: @@ -65,35 +84,114 @@ def validate_path_within_base(candidate: str, base: str) -> None: raise ValueError("destination escapes base directory") -def compute_relative_filename(file_path: str) -> str | None: - """ - Return the model's path relative to the last well-known folder (the model category), - using forward slashes, eg: - /.../models/checkpoints/flux/123/flux.safetensors -> "flux/123/flux.safetensors" - /.../models/text_encoders/clip_g.safetensors -> "clip_g.safetensors" +def compute_asset_response_paths( + file_path: str, +) -> tuple[str, str | None] | None: + """Compute (file_path, display_name) for an Asset response. - For non-model paths, returns None. + `file_path` is a logical namespace key: `/` for input/output/temp + assets and `models//` for files under ComfyUI's registered + model-folder paths. `display_name` is the path below that root or registered + folder name, suitable for UI labels. Returns None when the absolute path is + not under a known asset root or registered model-folder path. """ try: - root_category, rel_path = get_asset_category_and_relative_path(file_path) + root, folder_name, rel = get_asset_root_folder_name_and_filepath(file_path) except ValueError: return None - p = Path(rel_path) - parts = [seg for seg in p.parts if seg not in (".", "..", p.anchor)] - if not parts: - return None + display_name = rel or None + if folder_name is None: + response_file_path = f"{root}/{rel}" if rel else root + else: + response_file_path = f"{root}/{folder_name}/{rel}" if rel else f"{root}/{folder_name}" + return response_file_path, display_name - if root_category == "models": - # parts[0] is the category ("checkpoints", "vae", etc) – drop it - inside = parts[1:] if len(parts) > 1 else [parts[0]] - return "/".join(inside) - return "/".join(parts) # input/output: keep all parts + +def compute_display_name(file_path: str) -> str | None: + """Return the asset's `display_name`, or None for unknown paths.""" + result = compute_asset_response_paths(file_path) + return result[1] if result else None + + +def compute_file_path(file_path: str) -> str | None: + """Return the asset's logical `file_path`, or None for unknown paths.""" + result = compute_asset_response_paths(file_path) + return result[0] if result else None + + +def compute_relative_filename(file_path: str) -> str | None: + """ + Return the path relative to the asset root or model folder name, using forward slashes, eg: + /.../models/checkpoints/flux/123/flux.safetensors -> "flux/123/flux.safetensors" + /.../models/text_encoders/clip_g.safetensors -> "clip_g.safetensors" + /.../input/sub/image.png -> "sub/image.png" + + For unknown paths, returns None. + """ + return compute_display_name(file_path) + + +def get_asset_root_folder_name_and_filepath( + file_path: str, +) -> tuple[AssetRoot, str | None, str]: + """Decompose an absolute path into (root, registered folder name, path-under-root). + + `folder_name` is set only when the path is under a ComfyUI registered + model-folder path from `folder_names_and_paths`. The returned relative path + always uses `/` separators and is empty when the path is exactly the matched + root. + + Raises: + ValueError: path does not belong to any known root. + """ + fp_abs = os.path.abspath(file_path) + + def _check_is_within(child: str, parent: str) -> bool: + return Path(child).is_relative_to(parent) + + def _compute_relative(child: str, parent: str) -> str: + # Normalize relative path, stripping any leading ".." components + # by anchoring to root (os.sep) then computing relpath back from it. + rel = os.path.relpath( + os.path.join(os.sep, os.path.relpath(child, parent)), os.sep + ) + return "" if rel == "." else rel.replace(os.sep, "/") + + # Registered model folders define ComfyUI's model namespace. Check these + # first so output-backed or external model paths become + # models// rather than physical output/... paths. + best_model: tuple[int, str, str] | None = None + for folder_name, bases in get_comfy_models_folders(): + for b in bases: + base_abs = os.path.abspath(b) + if not _check_is_within(fp_abs, base_abs): + continue + cand = (len(base_abs), folder_name, _compute_relative(fp_abs, base_abs)) + if best_model is None or cand[0] > best_model[0]: + best_model = cand + + if best_model is not None: + _, folder_name, rel_inside = best_model + return "models", folder_name, rel_inside + + for root_tag, getter in ( + ("input", folder_paths.get_input_directory), + ("output", folder_paths.get_output_directory), + ("temp", folder_paths.get_temp_directory), + ): + base = os.path.abspath(getter()) + if _check_is_within(fp_abs, base): + return root_tag, None, _compute_relative(fp_abs, base) + + raise ValueError( + f"Path is not within input, output, temp, or configured model bases: {file_path}" + ) def get_asset_category_and_relative_path( file_path: str, -) -> tuple[Literal["input", "output", "temp", "models"], str]: +) -> tuple[AssetRoot, str]: """Determine which root category a file path belongs to. Categories: @@ -108,52 +206,12 @@ def get_asset_category_and_relative_path( Raises: ValueError: path does not belong to any known root. """ - fp_abs = os.path.abspath(file_path) + root, folder_name, rel = get_asset_root_folder_name_and_filepath(file_path) + if folder_name is None: + return root, rel - def _check_is_within(child: str, parent: str) -> bool: - return Path(child).is_relative_to(parent) - - def _compute_relative(child: str, parent: str) -> str: - # Normalize relative path, stripping any leading ".." components - # by anchoring to root (os.sep) then computing relpath back from it. - return os.path.relpath( - os.path.join(os.sep, os.path.relpath(child, parent)), os.sep - ) - - # 1) input - input_base = os.path.abspath(folder_paths.get_input_directory()) - if _check_is_within(fp_abs, input_base): - return "input", _compute_relative(fp_abs, input_base) - - # 2) output - output_base = os.path.abspath(folder_paths.get_output_directory()) - if _check_is_within(fp_abs, output_base): - return "output", _compute_relative(fp_abs, output_base) - - # 3) temp - temp_base = os.path.abspath(folder_paths.get_temp_directory()) - if _check_is_within(fp_abs, temp_base): - return "temp", _compute_relative(fp_abs, temp_base) - - # 4) models (check deepest matching base to avoid ambiguity) - best: tuple[int, str, str] | None = None # (base_len, bucket, rel_inside_bucket) - for bucket, bases in get_comfy_models_folders(): - for b in bases: - base_abs = os.path.abspath(b) - if not _check_is_within(fp_abs, base_abs): - continue - cand = (len(base_abs), bucket, _compute_relative(fp_abs, base_abs)) - if best is None or cand[0] > best[0]: - best = cand - - if best is not None: - _, bucket, rel_inside = best - combined = os.path.join(bucket, rel_inside) - return "models", os.path.relpath(os.path.join(os.sep, combined), os.sep) - - raise ValueError( - f"Path is not within input, output, temp, or configured model bases: {file_path}" - ) + combined = os.path.join(folder_name, rel) + return root, os.path.relpath(os.path.join(os.sep, combined), os.sep) def get_name_and_tags_from_asset_path(file_path: str) -> tuple[str, list[str]]: diff --git a/openapi.yaml b/openapi.yaml index 214962c5c..c62da5d2d 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -1667,8 +1667,9 @@ paths: post: operationId: createAssetFromHash tags: [assets] + deprecated: true summary: Create an asset reference from an existing hash - description: Registers a new asset that references existing content by hash, without re-uploading the bytes. + description: Deprecated. Registers a new asset that references existing content by hash, without re-uploading the bytes. x-feature-gate: enable-assets requestBody: required: true @@ -6328,7 +6329,16 @@ components: description: Unique identifier for the asset name: type: string - description: Name of the asset file + deprecated: true + description: "Deprecated: use `file_path` or `display_name`. Free user-chosen label with no path semantics." + file_path: + type: string + nullable: true + description: "Logical namespace key for the asset. Relative path only: no './' or '../' segments, no leading or trailing '/', and SHOULD include a file extension. For files under ComfyUI registered model-folder paths, uses `models//`. It is not a physical path, not stable across model folder configuration changes, and not a unique reference key; duplicate model names may resolve to a different physical file according to ComfyUI's model-folder search order. Use `id` for exact asset identity." + display_name: + type: string + nullable: true + description: "Human-facing display label for the asset, usually the filename relative to its root or registered model-folder name. Not a unique reference key; use `id` for identity." hash: type: string nullable: true @@ -8109,4 +8119,4 @@ components: items: $ref: "#/components/schemas/TaskEntry" pagination: - $ref: "#/components/schemas/PaginationInfo" \ No newline at end of file + $ref: "#/components/schemas/PaginationInfo" diff --git a/tests-unit/assets_test/services/test_path_utils.py b/tests-unit/assets_test/services/test_path_utils.py index 3fa905f9a..82f96c52b 100644 --- a/tests-unit/assets_test/services/test_path_utils.py +++ b/tests-unit/assets_test/services/test_path_utils.py @@ -6,7 +6,13 @@ from unittest.mock import patch import pytest -from app.assets.services.path_utils import get_asset_category_and_relative_path +from app.assets.services.path_utils import ( + compute_display_name, + compute_file_path, + get_asset_category_and_relative_path, + get_comfy_models_folders, + get_name_and_tags_from_asset_path, +) @pytest.fixture @@ -17,24 +23,47 @@ def fake_dirs(): input_dir = root_path / "input" output_dir = root_path / "output" temp_dir = root_path / "temp" - models_dir = root_path / "models" / "checkpoints" - for d in (input_dir, output_dir, temp_dir, models_dir): + models_dir = root_path / "models" + checkpoints_dir = models_dir / "checkpoints" + output_checkpoints_dir = output_dir / "checkpoints" + external_checkpoints_dir = root_path / "external" / "not_named_like_category" + for d in ( + input_dir, + output_dir, + temp_dir, + checkpoints_dir, + output_checkpoints_dir, + external_checkpoints_dir, + ): d.mkdir(parents=True) with patch("app.assets.services.path_utils.folder_paths") as mock_fp: mock_fp.get_input_directory.return_value = str(input_dir) mock_fp.get_output_directory.return_value = str(output_dir) mock_fp.get_temp_directory.return_value = str(temp_dir) + mock_fp.models_dir = str(models_dir) with patch( "app.assets.services.path_utils.get_comfy_models_folders", - return_value=[("checkpoints", [str(models_dir)])], + return_value=[ + ( + "checkpoints", + [ + str(checkpoints_dir), + str(output_checkpoints_dir), + str(external_checkpoints_dir), + ], + ) + ], ): yield { "input": input_dir, "output": output_dir, "temp": temp_dir, "models": models_dir, + "checkpoints": checkpoints_dir, + "output_checkpoints": output_checkpoints_dir, + "external_checkpoints": external_checkpoints_dir, } @@ -71,7 +100,7 @@ class TestGetAssetCategoryAndRelativePath: assert os.path.normpath(rel) == os.path.normpath("sub/ComfyUI_temp_tczip_00004_.png") def test_model_file(self, fake_dirs): - f = fake_dirs["models"] / "model.safetensors" + f = fake_dirs["checkpoints"] / "model.safetensors" f.touch() cat, rel = get_asset_category_and_relative_path(str(f)) assert cat == "models" @@ -79,3 +108,195 @@ class TestGetAssetCategoryAndRelativePath: def test_unknown_path_raises(self, fake_dirs): with pytest.raises(ValueError, match="not within"): get_asset_category_and_relative_path("/some/random/path.png") + + +class TestResponsePaths: + def test_get_comfy_models_folders_excludes_core_infrastructure(self, tmp_path: Path): + controlnet_dir = tmp_path / "models" / "controlnet" + configs_dir = tmp_path / "models" / "configs" + custom_nodes_dir = tmp_path / "custom_nodes" + for directory in (controlnet_dir, configs_dir, custom_nodes_dir): + directory.mkdir(parents=True) + + with patch("app.assets.services.path_utils.folder_paths") as mock_fp: + mock_fp.folder_names_and_paths = { + "controlnet": ([str(controlnet_dir)], {".safetensors"}), + "configs": ([str(configs_dir)], {".yaml"}), + "custom_nodes": ([str(custom_nodes_dir)], set()), + } + + folders = get_comfy_models_folders() + + assert folders == [("controlnet", [str(controlnet_dir)])] + + def test_input_file_path_and_display_name_include_subfolder(self, fake_dirs): + sub = fake_dirs["input"] / "some" / "folder" + sub.mkdir(parents=True) + f = sub / "image.png" + f.touch() + + assert compute_file_path(str(f)) == "input/some/folder/image.png" + assert compute_display_name(str(f)) == "some/folder/image.png" + + def test_model_file_path_is_relative_to_physical_models_root(self, fake_dirs): + sub = fake_dirs["checkpoints"] / "flux" + sub.mkdir() + f = sub / "model.safetensors" + f.touch() + + assert compute_file_path(str(f)) == "models/checkpoints/flux/model.safetensors" + assert compute_display_name(str(f)) == "flux/model.safetensors" + + @pytest.mark.parametrize( + "folder_name", + ["checkpoints", "clip", "vae", "diffusion_models", "loras"], + ) + def test_output_model_folder_uses_model_namespace_file_path(self, fake_dirs, folder_name): + output_model_dir = fake_dirs["output"] / folder_name + output_model_dir.mkdir(exist_ok=True) + default_model_dir = fake_dirs["models"] / folder_name + default_model_dir.mkdir(exist_ok=True) + f = output_model_dir / "saved.safetensors" + f.touch() + + with patch( + "app.assets.services.path_utils.get_comfy_models_folders", + return_value=[(folder_name, [str(default_model_dir), str(output_model_dir)])], + ): + cat, rel = get_asset_category_and_relative_path(str(f)) + assert cat == "models" + assert os.path.normpath(rel) == os.path.normpath(f"{folder_name}/saved.safetensors") + + assert compute_file_path(str(f)) == f"models/{folder_name}/saved.safetensors" + assert compute_display_name(str(f)) == "saved.safetensors" + + name, tags = get_name_and_tags_from_asset_path(str(f)) + assert name == "saved.safetensors" + assert tags[:2] == ["models", folder_name] + + def test_output_model_subfolder_uses_model_namespace_file_path(self, fake_dirs): + folder_name = "loras" + output_model_dir = fake_dirs["output"] / folder_name + subdir = output_model_dir / "experiments" + subdir.mkdir(parents=True) + f = subdir / "my_lora.safetensors" + f.touch() + + with patch( + "app.assets.services.path_utils.get_comfy_models_folders", + return_value=[(folder_name, [str(output_model_dir)])], + ): + cat, rel = get_asset_category_and_relative_path(str(f)) + assert cat == "models" + assert os.path.normpath(rel) == os.path.normpath( + "loras/experiments/my_lora.safetensors" + ) + + assert ( + compute_file_path(str(f)) + == "models/loras/experiments/my_lora.safetensors" + ) + assert compute_display_name(str(f)) == "experiments/my_lora.safetensors" + + name, tags = get_name_and_tags_from_asset_path(str(f)) + assert name == "my_lora.safetensors" + assert tags[:3] == ["models", "loras", "experiments"] + + def test_external_model_folder_uses_registered_folder_name_namespace(self, fake_dirs): + f = fake_dirs["external_checkpoints"] / "external.safetensors" + f.touch() + + assert compute_file_path(str(f)) == "models/checkpoints/external.safetensors" + assert compute_display_name(str(f)) == "external.safetensors" + + def test_nested_registered_bases_for_same_model_folder_use_deepest_match(self, tmp_path: Path): + llm_dir = tmp_path / "models" / "LLM" + llm_checkpoints_dir = llm_dir / "checkpoints" + llm_checkpoints_dir.mkdir(parents=True) + checkpoint = llm_checkpoints_dir / "model.safetensors" + checkpoint.touch() + tokenizer = llm_dir / "tokenizer.json" + tokenizer.touch() + + with patch( + "app.assets.services.path_utils.get_comfy_models_folders", + return_value=[("LLM", [str(llm_checkpoints_dir), str(llm_dir)])], + ): + checkpoint_cat, checkpoint_rel = get_asset_category_and_relative_path( + str(checkpoint) + ) + tokenizer_cat, tokenizer_rel = get_asset_category_and_relative_path( + str(tokenizer) + ) + + assert checkpoint_cat == tokenizer_cat == "models" + assert checkpoint_rel == "LLM/model.safetensors" + assert tokenizer_rel == "LLM/tokenizer.json" + assert compute_file_path(str(checkpoint)) == "models/LLM/model.safetensors" + assert compute_display_name(str(checkpoint)) == "model.safetensors" + assert compute_file_path(str(tokenizer)) == "models/LLM/tokenizer.json" + assert compute_display_name(str(tokenizer)) == "tokenizer.json" + + def test_same_relative_model_file_under_multiple_roots_shares_logical_file_path( + self, tmp_path: Path + ): + foo_dir = tmp_path / "foo" + bar_dir = tmp_path / "bar" + foo_dir.mkdir() + bar_dir.mkdir() + foo_file = foo_dir / "baz.safetensors" + bar_file = bar_dir / "baz.safetensors" + foo_file.touch() + bar_file.touch() + + with patch( + "app.assets.services.path_utils.get_comfy_models_folders", + return_value=[("checkpoints", [str(foo_dir), str(bar_dir)])], + ): + assert compute_file_path(str(foo_file)) == "models/checkpoints/baz.safetensors" + assert compute_file_path(str(bar_file)) == "models/checkpoints/baz.safetensors" + assert compute_display_name(str(foo_file)) == "baz.safetensors" + assert compute_display_name(str(bar_file)) == "baz.safetensors" + + def test_output_clip_folder_uses_canonical_text_encoders_folder_name(self, fake_dirs): + output_clip_dir = fake_dirs["output"] / "clip" + output_clip_dir.mkdir() + f = output_clip_dir / "clip_l.safetensors" + f.touch() + + with patch( + "app.assets.services.path_utils.get_comfy_models_folders", + return_value=[("text_encoders", [str(output_clip_dir)])], + ): + assert compute_file_path(str(f)) == "models/text_encoders/clip_l.safetensors" + assert compute_display_name(str(f)) == "clip_l.safetensors" + + def test_physical_unet_folder_uses_diffusion_models_namespace(self, fake_dirs): + unet_dir = fake_dirs["models"] / "unet" + diffusion_models_dir = fake_dirs["models"] / "diffusion_models" + unet_dir.mkdir() + diffusion_models_dir.mkdir() + f = unet_dir / "wan.safetensors" + f.touch() + + with patch( + "app.assets.services.path_utils.get_comfy_models_folders", + return_value=[("diffusion_models", [str(unet_dir), str(diffusion_models_dir)])], + ): + cat, rel = get_asset_category_and_relative_path(str(f)) + assert cat == "models" + assert rel == "diffusion_models/wan.safetensors" + assert compute_file_path(str(f)) == "models/diffusion_models/wan.safetensors" + assert compute_display_name(str(f)) == "wan.safetensors" + + def test_unregistered_file_under_physical_models_root_has_no_file_path(self, fake_dirs): + f = fake_dirs["models"] / "not_registered" / "orphan.bin" + f.parent.mkdir() + f.touch() + + assert compute_file_path(str(f)) is None + assert compute_display_name(str(f)) is None + + def test_unknown_path_returns_none(self, fake_dirs): + assert compute_file_path("/some/random/path.png") is None + assert compute_display_name("/some/random/path.png") is None diff --git a/tests-unit/assets_test/test_uploads.py b/tests-unit/assets_test/test_uploads.py index 0f2b124a3..d487e55d2 100644 --- a/tests-unit/assets_test/test_uploads.py +++ b/tests-unit/assets_test/test_uploads.py @@ -5,6 +5,8 @@ from concurrent.futures import ThreadPoolExecutor import requests import pytest +from helpers import get_asset_filename + def test_upload_ok_duplicate_reference(http: requests.Session, api_base: str, make_asset_bytes): name = "dup_a.safetensors" @@ -63,6 +65,14 @@ def test_upload_fastpath_from_existing_hash_no_file(http: requests.Session, api_ 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.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_upload_fastpath_with_known_hash_and_file( @@ -107,6 +117,54 @@ def test_upload_multiple_tags_fields_are_merged(http: requests.Session, api_base assert {"models", "checkpoints", "unit-tests", "alpha"}.issubset(tags) +@pytest.mark.parametrize( + ("tags", "extension", "expected_prefix", "expected_display_prefix"), + [ + (["input", "unit-tests"], ".png", "input", ""), + (["models", "checkpoints", "unit-tests"], ".safetensors", "models/checkpoints", ""), + ], +) +def test_upload_response_includes_file_path_and_display_name( + tags: list[str], + extension: str, + expected_prefix: str, + expected_display_prefix: str, + http: requests.Session, + api_base: str, + asset_factory, + make_asset_bytes, +): + scope = f"response-paths-{uuid.uuid4().hex[:6]}" + scoped_tags = [*tags, scope] + name = f"asset_response_path{extension}" + + created = asset_factory(name, scoped_tags, {}, make_asset_bytes(name, 1024)) + stored_filename = get_asset_filename(created["asset_hash"], extension) + expected_suffix = f"unit-tests/{scope}/{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, @@ -225,6 +283,16 @@ def test_upload_models_unknown_category(http: requests.Session, api_base: str): assert body["error"]["message"].startswith("unknown models category") +def test_upload_models_rejects_non_model_folder_name(http: requests.Session, api_base: str): + files = {"file": ("node.py", b"print('not a model')", "text/x-python")} + form = {"tags": json.dumps(["models", "custom_nodes", "unit-tests"]), "name": "node.py"} + 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" + assert body["error"]["message"].startswith("unknown models category") + + def test_upload_models_requires_category(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({})}