tests: chroot and authenticate the asyncssh fixture; pin write-through semantics

The fixture no longer serves the whole filesystem without
authentication: it is chrooted to a fresh directory, requires the
test user key, and shuts down by closing the client connection and
draining the loop's tasks (the suite previously ended with 'Task was
destroyed but it is pending!').

The functional tests now pin the write-through contract: hardlink
peers of the destination see updates and source hardlinks keep their
inode, an existing destination keeps its mode, a new file's mode is
umask-filtered, a read-only destination is refused and unchanged, and
a trailing slash on a file destination is rejected.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: shcheklein

tests/conftest.py

@@ -1,5 +1,6 @@
 import asyncio
 import threading
+from contextlib import suppress
 from pathlib import Path
 from queue import Queue
 
@@ -8,18 +9,27 @@
 import pytest
 
 _STATIC = (Path(__file__).parent / "static").resolve()
+_USER_KEY = asyncssh.read_private_key(str(_STATIC / "user.key"))
 
 
-class _NoAuthSSHServer(asyncssh.SSHServer):
+class _TestSSHServer(asyncssh.SSHServer):
     def begin_auth(self, username):
-        return False
+        return True
+
+    def public_key_auth_supported(self):
+        return True
+
+    def validate_public_key(self, username, key):
+        return key == _USER_KEY.convert_to_public()
 
 
 @pytest.fixture(scope="session")
-def asyncssh_server():
+def asyncssh_server(tmp_path_factory):
     """SFTP server that, unlike the paramiko-based mockssh fixture,
-    implements the copy-data and limits extensions. Serves the local
-    filesystem as the current user; yields (host, port)."""
+    implements the copy-data and limits extensions. Authenticated with
+    the test user key and chrooted to a fresh directory; yields
+    (host, port, root) where the remote "/" maps to root."""
+    root = tmp_path_factory.mktemp("asyncssh-root")
     loop = asyncio.new_event_loop()
     thread = threading.Thread(target=loop.run_forever, daemon=True)
     thread.start()
@@ -28,18 +38,36 @@ async def _listen():
         return await asyncssh.listen(
             "127.0.0.1",
             0,
-            server_host_keys=[str(_STATIC / "user.key")],
-            server_factory=_NoAuthSSHServer,
-            sftp_factory=asyncssh.SFTPServer,
+            server_host_keys=[_USER_KEY],
+            server_factory=_TestSSHServer,
+            sftp_factory=lambda chan: asyncssh.SFTPServer(
+                chan, chroot=str(root)
+            ),
         )
 
     server = asyncio.run_coroutine_threadsafe(_listen(), loop).result(30)
     try:
-        yield "127.0.0.1", server.get_port()
+        yield "127.0.0.1", server.get_port(), root
     finally:
-        loop.call_soon_threadsafe(server.close)
+
+        async def _shutdown():
+            server.close()
+            await server.wait_closed()
+            tasks = [
+                task
+                for task in asyncio.all_tasks()
+                if task is not asyncio.current_task()
+            ]
+            for task in tasks:
+                task.cancel()
+            if tasks:
+                await asyncio.wait(tasks, timeout=5)
+
+        with suppress(Exception):
+            asyncio.run_coroutine_threadsafe(_shutdown(), loop).result(30)
         loop.call_soon_threadsafe(loop.stop)
         thread.join(timeout=5)
+        loop.close()
 
 
 def _handler_run(self):

tests/test_sshfs.py

@@ -6,13 +6,15 @@
 import tempfile
 import warnings
 from concurrent import futures
+from contextlib import suppress
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from types import SimpleNamespace
 
 import fsspec
 import pytest
-from asyncssh.sftp import SFTPAttrs, SFTPFailure, SFTPOpUnsupported
+from asyncssh.sftp import SFTPAttrs, SFTPError, SFTPFailure, SFTPOpUnsupported
+from fsspec.asyn import sync
 from importlib_metadata import entry_points
 
 from sshfs import SSHFileSystem
@@ -234,73 +236,121 @@ async def __aexit__(self, *exc):
         return _Ctx()
 
 
-@pytest.fixture
+@pytest.fixture(scope="session")
 def copydata_fs(asyncssh_server):
-    host, port = asyncssh_server
-    yield SSHFileSystem(host=host, port=port, username="user")
+    host, port, _root = asyncssh_server
+    fs = SSHFileSystem(
+        host=host,
+        port=port,
+        username="user",
+        client_keys=[USERS["user"]],
+    )
+    yield fs
+    # Close the connection so the server fixture can shut its loop
+    # down without cancelling live connection tasks.
+    with suppress(Exception):
+        sync(fs.loop, fs._stack.aclose, timeout=5)
 
 
-def test_cp_file_copy_data(copydata_fs, tmp_path):
-    fs = copydata_fs
-    src = tmp_path / "src"
-    src.write_bytes(b"payload")
-    src.chmod(0o640)
+@pytest.fixture
+def copydata_dir(asyncssh_server, request):
+    _host, _port, root = asyncssh_server
+    local = root / request.node.name
+    local.mkdir()
+    # the server is chrooted to `root`, so `local` is served as this
+    # remote path
+    yield local, "/" + local.name
 
-    dst = tmp_path / "dst"
-    fs.cp_file(str(src), str(dst))
-    # the copy-data path was actually taken, not the shell fallback
-    assert fs._supports_remote_copy is True
-    assert dst.read_bytes() == b"payload"
-    # a new destination gets the source's mode
-    assert (dst.stat().st_mode & 0o7777) == 0o640
 
-    # an existing destination keeps its own mode, like cp
-    dst.chmod(0o600)
-    fs.cp_file(str(src), str(dst))
-    assert dst.read_bytes() == b"payload"
-    assert (dst.stat().st_mode & 0o7777) == 0o600
+def test_cp_file_copy_data(copydata_fs, copydata_dir):
+    fs = copydata_fs
+    local, remote = copydata_dir
+    (local / "src").write_bytes(b"payload")
+    (local / "src").chmod(0o666)
 
+    umask = os.umask(0)
+    os.umask(umask)
 
-def test_cp_file_copy_data_aliases(copydata_fs, tmp_path):
+    fs.cp_file(remote + "/src", remote + "/dst")
+    # the copy-data path was actually taken, not the shell fallback
+    assert fs._supports_remote_copy is True
+    assert (local / "dst").read_bytes() == b"payload"
+    # a new file gets the source's mode filtered by the server's
+    # umask, like cp
+    assert ((local / "dst").stat().st_mode & 0o7777) == 0o666 & ~umask
+
+    # an existing destination keeps its inode: its own mode survives
+    # and hardlink peers see the update, like cp writing through the
+    # file
+    (local / "dst").chmod(0o600)
+    os.link(local / "dst", local / "peer")
+    (local / "src").write_bytes(b"new payload")
+    fs.cp_file(remote + "/src", remote + "/dst")
+    assert (local / "dst").read_bytes() == b"new payload"
+    assert ((local / "dst").stat().st_mode & 0o7777) == 0o600
+    assert (local / "peer").read_bytes() == b"new payload"
+
+
+def test_cp_file_copy_data_aliases(copydata_fs, copydata_dir):
     fs = copydata_fs
-    src = tmp_path / "src"
+    local, remote = copydata_dir
+    src = local / "src"
     src.write_bytes(b"payload")
 
     with pytest.raises(shutil.SameFileError):
-        fs.cp_file(str(src), str(src))
+        fs.cp_file(remote + "/src", remote + "/src")
 
     # bytes and str spellings of the same path are still aliases
     with pytest.raises(shutil.SameFileError):
-        fs.cp_file(str(src).encode(), str(src))
+        fs.cp_file((remote + "/src").encode(), remote + "/src")
 
-    link = tmp_path / "link"
-    link.symlink_to(src)
+    (local / "link").symlink_to("src")
     with pytest.raises(shutil.SameFileError):
-        fs.cp_file(str(src), str(link))
+        fs.cp_file(remote + "/src", remote + "/link")
 
-    # hardlink aliases cannot be detected over SFTP; the copy must
-    # still never destroy the source
-    hard = tmp_path / "hard"
-    os.link(src, hard)
-    fs.cp_file(str(src), str(hard))
+    # hardlink aliases cannot be detected over SFTP: the write-through
+    # copy puts the bytes over themselves and must leave the file,
+    # its content and the link intact
+    os.link(src, local / "hard")
+    fs.cp_file(remote + "/src", remote + "/hard")
     assert src.read_bytes() == b"payload"
-    assert hard.read_bytes() == b"payload"
+    assert (local / "hard").stat().st_ino == src.stat().st_ino
 
 
-def test_cp_file_copy_data_directory_destination(copydata_fs, tmp_path):
+def test_cp_file_copy_data_directory_destination(copydata_fs, copydata_dir):
     fs = copydata_fs
-    src = tmp_path / "src"
-    src.write_bytes(b"payload")
+    local, remote = copydata_dir
+    (local / "src").write_bytes(b"payload")
+    (local / "d").mkdir()
 
-    directory = tmp_path / "directory"
-    directory.mkdir()
-    fs.cp_file(str(src), str(directory))
-    assert (directory / "src").read_bytes() == b"payload"
+    fs.cp_file(remote + "/src", remote + "/d")
+    assert (local / "d" / "src").read_bytes() == b"payload"
 
     # "copy into" resolving to the source itself is an alias
     with pytest.raises(shutil.SameFileError):
-        fs.cp_file(str(directory / "src"), str(directory))
-    assert (directory / "src").read_bytes() == b"payload"
+        fs.cp_file(remote + "/d/src", remote + "/d")
+    assert (local / "d" / "src").read_bytes() == b"payload"
+
+
+def test_cp_file_copy_data_destination_errors(copydata_fs, copydata_dir):
+    fs = copydata_fs
+    local, remote = copydata_dir
+    (local / "src").write_bytes(b"payload")
+
+    # a read-only destination is refused at open, like cp, and stays
+    # untouched
+    ro = local / "ro"
+    ro.write_bytes(b"old")
+    ro.chmod(0o444)
+    with pytest.raises(PermissionError):
+        fs.cp_file(remote + "/src", remote + "/ro")
+    assert ro.read_bytes() == b"old"
+
+    # a trailing slash on a file destination is not a directory (the
+    # server rejects it; like mkdir, the SFTP error is passed through)
+    with pytest.raises((OSError, SFTPError)):
+        fs.cp_file(remote + "/src", remote + "/ro/")
+    assert ro.read_bytes() == b"old"
 
 
 def test_mv_fallback_keeps_source_on_copy_failure(fs, monkeypatch):