mirror of
https://github.com/comfyanonymous/ComfyUI.git
synced 2026-03-04 16:57:31 +08:00
fix: follow symlinks in list_files_recursively with cycle detection
list_files_recursively now uses followlinks=True so symlinked directories under input/ and output/ roots are traversed, matching the existing behavior of folder_paths.recursive_search for models. Tracks (st_dev, st_ino) pairs of visited directories to detect and break circular symlink loops safely. Amp-Thread-ID: https://ampcode.com/threads/T-019c9220-21b8-7678-b428-9215ff1bb011 Co-authored-by: Amp <amp@ampcode.com>
This commit is contained in:
parent
3ff3a64987
commit
7daf360dfa
@ -42,14 +42,26 @@ def is_visible(name: str) -> bool:
|
|||||||
|
|
||||||
|
|
||||||
def list_files_recursively(base_dir: str) -> list[str]:
|
def list_files_recursively(base_dir: str) -> list[str]:
|
||||||
"""Recursively list all files in a directory."""
|
"""Recursively list all files in a directory, following symlinks."""
|
||||||
out: list[str] = []
|
out: list[str] = []
|
||||||
base_abs = os.path.abspath(base_dir)
|
base_abs = os.path.abspath(base_dir)
|
||||||
if not os.path.isdir(base_abs):
|
if not os.path.isdir(base_abs):
|
||||||
return out
|
return out
|
||||||
|
# Track seen real directory identities to prevent circular symlink loops
|
||||||
|
seen_dirs: set[tuple[int, int]] = set()
|
||||||
for dirpath, subdirs, filenames in os.walk(
|
for dirpath, subdirs, filenames in os.walk(
|
||||||
base_abs, topdown=True, followlinks=False
|
base_abs, topdown=True, followlinks=True
|
||||||
):
|
):
|
||||||
|
try:
|
||||||
|
st = os.stat(dirpath)
|
||||||
|
dir_id = (st.st_dev, st.st_ino)
|
||||||
|
except OSError:
|
||||||
|
subdirs.clear()
|
||||||
|
continue
|
||||||
|
if dir_id in seen_dirs:
|
||||||
|
subdirs.clear()
|
||||||
|
continue
|
||||||
|
seen_dirs.add(dir_id)
|
||||||
subdirs[:] = [d for d in subdirs if is_visible(d)]
|
subdirs[:] = [d for d in subdirs if is_visible(d)]
|
||||||
for name in filenames:
|
for name in filenames:
|
||||||
if not is_visible(name):
|
if not is_visible(name):
|
||||||
|
|||||||
@ -1,3 +1,8 @@
|
|||||||
|
import os
|
||||||
|
import sys
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
from app.assets.services.file_utils import is_visible, list_files_recursively
|
from app.assets.services.file_utils import is_visible, list_files_recursively
|
||||||
|
|
||||||
|
|
||||||
@ -53,3 +58,64 @@ class TestListFilesRecursively:
|
|||||||
def test_nonexistent_directory(self, tmp_path):
|
def test_nonexistent_directory(self, tmp_path):
|
||||||
result = list_files_recursively(str(tmp_path / "nonexistent"))
|
result = list_files_recursively(str(tmp_path / "nonexistent"))
|
||||||
assert result == []
|
assert result == []
|
||||||
|
|
||||||
|
@pytest.mark.skipif(sys.platform == "win32", reason="symlinks need privileges on Windows")
|
||||||
|
def test_follows_symlinked_directories(self, tmp_path):
|
||||||
|
target = tmp_path / "real_dir"
|
||||||
|
target.mkdir()
|
||||||
|
(target / "model.safetensors").write_text("data")
|
||||||
|
|
||||||
|
root = tmp_path / "root"
|
||||||
|
root.mkdir()
|
||||||
|
(root / "link").symlink_to(target)
|
||||||
|
|
||||||
|
result = list_files_recursively(str(root))
|
||||||
|
|
||||||
|
assert len(result) == 1
|
||||||
|
assert result[0].endswith("model.safetensors")
|
||||||
|
assert "link" in result[0]
|
||||||
|
|
||||||
|
@pytest.mark.skipif(sys.platform == "win32", reason="symlinks need privileges on Windows")
|
||||||
|
def test_follows_symlinked_files(self, tmp_path):
|
||||||
|
real_file = tmp_path / "real.txt"
|
||||||
|
real_file.write_text("content")
|
||||||
|
|
||||||
|
root = tmp_path / "root"
|
||||||
|
root.mkdir()
|
||||||
|
(root / "link.txt").symlink_to(real_file)
|
||||||
|
|
||||||
|
result = list_files_recursively(str(root))
|
||||||
|
|
||||||
|
assert len(result) == 1
|
||||||
|
assert result[0].endswith("link.txt")
|
||||||
|
|
||||||
|
@pytest.mark.skipif(sys.platform == "win32", reason="symlinks need privileges on Windows")
|
||||||
|
def test_circular_symlinks_do_not_loop(self, tmp_path):
|
||||||
|
dir_a = tmp_path / "a"
|
||||||
|
dir_a.mkdir()
|
||||||
|
(dir_a / "file.txt").write_text("a")
|
||||||
|
# a/b -> a (circular)
|
||||||
|
(dir_a / "b").symlink_to(dir_a)
|
||||||
|
|
||||||
|
result = list_files_recursively(str(dir_a))
|
||||||
|
|
||||||
|
assert len(result) == 1
|
||||||
|
assert result[0].endswith("file.txt")
|
||||||
|
|
||||||
|
@pytest.mark.skipif(sys.platform == "win32", reason="symlinks need privileges on Windows")
|
||||||
|
def test_mutual_circular_symlinks(self, tmp_path):
|
||||||
|
dir_a = tmp_path / "a"
|
||||||
|
dir_b = tmp_path / "b"
|
||||||
|
dir_a.mkdir()
|
||||||
|
dir_b.mkdir()
|
||||||
|
(dir_a / "file_a.txt").write_text("a")
|
||||||
|
(dir_b / "file_b.txt").write_text("b")
|
||||||
|
# a/link_b -> b and b/link_a -> a
|
||||||
|
(dir_a / "link_b").symlink_to(dir_b)
|
||||||
|
(dir_b / "link_a").symlink_to(dir_a)
|
||||||
|
|
||||||
|
result = list_files_recursively(str(dir_a))
|
||||||
|
basenames = sorted(os.path.basename(p) for p in result)
|
||||||
|
|
||||||
|
assert "file_a.txt" in basenames
|
||||||
|
assert "file_b.txt" in basenames
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user