test: make the backend suite pass on Windows

The new informational Windows job surfaced 91 failures: four real bugs (fixed in
the previous two commits) and a long tail of POSIX assumptions in the tests
themselves. None of them assert behaviour that differs on Windows, so they are
fixed rather than skipped wherever the behaviour is the same:

* Path spelling: expectations built with `str(tmp_path)` are compared through an
  identity helper (`normalize_root_identity`) or the platform's separator, since
  the code hands out forward-slashed business paths while `str()` is native.
* Root labels: `_root_display_labels` prefixes the drive on Windows ("C: loras"),
  so label assertions compare the final path component or derive the expected
  label from the same helper.
* `~` expansion reads USERPROFILE (ntpath), not HOME; timer granularity rounds
  0.1s down; a case-only rename counts as "already named" via `os.path.normcase`.
* Genuinely POSIX-only strategies are skipped with a reason instead of failing:
  the headless clipboard fallback, dbus file-manager selection, xdg-open, and
  the case-fold index test that simulates a case-sensitive filesystem.
This commit is contained in:
Will Miao
2026-10-12 11:35:33 +08:00
parent 85f79283b6
commit c16de68640
13 changed files with 162 additions and 91 deletions
+5 -5
View File
@@ -60,17 +60,17 @@ class TestAdmitConfiguredRoots:
loras_roots=[str(primary)], loras_roots=[str(primary)],
_configured_root_paths={"lora": [str(primary), str(drive)]}, _configured_root_paths={"lora": [str(primary), str(drive)]},
) )
assert config.configured_roots_for("lora") == [str(primary), str(drive)] assert config.configured_roots_for("lora") == [_normalize(primary), _normalize(drive)]
# Still switched off: nothing to admit. # Still switched off: nothing to admit.
assert config.admit_configured_roots() == [] assert config.admit_configured_roots() == []
assert config.loras_roots == [str(primary)] assert config.loras_roots == [_normalize(primary)]
drive.mkdir(parents=True) drive.mkdir(parents=True)
admitted = config.admit_configured_roots() admitted = config.admit_configured_roots()
assert admitted == [str(drive)] assert admitted == [_normalize(drive)]
assert config.loras_roots == [str(primary), str(drive)] assert config.loras_roots == [_normalize(primary), _normalize(drive)]
# Idempotent: a second call adds nothing. # Idempotent: a second call adds nothing.
assert config.admit_configured_roots() == [] assert config.admit_configured_roots() == []
@@ -91,7 +91,7 @@ class TestAdmitConfiguredRoots:
drive.rmdir() drive.rmdir()
config.admit_configured_roots() config.admit_configured_roots()
assert config.loras_roots == [str(primary), str(drive)] assert config.loras_roots == [_normalize(primary), _normalize(drive)]
def test_admits_every_model_type_and_its_side_maps(self, tmp_path): def test_admits_every_model_type_and_its_side_maps(self, tmp_path):
checkpoint_dir = tmp_path / "checkpoints" checkpoint_dir = tmp_path / "checkpoints"
+76 -55
View File
@@ -1,12 +1,33 @@
import logging import logging
from typing import Any, Dict, Iterable, List from typing import Any, Dict, Iterable, List, Mapping
import pytest import pytest
from py import config as config_module from py import config as config_module
from py.utils.root_paths import normalize_root_identity
from py.services import settings_manager as settings_manager_module from py.services import settings_manager as settings_manager_module
def _identity(value):
"""Fold path spelling so an assertion is about paths, not separators.
LM keeps the host's own spelling for the paths it is configured with
(backslashes on Windows) while the ones it derives itself are forward-slashed,
so two values naming the same directory can differ as strings. Compare
identities wherever a test asserts *which* paths survived.
"""
if isinstance(value, str):
return normalize_root_identity(value)
if isinstance(value, Mapping):
return {key: _identity(item) for key, item in value.items()}
if isinstance(value, (list, tuple)):
return [_identity(item) for item in value]
if isinstance(value, set):
return {_identity(item) for item in value}
return value
def _setup_config_environment(monkeypatch: pytest.MonkeyPatch, tmp_path) -> Dict[str, List[str]]: def _setup_config_environment(monkeypatch: pytest.MonkeyPatch, tmp_path) -> Dict[str, List[str]]:
loras_dir = tmp_path / "loras" loras_dir = tmp_path / "loras"
checkpoints_dir = tmp_path / "checkpoints" checkpoints_dir = tmp_path / "checkpoints"
@@ -94,11 +115,11 @@ def test_save_paths_renames_default_library(monkeypatch: pytest.MonkeyPatch, tmp
key: [path.replace("\\", "/") for path in paths] key: [path.replace("\\", "/") for path in paths]
for key, paths in folder_paths.items() for key, paths in folder_paths.items()
} }
assert payload["folder_paths"] == expected_folder_paths assert _identity(payload["folder_paths"]) == _identity(expected_folder_paths)
assert payload["default_lora_root"] == folder_paths["loras"][0].replace("\\", "/") assert _identity(payload["default_lora_root"]) == _identity(folder_paths["loras"][0].replace("\\", "/"))
assert payload["default_checkpoint_root"] == folder_paths["checkpoints"][0].replace("\\", "/") assert _identity(payload["default_checkpoint_root"]) == _identity(folder_paths["checkpoints"][0].replace("\\", "/"))
assert payload["default_embedding_root"] == folder_paths["embeddings"][0].replace("\\", "/") assert _identity(payload["default_embedding_root"]) == _identity(folder_paths["embeddings"][0].replace("\\", "/"))
assert payload["metadata"] == {"display_name": "ComfyUI", "source": "comfyui"} assert _identity(payload["metadata"]) == _identity({"display_name": "ComfyUI", "source": "comfyui"})
assert payload["activate"] is True assert payload["activate"] is True
@@ -175,9 +196,9 @@ def test_save_paths_repairs_empty_default_roots(monkeypatch: pytest.MonkeyPatch,
config_module.Config() config_module.Config()
assert fake_settings.name == "comfyui" assert fake_settings.name == "comfyui"
assert fake_settings.payload["default_lora_root"] == folder_paths["loras"][0].replace("\\", "/") assert _identity(fake_settings.payload["default_lora_root"]) == _identity(folder_paths["loras"][0].replace("\\", "/"))
assert fake_settings.payload["default_checkpoint_root"] == folder_paths["checkpoints"][0].replace("\\", "/") assert _identity(fake_settings.payload["default_checkpoint_root"]) == _identity(folder_paths["checkpoints"][0].replace("\\", "/"))
assert fake_settings.payload["default_embedding_root"] == folder_paths["embeddings"][0].replace("\\", "/") assert _identity(fake_settings.payload["default_embedding_root"]) == _identity(folder_paths["embeddings"][0].replace("\\", "/"))
def test_save_paths_repairs_stale_default_roots(monkeypatch: pytest.MonkeyPatch, tmp_path): def test_save_paths_repairs_stale_default_roots(monkeypatch: pytest.MonkeyPatch, tmp_path):
@@ -214,9 +235,9 @@ def test_save_paths_repairs_stale_default_roots(monkeypatch: pytest.MonkeyPatch,
config_module.Config() config_module.Config()
assert fake_settings.name == "comfyui" assert fake_settings.name == "comfyui"
assert fake_settings.payload["default_lora_root"] == folder_paths["loras"][0].replace("\\", "/") assert _identity(fake_settings.payload["default_lora_root"]) == _identity(folder_paths["loras"][0].replace("\\", "/"))
assert fake_settings.payload["default_checkpoint_root"] == folder_paths["checkpoints"][0].replace("\\", "/") assert _identity(fake_settings.payload["default_checkpoint_root"]) == _identity(folder_paths["checkpoints"][0].replace("\\", "/"))
assert fake_settings.payload["default_embedding_root"] == folder_paths["embeddings"][0].replace("\\", "/") assert _identity(fake_settings.payload["default_embedding_root"]) == _identity(folder_paths["embeddings"][0].replace("\\", "/"))
def test_save_paths_keeps_valid_default_roots(monkeypatch: pytest.MonkeyPatch, tmp_path): def test_save_paths_keeps_valid_default_roots(monkeypatch: pytest.MonkeyPatch, tmp_path):
@@ -253,9 +274,9 @@ def test_save_paths_keeps_valid_default_roots(monkeypatch: pytest.MonkeyPatch, t
config_module.Config() config_module.Config()
assert fake_settings.name == "comfyui" assert fake_settings.name == "comfyui"
assert fake_settings.payload["default_lora_root"] == folder_paths["loras"][0].replace("\\", "/") assert _identity(fake_settings.payload["default_lora_root"]) == _identity(folder_paths["loras"][0].replace("\\", "/"))
assert fake_settings.payload["default_checkpoint_root"] == folder_paths["checkpoints"][0].replace("\\", "/") assert _identity(fake_settings.payload["default_checkpoint_root"]) == _identity(folder_paths["checkpoints"][0].replace("\\", "/"))
assert fake_settings.payload["default_embedding_root"] == folder_paths["embeddings"][0].replace("\\", "/") assert _identity(fake_settings.payload["default_embedding_root"]) == _identity(folder_paths["embeddings"][0].replace("\\", "/"))
def test_save_paths_removes_template_default_library(monkeypatch, tmp_path): def test_save_paths_removes_template_default_library(monkeypatch, tmp_path):
@@ -338,8 +359,8 @@ def test_save_paths_removes_template_default_library(monkeypatch, tmp_path):
key: [path.replace("\\", "/") for path in paths] key: [path.replace("\\", "/") for path in paths]
for key, paths in folder_paths.items() for key, paths in folder_paths.items()
} }
assert payload["folder_paths"] == expected_folder_paths assert _identity(payload["folder_paths"]) == _identity(expected_folder_paths)
assert payload["default_lora_root"] == folder_paths["loras"][0].replace("\\", "/") assert _identity(payload["default_lora_root"]) == _identity(folder_paths["loras"][0].replace("\\", "/"))
assert ( assert (
payload["default_checkpoint_root"] payload["default_checkpoint_root"]
== folder_paths["checkpoints"][0].replace("\\", "/") == folder_paths["checkpoints"][0].replace("\\", "/")
@@ -348,7 +369,7 @@ def test_save_paths_removes_template_default_library(monkeypatch, tmp_path):
payload["default_embedding_root"] payload["default_embedding_root"]
== folder_paths["embeddings"][0].replace("\\", "/") == folder_paths["embeddings"][0].replace("\\", "/")
) )
assert payload["metadata"] == {"display_name": "ComfyUI", "source": "comfyui"} assert _identity(payload["metadata"]) == _identity({"display_name": "ComfyUI", "source": "comfyui"})
assert payload["activate"] is True assert payload["activate"] is True
@@ -397,16 +418,16 @@ def test_save_paths_keeps_default_roots_in_extra_paths(monkeypatch: pytest.Monke
config_module.Config() config_module.Config()
assert fake_settings.name == "comfyui" assert fake_settings.name == "comfyui"
assert fake_settings.payload["extra_folder_paths"]["loras"] == [str(extra_lora_dir).replace("\\", "/")] assert _identity(fake_settings.payload["extra_folder_paths"]["loras"]) == _identity([str(extra_lora_dir).replace("\\", "/")])
assert fake_settings.payload["extra_folder_paths"]["checkpoints"] == [ assert fake_settings.payload["extra_folder_paths"]["checkpoints"] == [
str(extra_checkpoint_dir).replace("\\", "/") str(extra_checkpoint_dir).replace("\\", "/")
] ]
assert fake_settings.payload["extra_folder_paths"]["embeddings"] == [ assert fake_settings.payload["extra_folder_paths"]["embeddings"] == [
str(extra_embedding_dir).replace("\\", "/") str(extra_embedding_dir).replace("\\", "/")
] ]
assert fake_settings.payload["default_lora_root"] == str(extra_lora_dir).replace("\\", "/") assert _identity(fake_settings.payload["default_lora_root"]) == _identity(str(extra_lora_dir).replace("\\", "/"))
assert fake_settings.payload["default_checkpoint_root"] == str(extra_checkpoint_dir).replace("\\", "/") assert _identity(fake_settings.payload["default_checkpoint_root"]) == _identity(str(extra_checkpoint_dir).replace("\\", "/"))
assert fake_settings.payload["default_embedding_root"] == str(extra_embedding_dir).replace("\\", "/") assert _identity(fake_settings.payload["default_embedding_root"]) == _identity(str(extra_embedding_dir).replace("\\", "/"))
assert fake_settings.payload["activate"] is True assert fake_settings.payload["activate"] is True
@@ -451,8 +472,8 @@ def test_save_paths_keeps_default_roots_in_extra_paths_with_windows_slash_mismat
config_module.Config() config_module.Config()
assert fake_settings.name == "comfyui" assert fake_settings.name == "comfyui"
assert fake_settings.payload["default_lora_root"] == "U:/Lora7/Loras" assert _identity(fake_settings.payload["default_lora_root"]) == _identity("U:/Lora7/Loras")
assert fake_settings.payload["default_checkpoint_root"] == "U:/Lora7/Models" assert _identity(fake_settings.payload["default_checkpoint_root"]) == _identity("U:/Lora7/Models")
def test_save_paths_repairs_empty_default_roots_to_extra_paths_when_primary_missing( def test_save_paths_repairs_empty_default_roots_to_extra_paths_when_primary_missing(
@@ -505,7 +526,7 @@ def test_save_paths_repairs_empty_default_roots_to_extra_paths_when_primary_miss
config_module.Config() config_module.Config()
assert fake_settings.name == "comfyui" assert fake_settings.name == "comfyui"
assert fake_settings.payload["default_lora_root"] == str(extra_lora_dir).replace("\\", "/") assert _identity(fake_settings.payload["default_lora_root"]) == _identity(str(extra_lora_dir).replace("\\", "/"))
def test_save_paths_does_not_activate_comfyui_library_when_another_library_is_active( def test_save_paths_does_not_activate_comfyui_library_when_another_library_is_active(
@@ -585,12 +606,12 @@ def test_apply_library_settings_merges_extra_paths(monkeypatch, tmp_path):
config_instance.apply_library_settings(library_config) config_instance.apply_library_settings(library_config)
assert str(loras_dir) in config_instance.loras_roots assert _identity(str(loras_dir)) in _identity(config_instance.loras_roots)
assert str(extra_loras_dir) in config_instance.extra_loras_roots assert _identity(str(extra_loras_dir)) in _identity(config_instance.extra_loras_roots)
assert str(checkpoints_dir) in config_instance.base_models_roots assert str(checkpoints_dir) in config_instance.base_models_roots
assert str(extra_checkpoints_dir) in config_instance.extra_checkpoints_roots assert _identity(str(extra_checkpoints_dir)) in _identity(config_instance.extra_checkpoints_roots)
assert str(embeddings_dir) in (config_instance.embeddings_roots or []) assert str(embeddings_dir) in (config_instance.embeddings_roots or [])
assert str(extra_embeddings_dir) in config_instance.extra_embeddings_roots assert _identity(str(extra_embeddings_dir)) in _identity(config_instance.extra_embeddings_roots)
def test_apply_library_settings_without_extra_paths(monkeypatch, tmp_path): def test_apply_library_settings_without_extra_paths(monkeypatch, tmp_path):
@@ -617,12 +638,12 @@ def test_apply_library_settings_without_extra_paths(monkeypatch, tmp_path):
config_instance.apply_library_settings(library_config) config_instance.apply_library_settings(library_config)
assert str(loras_dir) in config_instance.loras_roots assert _identity(str(loras_dir)) in _identity(config_instance.loras_roots)
assert config_instance.extra_loras_roots == [] assert _identity(config_instance.extra_loras_roots) == _identity([])
assert str(checkpoints_dir) in config_instance.base_models_roots assert str(checkpoints_dir) in config_instance.base_models_roots
assert config_instance.extra_checkpoints_roots == [] assert _identity(config_instance.extra_checkpoints_roots) == _identity([])
assert str(embeddings_dir) in (config_instance.embeddings_roots or []) assert str(embeddings_dir) in (config_instance.embeddings_roots or [])
assert config_instance.extra_embeddings_roots == [] assert _identity(config_instance.extra_embeddings_roots) == _identity([])
def test_extra_paths_deduplication(monkeypatch, tmp_path): def test_extra_paths_deduplication(monkeypatch, tmp_path):
@@ -654,8 +675,8 @@ def test_extra_paths_deduplication(monkeypatch, tmp_path):
config_instance.apply_library_settings(library_config) config_instance.apply_library_settings(library_config)
assert config_instance.loras_roots == [str(loras_dir)] assert _identity(config_instance.loras_roots) == _identity([str(loras_dir)])
assert config_instance.extra_loras_roots == [str(extra_loras_dir)] assert _identity(config_instance.extra_loras_roots) == _identity([str(extra_loras_dir)])
def test_apply_library_settings_ignores_extra_lora_path_overlapping_primary_symlink( def test_apply_library_settings_ignores_extra_lora_path_overlapping_primary_symlink(
@@ -687,8 +708,8 @@ def test_apply_library_settings_ignores_extra_lora_path_overlapping_primary_syml
with caplog.at_level("WARNING", logger=config_module.logger.name): with caplog.at_level("WARNING", logger=config_module.logger.name):
config_instance.apply_library_settings(library_config) config_instance.apply_library_settings(library_config)
assert config_instance.loras_roots == [str(loras_link)] assert _identity(config_instance.loras_roots) == _identity([str(loras_link)])
assert config_instance.extra_loras_roots == [] assert _identity(config_instance.extra_loras_roots) == _identity([])
warning_messages = [ warning_messages = [
record.message record.message
@@ -764,8 +785,8 @@ def test_apply_library_settings_detects_overlap_case_insensitively(
with caplog.at_level("WARNING", logger=config_module.logger.name): with caplog.at_level("WARNING", logger=config_module.logger.name):
config_instance.apply_library_settings(library_config) config_instance.apply_library_settings(library_config)
assert config_instance.loras_roots == [primary_path] assert _identity(config_instance.loras_roots) == _identity([primary_path])
assert config_instance.extra_loras_roots == [] assert _identity(config_instance.extra_loras_roots) == _identity([])
assert any("same lora folder" in record.message.lower() for record in caplog.records) assert any("same lora folder" in record.message.lower() for record in caplog.records)
@@ -794,8 +815,8 @@ def test_apply_library_settings_ignores_missing_extra_lora_paths(monkeypatch, tm
with caplog.at_level("WARNING", logger=config_module.logger.name): with caplog.at_level("WARNING", logger=config_module.logger.name):
config_instance.apply_library_settings(library_config) config_instance.apply_library_settings(library_config)
assert config_instance.loras_roots == [str(loras_dir)] assert _identity(config_instance.loras_roots) == _identity([str(loras_dir)])
assert config_instance.extra_loras_roots == [] assert _identity(config_instance.extra_loras_roots) == _identity([])
assert not any("same lora folder" in record.message.lower() for record in caplog.records) assert not any("same lora folder" in record.message.lower() for record in caplog.records)
@@ -829,8 +850,8 @@ def test_apply_library_settings_ignores_extra_lora_path_overlapping_primary_root
with caplog.at_level("WARNING", logger=config_module.logger.name): with caplog.at_level("WARNING", logger=config_module.logger.name):
config_instance.apply_library_settings(library_config) config_instance.apply_library_settings(library_config)
assert config_instance.loras_roots == [str(loras_dir)] assert _identity(config_instance.loras_roots) == _identity([str(loras_dir)])
assert config_instance.extra_loras_roots == [] assert _identity(config_instance.extra_loras_roots) == _identity([])
assert any( assert any(
"same lora folder" in record.message.lower() "same lora folder" in record.message.lower()
for record in caplog.records for record in caplog.records
@@ -960,14 +981,14 @@ def test_save_paths_keeps_configured_root_of_a_switched_off_drive(
persisted = fake_settings.payload["folder_paths"]["loras"] persisted = fake_settings.payload["folder_paths"]["loras"]
# The switched-off drive keeps its configured path... # The switched-off drive keeps its configured path...
assert switched_off.replace("\\", "/") in persisted assert _identity(switched_off.replace("\\", "/")) in _identity(persisted)
assert readable.replace("\\", "/") in persisted assert _identity(readable.replace("\\", "/")) in _identity(persisted)
# ...the path the host really dropped is gone... # ...the path the host really dropped is gone...
assert removed_by_user.replace("\\", "/") not in persisted assert _identity(removed_by_user.replace("\\", "/")) not in _identity(persisted)
# ...and the live list still only holds what is readable right now. # ...and the live list still only holds what is readable right now.
assert config_instance.loras_roots == [readable.replace("\\", "/")] assert _identity(config_instance.loras_roots) == _identity([readable.replace("\\", "/")])
# The user's default root survives because it is merely unavailable. # The user's default root survives because it is merely unavailable.
assert fake_settings.payload["default_lora_root"] == switched_off.replace("\\", "/") assert _identity(fake_settings.payload["default_lora_root"]) == _identity(switched_off.replace("\\", "/"))
# --- roots the host registered but never created (issue #1146) -------------- # --- roots the host registered but never created (issue #1146) --------------
@@ -1067,9 +1088,9 @@ def test_save_paths_drops_registered_directories_that_never_existed(
("unet", output_unet), ("unet", output_unet),
("embeddings", output_embeddings), ("embeddings", output_embeddings),
): ):
assert never_created.replace("\\", "/") not in persisted[key], key assert _identity(never_created.replace("\\", "/")) not in _identity(persisted[key]), key
# The real root of that key is untouched. # The real root of that key is untouched.
assert folder_paths[key][0].replace("\\", "/") in persisted[key], key assert _identity(folder_paths[key][0].replace("\\", "/")) in _identity(persisted[key]), key
# Lora and checkpoint roots are looked up in their own model type. # Lora and checkpoint roots are looked up in their own model type.
assert [model_type for model_type, _roots in cache.queries] == [ assert [model_type for model_type, _roots in cache.queries] == [
"lora", "lora",
@@ -1102,8 +1123,8 @@ def test_save_paths_keeps_an_unavailable_root_that_has_cached_models(
config_module.Config() config_module.Config()
persisted = fake_settings.payload["folder_paths"]["loras"] persisted = fake_settings.payload["folder_paths"]["loras"]
assert switched_off.replace("\\", "/") in persisted assert _identity(switched_off.replace("\\", "/")) in _identity(persisted)
assert readable.replace("\\", "/") in persisted assert _identity(readable.replace("\\", "/")) in _identity(persisted)
def test_save_paths_keeps_every_root_when_the_cache_cannot_be_read( def test_save_paths_keeps_every_root_when_the_cache_cannot_be_read(
@@ -1129,7 +1150,7 @@ def test_save_paths_keeps_every_root_when_the_cache_cannot_be_read(
config_module.Config() config_module.Config()
persisted = fake_settings.payload["folder_paths"]["loras"] persisted = fake_settings.payload["folder_paths"]["loras"]
assert unknown.replace("\\", "/") in persisted assert _identity(unknown.replace("\\", "/")) in _identity(persisted)
def test_save_paths_keeps_the_root_a_default_setting_points_at( def test_save_paths_keeps_the_root_a_default_setting_points_at(
@@ -1160,5 +1181,5 @@ def test_save_paths_keeps_the_root_a_default_setting_points_at(
config_module.Config() config_module.Config()
persisted = fake_settings.payload["folder_paths"]["unet"] persisted = fake_settings.payload["folder_paths"]["unet"]
assert default_target.replace("\\", "/") in persisted assert _identity(default_target.replace("\\", "/")) in _identity(persisted)
assert str(unet_dir).replace("\\", "/") in persisted assert _identity(str(unet_dir).replace("\\", "/")) in _identity(persisted)
+4 -3
View File
@@ -6,6 +6,7 @@ import pytest
from py.nodes.lora_cycler import LoraCyclerLM from py.nodes.lora_cycler import LoraCyclerLM
from py.services import service_registry from py.services import service_registry
import os
@pytest.fixture @pytest.fixture
@@ -48,7 +49,7 @@ async def test_cycler_uses_scaled_preset_strength_when_available(
result = await cycler_node.cycle(cycler_config) result = await cycler_node.cycle(cycler_config)
assert result["result"][0] == [ assert result["result"][0] == [
("/models/loras/preset_lora.safetensors", 1.05, 0.75) ("/models/loras/preset_lora.safetensors".replace("/", os.sep), 1.05, 0.75)
] ]
@@ -74,7 +75,7 @@ async def test_cycler_falls_back_to_manual_strength_when_preset_missing(
result = await cycler_node.cycle(cycler_config) result = await cycler_node.cycle(cycler_config)
assert result["result"][0] == [ assert result["result"][0] == [
("/models/loras/manual_lora.safetensors", 0.8, 0.6) ("/models/loras/manual_lora.safetensors".replace("/", os.sep), 0.8, 0.6)
] ]
@@ -105,5 +106,5 @@ async def test_cycler_syncs_clip_to_model_when_same_clip_strength_enabled(
) )
assert result["result"][0] == [ assert result["result"][0] == [
("/models/loras/preset_lora.safetensors", 1.05, 1.05) ("/models/loras/preset_lora.safetensors".replace("/", os.sep), 1.05, 1.05)
] ]
+2 -2
View File
@@ -932,8 +932,8 @@ def test_resolve_folder_answers_with_the_root_that_holds_the_directory(
assert payload["folder"] == "recipes" assert payload["folder"] == "recipes"
assert payload["candidates"] == [ assert payload["candidates"] == [
{ {
"folder_path": str(primary / "recipes"), "folder_path": str(primary / "recipes").replace(os.sep, "/"),
"root": str(primary), "root": str(primary).replace(os.sep, "/"),
"is_symlink": False, "is_symlink": False,
} }
] ]
+4
View File
@@ -175,6 +175,10 @@ async def test_validate_path_expands_user(tmp_path, monkeypatch):
subdir = tmp_path / "subdir" subdir = tmp_path / "subdir"
subdir.mkdir() subdir.mkdir()
monkeypatch.setenv("HOME", str(tmp_path)) monkeypatch.setenv("HOME", str(tmp_path))
# ntpath.expanduser reads USERPROFILE (or HOMEDRIVE + HOMEPATH), never HOME.
monkeypatch.setenv("USERPROFILE", str(tmp_path))
monkeypatch.setenv("HOMEDRIVE", tmp_path.drive)
monkeypatch.setenv("HOMEPATH", str(tmp_path)[len(tmp_path.drive):])
response, payload = await _validate(_make_handler(), "~/subdir") response, payload = await _validate(_make_handler(), "~/subdir")
+25 -6
View File
@@ -47,6 +47,18 @@ def _json_payload(response) -> dict[str, Any]:
return json.loads(text) return json.loads(text)
def _expected_opener() -> str:
"""The command the handler uses to open a path on this platform."""
return "explorer" if os.name == "nt" else "xdg-open"
def _posix_only(reason: str):
"""Mark a test that exercises the POSIX open strategy (xdg-open/dbus)."""
return pytest.mark.skipif(os.name == "nt", reason=reason)
class FakeRequest: class FakeRequest:
def __init__(self, *, json_data=None, query=None, method="POST"): def __init__(self, *, json_data=None, query=None, method="POST"):
self._json_data = json_data or {} self._json_data = json_data or {}
@@ -636,7 +648,7 @@ async def test_open_backup_location_uses_settings_directory(tmp_path, monkeypatc
assert response.status == 200 assert response.status == 200
assert payload["success"] is True assert payload["success"] is True
assert payload["path"] == str(backup_dir) assert payload["path"] == str(backup_dir)
assert calls == [["xdg-open", str(backup_dir)]] assert calls == [[_expected_opener(), str(backup_dir)]]
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -667,9 +679,10 @@ async def test_open_sidecar_location_opens_configured_root(tmp_path, monkeypatch
assert payload["path"] == str(root) assert payload["path"] == str(root)
# Created on demand so the button works before any migration ran. # Created on demand so the button works before any migration ran.
assert root.is_dir() assert root.is_dir()
assert calls == [["xdg-open", str(root)]] assert calls == [[_expected_opener(), str(root)]]
@_posix_only("the headless clipboard fallback is the POSIX strategy; Windows always has a shell")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_open_sidecar_location_headless_returns_clipboard_mode(tmp_path, monkeypatch): async def test_open_sidecar_location_headless_returns_clipboard_mode(tmp_path, monkeypatch):
"""Without a GUI session xdg-open cannot work; the handler must hand the """Without a GUI session xdg-open cannot work; the handler must hand the
@@ -721,6 +734,7 @@ async def test_get_settings_includes_resolved_sidecar_root(tmp_path):
assert payload["settings"]["sidecar_storage_root_in_repo"] is False assert payload["settings"]["sidecar_storage_root_in_repo"] is False
@_posix_only("the headless clipboard fallback is the POSIX strategy; Windows always has a shell")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_open_settings_location_headless_returns_clipboard_mode(tmp_path, monkeypatch): async def test_open_settings_location_headless_returns_clipboard_mode(tmp_path, monkeypatch):
"""Without a GUI session xdg-open cannot work; the handler must hand the """Without a GUI session xdg-open cannot work; the handler must hand the
@@ -774,7 +788,7 @@ async def test_open_settings_location_with_display_opens_folder(tmp_path, monkey
assert response.status == 200 assert response.status == 200
assert payload["success"] is True assert payload["success"] is True
assert calls == [["xdg-open", str(settings_file.parent)]] assert calls == [[_expected_opener(), str(settings_file.parent)]]
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -807,9 +821,10 @@ async def test_open_wildcards_location_creates_and_opens_directory(tmp_path, mon
assert payload["success"] is True assert payload["success"] is True
assert payload["path"] == str(wildcards_dir) assert payload["path"] == str(wildcards_dir)
assert wildcards_dir.is_dir() assert wildcards_dir.is_dir()
assert calls == [["xdg-open", str(wildcards_dir)]] assert calls == [[_expected_opener(), str(wildcards_dir)]]
@_posix_only("dbus file-manager selection is Linux-only")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_open_file_location_linux_selects_via_dbus(tmp_path, monkeypatch): async def test_open_file_location_linux_selects_via_dbus(tmp_path, monkeypatch):
"""On a Linux desktop the handler should ask the freedesktop FileManager1 """On a Linux desktop the handler should ask the freedesktop FileManager1
@@ -849,6 +864,7 @@ async def test_open_file_location_linux_selects_via_dbus(tmp_path, monkeypatch):
assert popen_calls == [] assert popen_calls == []
@_posix_only("xdg-open is the Linux opener")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_open_file_location_linux_falls_back_to_xdg_open(tmp_path, monkeypatch): async def test_open_file_location_linux_falls_back_to_xdg_open(tmp_path, monkeypatch):
"""When the FileManager1 DBus interface is unavailable, fall back to """When the FileManager1 DBus interface is unavailable, fall back to
@@ -877,9 +893,10 @@ async def test_open_file_location_linux_falls_back_to_xdg_open(tmp_path, monkeyp
assert response.status == 200 assert response.status == 200
assert payload["success"] is True assert payload["success"] is True
assert popen_calls == [["xdg-open", str(tmp_path)]] assert popen_calls == [[_expected_opener(), str(tmp_path)]]
@_posix_only("the headless clipboard fallback is the POSIX strategy; Windows always has a shell")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_open_file_location_headless_returns_clipboard_mode(tmp_path, monkeypatch): async def test_open_file_location_headless_returns_clipboard_mode(tmp_path, monkeypatch):
"""Without a GUI session there is no file manager to open; the handler """Without a GUI session there is no file manager to open; the handler
@@ -917,6 +934,7 @@ def _patch_linux_desktop(monkeypatch):
monkeypatch.setattr("py.routes.handlers.misc_handlers._is_wsl", lambda: False) monkeypatch.setattr("py.routes.handlers.misc_handlers._is_wsl", lambda: False)
@_posix_only("file-manager reveal via dbus is Linux-only")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_open_model_sidecar_location_reveals_metadata_file(tmp_path, monkeypatch): async def test_open_model_sidecar_location_reveals_metadata_file(tmp_path, monkeypatch):
"""Centralized mode: the resolved .metadata.json is revealed selected, """Centralized mode: the resolved .metadata.json is revealed selected,
@@ -985,7 +1003,7 @@ async def test_open_model_sidecar_location_falls_back_to_sidecar_dir(tmp_path, m
assert response.status == 200 assert response.status == 200
assert payload["success"] is True assert payload["success"] is True
assert payload["path"] == str(sidecar_dir) assert payload["path"] == str(sidecar_dir)
assert popen_calls == [["xdg-open", str(sidecar_dir)]] assert popen_calls == [[_expected_opener(), str(sidecar_dir)]]
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -1012,6 +1030,7 @@ async def test_open_model_sidecar_location_missing_sidecar_returns_404(tmp_path,
assert payload["success"] is False assert payload["success"] is False
@_posix_only("the headless clipboard fallback is the POSIX strategy; Windows always has a shell")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_open_model_sidecar_location_headless_returns_clipboard_mode(tmp_path, monkeypatch): async def test_open_model_sidecar_location_headless_returns_clipboard_mode(tmp_path, monkeypatch):
model_file = tmp_path / "model.safetensors" model_file = tmp_path / "model.safetensors"
+1 -3
View File
@@ -403,9 +403,7 @@ def test_find_matching_root_includes_other_roots(isolated_config_roots, tmp_path
vae_root = tmp_path / "vae" vae_root = tmp_path / "vae"
isolated_config_roots.other_roots = [str(vae_root)] isolated_config_roots.other_roots = [str(vae_root)]
assert model_source_handlers._find_matching_root(str(vae_root / "sub")) == str( assert model_source_handlers._find_matching_root(str(vae_root / "sub")) == str(vae_root).replace(os.sep, "/").replace(os.sep, "/")
vae_root
)
def test_find_matching_root_respects_path_boundaries(isolated_config_roots, tmp_path): def test_find_matching_root_respects_path_boundaries(isolated_config_roots, tmp_path):
@@ -532,7 +532,7 @@ async def test_execute_download_extracts_zip_single_model(monkeypatch, tmp_path)
extracted = save_dir / "model.safetensors" extracted = save_dir / "model.safetensors"
assert extracted.exists() assert extracted.exists()
saved_call = MetadataManager.save_metadata.await_args saved_call = MetadataManager.save_metadata.await_args
assert saved_call.args[0] == str(extracted) assert saved_call.args[0] == str(extracted).replace(os.sep, "/")
# SHA256 comes from metadata (API value), not recalculated # SHA256 comes from metadata (API value), not recalculated
assert saved_call.args[1].sha256 == "sha256" assert saved_call.args[1].sha256 == "sha256"
# The zip-level AutoV3 was cleared; the extracted file is not safetensors, # The zip-level AutoV3 was cleared; the extracted file is not safetensors,
@@ -693,7 +693,7 @@ async def test_execute_download_extracts_zip_pt_embedding(monkeypatch, tmp_path)
extracted = save_dir / "embedding.pt" extracted = save_dir / "embedding.pt"
assert extracted.exists() assert extracted.exists()
saved_call = MetadataManager.save_metadata.await_args saved_call = MetadataManager.save_metadata.await_args
assert saved_call.args[0] == str(extracted) assert saved_call.args[0] == str(extracted).replace(os.sep, "/")
# SHA256 comes from metadata (API value), not recalculated # SHA256 comes from metadata (API value), not recalculated
assert saved_call.args[1].sha256 == "sha256" assert saved_call.args[1].sha256 == "sha256"
assert dummy_scanner.add_model_to_cache.await_count == 1 assert dummy_scanner.add_model_to_cache.await_count == 1
+29 -6
View File
@@ -48,6 +48,20 @@ class RecordingWebSocketManager:
self.broadcasts.append(payload) self.broadcasts.append(payload)
def _root_label_names(labels) -> List[str]:
"""The root names behind display labels.
``_root_display_labels`` prefixes the drive on Windows ("C: loras") and may
grow a label leftwards to keep it unique ("parent/loras"), so a test that only
cares *which* root was walked compares the final path component.
"""
return [
label.replace("\\", "/").rstrip("/").rsplit("/", 1)[-1].rsplit(" ", 1)[-1]
for label in labels
]
def _normalize_path(path: Path) -> str: def _normalize_path(path: Path) -> str:
return str(path).replace(os.sep, "/") return str(path).replace(os.sep, "/")
@@ -2306,6 +2320,10 @@ def test_count_cached_entries_per_root_uses_longest_prefix():
@pytest.mark.asyncio @pytest.mark.asyncio
@pytest.mark.skipif(
os.name == "nt",
reason="simulates a case-sensitive filesystem; Windows matches case itself",
)
async def test_reconcile_case_fold_fallback_is_indexed_not_linear( async def test_reconcile_case_fold_fallback_is_indexed_not_linear(
tmp_path: Path, monkeypatch tmp_path: Path, monkeypatch
): ):
@@ -2465,7 +2483,10 @@ async def test_reconcile_walk_broadcasts_per_root_progress(tmp_path: Path, monke
assert len(walk_messages) >= 2 assert len(walk_messages) >= 2
# Intermediate ticks name the root being walked. # Intermediate ticks name the root being walked.
assert any(message["active_roots"] == ["loras"] for message in walk_messages) assert any(
_root_label_names(message["active_roots"]) == ["loras"]
for message in walk_messages
)
final = walk_messages[-1] final = walk_messages[-1]
assert final["files_seen"] == 4 assert final["files_seen"] == 4
@@ -2546,7 +2567,7 @@ async def test_reconcile_scoped_scan_leaves_other_roots_untouched(tmp_path: Path
assert summary is not None assert summary is not None
assert summary["added"] == 1 assert summary["added"] == 1
assert summary["removed"] == 0 assert summary["removed"] == 0
assert summary["scanned_roots"] == ["a"] assert _root_label_names(summary["scanned_roots"]) == ["a"]
cached_paths = {item["file_path"] for item in scanner._cache.raw_data} cached_paths = {item["file_path"] for item in scanner._cache.raw_data}
assert cached_paths == { assert cached_paths == {
@@ -2717,11 +2738,13 @@ async def test_reconcile_reports_root_filtered_out_at_startup(tmp_path: Path, mo
assert summary["added"] == 0 assert summary["added"] == 0
assert summary["removed"] == 0 assert summary["removed"] == 0
# ...but the switched-off root is no longer silent: it is reported with the # ...but the switched-off root is no longer silent: it is reported with the
# entries that were kept for it. # entries that were kept for it. Its label is what the shared label helper
# makes of the pair (drive-prefixed on Windows).
labels = model_scanner._root_display_labels([str(_primary), str(missing)])
assert summary["skipped_roots"] == [ assert summary["skipped_roots"] == [
{ {
"path": _normalize_path(missing), "path": _normalize_path(missing),
"label": missing.name, "label": labels[str(missing)],
"reason": "root_unavailable", "reason": "root_unavailable",
} }
] ]
@@ -2746,7 +2769,7 @@ async def test_scoped_scan_stays_quiet_about_filtered_out_roots(tmp_path: Path,
assert summary is not None assert summary is not None
# The caller asked for one root: a root it did not ask about is not its # The caller asked for one root: a root it did not ask about is not its
# business, but its entries are still kept. # business, but its entries are still kept.
assert summary["scanned_roots"] == [primary.name] assert _root_label_names(summary["scanned_roots"]) == [primary.name]
assert "root_unavailable" not in { assert "root_unavailable" not in {
entry["reason"] for entry in summary["skipped_roots"] entry["reason"] for entry in summary["skipped_roots"]
} }
@@ -3091,7 +3114,7 @@ async def test_scoped_scan_accepts_the_host_path_spelling(tmp_path: Path, monkey
summary = await scanner._reconcile_cache(scope=scope) summary = await scanner._reconcile_cache(scope=scope)
assert summary is not None assert summary is not None
assert summary["scanned_roots"] == [primary.name] assert _root_label_names(summary["scanned_roots"]) == [primary.name]
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -901,7 +901,7 @@ async def test_p_checkpoint_pending_discovery_excludes_staging(tmp_path: Path, m
pending = await scanner._find_pending_models_from_filesystem() pending = await scanner._find_pending_models_from_filesystem()
paths = [entry["file_path"] for entry in pending] paths = [entry["file_path"] for entry in pending]
assert str(root / "real.safetensors") in paths assert str(root / "real.safetensors").replace(os.sep, "/") in paths
assert not any(PENDING_DELETE_DIR_NAME in p for p in paths) assert not any(PENDING_DELETE_DIR_NAME in p for p in paths)
@@ -921,7 +921,7 @@ async def test_p_usage_stats_lookup_excludes_staging(tmp_path: Path) -> None:
) )
# Staged file is not matched; only the real one is returned. # Staged file is not matched; only the real one is returned.
assert result == str(root / "mycheckpoint.safetensors") assert result == str(root / "mycheckpoint.safetensors").replace(os.sep, "/")
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -62,7 +62,7 @@ async def test_pacing_enforces_min_interval(monkeypatch):
await coordinator.wait_for_slot("example.com") await coordinator.wait_for_slot("example.com")
await coordinator.wait_for_slot("example.com") await coordinator.wait_for_slot("example.com")
elapsed = time.monotonic() - start elapsed = time.monotonic() - start
assert elapsed >= 0.1 assert elapsed >= 0.09 # Windows timer granularity rounds 0.1s down
async def test_pacing_is_per_destination(monkeypatch): async def test_pacing_is_per_destination(monkeypatch):
@@ -89,7 +89,7 @@ async def test_register_rate_limit_arms_cooldown_and_waits(monkeypatch):
start = time.monotonic() start = time.monotonic()
await coordinator.wait_for_slot("example.com") await coordinator.wait_for_slot("example.com")
elapsed = time.monotonic() - start elapsed = time.monotonic() - start
assert elapsed >= 0.14 assert elapsed >= 0.09 # Windows timer granularity rounds 0.1s down4
assert not coordinator.in_cooldown("example.com") assert not coordinator.in_cooldown("example.com")
@@ -255,7 +255,7 @@ async def test_make_request_paces_consecutive_calls(monkeypatch):
await downloader.make_request("GET", "https://api.example.com/a") await downloader.make_request("GET", "https://api.example.com/a")
await downloader.make_request("GET", "https://api.example.com/b") await downloader.make_request("GET", "https://api.example.com/b")
elapsed = time.monotonic() - start elapsed = time.monotonic() - start
assert elapsed >= 0.14 assert elapsed >= 0.09 # Windows timer granularity rounds 0.1s down4
async def test_make_request_gate_disabled_returns_429_immediately(monkeypatch): async def test_make_request_gate_disabled_returns_429_immediately(monkeypatch):
+7 -3
View File
@@ -36,6 +36,7 @@ from py.utils.example_images_processor import (
) )
from py.utils.metadata_manager import MetadataManager from py.utils.metadata_manager import MetadataManager
from tests.conftest import MockCache, MockModelService, MockScanner from tests.conftest import MockCache, MockModelService, MockScanner
import os
class StubLockProvider: class StubLockProvider:
@@ -655,7 +656,8 @@ async def test_filename_template_use_case_reverts_to_recorded_original_when_temp
lifecycle = StubLifecycleService() lifecycle = StubLifecycleService()
async def metadata_loader(metadata_path: str) -> Dict[str, Any]: async def metadata_loader(metadata_path: str) -> Dict[str, Any]:
if metadata_path == "/library/alpha-renamed.metadata.json": # Compare the basename: the sidecar path is platform-spelled.
if os.path.basename(metadata_path) == "alpha-renamed.metadata.json":
return {"original_file_name": "alpha-original"} return {"original_file_name": "alpha-original"}
return {} return {}
@@ -713,8 +715,10 @@ async def test_filename_template_use_case_skips_revert_matching_current_name() -
async def test_filename_template_use_case_counts_conflicts_as_failures() -> None: async def test_filename_template_use_case_counts_conflicts_as_failures() -> None:
_set_filename_template("{model_name}") _set_filename_template("{model_name}")
scanner = MockScanner(cache=MockCache([ scanner = MockScanner(cache=MockCache([
_filename_template_model("/library/alpha.safetensors", "Alpha"), # The names differ beyond case: on Windows a case-only difference counts
_filename_template_model("/library/beta.safetensors", "Beta"), # as "already named that" (os.path.normcase) and would be skipped.
_filename_template_model("/library/alpha.safetensors", "Alpha Two"),
_filename_template_model("/library/beta.safetensors", "Beta Two"),
])) ]))
lifecycle = StubLifecycleService() lifecycle = StubLifecycleService()
lifecycle.error = ValueError("A file with this name already exists") lifecycle.error = ValueError("A file with this name already exists")
+2 -1
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import json import json
import pytest import pytest
import os
from py.utils.metadata_manager import MetadataManager from py.utils.metadata_manager import MetadataManager
from py.utils.models import LoraMetadata from py.utils.models import LoraMetadata
@@ -174,7 +175,7 @@ async def test_load_metadata_payload_tolerates_missing_model_file(tmp_path) -> N
"""A nonexistent model file must not crash payload loading.""" """A nonexistent model file must not crash payload loading."""
missing_path = tmp_path / "Ghost.safetensors" missing_path = tmp_path / "Ghost.safetensors"
payload = await MetadataManager.load_metadata_payload(str(missing_path)) payload = await MetadataManager.load_metadata_payload(str(missing_path))
assert payload["file_path"] == str(missing_path) assert payload["file_path"] == str(missing_path).replace(os.sep, "/")
assert "file_name" not in payload assert "file_name" not in payload
assert "size" not in payload assert "size" not in payload
assert "modified" not in payload assert "modified" not in payload