diff --git a/app/assets/services/path_utils.py b/app/assets/services/path_utils.py index c13b86a45..7c27c8878 100644 --- a/app/assets/services/path_utils.py +++ b/app/assets/services/path_utils.py @@ -209,8 +209,14 @@ def get_asset_category_and_relative_path( return "temp", _compute_relative(fp_abs, temp_base) # 4) models (check deepest matching base to avoid ambiguity) + ext = os.path.splitext(fp_abs)[1].lower() best: tuple[int, str, str] | None = None # (base_len, bucket, rel_inside_bucket) - for bucket, bases, _exts in get_comfy_models_folders(): + for bucket, bases, extensions in get_comfy_models_folders(): + # A bucket only lists files within its extension set (empty set + # accepts any extension), so a bucket that cannot load the file + # must not contribute a loader path. + if extensions and ext not in extensions: + continue for b in bases: base_abs = os.path.abspath(b) if not _check_is_within(fp_abs, base_abs): diff --git a/tests-unit/assets_test/services/test_path_utils.py b/tests-unit/assets_test/services/test_path_utils.py index ad769d209..ddf23c676 100644 --- a/tests-unit/assets_test/services/test_path_utils.py +++ b/tests-unit/assets_test/services/test_path_utils.py @@ -523,6 +523,42 @@ class TestLoaderPath: assert compute_logical_path(str(f)) == "models/not_registered/orphan.bin" assert compute_loader_path(str(f)) is None + def test_extension_mismatch_in_registered_bucket_has_no_loader_path(self, fake_dirs): + # Inside a registered bucket, but the bucket's extension set cannot + # load it: no model_type tag, and no loader path either. + f = fake_dirs["models"] / "notes.txt" + f.touch() + + assert compute_logical_path(str(f)) == "models/checkpoints/notes.txt" + assert compute_loader_path(str(f)) is None + + def test_shared_base_loader_path_uses_extension_matching_bucket(self, fake_dirs): + shared_root = fake_dirs["models"].parent / "unet" + shared_root.mkdir() + f = shared_root / "wan.gguf" + f.touch() + + with patch( + "app.assets.services.path_utils.get_comfy_models_folders", + return_value=[ + ("diffusion_models", [str(shared_root)], {".safetensors"}), + ("unet_gguf", [str(shared_root)], {".gguf"}), + ], + ): + assert compute_loader_path(str(f)) == "wan.gguf" + + def test_match_all_bucket_provides_loader_path_for_any_extension(self, fake_dirs): + custom_root = fake_dirs["models"].parent / "custom_bucket" + custom_root.mkdir() + f = custom_root / "weights.bin" + f.touch() + + with patch( + "app.assets.services.path_utils.get_comfy_models_folders", + return_value=[("custom_bucket", [str(custom_root)], set())], + ): + assert compute_loader_path(str(f)) == "weights.bin" + def test_extra_path_model_has_loader_path_but_no_logical_path(self, tmp_path: Path): """Registered category base outside models_dir (extra_model_paths style).