diff --git a/sshfs/spec.py b/sshfs/spec.py index f8f5940..8ed0558 100644 --- a/sshfs/spec.py +++ b/sshfs/spec.py @@ -16,7 +16,7 @@ sync, sync_wrapper, ) -from fsspec.utils import infer_storage_options +from fsspec.utils import infer_storage_options, stringify_path from sshfs.file import SSHFile from sshfs.pools import SFTPSoftChannelPool @@ -63,6 +63,10 @@ def __init__( Options passed to `SSHClientConnection.start_sftp_client` (e.g. env, send_env, path_encoding, path_errors, sftp_version). + + Note + ---- + when ``sftp_client_kwargs`` [ ``path_encoding`` ] = ``None`` , protocol-prefixed paths are not supported (e.g. ``b"ssh://host/path/to/file"``) """ super().__init__(self, **kwargs) @@ -95,6 +99,12 @@ def __init__( @classmethod def _strip_protocol(cls, path): + # Note: when `path_encoding` is None, protocol-prefixed paths are not supported + # Raw SFTP paths must retain their encoding (path_encoding=None). + if isinstance(path, bytes): + return path + # convert `pathlib.Path` and `os.PathLike` etc. + path = stringify_path(path) # Remove components such as host and username from path. inferred_path = infer_storage_options(path)["path"] stripped = super()._strip_protocol(inferred_path) @@ -215,6 +225,8 @@ async def _modified(self, path: str, **kwargs) -> datetime: @wrap_exceptions async def _mv(self, lpath, rpath, **kwargs): + lpath = self._strip_protocol(lpath) + rpath = self._strip_protocol(rpath) async with self._pool.get() as channel: with suppress(SFTPOpUnsupported): return await channel.posix_rename(lpath, rpath) @@ -238,6 +250,7 @@ async def _put_file( callback=None, **kwargs, ): + rpath = self._strip_protocol(rpath) await self._makedirs(self._parent(rpath), exist_ok=True) async with self._pool.get() as channel: await channel.put( @@ -251,6 +264,7 @@ async def _put_file( async def _get_file( self, lpath, rpath, block_size=READ_BLOCK_SIZE, callback=None, **kwargs ): + lpath = self._strip_protocol(lpath) async with self._pool.get() as channel: await channel.get( lpath, @@ -261,11 +275,14 @@ async def _get_file( @wrap_exceptions async def _cp_file(self, lpath, rpath, **kwargs): + lpath = self._strip_protocol(lpath) + rpath = self._strip_protocol(rpath) cmd = f"cp {shlex.quote(lpath)} {shlex.quote(rpath)}" await self._execute(cmd) @wrap_exceptions async def _ls(self, path, detail=True, **kwargs): + path = self._strip_protocol(path) async with self._pool.get() as channel: file_attrs = await channel.readdir(path) @@ -287,6 +304,7 @@ async def _ls(self, path, detail=True, **kwargs): async def _mkdir( self, path, *, create_parents=True, permissions=511, **kwargs ): + path = self._strip_protocol(path) if create_parents: return await self._makedirs(path, exist_ok=True) @@ -298,6 +316,7 @@ async def _mkdir( async def _makedirs( self, path, *, exist_ok=False, permissions=511, **kwargs ): + path = self._strip_protocol(path) attrs = asyncssh.SFTPAttrs(permissions=permissions) async with self._pool.get() as channel: await channel.makedirs(path, exist_ok=exist_ok, attrs=attrs) @@ -307,6 +326,7 @@ async def _makedirs( @wrap_exceptions async def _rm_file(self, path, **kwargs): + path = self._strip_protocol(path) async with self._pool.get() as channel: await channel.unlink(path) @@ -319,6 +339,7 @@ async def _rmdir( on_error=None, **kwargs, ): + path = self._strip_protocol(path) async with self._pool.get() as channel: if recursive: await channel.rmtree( @@ -343,6 +364,7 @@ async def _rm(self, path, recursive=False, **kwargs): @wrap_exceptions async def _checksum(self, path): + path = self._strip_protocol(path) system = await self._get_system() if system == "Linux": command = "md5sum" @@ -381,11 +403,13 @@ async def _execute(self, *args, **kwargs): execute = sync_wrapper(_execute) def _open(self, path, *args, **kwargs): + path = self._strip_protocol(path) return SSHFile(self, path, *args, **kwargs) @wrap_exceptions async def _cat_file(self, path, **kwargs): """Asynchronously fetch the contents of a file""" + path = self._strip_protocol(path) async with self._pool.get() as channel: async with channel.open(path, "rb") as f: return await f.read() @@ -393,6 +417,7 @@ async def _cat_file(self, path, **kwargs): @wrap_exceptions async def _pipe_file(self, path, data, chunksize=50 * 2**20, **kwargs): """Asynchronously writes the given data to a remote file in chunks.""" + path = self._strip_protocol(path) await self._makedirs(self._parent(path), exist_ok=True) async with self._pool.get() as channel: diff --git a/tests/test_sshfs.py b/tests/test_sshfs.py index 141baf7..6ec1f16 100644 --- a/tests/test_sshfs.py +++ b/tests/test_sshfs.py @@ -35,17 +35,22 @@ def ssh_server(): yield server -@pytest.fixture(scope="session") -def base_remote_dir(): - with tempfile.TemporaryDirectory() as path: - yield path - - @pytest.fixture -def remote_dir(fs, base_remote_dir, request): - directory = posixpath.join(base_remote_dir, request.node.name).rstrip("/") +def plain_remote_dir(fs, tmp_path): + directory = str(tmp_path / "remote") fs.mkdir(directory) - yield directory + return directory + + +@pytest.fixture(params=["plain", "ssh", "sftp"]) +def remote_dir(plain_remote_dir, ssh_server, request): + if request.param == "plain": + return plain_remote_dir + + return ( + f"{request.param}://user@" + f"{ssh_server.host}:{ssh_server.port}{plain_remote_dir}" + ) @pytest.fixture @@ -88,6 +93,7 @@ def test_fsspec_registration(ssh_server): assert isinstance(fs, SSHFileSystem) +@pytest.mark.parametrize("remote_dir", ["plain"], indirect=True) def test_fsspec_url_parsing(ssh_server, remote_dir, user="user"): for ep in list(entry_points(group="fsspec.specs")): url = f"{ep.name}://{user}@{ssh_server.host}:{ssh_server.port}/{remote_dir}/file" @@ -103,7 +109,7 @@ def test_fsspec_url_parsing(ssh_server, remote_dir, user="user"): } -def test_sftp_client_kwargs(ssh_server, base_remote_dir, user="user"): +def test_sftp_client_kwargs(ssh_server, tmp_path, user="user"): fs = SSHFileSystem( host=ssh_server.host, port=ssh_server.port, @@ -113,14 +119,12 @@ def test_sftp_client_kwargs(ssh_server, base_remote_dir, user="user"): ) assert fs._pool.sftp_client_kwargs == {"sftp_version": 3} - file = posixpath.join(base_remote_dir, "sftp_client_kwargs_probe") + file = posixpath.join(tmp_path, "sftp_client_kwargs_probe") fs.touch(file) assert fs.exists(file) -def test_sftp_client_kwargs_path_encoding( - ssh_server, base_remote_dir, user="user" -): +def test_sftp_client_kwargs_path_encoding(ssh_server, tmp_path, user="user"): # path_encoding=None makes asyncssh deliver remote paths as raw # bytes instead of str, for servers with non-UTF-8 file names (#39). fs = SSHFileSystem( @@ -128,10 +132,11 @@ def test_sftp_client_kwargs_path_encoding( port=ssh_server.port, username=user, client_keys=[USERS[user]], + # Note: when `path_encoding` is None, protocol-prefixed paths are not supported sftp_client_kwargs={"path_encoding": None}, ) - directory = Path(base_remote_dir) / "path_encoding_probe" + directory = tmp_path / "path_encoding_probe" directory.mkdir() (directory / "data.txt").write_bytes(b"data") @@ -140,22 +145,24 @@ def test_sftp_client_kwargs_path_encoding( assert fs.cat_file(encoded + b"/data.txt") == b"data" -def test_info(fs, remote_dir): +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) +def test_info(fs, remote_dir, plain_remote_dir): fs.touch(remote_dir + "/a.txt") details = fs.info(remote_dir + "/a.txt") assert details["type"] == "file" - assert details["name"] == remote_dir + "/a.txt" + assert details["name"] == plain_remote_dir + "/a.txt" assert details["size"] == 0 fs.mkdir(remote_dir + "/dir") details = fs.info(remote_dir + "/dir") assert details["type"] == "directory" - assert details["name"] == remote_dir + "/dir/" + assert details["name"] == plain_remote_dir + "/dir/" details = fs.info(remote_dir + "/dir/") - assert details["name"] == remote_dir + "/dir/" + assert details["name"] == plain_remote_dir + "/dir/" +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) def test_info_timestamps_are_tz_aware_utc(fs, remote_dir): fs.touch(remote_dir + "/a.txt") details = fs.info(remote_dir + "/a.txt") @@ -172,6 +179,7 @@ def test_decode_attributes_missing_timestamps(fs): assert details["mtime"] is None +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) def test_move(fs, remote_dir): fs.touch(remote_dir + "/a.txt") initial_info = fs.info(remote_dir + "/a.txt") @@ -198,6 +206,7 @@ def test_move(fs, remote_dir): assert initial_info == secondary_info +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) def test_copy(fs, remote_dir): fs.touch(remote_dir + "/a.txt") initial_info = fs.info(remote_dir + "/a.txt") @@ -211,6 +220,7 @@ def test_copy(fs, remote_dir): assert strip_keys(initial_info) == strip_keys(secondary_info) +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) def test_rm(fs, remote_dir): fs.touch(remote_dir + "/a.txt") fs.rm(remote_dir + "/a.txt") @@ -229,6 +239,7 @@ def test_rm(fs, remote_dir): assert not fs.exists(remote_dir + "/dir") +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) def test_checksum(fs, remote_dir): data = b"iterative.ai" @@ -239,15 +250,19 @@ def test_checksum(fs, remote_dir): assert fs.checksum(remote_dir + "/a.txt") == checksum -def test_ls(fs, remote_dir): +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) +def test_ls(fs, remote_dir, plain_remote_dir): fs.mkdir(remote_dir + "dir/") files = set() + plain_files = set() for no in range(8): file = remote_dir + f"dir/test_{no}" + plain_file = plain_remote_dir + f"dir/test_{no}" fs.touch(file) files.add(file) + plain_files.add(plain_file) - assert set(fs.ls(remote_dir + "dir/", detail=False)) == files + assert set(fs.ls(remote_dir + "dir/", detail=False)) == plain_files dirs = fs.ls(remote_dir + "dir/") expected = [fs.info(file) for file in files] @@ -259,7 +274,8 @@ def test_ls(fs, remote_dir): assert dirs == expected -def test_walk(fs, remote_dir): +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) +def test_walk(fs, remote_dir, plain_remote_dir): fs.mkdir(remote_dir + "/a") fs.mkdir(remote_dir + "/a/b") fs.touch(remote_dir + "/a/f1") @@ -270,8 +286,8 @@ def test_walk(fs, remote_dir): for root, dirs, files in fs.walk(remote_dir + "/a") } assert result == { - remote_dir + "/a": (["b"], ["f1"]), - remote_dir + "/a/b": ([], ["f2"]), + plain_remote_dir + "/a": (["b"], ["f1"]), + plain_remote_dir + "/a/b": ([], ["f2"]), } @@ -288,6 +304,7 @@ def test_strip_protocol(): # An explicit absolute root is preserved so walk("/") lists from the root. assert strip("/") == "/" assert strip("ssh://host/") == "/" + assert strip("sftp://host/") == "/" # Empty and relative paths still resolve against the home directory # instead of being redirected to the root. assert strip("") == "" @@ -296,8 +313,13 @@ def test_strip_protocol(): # Regular absolute paths are unaffected. assert strip("/foo/bar") == "/foo/bar" assert strip("ssh://host/foo/bar") == "/foo/bar" + assert strip("sftp://user:secret@host:2222/foo/bar") == "/foo/bar" + # Note: when `path_encoding` is None, protocol-prefixed paths are not supported + # Raw paths are needed when the SFTP client's path_encoding is None. + assert strip(b"/foo/\xff") == b"/foo/\xff" +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) def test_mkdir(fs, remote_dir): fs.mkdir(remote_dir + "dir/") assert fs.isdir(remote_dir + "dir/") @@ -314,6 +336,7 @@ def test_mkdir(fs, remote_dir): assert fs.isdir(remote_dir + "dir/sub") +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) def test_makedirs(fs, remote_dir): fs.makedirs(remote_dir + "dir/a/b/c/") assert fs.isdir(remote_dir + "dir/a/b/c/") @@ -326,6 +349,7 @@ def test_makedirs(fs, remote_dir): fs.makedirs(remote_dir + "dir/a/b/c/", exist_ok=True) +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) def test_exceptions(fs, remote_dir): with pytest.raises(FileNotFoundError): with fs.open(remote_dir + "/a.txt"): @@ -339,6 +363,7 @@ def test_exceptions(fs, remote_dir): fs.makedirs(remote_dir + "/dir/a/b/c") +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) def test_open_block_size(fs, remote_dir): # mockssh (paramiko) does not implement limits@openssh.com, so the # tuned defaults must survive asyncssh's synthesized 16 KiB floors. @@ -383,6 +408,7 @@ def test_determine_block_size(): assert determine(writer, SimpleNamespace()) == WRITE_BLOCK_SIZE +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) def test_open_rw(fs, remote_dir): data = b"dvc.org" @@ -393,6 +419,7 @@ def test_open_rw(fs, remote_dir): assert stream.read() == data +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) def test_open_rw_flush(fs, remote_dir): data = b"dvc.org" @@ -406,6 +433,7 @@ def test_open_rw_flush(fs, remote_dir): assert stream.read() == data * 400 +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) def test_open_rwa(fs, remote_dir): data = b"dvc.org" @@ -421,6 +449,7 @@ def test_open_rwa(fs, remote_dir): assert stream.read() == data * 400 +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) def test_open_r_seek(fs, remote_dir): data = b"dvc.org" @@ -433,6 +462,7 @@ def test_open_r_seek(fs, remote_dir): assert stream.read() == data * 100 +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) @pytest.mark.parametrize("fs", [fs, fs_hard_queue], indirect=True) def test_concurrent_operations(fs, remote_dir): def create_random_file(): @@ -461,6 +491,7 @@ def read_random_file(name): assert write_names == read_names +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) @pytest.mark.parametrize("file_path", ["a.txt", "dir/a.txt"]) def test_put_file(fs, remote_dir, file_path): with tempfile.NamedTemporaryFile() as file: @@ -472,6 +503,7 @@ def test_put_file(fs, remote_dir, file_path): assert stream.read() == b"data" +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) def test_concurrency_for_raw_commands(fs, remote_dir): with fs.open(remote_dir + "/cp_data", "wb") as stream: stream.write(b"hello!") @@ -489,18 +521,21 @@ def test_concurrency_for_raw_commands(fs, remote_dir): future.result() +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) def test_modified(fs, remote_dir): modified = fs.modified(remote_dir) assert isinstance(modified, datetime) -def test_cat_file_sync(fs, remote_dir): +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) +def test_cat_file_sync(fs, remote_dir, plain_remote_dir): # Define the content to write to the test file test_content = b"Test content for cat_file" test_file_path = remote_dir + "/test_file.txt" + test_plain_file_path = plain_remote_dir + "/test_file.txt" # Write content to the file synchronously - with open(test_file_path, "wb") as f: + with open(test_plain_file_path, "wb") as f: f.write(test_content) # Use the cat_file method to read the content back synchronously @@ -512,6 +547,7 @@ def test_cat_file_sync(fs, remote_dir): ), "The content read from the file does not match the content written." +@pytest.mark.parametrize("remote_dir", ["plain", "sftp", "ssh"], indirect=True) def test_pipe_file(fs, remote_dir): test_data = b"Test data for pipe_file" * (2**20) # 1 MB of test data test_file_path = remote_dir + "/test_pipe_file.txt"