Download ComfyUI/custom_nodes/koolook/tests/server/test_published_setups_relocation.py from vidfom/aimusic: direct link, hf CLI and curl.
- Browser
- Download file 7.24 kB
-
https://huggingface.co/vidfom/aimusic/resolve/main/ComfyUI/custom_nodes/koolook/tests/server/test_published_setups_relocation.py
- Command line
-
hf download hf://vidfom/aimusic/ComfyUI/custom_nodes/koolook/tests/server/test_published_setups_relocation.py
-
curl -L -o test_published_setups_relocation.py https://huggingface.co/vidfom/aimusic/resolve/main/ComfyUI/custom_nodes/koolook/tests/server/test_published_setups_relocation.py
7.24 kB
| from __future__ import annotations | |
| import koolook_routes | |
| from koolook_setups import DEFAULT_SETUPS_FILENAME, DEFAULT_SETUPS_SUBDIR | |
| def test_published_setups_path_sits_beside_snapshot_library(monkeypatch, tmp_path) -> None: | |
| """Published setups live as a sibling of the configured snapshot library | |
| (so they follow libraryPath / KFORGELABS_PRESETS) instead of a fixed | |
| user-dir folder (issue #227).""" | |
| library = tmp_path / "kforge" / "koolook-presets" | |
| monkeypatch.setattr(koolook_routes, "_configured_dir", lambda: (library, "settings")) | |
| assert koolook_routes._published_setups_path() == ( | |
| tmp_path / "kforge" / DEFAULT_SETUPS_SUBDIR / DEFAULT_SETUPS_FILENAME | |
| ) | |
| def test_migrate_legacy_published_setups_copies_when_target_absent(tmp_path) -> None: | |
| """A pre-relocation registry at the old user-dir path is copied to the new | |
| location when the new one has none yet — non-destructively (old file kept).""" | |
| legacy = tmp_path / "user" / "koolook-published-setups" / "setups.json" | |
| legacy.parent.mkdir(parents=True) | |
| legacy.write_text('{"setups": [{"id": "carried-over"}]}', encoding="utf-8") | |
| target = tmp_path / "kforge" / "koolook-published-setups" / "setups.json" | |
| koolook_routes._migrate_legacy_published_setups(legacy, target) | |
| assert target.is_file() | |
| assert target.read_text(encoding="utf-8") == legacy.read_text(encoding="utf-8") | |
| assert legacy.is_file() # non-destructive: original is left in place | |
| def test_migrate_legacy_published_setups_noop_when_target_present(tmp_path) -> None: | |
| """Migration never overwrites an existing registry at the new location.""" | |
| legacy = tmp_path / "user" / "koolook-published-setups" / "setups.json" | |
| legacy.parent.mkdir(parents=True) | |
| legacy.write_text('{"setups": [{"id": "old"}]}', encoding="utf-8") | |
| target = tmp_path / "kforge" / "koolook-published-setups" / "setups.json" | |
| target.parent.mkdir(parents=True) | |
| target.write_text('{"setups": [{"id": "current"}]}', encoding="utf-8") | |
| koolook_routes._migrate_legacy_published_setups(legacy, target) | |
| assert '"current"' in target.read_text(encoding="utf-8") | |
| def test_migrate_legacy_published_setups_noop_when_no_legacy(tmp_path) -> None: | |
| """Fresh install (no old file) migrates nothing and does not create the | |
| target — the registry's own fallback handles an empty location.""" | |
| legacy = tmp_path / "user" / "koolook-published-setups" / "setups.json" | |
| target = tmp_path / "kforge" / "koolook-published-setups" / "setups.json" | |
| koolook_routes._migrate_legacy_published_setups(legacy, target) | |
| assert not target.exists() | |
| def _point_registry(monkeypatch, tmp_path): | |
| """Wire the default registry factory at tmp paths and return (legacy, target).""" | |
| library = tmp_path / "lib" / "koolook-presets" | |
| monkeypatch.setattr(koolook_routes, "_configured_dir", lambda: (library, "settings")) | |
| legacy = tmp_path / "user" / "koolook-published-setups" / "setups.json" | |
| legacy.parent.mkdir(parents=True) | |
| legacy.write_text('{"setups": []}', encoding="utf-8") | |
| monkeypatch.setattr(koolook_routes, "default_storage_path", lambda: legacy) | |
| target = tmp_path / "lib" / "koolook-published-setups" / "setups.json" | |
| return legacy, target | |
| def test_default_registry_keeps_legacy_data_when_migration_fails(monkeypatch, tmp_path) -> None: | |
| """If the legacy->new copy fails, the registry keeps reading the real legacy | |
| file rather than masking it with the bundled sample (PR #239 review, HIGH).""" | |
| legacy, target = _point_registry(monkeypatch, tmp_path) | |
| def boom(*_a, **_k): | |
| raise OSError("disk full") | |
| monkeypatch.setattr(koolook_routes.shutil, "copy2", boom) | |
| registry = koolook_routes._default_published_setup_registry() | |
| # Migration could not produce the new file, so the primary storage stays on | |
| # the legacy file (real data) — not the new path that would sample-fallback. | |
| assert not target.exists() | |
| assert registry.storage_path == legacy | |
| def test_default_registry_uses_new_path_after_successful_migration(monkeypatch, tmp_path) -> None: | |
| """On a clean migration the registry uses the relocated path.""" | |
| legacy, target = _point_registry(monkeypatch, tmp_path) | |
| registry = koolook_routes._default_published_setup_registry() | |
| assert target.is_file() | |
| assert registry.storage_path == target | |
| def test_default_registry_prefers_intact_legacy_over_corrupt_target(monkeypatch, tmp_path) -> None: | |
| """A partial/corrupt relocated file (e.g. an interrupted copy) must not | |
| mask intact legacy data: existence alone is not proof migration succeeded | |
| (PR #239 re-review, HIGH). The corrupt file is left in place untouched.""" | |
| legacy, target = _point_registry(monkeypatch, tmp_path) | |
| legacy.write_text('{"setups": [{"id": "real-data"}]}', encoding="utf-8") | |
| target.parent.mkdir(parents=True) | |
| target.write_text('{"setups": [{"id": "trunc', encoding="utf-8") | |
| registry = koolook_routes._default_published_setup_registry() | |
| assert registry.storage_path == legacy | |
| # Non-destructive: the corrupt file is evidence, not something to overwrite. | |
| assert target.read_text(encoding="utf-8") == '{"setups": [{"id": "trunc' | |
| def test_default_registry_keeps_valid_target_authoritative(monkeypatch, tmp_path) -> None: | |
| """A readable relocated registry stays primary even while the legacy file | |
| still exists — preferring legacy is strictly a corruption fallback.""" | |
| legacy, target = _point_registry(monkeypatch, tmp_path) | |
| target.parent.mkdir(parents=True) | |
| target.write_text('{"setups": [{"id": "migrated"}]}', encoding="utf-8") | |
| registry = koolook_routes._default_published_setup_registry() | |
| assert registry.storage_path == target | |
| def test_default_registry_keeps_target_when_both_files_corrupt(monkeypatch, tmp_path) -> None: | |
| """If legacy is corrupt too there is nothing intact to protect; stay on the | |
| relocated path so storage diagnostics point at the current location.""" | |
| legacy, target = _point_registry(monkeypatch, tmp_path) | |
| legacy.write_text("not json either", encoding="utf-8") | |
| target.parent.mkdir(parents=True) | |
| target.write_text("not json", encoding="utf-8") | |
| registry = koolook_routes._default_published_setup_registry() | |
| assert registry.storage_path == target | |
| def test_migration_interrupted_copy_leaves_no_partial_target(monkeypatch, tmp_path) -> None: | |
| """An interrupted copy must never leave a partial setups.json at the new | |
| location (it would shadow the legacy registry on every later boot). The | |
| migration writes via temp file + atomic replace and cleans up after a | |
| failure.""" | |
| legacy, target = _point_registry(monkeypatch, tmp_path) | |
| real_open = open | |
| def partial_copy(src, dst, *args, **kwargs): | |
| with real_open(dst, "w", encoding="utf-8") as f: | |
| f.write('{"setups": [{"id"') # simulate dying mid-write | |
| raise OSError("interrupted") | |
| monkeypatch.setattr(koolook_routes.shutil, "copy2", partial_copy) | |
| registry = koolook_routes._default_published_setup_registry() | |
| assert not target.exists() | |
| assert list(target.parent.glob("*.tmp")) == [] | |
| assert registry.storage_path == legacy | |