mirror of
https://github.com/comfyanonymous/ComfyUI.git
synced 2026-06-06 06:17:38 +08:00
feat(assets): add file_path and display_name fields
Co-authored-by: Amp <amp@ampcode.com> Amp-Thread-ID: https://ampcode.com/threads/T-019e9643-8c7f-7672-8078-e1daae84dab6
This commit is contained in:
parent
df2454b47e
commit
c533a88149
@ -10,7 +10,6 @@ from typing import Any
|
|||||||
from aiohttp import web
|
from aiohttp import web
|
||||||
from pydantic import ValidationError
|
from pydantic import ValidationError
|
||||||
|
|
||||||
import folder_paths
|
|
||||||
from app import user_manager
|
from app import user_manager
|
||||||
from app.assets.api import schemas_in, schemas_out
|
from app.assets.api import schemas_in, schemas_out
|
||||||
from app.assets.services import schemas
|
from app.assets.services import schemas
|
||||||
@ -39,6 +38,10 @@ from app.assets.services import (
|
|||||||
update_asset_metadata,
|
update_asset_metadata,
|
||||||
upload_from_temp_path,
|
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
|
from app.assets.services.tagging import list_tag_histogram
|
||||||
|
|
||||||
ROUTES = web.RouteTableDef()
|
ROUTES = web.RouteTableDef()
|
||||||
@ -160,9 +163,16 @@ def _build_asset_response(result: schemas.AssetDetailResult | schemas.UploadResu
|
|||||||
preview_url = None
|
preview_url = None
|
||||||
else:
|
else:
|
||||||
preview_url = _build_preview_url_from_view(result.tags, result.ref.user_metadata)
|
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(
|
return schemas_out.Asset(
|
||||||
id=result.ref.id,
|
id=result.ref.id,
|
||||||
name=result.ref.name,
|
name=result.ref.name,
|
||||||
|
file_path=file_path,
|
||||||
|
display_name=display_name,
|
||||||
asset_hash=result.asset.hash if result.asset else None,
|
asset_hash=result.asset.hash if result.asset else None,
|
||||||
size=int(result.asset.size_bytes) 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,
|
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 spec.tags and spec.tags[0] == "models":
|
||||||
if (
|
if len(spec.tags) < 2 or not is_comfy_model_folder_name(spec.tags[1]):
|
||||||
len(spec.tags) < 2
|
|
||||||
or spec.tags[1] not in folder_paths.folder_names_and_paths
|
|
||||||
):
|
|
||||||
delete_temp_file_if_exists(parsed.tmp_path)
|
delete_temp_file_if_exists(parsed.tmp_path)
|
||||||
category = spec.tags[1] if len(spec.tags) >= 2 else ""
|
category = spec.tags[1] if len(spec.tags) >= 2 else ""
|
||||||
return _build_error_response(
|
return _build_error_response(
|
||||||
|
|||||||
@ -9,7 +9,19 @@ class Asset(BaseModel):
|
|||||||
``id`` here is the AssetReference id, not the content-addressed Asset id."""
|
``id`` here is the AssetReference id, not the content-addressed Asset id."""
|
||||||
|
|
||||||
id: str
|
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/<folder_name>/<filename>`; 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
|
asset_hash: str | None = None
|
||||||
size: int | None = None
|
size: int | None = None
|
||||||
mime_type: str | None = None
|
mime_type: str | None = None
|
||||||
|
|||||||
@ -6,15 +6,21 @@ import folder_paths
|
|||||||
from app.assets.helpers import normalize_tags
|
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]]]:
|
def get_comfy_models_folders() -> list[tuple[str, list[str]]]:
|
||||||
"""Build list of (folder_name, base_paths[]) for all model locations.
|
"""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,
|
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]]] = []
|
targets: list[tuple[str, list[str]]] = []
|
||||||
for name, values in folder_paths.folder_names_and_paths.items():
|
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
|
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]]:
|
def resolve_destination_from_tags(tags: list[str]) -> tuple[str, list[str]]:
|
||||||
"""Validates and maps tags -> (base_dir, subdirs_for_fs)"""
|
"""Validates and maps tags -> (base_dir, subdirs_for_fs)"""
|
||||||
if not tags:
|
if not tags:
|
||||||
@ -34,8 +50,11 @@ def resolve_destination_from_tags(tags: list[str]) -> tuple[str, list[str]]:
|
|||||||
if root == "models":
|
if root == "models":
|
||||||
if len(tags) < 2:
|
if len(tags) < 2:
|
||||||
raise ValueError("at least two tags required for model asset")
|
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:
|
try:
|
||||||
bases = folder_paths.folder_names_and_paths[tags[1]][0]
|
bases = folder_paths.folder_names_and_paths[folder_name][0]
|
||||||
except KeyError:
|
except KeyError:
|
||||||
raise ValueError(f"unknown model category '{tags[1]}'")
|
raise ValueError(f"unknown model category '{tags[1]}'")
|
||||||
if not bases:
|
if not bases:
|
||||||
@ -65,35 +84,114 @@ def validate_path_within_base(candidate: str, base: str) -> None:
|
|||||||
raise ValueError("destination escapes base directory")
|
raise ValueError("destination escapes base directory")
|
||||||
|
|
||||||
|
|
||||||
def compute_relative_filename(file_path: str) -> str | None:
|
def compute_asset_response_paths(
|
||||||
"""
|
file_path: str,
|
||||||
Return the model's path relative to the last well-known folder (the model category),
|
) -> tuple[str, str | None] | None:
|
||||||
using forward slashes, eg:
|
"""Compute (file_path, display_name) for an Asset response.
|
||||||
/.../models/checkpoints/flux/123/flux.safetensors -> "flux/123/flux.safetensors"
|
|
||||||
/.../models/text_encoders/clip_g.safetensors -> "clip_g.safetensors"
|
|
||||||
|
|
||||||
For non-model paths, returns None.
|
`file_path` is a logical namespace key: `<root>/<rel>` for input/output/temp
|
||||||
|
assets and `models/<folder_name>/<rel>` 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:
|
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:
|
except ValueError:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
p = Path(rel_path)
|
display_name = rel or None
|
||||||
parts = [seg for seg in p.parts if seg not in (".", "..", p.anchor)]
|
if folder_name is None:
|
||||||
if not parts:
|
response_file_path = f"{root}/{rel}" if rel else root
|
||||||
return None
|
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
|
def compute_display_name(file_path: str) -> str | None:
|
||||||
inside = parts[1:] if len(parts) > 1 else [parts[0]]
|
"""Return the asset's `display_name`, or None for unknown paths."""
|
||||||
return "/".join(inside)
|
result = compute_asset_response_paths(file_path)
|
||||||
return "/".join(parts) # input/output: keep all parts
|
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/<folder_name>/<relative-path> 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(
|
def get_asset_category_and_relative_path(
|
||||||
file_path: str,
|
file_path: str,
|
||||||
) -> tuple[Literal["input", "output", "temp", "models"], str]:
|
) -> tuple[AssetRoot, str]:
|
||||||
"""Determine which root category a file path belongs to.
|
"""Determine which root category a file path belongs to.
|
||||||
|
|
||||||
Categories:
|
Categories:
|
||||||
@ -108,52 +206,12 @@ def get_asset_category_and_relative_path(
|
|||||||
Raises:
|
Raises:
|
||||||
ValueError: path does not belong to any known root.
|
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:
|
combined = os.path.join(folder_name, rel)
|
||||||
return Path(child).is_relative_to(parent)
|
return root, os.path.relpath(os.path.join(os.sep, combined), os.sep)
|
||||||
|
|
||||||
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}"
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def get_name_and_tags_from_asset_path(file_path: str) -> tuple[str, list[str]]:
|
def get_name_and_tags_from_asset_path(file_path: str) -> tuple[str, list[str]]:
|
||||||
|
|||||||
16
openapi.yaml
16
openapi.yaml
@ -1667,8 +1667,9 @@ paths:
|
|||||||
post:
|
post:
|
||||||
operationId: createAssetFromHash
|
operationId: createAssetFromHash
|
||||||
tags: [assets]
|
tags: [assets]
|
||||||
|
deprecated: true
|
||||||
summary: Create an asset reference from an existing hash
|
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
|
x-feature-gate: enable-assets
|
||||||
requestBody:
|
requestBody:
|
||||||
required: true
|
required: true
|
||||||
@ -6328,7 +6329,16 @@ components:
|
|||||||
description: Unique identifier for the asset
|
description: Unique identifier for the asset
|
||||||
name:
|
name:
|
||||||
type: string
|
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/<folder_name>/<filename>`. 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:
|
hash:
|
||||||
type: string
|
type: string
|
||||||
nullable: true
|
nullable: true
|
||||||
@ -8109,4 +8119,4 @@ components:
|
|||||||
items:
|
items:
|
||||||
$ref: "#/components/schemas/TaskEntry"
|
$ref: "#/components/schemas/TaskEntry"
|
||||||
pagination:
|
pagination:
|
||||||
$ref: "#/components/schemas/PaginationInfo"
|
$ref: "#/components/schemas/PaginationInfo"
|
||||||
|
|||||||
@ -6,7 +6,13 @@ from unittest.mock import patch
|
|||||||
|
|
||||||
import pytest
|
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
|
@pytest.fixture
|
||||||
@ -17,24 +23,47 @@ def fake_dirs():
|
|||||||
input_dir = root_path / "input"
|
input_dir = root_path / "input"
|
||||||
output_dir = root_path / "output"
|
output_dir = root_path / "output"
|
||||||
temp_dir = root_path / "temp"
|
temp_dir = root_path / "temp"
|
||||||
models_dir = root_path / "models" / "checkpoints"
|
models_dir = root_path / "models"
|
||||||
for d in (input_dir, output_dir, temp_dir, models_dir):
|
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)
|
d.mkdir(parents=True)
|
||||||
|
|
||||||
with patch("app.assets.services.path_utils.folder_paths") as mock_fp:
|
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_input_directory.return_value = str(input_dir)
|
||||||
mock_fp.get_output_directory.return_value = str(output_dir)
|
mock_fp.get_output_directory.return_value = str(output_dir)
|
||||||
mock_fp.get_temp_directory.return_value = str(temp_dir)
|
mock_fp.get_temp_directory.return_value = str(temp_dir)
|
||||||
|
mock_fp.models_dir = str(models_dir)
|
||||||
|
|
||||||
with patch(
|
with patch(
|
||||||
"app.assets.services.path_utils.get_comfy_models_folders",
|
"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 {
|
yield {
|
||||||
"input": input_dir,
|
"input": input_dir,
|
||||||
"output": output_dir,
|
"output": output_dir,
|
||||||
"temp": temp_dir,
|
"temp": temp_dir,
|
||||||
"models": models_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")
|
assert os.path.normpath(rel) == os.path.normpath("sub/ComfyUI_temp_tczip_00004_.png")
|
||||||
|
|
||||||
def test_model_file(self, fake_dirs):
|
def test_model_file(self, fake_dirs):
|
||||||
f = fake_dirs["models"] / "model.safetensors"
|
f = fake_dirs["checkpoints"] / "model.safetensors"
|
||||||
f.touch()
|
f.touch()
|
||||||
cat, rel = get_asset_category_and_relative_path(str(f))
|
cat, rel = get_asset_category_and_relative_path(str(f))
|
||||||
assert cat == "models"
|
assert cat == "models"
|
||||||
@ -79,3 +108,195 @@ class TestGetAssetCategoryAndRelativePath:
|
|||||||
def test_unknown_path_raises(self, fake_dirs):
|
def test_unknown_path_raises(self, fake_dirs):
|
||||||
with pytest.raises(ValueError, match="not within"):
|
with pytest.raises(ValueError, match="not within"):
|
||||||
get_asset_category_and_relative_path("/some/random/path.png")
|
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
|
||||||
|
|||||||
@ -5,6 +5,8 @@ from concurrent.futures import ThreadPoolExecutor
|
|||||||
import requests
|
import requests
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
|
from helpers import get_asset_filename
|
||||||
|
|
||||||
|
|
||||||
def test_upload_ok_duplicate_reference(http: requests.Session, api_base: str, make_asset_bytes):
|
def test_upload_ok_duplicate_reference(http: requests.Session, api_base: str, make_asset_bytes):
|
||||||
name = "dup_a.safetensors"
|
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 r2.status_code == 200, b2 # fast path returns 200 with created_new == False
|
||||||
assert b2["created_new"] is False
|
assert b2["created_new"] is False
|
||||||
assert b2["asset_hash"] == h
|
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(
|
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)
|
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"])
|
@pytest.mark.parametrize("root", ["input", "output"])
|
||||||
def test_concurrent_upload_identical_bytes_different_names(
|
def test_concurrent_upload_identical_bytes_different_names(
|
||||||
root: str,
|
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")
|
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):
|
def test_upload_models_requires_category(http: requests.Session, api_base: str):
|
||||||
files = {"file": ("nocat.safetensors", b"A" * 64, "application/octet-stream")}
|
files = {"file": ("nocat.safetensors", b"A" * 64, "application/octet-stream")}
|
||||||
form = {"tags": json.dumps(["models"]), "name": "nocat.safetensors", "user_metadata": json.dumps({})}
|
form = {"tags": json.dumps(["models"]), "name": "nocat.safetensors", "user_metadata": json.dumps({})}
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user