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
41 changes: 41 additions & 0 deletions echopype/tests/utils/test_utils_io.py
Original file line number Diff line number Diff line change
Expand Up @@ -325,3 +325,44 @@ def test_init_ep_dir(monkeypatch):
assert echopype.utils.io.ECHOPYPE_DIR.exists() is True

temp_user_dir.cleanup()


def test_check_file_permissions_fsmap_writes_inside_target(tmp_path):
"""check_file_permissions must probe inside the target store, not its parent (#1098)."""
from echopype.utils.io import check_file_permissions

fs = fsspec.filesystem('file')
target = tmp_path / 'combined_files'
target.mkdir()

# record where the probe file is opened (it is deleted again, so it can't
# be observed after the fact)
opened = []
real_open = fs.open

def _record_open(path, *args, **kwargs):
opened.append(path)
return real_open(path, *args, **kwargs)

fs.open = _record_open
check_file_permissions(fsspec.FSMap(str(target), fs))

assert opened
# the probe path uses fsspec's forward-slash convention; normalize both
# sides so the parent-dir check is OS-independent (Windows uses os.sep '\\').
target_dir = str(target).replace(os.sep, '/')
assert all(p.replace(os.sep, '/').rsplit('/', 1)[0] == target_dir for p in opened)
assert list(target.iterdir()) == []


def test_check_file_permissions_fsmap_creates_missing_store(tmp_path):
"""A not-yet-created store should be created and pass the permission check (#1098)."""
from echopype.utils.io import check_file_permissions

fs = fsspec.filesystem('file')
target = tmp_path / 'new_store.zarr'

check_file_permissions(fsspec.FSMap(str(target), fs))

assert target.is_dir()
assert list(target.iterdir()) == []
9 changes: 5 additions & 4 deletions echopype/utils/io.py
Original file line number Diff line number Diff line change
Expand Up @@ -331,10 +331,11 @@ def check_file_permissions(FILE_DIR):
try:
fname = "." + str(uuid.uuid4())
if isinstance(FILE_DIR, FSMap):
base_dir = os.path.dirname(FILE_DIR.root)
if not base_dir:
base_dir = FILE_DIR.root
TEST_FILE = os.path.join(base_dir, fname).replace("\\", "/")
# Write the probe file inside the target store, not its parent
# directory (#1098). Create the store first if needed, mirroring
# the Path/str branch below.
TEST_FILE = os.path.join(FILE_DIR.root, fname).replace("\\", "/")
FILE_DIR.fs.makedirs(FILE_DIR.root, exist_ok=True)
with FILE_DIR.fs.open(TEST_FILE, "w") as f:
f.write("testing\n")
FILE_DIR.fs.delete(TEST_FILE)
Expand Down
Loading