"""Detailed file system utilities tests matching TypeScript version granularity."""
import asyncio
import os
import shutil
import tempfile
from pathlib import Path
import pytest
from wassden.lib import fs_utils
# Test constants
LARGE_FILE_SIZE = 10000
VERY_LARGE_FILE_SIZE = 50000
class TestFileReadOperations:
"""Detailed file read operation tests."""
def setup_method(self):
"""Set up test environment for each test."""
self.temp_dir = Path(tempfile.mkdtemp())
self.original_cwd = Path.cwd()
os.chdir(self.temp_dir)
def teardown_method(self):
"""Clean up test environment after each test."""
os.chdir(self.original_cwd)
shutil.rmtree(self.temp_dir)
@pytest.mark.asyncio
async def test_read_existing_file(self):
"""Test reading an existing file."""
test_file = self.temp_dir / "test.txt"
test_content = "Hello, World!"
test_file.write_text(test_content)
result = await fs_utils.read_file(test_file)
assert result == test_content
@pytest.mark.asyncio
async def test_read_nonexistent_file(self):
"""Test reading a non-existent file."""
nonexistent_file = self.temp_dir / "nonexistent.txt"
with pytest.raises(FileNotFoundError):
await fs_utils.read_file(nonexistent_file)
@pytest.mark.asyncio
async def test_read_empty_file(self):
"""Test reading an empty file."""
empty_file = self.temp_dir / "empty.txt"
empty_file.write_text("")
result = await fs_utils.read_file(empty_file)
assert result == ""
@pytest.mark.asyncio
async def test_read_large_file(self):
"""Test reading a large file."""
large_file = self.temp_dir / "large.txt"
large_content = "A" * 10000 # 10KB of 'A's
large_file.write_text(large_content)
result = await fs_utils.read_file(large_file)
assert result == large_content
assert len(result) == LARGE_FILE_SIZE
@pytest.mark.asyncio
async def test_read_file_with_special_characters(self):
"""Test reading file with special characters."""
special_file = self.temp_dir / "special.txt"
special_content = "Special chars: !@#$%^&*()_+-={}[]|\\:;\"'<>?,./"
special_file.write_text(special_content)
result = await fs_utils.read_file(special_file)
assert result == special_content
@pytest.mark.asyncio
async def test_read_file_with_unicode(self):
"""Test reading file with Unicode characters."""
unicode_file = self.temp_dir / "unicode.txt"
unicode_content = "Unicode: 日本語, 한국어, العربية, русский, français"
unicode_file.write_text(unicode_content, encoding="utf-8")
result = await fs_utils.read_file(unicode_file)
assert result == unicode_content
@pytest.mark.asyncio
async def test_read_file_with_newlines(self):
"""Test reading file with various newline characters."""
newline_file = self.temp_dir / "newlines.txt"
newline_content = "Line 1\nLine 2\r\nLine 3\rLine 4"
# Write in binary mode to preserve exact newlines
newline_file.write_bytes(newline_content.encode("utf-8"))
result = await fs_utils.read_file(newline_file)
# On some platforms, newlines might be normalized
assert "Line 1" in result
assert "Line 2" in result
assert "Line 3" in result
assert "Line 4" in result
@pytest.mark.asyncio
async def test_read_file_relative_path(self):
"""Test reading file with relative path."""
test_file = Path("relative_test.txt")
test_content = "Relative path content"
test_file.write_text(test_content)
result = await fs_utils.read_file(Path("relative_test.txt"))
assert result == test_content
@pytest.mark.asyncio
async def test_read_file_absolute_path(self):
"""Test reading file with absolute path."""
test_file = self.temp_dir / "absolute_test.txt"
test_content = "Absolute path content"
test_file.write_text(test_content)
result = await fs_utils.read_file(test_file.absolute())
assert result == test_content
@pytest.mark.asyncio
async def test_read_file_with_spaces_in_name(self):
"""Test reading file with spaces in filename."""
spaced_file = self.temp_dir / "file with spaces.txt"
spaced_content = "Content in file with spaces"
spaced_file.write_text(spaced_content)
result = await fs_utils.read_file(spaced_file)
assert result == spaced_content
class TestFileExistsOperations:
"""Detailed file existence check tests."""
def setup_method(self):
"""Set up test environment for each test."""
self.temp_dir = Path(tempfile.mkdtemp())
self.original_cwd = Path.cwd()
os.chdir(self.temp_dir)
def teardown_method(self):
"""Clean up test environment after each test."""
os.chdir(self.original_cwd)
shutil.rmtree(self.temp_dir)
@pytest.mark.asyncio
async def test_file_exists_true(self):
"""Test file_exists returns True for existing file."""
existing_file = self.temp_dir / "exists.txt"
existing_file.write_text("Content")
result = await fs_utils.file_exists(existing_file)
assert result is True
@pytest.mark.asyncio
async def test_file_exists_false(self):
"""Test file_exists returns False for non-existing file."""
nonexistent_file = self.temp_dir / "does_not_exist.txt"
result = await fs_utils.file_exists(nonexistent_file)
assert result is False
@pytest.mark.asyncio
async def test_file_exists_directory(self):
"""Test file_exists behavior with directories."""
test_dir = self.temp_dir / "test_directory"
test_dir.mkdir()
result = await fs_utils.file_exists(test_dir)
# Depending on implementation, might return True for directories
assert isinstance(result, bool)
@pytest.mark.asyncio
async def test_file_exists_empty_file(self):
"""Test file_exists returns True for empty file."""
empty_file = self.temp_dir / "empty.txt"
empty_file.write_text("")
result = await fs_utils.file_exists(empty_file)
assert result is True
@pytest.mark.asyncio
async def test_file_exists_relative_path(self):
"""Test file_exists with relative path."""
relative_file = Path("relative_exists.txt")
relative_file.write_text("Content")
result = await fs_utils.file_exists(Path("relative_exists.txt"))
assert result is True
@pytest.mark.asyncio
async def test_file_exists_absolute_path(self):
"""Test file_exists with absolute path."""
absolute_file = self.temp_dir / "absolute_exists.txt"
absolute_file.write_text("Content")
result = await fs_utils.file_exists(absolute_file.absolute())
assert result is True
@pytest.mark.asyncio
async def test_file_exists_special_characters_in_name(self):
"""Test file_exists with special characters in filename."""
special_file = self.temp_dir / "file!@#$%^&*().txt"
special_file.write_text("Content")
result = await fs_utils.file_exists(special_file)
assert result is True
@pytest.mark.asyncio
async def test_file_exists_unicode_filename(self):
"""Test file_exists with Unicode filename."""
unicode_file = self.temp_dir / "ファイル名.txt"
unicode_file.write_text("Content")
result = await fs_utils.file_exists(unicode_file)
assert result is True
@pytest.mark.asyncio
async def test_file_exists_long_path(self):
"""Test file_exists with very long path."""
# Create nested directories to make a long path
long_path = self.temp_dir
for i in range(10):
long_path = long_path / f"level{i}"
long_path.mkdir(parents=True)
long_file = long_path / "long_path_file.txt"
long_file.write_text("Content")
result = await fs_utils.file_exists(long_file)
assert result is True
class TestProjectRootDetection:
"""Detailed project root detection tests."""
def setup_method(self):
"""Set up test environment for each test."""
self.temp_dir = Path(tempfile.mkdtemp())
self.original_cwd = Path.cwd()
def teardown_method(self):
"""Clean up test environment after each test."""
os.chdir(self.original_cwd)
shutil.rmtree(self.temp_dir)
def test_get_project_root_with_pyproject_toml(self):
"""Test project root detection with pyproject.toml."""
# Create pyproject.toml in temp directory
pyproject_file = self.temp_dir / "pyproject.toml"
pyproject_file.write_text("[tool.poetry]\nname = 'test'\n")
# Create nested directory and change to it
nested_dir = self.temp_dir / "src" / "package"
nested_dir.mkdir(parents=True)
os.chdir(nested_dir)
result = fs_utils.get_project_root()
# Resolve paths to handle symlinks (macOS /var -> /private/var)
assert result.resolve() == self.temp_dir.resolve()
def test_get_project_root_with_git_directory(self):
"""Test project root detection with .git directory."""
# Create .git directory
git_dir = self.temp_dir / ".git"
git_dir.mkdir()
# Create nested directory and change to it
nested_dir = self.temp_dir / "src" / "package"
nested_dir.mkdir(parents=True)
os.chdir(nested_dir)
result = fs_utils.get_project_root()
# Resolve paths to handle symlinks (macOS /var -> /private/var)
assert result.resolve() == self.temp_dir.resolve()
def test_get_project_root_with_package_json(self):
"""Test project root detection with package.json."""
# Create package.json
package_file = self.temp_dir / "package.json"
package_file.write_text('{"name": "test-package"}')
# Create nested directory and change to it
nested_dir = self.temp_dir / "src" / "components"
nested_dir.mkdir(parents=True)
os.chdir(nested_dir)
result = fs_utils.get_project_root()
# Resolve paths to handle symlinks (macOS /var -> /private/var)
assert result.resolve() == self.temp_dir.resolve()
def test_get_project_root_with_setup_py(self):
"""Test project root detection with setup.py."""
# Create setup.py
setup_file = self.temp_dir / "setup.py"
setup_file.write_text("from setuptools import setup\nsetup()")
# Create nested directory and change to it
nested_dir = self.temp_dir / "src" / "module"
nested_dir.mkdir(parents=True)
os.chdir(nested_dir)
result = fs_utils.get_project_root()
# Resolve paths to handle symlinks (macOS /var -> /private/var)
assert result.resolve() == self.temp_dir.resolve()
def test_get_project_root_with_multiple_indicators(self):
"""Test project root detection with multiple indicators."""
# Create multiple project indicators
(self.temp_dir / "pyproject.toml").write_text("[tool.poetry]")
(self.temp_dir / ".git").mkdir()
(self.temp_dir / "package.json").write_text('{"name": "test"}')
# Create deeply nested directory
deep_nested = self.temp_dir / "a" / "b" / "c" / "d" / "e"
deep_nested.mkdir(parents=True)
os.chdir(deep_nested)
result = fs_utils.get_project_root()
# Resolve paths to handle symlinks (macOS /var -> /private/var)
assert result.resolve() == self.temp_dir.resolve()
def test_get_project_root_no_indicators(self):
"""Test project root detection when no indicators are found."""
# Create directory without any project indicators
nested_dir = self.temp_dir / "some" / "nested" / "path"
nested_dir.mkdir(parents=True)
os.chdir(nested_dir)
result = fs_utils.get_project_root()
# Should return current working directory when no indicators found
assert result.resolve() == nested_dir.resolve()
def test_get_project_root_at_root_level(self):
"""Test project root detection when already at project root."""
# Create project indicators
(self.temp_dir / "pyproject.toml").write_text("[tool.poetry]")
os.chdir(self.temp_dir)
result = fs_utils.get_project_root()
# Resolve paths to handle symlinks (macOS /var -> /private/var)
assert result.resolve() == self.temp_dir.resolve()
def test_get_project_root_mixed_indicators_different_levels(self):
"""Test project root detection with indicators at different levels."""
# Create .git at temp_dir level
(self.temp_dir / ".git").mkdir()
# Create pyproject.toml at nested level
nested_dir = self.temp_dir / "project"
nested_dir.mkdir()
(nested_dir / "pyproject.toml").write_text("[tool.poetry]")
# Change to deeper nested directory
deeper_dir = nested_dir / "src" / "module"
deeper_dir.mkdir(parents=True)
os.chdir(deeper_dir)
result = fs_utils.get_project_root()
# Should find the closest indicator (pyproject.toml)
assert result.resolve() == nested_dir.resolve()
class TestAsyncFileOperationsConcurrency:
"""Test concurrent async file operations."""
def setup_method(self):
"""Set up test environment for each test."""
self.temp_dir = Path(tempfile.mkdtemp())
self.original_cwd = Path.cwd()
os.chdir(self.temp_dir)
def teardown_method(self):
"""Clean up test environment after each test."""
os.chdir(self.original_cwd)
shutil.rmtree(self.temp_dir)
@pytest.mark.asyncio
async def test_concurrent_file_reads(self):
"""Test concurrent reading of multiple files."""
# Create multiple test files
files_data = {}
for i in range(10):
file_path = self.temp_dir / f"concurrent_read_{i}.txt"
content = f"Content of file {i}"
file_path.write_text(content)
files_data[file_path] = content
# Read all files concurrently
tasks = [fs_utils.read_file(path) for path in files_data]
results = await asyncio.gather(*tasks)
# Verify all files were read correctly
for path, result in zip(files_data.keys(), results, strict=False):
assert result == files_data[path]
@pytest.mark.asyncio
async def test_concurrent_file_reads_more(self):
"""Test more concurrent reading to replace removed write tests."""
# Prepare file data
files_data = {}
for i in range(10):
file_path = self.temp_dir / f"concurrent_write_{i}.txt"
content = f"Concurrent write content {i}"
files_data[file_path] = content
file_path.write_text(content)
# Read all files concurrently
tasks = [fs_utils.read_file(path) for path in files_data]
results = await asyncio.gather(*tasks)
# Verify all files were read correctly
for path, result in zip(files_data.keys(), results, strict=False):
assert result == files_data[path]
@pytest.mark.asyncio
async def test_concurrent_file_existence_checks(self):
"""Test concurrent file existence checks."""
# Create some files, leave others non-existent
existing_files = []
nonexistent_files = []
for i in range(5):
existing_file = self.temp_dir / f"exists_{i}.txt"
existing_file.write_text(f"Content {i}")
existing_files.append(existing_file)
nonexistent_file = self.temp_dir / f"not_exists_{i}.txt"
nonexistent_files.append(nonexistent_file)
all_files = existing_files + nonexistent_files
# Check existence concurrently
tasks = [fs_utils.file_exists(path) for path in all_files]
results = await asyncio.gather(*tasks)
# Verify results
for i, path in enumerate(all_files):
if path in existing_files:
assert results[i] is True
else:
assert results[i] is False
@pytest.mark.asyncio
async def test_mixed_concurrent_operations(self):
"""Test mixed concurrent file operations."""
# Setup files for reading
read_file = self.temp_dir / "to_read.txt"
read_file.write_text("Read this content")
# Prepare for checking
check_file = self.temp_dir / "to_check.txt"
check_file.write_text("Check this")
check_file2 = self.temp_dir / "to_check2.txt"
check_file2.write_text("Check content")
# Run mixed operations concurrently
read_task = fs_utils.read_file(read_file)
exists_task = fs_utils.file_exists(check_file)
exists_task2 = fs_utils.file_exists(check_file2)
read_result, exists_result, exists_result2 = await asyncio.gather(read_task, exists_task, exists_task2)
# Verify results
assert read_result == "Read this content"
assert exists_result is True
assert exists_result2 is True
class TestErrorHandlingInFileOperations:
"""Test error handling in file operations."""
def setup_method(self):
"""Set up test environment for each test."""
self.temp_dir = Path(tempfile.mkdtemp())
self.original_cwd = Path.cwd()
os.chdir(self.temp_dir)
def teardown_method(self):
"""Clean up test environment after each test."""
os.chdir(self.original_cwd)
shutil.rmtree(self.temp_dir)
@pytest.mark.asyncio
async def test_read_file_permission_error(self):
"""Test reading file with permission error."""
if os.name == "nt": # Windows
pytest.skip("Permission tests not reliable on Windows")
if os.geteuid() == 0: # Running as root
pytest.skip("Permission tests not reliable when running as root")
restricted_file = self.temp_dir / "restricted.txt"
restricted_file.write_text("Restricted content")
restricted_file.chmod(0o000) # Remove all permissions
try:
with pytest.raises(PermissionError):
await fs_utils.read_file(restricted_file)
finally:
restricted_file.chmod(0o644) # Restore permissions for cleanup
@pytest.mark.asyncio
async def test_read_file_is_directory_error(self):
"""Test reading when path points to a directory."""
test_dir = self.temp_dir / "is_directory"
test_dir.mkdir()
with pytest.raises((IsADirectoryError, PermissionError)):
await fs_utils.read_file(test_dir)
@pytest.mark.asyncio
async def test_invalid_file_path_characters(self):
"""Test file operations with invalid path characters."""
# This test might be platform-specific
invalid_chars = ["<", ">", ":", '"', "|", "?", "*"] if os.name == "nt" else ["\0"]
for char in invalid_chars:
invalid_path = self.temp_dir / f"invalid{char}file.txt"
# Should handle invalid characters gracefully
try:
Path(str(invalid_path)).write_text("test")
await fs_utils.read_file(invalid_path)
except (OSError, ValueError):
# Expected on some platforms
pass
