Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions src/pytest_data_loader/loaders/impl.py
Original file line number Diff line number Diff line change
Expand Up @@ -518,6 +518,14 @@ def load(self) -> Iterable[LoadedData | LazyLoadedData]: # type: ignore[overrid
"""Load multiple files from a directory"""

def load_files(dir_path: Path) -> None:
if self.load_attrs.recursive:
# Detect directory traversal cycles by tracking visited inodes
stat = dir_path.stat()
inode = (stat.st_dev, stat.st_ino)
if inode in visited_dirs:
raise RuntimeError(f"Detected a circular symlink: {dir_path}")
visited_dirs.add(inode)

for p in sorted(dir_path.iterdir()):
if p.is_dir():
if self.load_attrs.recursive:
Expand Down Expand Up @@ -546,6 +554,7 @@ def load_files(dir_path: Path) -> None:
assert isinstance(loaded_data, LoadedData | LazyLoadedData), type(loaded_data)
loaded_files.append(loaded_data)

visited_dirs: set[tuple[int, int]] = set()
loaded_files: list[LoadedData | LazyLoadedData] = []
load_files(self.path)
return loaded_files
2 changes: 1 addition & 1 deletion src/pytest_data_loader/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -178,7 +178,7 @@ def validate_loader_func_args_and_normalize(


def check_circular_symlink(path: Path) -> None:
"""Detect a circular symlink
"""Detect a circular symlink resolution loops (ELOOP)

:param path: The path to check
"""
Expand Down
1 change: 1 addition & 0 deletions tests/data/symlinks/dir/symlink
1 change: 0 additions & 1 deletion tests/data/symlinks/symlink

This file was deleted.

3 changes: 3 additions & 0 deletions tests/tests_plugin/helper.py
Original file line number Diff line number Diff line change
Expand Up @@ -174,6 +174,7 @@ def run_pytest_with_context(
read_option_func_def: str | None = None,
collect_only: bool = False,
check_test_id: bool = False,
**other_loader_options: Any,
) -> RunResult:
"""Common test logic that runs pytest via pytester with various test context and checks the basic functionality
of pytest-data-loader plugin
Expand Down Expand Up @@ -206,6 +207,8 @@ def run_pytest_with_context(
loader_options.append(f"{DataLoaderFunctionType.FILE_READER_FUNC}={file_reader_func_def}")
if read_option_func_def:
loader_options.append(f"{DataLoaderFunctionType.READ_OPTION_FUNC}={read_option_func_def}")
if other_loader_options:
loader_options.extend(f"{k}={v!r}" for k, v in other_loader_options.items())
loader_options_str = ", " + ", ".join(loader_options) if loader_options else ""

is_abs_path = Path(test_context.path).is_absolute()
Expand Down
27 changes: 13 additions & 14 deletions tests/tests_plugin/test_data_path.py
Original file line number Diff line number Diff line change
Expand Up @@ -92,28 +92,27 @@ def test_parametrize_dir_loader_with_no_file(test_context: TestContext, loader:
@pytest.mark.parametrize("is_abs_path", [False, True])
@pytest.mark.parametrize("is_circular", [False, True])
def test_symlink(test_context: TestContext, is_circular: bool, is_abs_path: bool, collect_only: bool) -> None:
"""Test that symlinks are handled properly, including circular symlinks"""
symlink_data_dir_name = "symlinks"
src_symlink_data_dir = ABS_PATH_LOADER_DIR / symlink_data_dir_name
dst = Path(test_context.data_dir) / symlink_data_dir_name
if is_circular:
src_symlink_data_dir /= "circular"
dst.mkdir(exist_ok=True, parents=True)
dst /= "circular"

"""Test that symlinks are handled properly, including circular symlinks (ELOOP and directory traversal cycles)"""
src_symlink_data_dir = ABS_PATH_LOADER_DIR / "symlinks"
dst = Path(test_context.data_dir) / src_symlink_data_dir.name
dst.symlink_to(src_symlink_data_dir, target_is_directory=True)

kwargs = {}
if test_context.loader.is_file_loader:
dir_or_filename = "symlink.txt"
dir_or_file = Path("symlink.txt")
else:
dir_or_filename = "symlink"
dir_or_file = Path("dir", "symlink")
kwargs.update(recursive=True)

if is_circular:
dir_or_file = Path("circular", dir_or_file)

if is_abs_path:
path = dst / dir_or_filename
path = dst / dir_or_file
else:
path = dst.relative_to(test_context.data_dir) / dir_or_filename
path = dst.relative_to(test_context.data_dir) / dir_or_file

result = run_pytest_with_context(test_context, path=path, collect_only=collect_only)
result = run_pytest_with_context(test_context, path=path, collect_only=collect_only, **kwargs)
if is_circular:
assert result.ret == ExitCode.INTERRUPTED
assert "Detected a circular symlink" in str(result.stdout)
Expand Down