diff --git a/echopype/tests/utils/test_utils_io.py b/echopype/tests/utils/test_utils_io.py index 2dc13e83a..d722d3198 100644 --- a/echopype/tests/utils/test_utils_io.py +++ b/echopype/tests/utils/test_utils_io.py @@ -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()) == [] diff --git a/echopype/utils/io.py b/echopype/utils/io.py index f311dfe4e..10d14ad2c 100644 --- a/echopype/utils/io.py +++ b/echopype/utils/io.py @@ -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)