From 7f06b3d56b2655dd426fbc6c5ddf68a886438420 Mon Sep 17 00:00:00 2001
From: Ian Thomas <ianthomas23@gmail.com>
Date: Mon, 15 May 2023 08:54:47 +0100
Subject: [PATCH 1/8] AbstractGetTests

---
 .../tests/memory/memory_test.py               |  18 +-
 fsspec/tests/abstract/get.py                  | 282 ++++++++++++++++++
 2 files changed, 299 insertions(+), 1 deletion(-)

diff --git a/fsspec/implementations/tests/memory/memory_test.py b/fsspec/implementations/tests/memory/memory_test.py
index fd0ebaac8..98bf063ef 100644
--- a/fsspec/implementations/tests/memory/memory_test.py
+++ b/fsspec/implementations/tests/memory/memory_test.py
@@ -1,3 +1,5 @@
+import pytest
+
 import fsspec.tests.abstract as abstract
 from fsspec.implementations.tests.memory.memory_fixtures import MemoryFixtures
 
@@ -7,7 +9,21 @@ class TestMemoryCopy(abstract.AbstractCopyTests, MemoryFixtures):
 
 
 class TestMemoryGet(abstract.AbstractGetTests, MemoryFixtures):
-    pass
+    @pytest.mark.skip(reason="Bug: does not auto-create new directory")
+    def test_get_file_to_new_directory(self):
+        pass
+
+    @pytest.mark.skip(reason="Bug: does not auto-create new directory")
+    def test_get_file_to_file_in_new_directory(self):
+        pass
+
+    @pytest.mark.skip(reason="Bug: does not auto-create new directory")
+    def test_get_glob_to_new_directory(self):
+        pass
+
+    @pytest.mark.skip(reason="Bug: does not auto-create new directory")
+    def test_get_list_of_files_to_new_directory(self):
+        pass
 
 
 class TestMemoryPut(abstract.AbstractPutTests, MemoryFixtures):
diff --git a/fsspec/tests/abstract/get.py b/fsspec/tests/abstract/get.py
index 992e9c6ae..fd7a12994 100644
--- a/fsspec/tests/abstract/get.py
+++ b/fsspec/tests/abstract/get.py
@@ -1,4 +1,286 @@
 class AbstractGetTests:
+    def test_get_file_to_existing_directory(
+        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_path
+    ):
+        # Copy scenario 1a
+        source = fs_scenario_cp
+
+        target = local_join(local_path, "target")
+        local_fs.mkdir(target)
+        if not self.supports_empty_directories():
+            local_fs.touch(local_join(target, "dummy"))
+        assert local_fs.isdir(target)
+
+        target_file2 = local_join(target, "file2")
+        target_subfile1 = local_join(target, "subfile1")
+
+        # Copy from source directory
+        fs.get(fs_join(source, "file2"), target)
+        assert local_fs.isfile(target_file2)
+
+        # Copy from sub directory
+        fs.get(fs_join(source, "subdir", "subfile1"), target)
+        assert local_fs.isfile(target_subfile1)
+
+        # Remove copied files
+        local_fs.rm([target_file2, target_subfile1])
+        assert not local_fs.exists(target_file2)
+        assert not local_fs.exists(target_subfile1)
+
+        # Repeat with trailing slash on target
+        fs.get(fs_join(source, "file2"), target + "/")
+        assert local_fs.isdir(target)
+        assert local_fs.isfile(target_file2)
+
+        fs.get(fs_join(source, "subdir", "subfile1"), target + "/")
+        assert local_fs.isfile(target_subfile1)
+
+    def test_get_file_to_new_directory(
+        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_path
+    ):
+        # Copy scenario 1b
+        source = fs_scenario_cp
+
+        target = local_join(local_path, "target")
+        local_fs.mkdir(target)
+
+        fs.get(
+            fs_join(source, "subdir", "subfile1"), local_join(target, "newdir/")
+        )  # Note trailing slash
+
+        assert local_fs.isdir(target)
+        assert local_fs.isdir(local_join(target, "newdir"))
+        assert local_fs.isfile(local_join(target, "newdir", "subfile1"))
+
+    def test_get_file_to_file_in_existing_directory(
+        self, fs, fs_join, fs_path, fs_scenario_cp, local_fs, local_join, local_path
+    ):
+        # Copy scenario 1c
+        source = fs_scenario_cp
+
+        target = local_join(local_path, "target")
+        local_fs.mkdir(target)
+
+        fs.get(fs_join(source, "subdir", "subfile1"), local_join(target, "newfile"))
+        assert local_fs.isfile(local_join(target, "newfile"))
+
+    def test_get_file_to_file_in_new_directory(
+        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_path
+    ):
+        # Copy scenario 1d
+        source = fs_scenario_cp
+
+        target = local_join(local_path, "target")
+        local_fs.mkdir(target)
+
+        fs.get(
+            fs_join(source, "subdir", "subfile1"),
+            local_join(target, "newdir", "newfile"),
+        )
+        assert local_fs.isdir(local_join(target, "newdir"))
+        assert local_fs.isfile(local_join(target, "newdir", "newfile"))
+
+    def test_get_directory_to_existing_directory(
+        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_path
+    ):
+        # Copy scenario 1e
+        source = fs_scenario_cp
+
+        target = local_join(local_path, "target")
+        local_fs.mkdir(target)
+
+        for source_slash, target_slash in zip([False, True], [False, True]):
+            s = fs_join(source, "subdir")
+            if source_slash:
+                s += "/"
+            t = target + "/" if target_slash else target
+
+            # Without recursive does nothing
+            # ERROR: erroneously creates new directory
+            # fs.get(s, t)
+            # assert fs.ls(target) == []
+
+            # With recursive
+            fs.get(s, t, recursive=True)
+            if source_slash:
+                assert local_fs.isfile(local_join(target, "subfile1"))
+                assert local_fs.isfile(local_join(target, "subfile2"))
+                assert local_fs.isdir(local_join(target, "nesteddir"))
+                assert local_fs.isfile(local_join(target, "nesteddir", "nestedfile"))
+
+                local_fs.rm(
+                    [
+                        local_join(target, "subfile1"),
+                        local_join(target, "subfile2"),
+                        local_join(target, "nesteddir"),
+                    ],
+                    recursive=True,
+                )
+            else:
+                assert local_fs.isdir(local_join(target, "subdir"))
+                assert local_fs.isfile(local_join(target, "subdir", "subfile1"))
+                assert local_fs.isfile(local_join(target, "subdir", "subfile2"))
+                assert local_fs.isdir(local_join(target, "subdir", "nesteddir"))
+                assert local_fs.isfile(
+                    local_join(target, "subdir", "nesteddir", "nestedfile")
+                )
+
+                local_fs.rm(local_join(target, "subdir"), recursive=True)
+            assert local_fs.ls(target) == []
+
+            # Limit by maxdepth
+            # ERROR: maxdepth ignored here
+
+    def test_get_directory_to_new_directory(
+        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_path
+    ):
+        # Copy scenario 1f
+        source = fs_scenario_cp
+
+        target = local_join(local_path, "target")
+        local_fs.mkdir(target)
+
+        for source_slash, target_slash in zip([False, True], [False, True]):
+            s = fs_join(source, "subdir")
+            if source_slash:
+                s += "/"
+            t = local_join(target, "newdir")
+            if target_slash:
+                t += "/"
+
+            # Without recursive does nothing
+            # ERROR: erroneously creates new directory
+            # fs.get(s, t)
+            # assert fs.ls(target) == []
+
+            # With recursive
+            fs.get(s, t, recursive=True)
+            assert local_fs.isdir(local_join(target, "newdir"))
+            assert local_fs.isfile(local_join(target, "newdir", "subfile1"))
+            assert local_fs.isfile(local_join(target, "newdir", "subfile2"))
+            assert local_fs.isdir(local_join(target, "newdir", "nesteddir"))
+            assert local_fs.isfile(
+                local_join(target, "newdir", "nesteddir", "nestedfile")
+            )
+
+            local_fs.rm(local_join(target, "newdir"), recursive=True)
+            assert local_fs.ls(target) == []
+
+            # Limit by maxdepth
+            # ERROR: maxdepth ignored here
+
+    def test_get_glob_to_existing_directory(
+        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_path
+    ):
+        # Copy scenario 1g
+        source = fs_scenario_cp
+
+        target = local_join(local_path, "target")
+        local_fs.mkdir(target)
+
+        # for target_slash in [False, True]:
+        for target_slash in [False]:
+            t = target + "/" if target_slash else target
+
+            # Without recursive
+            fs.get(fs_join(source, "subdir", "*"), t)
+            assert local_fs.isfile(local_join(target, "subfile1"))
+            assert local_fs.isfile(local_join(target, "subfile2"))
+            # assert not local_fs.isdir(local_join(target, "nesteddir"))  # ERROR
+            assert not local_fs.isdir(local_join(target, "subdir"))
+
+            # With recursive
+
+            # Limit by maxdepth
+
+    def test_get_glob_to_new_directory(
+        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_path
+    ):
+        # Copy scenario 1h
+        source = fs_scenario_cp
+
+        target = local_join(local_path, "target")
+        local_fs.mkdir(target)
+
+        for target_slash in [False, True]:
+            t = fs_join(target, "newdir")
+            if target_slash:
+                t += "/"
+
+            # Without recursive
+            fs.get(fs_join(source, "subdir", "*"), t)
+            assert local_fs.isdir(local_join(target, "newdir"))
+            assert local_fs.isfile(local_join(target, "newdir", "subfile1"))
+            assert local_fs.isfile(local_join(target, "newdir", "subfile2"))
+            # ERROR - do not copy empty directory
+            # assert not local_fs.exists(local_join(target, "newdir", "nesteddir"))
+
+            local_fs.rm(local_join(target, "newdir"), recursive=True)
+            assert local_fs.ls(target) == []
+
+            # With recursive
+            fs.get(fs_join(source, "subdir", "*"), t, recursive=True)
+            assert local_fs.isdir(local_join(target, "newdir"))
+            assert local_fs.isfile(local_join(target, "newdir", "subfile1"))
+            assert local_fs.isfile(local_join(target, "newdir", "subfile2"))
+            assert local_fs.isdir(local_join(target, "newdir", "nesteddir"))
+            assert local_fs.isfile(
+                local_join(target, "newdir", "nesteddir", "nestedfile")
+            )
+
+            local_fs.rm(local_join(target, "newdir"), recursive=True)
+            assert local_fs.ls(target) == []
+
+            # Limit by maxdepth
+            # ERROR: this is not correct
+
+    def test_get_list_of_files_to_existing_directory(
+        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_path
+    ):
+        # Copy scenario 2a
+        source = fs_scenario_cp
+
+        target = local_join(local_path, "target")
+        local_fs.mkdir(target)
+
+        source_files = [
+            fs_join(source, "file1"),
+            fs_join(source, "file2"),
+            fs_join(source, "subdir", "subfile1"),
+        ]
+
+        for target_slash in [False, True]:
+            t = target + "/" if target_slash else target
+
+            fs.get(source_files, t)
+            assert local_fs.isfile(local_join(target, "file1"))
+            assert local_fs.isfile(local_join(target, "file2"))
+            assert local_fs.isfile(local_join(target, "subfile1"))
+
+            local_fs.rm(local_fs.find(target))
+            assert local_fs.ls(target) == []
+
+    def test_get_list_of_files_to_new_directory(
+        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_path
+    ):
+        # Copy scenario 2b
+        source = fs_scenario_cp
+
+        target = local_join(local_path, "target")
+        local_fs.mkdir(target)
+
+        source_files = [
+            fs_join(source, "file1"),
+            fs_join(source, "file2"),
+            fs_join(source, "subdir", "subfile1"),
+        ]
+
+        fs.get(source_files, local_join(target, "newdir") + "/")  # Note trailing slash
+        assert local_fs.isdir(local_join(target, "newdir"))
+        assert local_fs.isfile(local_join(target, "newdir", "file1"))
+        assert local_fs.isfile(local_join(target, "newdir", "file2"))
+        assert local_fs.isfile(local_join(target, "newdir", "subfile1"))
+
     def test_get_directory_recursive(
         self, fs, fs_join, fs_path, local_fs, local_join, local_path
     ):

From d3abf8bf236bcbdc5d7a935e3041c768e7019b40 Mon Sep 17 00:00:00 2001
From: Ian Thomas <ianthomas23@gmail.com>
Date: Mon, 15 May 2023 10:14:09 +0100
Subject: [PATCH 2/8] AbstractPutTests

---
 fsspec/implementations/tests/test_memory.py |  13 -
 fsspec/spec.py                              |   4 +-
 fsspec/tests/abstract/__init__.py           |  57 ++--
 fsspec/tests/abstract/put.py                | 329 ++++++++++++++++++++
 4 files changed, 364 insertions(+), 39 deletions(-)

diff --git a/fsspec/implementations/tests/test_memory.py b/fsspec/implementations/tests/test_memory.py
index 7e734da5f..bb6fd4bfb 100644
--- a/fsspec/implementations/tests/test_memory.py
+++ b/fsspec/implementations/tests/test_memory.py
@@ -23,19 +23,6 @@ def test_strip(m):
     assert m._strip_protocol("/b/c/") == "/b/c"
 
 
-def test_put_single(m, tmpdir):
-    fn = os.path.join(str(tmpdir), "dir")
-    os.mkdir(fn)
-    open(os.path.join(fn, "abc"), "w").write("text")
-    m.put(fn, "/test")  # no-op, no files
-    assert m.isdir("/test")
-    assert not m.exists("/test/abc")
-    assert not m.exists("/test/dir")
-    m.put(fn, "/test", recursive=True)
-    assert m.isdir("/test/dir")
-    assert m.cat("/test/dir/abc") == b"text"
-
-
 def test_ls(m):
     m.mkdir("/dir")
     m.mkdir("/dir/dir1")
diff --git a/fsspec/spec.py b/fsspec/spec.py
index 493faf4ee..53e87e4d4 100644
--- a/fsspec/spec.py
+++ b/fsspec/spec.py
@@ -983,16 +983,16 @@ def put(
         lpaths = fs.expand_path(lpath, recursive=recursive, maxdepth=maxdepth)
         if source_is_str and (not recursive or maxdepth is not None):
             # Non-recursive glob does not copy directories
-            lpaths = [p for p in lpaths if not (trailing_sep(p) or self.isdir(p))]
+            lpaths = [p for p in lpaths if not (trailing_sep(p) or fs.isdir(p))]
             if not lpaths:
                 return
 
+        isdir = isinstance(rpath, str) and (trailing_sep(rpath) or self.isdir(rpath))
         rpath = (
             self._strip_protocol(rpath)
             if isinstance(rpath, str)
             else [self._strip_protocol(p) for p in rpath]
         )
-        isdir = isinstance(rpath, str) and (trailing_sep(rpath) or self.isdir(rpath))
         rpaths = other_paths(
             lpaths,
             rpath,
diff --git a/fsspec/tests/abstract/__init__.py b/fsspec/tests/abstract/__init__.py
index 4301ac0d7..71199f3df 100644
--- a/fsspec/tests/abstract/__init__.py
+++ b/fsspec/tests/abstract/__init__.py
@@ -20,31 +20,10 @@ def fs_join():
         """
         return os.path.join
 
-    @staticmethod
     @pytest.fixture
-    def fs_scenario_cp(fs, fs_join, fs_path):
-        """
-        Scenario on remote filesystem that is used for many cp/get/put tests.
-
-        📁 source
-        ├── 📄 file1
-        ├── 📄 file2
-        └── 📁 subdir
-            ├── 📄 subfile1
-            ├── 📄 subfile2
-            └── 📁 nesteddir
-                └── 📄 nestedfile
-        """
-        source = fs_join(fs_path, "source")
-        subdir = fs_join(source, "subdir")
-        nesteddir = fs_join(subdir, "nesteddir")
-        fs.makedirs(nesteddir)
-        fs.touch(fs_join(source, "file1"))
-        fs.touch(fs_join(source, "file2"))
-        fs.touch(fs_join(subdir, "subfile1"))
-        fs.touch(fs_join(subdir, "subfile2"))
-        fs.touch(fs_join(nesteddir, "nestedfile"))
-        return source
+    def fs_scenario_cp(self, fs, fs_join, fs_path):
+        """Scenario on remote filesystem that is used for many cp/get/put tests."""
+        return self.scenario_cp(fs, fs_join, fs_path)
 
     @staticmethod
     @pytest.fixture
@@ -72,3 +51,33 @@ def supports_empty_directories(self):
         Return whether this implementation supports empty directories.
         """
         return True
+
+    @pytest.fixture
+    def local_scenario_cp(self, local_fs, local_join, local_path):
+        """Scenario on local filesystem that is used for many cp/get/put tests."""
+        return self.scenario_cp(local_fs, local_join, local_path)
+
+    def scenario_cp(self, some_fs, some_join, some_path):
+        """
+        Scenario that is used for many cp/get/put tests. Creates the following
+        directory and file structure:
+
+        📁 source
+        ├── 📄 file1
+        ├── 📄 file2
+        └── 📁 subdir
+            ├── 📄 subfile1
+            ├── 📄 subfile2
+            └── 📁 nesteddir
+                └── 📄 nestedfile
+        """
+        source = some_join(some_path, "source")
+        subdir = some_join(source, "subdir")
+        nesteddir = some_join(subdir, "nesteddir")
+        some_fs.makedirs(nesteddir)
+        some_fs.touch(some_join(source, "file1"))
+        some_fs.touch(some_join(source, "file2"))
+        some_fs.touch(some_join(subdir, "subfile1"))
+        some_fs.touch(some_join(subdir, "subfile2"))
+        some_fs.touch(some_join(nesteddir, "nestedfile"))
+        return source
diff --git a/fsspec/tests/abstract/put.py b/fsspec/tests/abstract/put.py
index cdb76a8b7..abc88a830 100644
--- a/fsspec/tests/abstract/put.py
+++ b/fsspec/tests/abstract/put.py
@@ -1,4 +1,333 @@
 class AbstractPutTests:
+    def test_put_file_to_existing_directory(
+        self,
+        fs,
+        fs_join,
+        fs_path,
+        local_scenario_cp,
+        local_join,
+    ):
+        # Copy scenario 1a
+        source = local_scenario_cp
+
+        target = fs_join(fs_path, "target")
+        fs.mkdir(target)
+        if not self.supports_empty_directories():
+            fs.touch(fs_join(target, "dummy"))
+        assert fs.isdir(target)
+
+        target_file2 = fs_join(target, "file2")
+        target_subfile1 = fs_join(target, "subfile1")
+
+        # Copy from source directory
+        fs.put(local_join(source, "file2"), target)
+        assert fs.isfile(target_file2)
+
+        # Copy from sub directory
+        fs.put(local_join(source, "subdir", "subfile1"), target)
+        assert fs.isfile(target_subfile1)
+
+        # Remove copied files
+        fs.rm([target_file2, target_subfile1])
+        assert not fs.exists(target_file2)
+        assert not fs.exists(target_subfile1)
+
+        # Repeat with trailing slash on target
+        fs.put(local_join(source, "file2"), target + "/")
+        assert fs.isdir(target)
+        assert fs.isfile(target_file2)
+
+        fs.put(local_join(source, "subdir", "subfile1"), target + "/")
+        assert fs.isfile(target_subfile1)
+
+    def test_put_file_to_new_directory(
+        self, fs, fs_join, fs_path, local_scenario_cp, local_join
+    ):
+        # Copy scenario 1b
+        source = local_scenario_cp
+
+        target = fs_join(fs_path, "target")
+        fs.mkdir(target)
+
+        fs.put(
+            local_join(source, "subdir", "subfile1"), fs_join(target, "newdir/")
+        )  # Note trailing slash
+        assert fs.isdir(target)
+        assert fs.isdir(fs_join(target, "newdir"))
+        assert fs.isfile(fs_join(target, "newdir", "subfile1"))
+
+    def test_put_file_to_file_in_existing_directory(
+        self, fs, fs_join, fs_path, local_scenario_cp, local_join
+    ):
+        # Copy scenario 1c
+        source = local_scenario_cp
+
+        target = fs_join(fs_path, "target")
+        fs.mkdir(target)
+
+        fs.put(local_join(source, "subdir", "subfile1"), fs_join(target, "newfile"))
+        assert fs.isfile(fs_join(target, "newfile"))
+
+    def test_put_file_to_file_in_new_directory(
+        self, fs, fs_join, fs_path, local_scenario_cp, local_join
+    ):
+        # Copy scenario 1d
+        source = local_scenario_cp
+
+        target = fs_join(fs_path, "target")
+        fs.mkdir(target)
+
+        fs.put(
+            local_join(source, "subdir", "subfile1"),
+            fs_join(target, "newdir", "newfile"),
+        )
+        assert fs.isdir(fs_join(target, "newdir"))
+        assert fs.isfile(fs_join(target, "newdir", "newfile"))
+
+    def test_put_directory_to_existing_directory(
+        self, fs, fs_join, fs_path, local_scenario_cp
+    ):
+        # Copy scenario 1e
+        source = local_scenario_cp
+
+        target = fs_join(fs_path, "target")
+        fs.mkdir(target)
+
+        for source_slash, target_slash in zip([False, True], [False, True]):
+            s = fs_join(source, "subdir")
+            if source_slash:
+                s += "/"
+            t = target + "/" if target_slash else target
+
+            # Without recursive does nothing
+            fs.put(s, t)
+            assert fs.ls(target) == []
+
+            # With recursive
+            fs.put(s, t, recursive=True)
+            if source_slash:
+                assert fs.isfile(fs_join(target, "subfile1"))
+                assert fs.isfile(fs_join(target, "subfile2"))
+                assert fs.isdir(fs_join(target, "nesteddir"))
+                assert fs.isfile(fs_join(target, "nesteddir", "nestedfile"))
+                assert not fs.exists(fs_join(target, "subdir"))
+
+                fs.rm(fs.ls(target, detail=False), recursive=True)
+            else:
+                assert fs.isdir(fs_join(target, "subdir"))
+                assert fs.isfile(fs_join(target, "subdir", "subfile1"))
+                assert fs.isfile(fs_join(target, "subdir", "subfile2"))
+                assert fs.isdir(fs_join(target, "subdir", "nesteddir"))
+                assert fs.isfile(fs_join(target, "subdir", "nesteddir", "nestedfile"))
+
+                fs.rm(fs_join(target, "subdir"), recursive=True)
+            assert fs.ls(target) == []
+
+            # Limit recursive by maxdepth
+            fs.put(s, t, recursive=True, maxdepth=1)
+            if source_slash:
+                assert fs.isfile(fs_join(target, "subfile1"))
+                assert fs.isfile(fs_join(target, "subfile2"))
+                assert not fs.exists(fs_join(target, "nesteddir"))
+                assert not fs.exists(fs_join(target, "subdir"))
+
+                fs.rm(fs.ls(target, detail=False), recursive=True)
+            else:
+                assert fs.isdir(fs_join(target, "subdir"))
+                assert fs.isfile(fs_join(target, "subdir", "subfile1"))
+                assert fs.isfile(fs_join(target, "subdir", "subfile2"))
+                assert not fs.exists(fs_join(target, "subdir", "nesteddir"))
+
+                fs.rm(fs_join(target, "subdir"), recursive=True)
+            assert fs.ls(target) == []
+
+    def test_put_directory_to_new_directory(
+        self, fs, fs_join, fs_path, local_scenario_cp
+    ):
+        # Copy scenario 1f
+        source = local_scenario_cp
+
+        target = fs_join(fs_path, "target")
+        fs.mkdir(target)
+
+        for source_slash, target_slash in zip([False, True], [False, True]):
+            s = fs_join(source, "subdir")
+            if source_slash:
+                s += "/"
+            t = fs_join(target, "newdir")
+            if target_slash:
+                t += "/"
+
+            # Without recursive does nothing
+            fs.put(s, t)
+            assert fs.ls(target) == []
+
+            # With recursive
+            fs.put(s, t, recursive=True)
+            assert fs.isdir(fs_join(target, "newdir"))
+            assert fs.isfile(fs_join(target, "newdir", "subfile1"))
+            assert fs.isfile(fs_join(target, "newdir", "subfile2"))
+            assert fs.isdir(fs_join(target, "newdir", "nesteddir"))
+            assert fs.isfile(fs_join(target, "newdir", "nesteddir", "nestedfile"))
+            assert not fs.exists(fs_join(target, "subdir"))
+
+            fs.rm(fs_join(target, "newdir"), recursive=True)
+            assert fs.ls(target) == []
+
+            # Limit recursive by maxdepth
+            fs.put(s, t, recursive=True, maxdepth=1)
+            assert fs.isdir(fs_join(target, "newdir"))
+            assert fs.isfile(fs_join(target, "newdir", "subfile1"))
+            assert fs.isfile(fs_join(target, "newdir", "subfile2"))
+            assert not fs.exists(fs_join(target, "newdir", "nesteddir"))
+            assert not fs.exists(fs_join(target, "subdir"))
+
+            fs.rm(fs_join(target, "newdir"), recursive=True)
+            assert fs.ls(target) == []
+
+    def test_put_glob_to_existing_directory(
+        self, fs, fs_join, fs_path, local_scenario_cp, local_join
+    ):
+        # Copy scenario 1g
+        source = local_scenario_cp
+
+        target = fs_join(fs_path, "target")
+        fs.mkdir(target)
+
+        for target_slash in [False, True]:
+            t = target + "/" if target_slash else target
+
+            # Without recursive
+            fs.put(local_join(source, "subdir", "*"), t)
+            assert fs.isfile(fs_join(target, "subfile1"))
+            assert fs.isfile(fs_join(target, "subfile2"))
+            assert not fs.isdir(fs_join(target, "nesteddir"))
+            assert not fs.exists(fs_join(target, "nesteddir", "nestedfile"))
+            assert not fs.exists(fs_join(target, "subdir"))
+
+            fs.rm(fs.ls(target, detail=False), recursive=True)
+            assert fs.ls(target) == []
+
+            # With recursive
+            fs.put(local_join(source, "subdir", "*"), t, recursive=True)
+            assert fs.isfile(fs_join(target, "subfile1"))
+            assert fs.isfile(fs_join(target, "subfile2"))
+            assert fs.isdir(fs_join(target, "nesteddir"))
+            assert fs.isfile(fs_join(target, "nesteddir", "nestedfile"))
+            assert not fs.exists(fs_join(target, "subdir"))
+
+            fs.rm(fs.ls(target, detail=False), recursive=True)
+            assert fs.ls(target) == []
+
+            # Limit recursive by maxdepth
+            fs.put(local_join(source, "subdir", "*"), t, recursive=True, maxdepth=1)
+            assert fs.isfile(fs_join(target, "subfile1"))
+            assert fs.isfile(fs_join(target, "subfile2"))
+            assert not fs.exists(fs_join(target, "nesteddir"))
+            assert not fs.exists(fs_join(target, "subdir"))
+
+            fs.rm(fs.ls(target, detail=False), recursive=True)
+            assert fs.ls(target) == []
+
+    def test_put_glob_to_new_directory(
+        self, fs, fs_join, fs_path, local_scenario_cp, local_join
+    ):
+        # Copy scenario 1h
+        source = local_scenario_cp
+
+        target = fs_join(fs_path, "target")
+        fs.mkdir(target)
+
+        for target_slash in [False, True]:
+            t = fs_join(target, "newdir")
+            if target_slash:
+                t += "/"
+
+            # Without recursive
+            fs.put(local_join(source, "subdir", "*"), t)
+            assert fs.isdir(fs_join(target, "newdir"))
+            assert fs.isfile(fs_join(target, "newdir", "subfile1"))
+            assert fs.isfile(fs_join(target, "newdir", "subfile2"))
+            assert not fs.exists(fs_join(target, "newdir", "nesteddir"))
+            assert not fs.exists(fs_join(target, "newdir", "nesteddir", "nestedfile"))
+            assert not fs.exists(fs_join(target, "subdir"))
+            assert not fs.exists(fs_join(target, "newdir", "subdir"))
+
+            fs.rm(fs_join(target, "newdir"), recursive=True)
+            assert fs.ls(target) == []
+
+            # With recursive
+            fs.put(local_join(source, "subdir", "*"), t, recursive=True)
+            assert fs.isdir(fs_join(target, "newdir"))
+            assert fs.isfile(fs_join(target, "newdir", "subfile1"))
+            assert fs.isfile(fs_join(target, "newdir", "subfile2"))
+            assert fs.isdir(fs_join(target, "newdir", "nesteddir"))
+            assert fs.isfile(fs_join(target, "newdir", "nesteddir", "nestedfile"))
+            assert not fs.exists(fs_join(target, "subdir"))
+            assert not fs.exists(fs_join(target, "newdir", "subdir"))
+
+            fs.rm(fs_join(target, "newdir"), recursive=True)
+            assert fs.ls(target) == []
+
+            # Limit recursive by maxdepth
+            fs.put(local_join(source, "subdir", "*"), t, recursive=True, maxdepth=1)
+            assert fs.isdir(fs_join(target, "newdir"))
+            assert fs.isfile(fs_join(target, "newdir", "subfile1"))
+            assert fs.isfile(fs_join(target, "newdir", "subfile2"))
+            assert not fs.exists(fs_join(target, "newdir", "nesteddir"))
+            assert not fs.exists(fs_join(target, "subdir"))
+            assert not fs.exists(fs_join(target, "newdir", "subdir"))
+
+            fs.rm(fs.ls(target, detail=False), recursive=True)
+            assert fs.ls(target) == []
+
+    def test_put_list_of_files_to_existing_directory(
+        self, fs, fs_join, fs_path, local_scenario_cp, local_join
+    ):
+        # Copy scenario 2a
+        source = local_scenario_cp
+
+        target = fs_join(fs_path, "target")
+        fs.mkdir(target)
+
+        source_files = [
+            local_join(source, "file1"),
+            local_join(source, "file2"),
+            local_join(source, "subdir", "subfile1"),
+        ]
+
+        for target_slash in [False, True]:
+            t = target + "/" if target_slash else target
+
+            fs.put(source_files, t)
+            assert fs.isfile(fs_join(target, "file1"))
+            assert fs.isfile(fs_join(target, "file2"))
+            assert fs.isfile(fs_join(target, "subfile1"))
+
+            fs.rm(fs.find(target))
+            assert fs.ls(target) == []
+
+    def test_put_list_of_files_to_new_directory(
+        self, fs, fs_join, fs_path, local_scenario_cp, local_join
+    ):
+        # Copy scenario 2b
+        source = local_scenario_cp
+
+        target = fs_join(fs_path, "target")
+        fs.mkdir(target)
+
+        source_files = [
+            local_join(source, "file1"),
+            local_join(source, "file2"),
+            local_join(source, "subdir", "subfile1"),
+        ]
+
+        fs.put(source_files, fs_join(target, "newdir") + "/")  # Note trailing slash
+        assert fs.isdir(fs_join(target, "newdir"))
+        assert fs.isfile(fs_join(target, "newdir", "file1"))
+        assert fs.isfile(fs_join(target, "newdir", "file2"))
+        assert fs.isfile(fs_join(target, "newdir", "subfile1"))
+
     def test_put_directory_recursive(
         self, fs, fs_join, fs_path, local_fs, local_join, local_path
     ):

From 2b16809259c6ddd37483642b4fda26c3c1a74c77 Mon Sep 17 00:00:00 2001
From: Ian Thomas <ianthomas23@gmail.com>
Date: Mon, 15 May 2023 10:27:19 +0100
Subject: [PATCH 3/8] Make abstract test fixtures member not static functions

---
 fsspec/implementations/tests/local/local_fixtures.py |  6 ++----
 .../implementations/tests/memory/memory_fixtures.py  |  9 +++------
 fsspec/tests/abstract/__init__.py                    | 12 ++++--------
 fsspec/tests/abstract/get.py                         |  2 --
 4 files changed, 9 insertions(+), 20 deletions(-)

diff --git a/fsspec/implementations/tests/local/local_fixtures.py b/fsspec/implementations/tests/local/local_fixtures.py
index aec1d2976..c0f07fc43 100644
--- a/fsspec/implementations/tests/local/local_fixtures.py
+++ b/fsspec/implementations/tests/local/local_fixtures.py
@@ -5,12 +5,10 @@
 
 
 class LocalFixtures(AbstractFixtures):
-    @staticmethod
     @pytest.fixture
-    def fs():
+    def fs(self):
         return LocalFileSystem(auto_mkdir=True)
 
-    @staticmethod
     @pytest.fixture
-    def fs_path(tmpdir):
+    def fs_path(self, tmpdir):
         return str(tmpdir)
diff --git a/fsspec/implementations/tests/memory/memory_fixtures.py b/fsspec/implementations/tests/memory/memory_fixtures.py
index d30521221..57c3c0868 100644
--- a/fsspec/implementations/tests/memory/memory_fixtures.py
+++ b/fsspec/implementations/tests/memory/memory_fixtures.py
@@ -5,9 +5,8 @@
 
 
 class MemoryFixtures(AbstractFixtures):
-    @staticmethod
     @pytest.fixture
-    def fs():
+    def fs(self):
         m = filesystem("memory")
         m.store.clear()
         m.pseudo_dirs.clear()
@@ -19,12 +18,10 @@ def fs():
             m.pseudo_dirs.clear()
             m.pseudo_dirs.append("")
 
-    @staticmethod
     @pytest.fixture
-    def fs_join():
+    def fs_join(self):
         return lambda *args: "/".join(args)
 
-    @staticmethod
     @pytest.fixture
-    def fs_path():
+    def fs_path(self):
         return ""
diff --git a/fsspec/tests/abstract/__init__.py b/fsspec/tests/abstract/__init__.py
index 71199f3df..95dbe060d 100644
--- a/fsspec/tests/abstract/__init__.py
+++ b/fsspec/tests/abstract/__init__.py
@@ -9,9 +9,8 @@
 
 
 class AbstractFixtures:
-    @staticmethod
     @pytest.fixture
-    def fs_join():
+    def fs_join(self):
         """
         Return a function that joins its arguments together into a path.
 
@@ -25,25 +24,22 @@ def fs_scenario_cp(self, fs, fs_join, fs_path):
         """Scenario on remote filesystem that is used for many cp/get/put tests."""
         return self.scenario_cp(fs, fs_join, fs_path)
 
-    @staticmethod
     @pytest.fixture
-    def local_fs():
+    def local_fs(self):
         # Maybe need an option for auto_mkdir=False?  This is only relevant
         # for certain implementations.
         return LocalFileSystem(auto_mkdir=True)
 
-    @staticmethod
     @pytest.fixture
-    def local_join():
+    def local_join(self):
         """
         Return a function that joins its arguments together into a path, on
         the local filesystem.
         """
         return os.path.join
 
-    @staticmethod
     @pytest.fixture
-    def local_path(tmpdir):
+    def local_path(self, tmpdir):
         return tmpdir
 
     def supports_empty_directories(self):
diff --git a/fsspec/tests/abstract/get.py b/fsspec/tests/abstract/get.py
index fd7a12994..f3ab98879 100644
--- a/fsspec/tests/abstract/get.py
+++ b/fsspec/tests/abstract/get.py
@@ -7,8 +7,6 @@ def test_get_file_to_existing_directory(
 
         target = local_join(local_path, "target")
         local_fs.mkdir(target)
-        if not self.supports_empty_directories():
-            local_fs.touch(local_join(target, "dummy"))
         assert local_fs.isdir(target)
 
         target_file2 = local_join(target, "file2")

From 09d3e5e0f6e4a1f9db612847473ca61d60366edb Mon Sep 17 00:00:00 2001
From: Ian Thomas <ianthomas23@gmail.com>
Date: Mon, 15 May 2023 14:38:12 +0100
Subject: [PATCH 4/8] Async cp/get/put

---
 fsspec/asyn.py                | 86 ++++++++++++++++++++++++++++-------
 fsspec/spec.py                |  4 +-
 fsspec/tests/abstract/copy.py | 12 +++--
 fsspec/tests/abstract/put.py  | 12 +++--
 4 files changed, 89 insertions(+), 25 deletions(-)

diff --git a/fsspec/asyn.py b/fsspec/asyn.py
index d4e309790..6505de33b 100644
--- a/fsspec/asyn.py
+++ b/fsspec/asyn.py
@@ -331,20 +331,30 @@ async def _copy(
         batch_size=None,
         **kwargs,
     ):
+        from .implementations.local import trailing_sep, trailing_sep_maybe_asterisk
+
         if on_error is None and recursive:
             on_error = "ignore"
         elif on_error is None:
             on_error = "raise"
 
+        source_is_str = isinstance(path1, str)
         paths = await self._expand_path(path1, maxdepth=maxdepth, recursive=recursive)
+        if source_is_str and (not recursive or maxdepth is not None):
+            # Non-recursive glob does not copy directories
+            paths = [p for p in paths if not (trailing_sep(p) or await self._isdir(p))]
+            if not paths:
+                return
+
         isdir = isinstance(path2, str) and (
-            path2.endswith("/") or await self._isdir(path2)
+            trailing_sep(path2) or await self._isdir(path2)
         )
         path2 = other_paths(
             paths,
             path2,
-            exists=isdir and isinstance(path1, str) and not path1.endswith("/"),
+            exists=isdir and source_is_str and not trailing_sep_maybe_asterisk(path1),
             is_dir=isdir,
+            flatten=not source_is_str,
         )
         batch_size = batch_size or self.batch_size
         coros = [self._cp_file(p1, p2, **kwargs) for p1, p2 in zip(paths, path2)]
@@ -466,6 +476,7 @@ async def _put(
         recursive=False,
         callback=_DEFAULT_CALLBACK,
         batch_size=None,
+        maxdepth=None,
         **kwargs,
     ):
         """Copy file(s) from local.
@@ -481,21 +492,34 @@ async def _put(
         constructor, or for all instances by setting the "gather_batch_size" key
         in ``fsspec.config.conf``, falling back to 1/8th of the system limit .
         """
-        from .implementations.local import LocalFileSystem, make_path_posix
+        from .implementations.local import (
+            LocalFileSystem,
+            make_path_posix,
+            trailing_sep,
+            trailing_sep_maybe_asterisk,
+        )
 
-        rpath = self._strip_protocol(rpath)
-        if isinstance(lpath, str):
+        source_is_str = isinstance(lpath, str)
+        if source_is_str:
             lpath = make_path_posix(lpath)
         fs = LocalFileSystem()
-        lpaths = fs.expand_path(lpath, recursive=recursive)
+        lpaths = fs.expand_path(lpath, recursive=recursive, maxdepth=maxdepth)
+        if source_is_str and (not recursive or maxdepth is not None):
+            # Non-recursive glob does not copy directories
+            lpaths = [p for p in lpaths if not (trailing_sep(p) or fs.isdir(p))]
+            if not lpaths:
+                return
+
         isdir = isinstance(rpath, str) and (
-            rpath.endswith("/") or await self._isdir(rpath)
+            trailing_sep(rpath) or await self._isdir(rpath)
         )
+        rpath = self._strip_protocol(rpath)
         rpaths = other_paths(
             lpaths,
             rpath,
-            exists=isdir and isinstance(lpath, str) and not lpath.endswith("/"),
+            exists=isdir and source_is_str and not trailing_sep_maybe_asterisk(lpath),
             is_dir=isdir,
+            flatten=not source_is_str,
         )
 
         is_dir = {l: os.path.isdir(l) for l in lpaths}
@@ -519,7 +543,13 @@ async def _get_file(self, rpath, lpath, **kwargs):
         raise NotImplementedError
 
     async def _get(
-        self, rpath, lpath, recursive=False, callback=_DEFAULT_CALLBACK, **kwargs
+        self,
+        rpath,
+        lpath,
+        recursive=False,
+        callback=_DEFAULT_CALLBACK,
+        maxdepth=None,
+        **kwargs,
     ):
         """Copy file(s) to local.
 
@@ -535,21 +565,38 @@ async def _get(
         constructor, or for all instances by setting the "gather_batch_size" key
         in ``fsspec.config.conf``, falling back to 1/8th of the system limit .
         """
-        from fsspec.implementations.local import LocalFileSystem, make_path_posix
+        from .implementations.local import (
+            LocalFileSystem,
+            make_path_posix,
+            trailing_sep,
+            trailing_sep_maybe_asterisk,
+        )
 
+        source_is_str = isinstance(rpath, str)
         # First check for rpath trailing slash as _strip_protocol removes it.
-        rpath_trailing_slash = isinstance(rpath, str) and rpath.endswith("/")
+        source_not_trailing_sep = source_is_str and not trailing_sep_maybe_asterisk(
+            rpath
+        )
         rpath = self._strip_protocol(rpath)
-        lpath = make_path_posix(lpath)
         rpaths = await self._expand_path(rpath, recursive=recursive)
+        if source_is_str and (not recursive or maxdepth is not None):
+            # Non-recursive glob does not copy directories
+            rpaths = [
+                p for p in rpaths if not (trailing_sep(p) or await self._isdir(p))
+            ]
+            if not rpaths:
+                return
+
+        lpath = make_path_posix(lpath)
         isdir = isinstance(lpath, str) and (
-            lpath.endswith("/") or LocalFileSystem().isdir(lpath)
+            trailing_sep(lpath) or LocalFileSystem().isdir(lpath)
         )
         lpaths = other_paths(
             rpaths,
             lpath,
-            exists=isdir and not rpath_trailing_slash,
+            exists=isdir and source_not_trailing_sep,
             is_dir=isdir,
+            flatten=not source_is_str,
         )
         [os.makedirs(os.path.dirname(lp), exist_ok=True) for lp in lpaths]
         batch_size = kwargs.pop("batch_size", self.batch_size)
@@ -766,9 +813,16 @@ async def _expand_path(self, path, recursive=False, maxdepth=None):
                     bit = set(await self._glob(p))
                     out |= bit
                     if recursive:
+                        # glob call above expanded one depth so if maxdepth is defined
+                        # then decrement it in expand_path call below. If it is zero
+                        # after decrementing then avoid expand_path call.
+                        if maxdepth is not None and maxdepth <= 1:
+                            continue
                         out |= set(
                             await self._expand_path(
-                                list(bit), recursive=recursive, maxdepth=maxdepth
+                                list(bit),
+                                recursive=recursive,
+                                maxdepth=maxdepth - 1 if maxdepth is not None else None,
                             )
                         )
                     continue
@@ -778,8 +832,6 @@ async def _expand_path(self, path, recursive=False, maxdepth=None):
                 if p not in out and (recursive is False or (await self._exists(p))):
                     # should only check once, for the root
                     out.add(p)
-            # reduce depth on each recursion level unless None or 0
-            maxdepth = maxdepth if not maxdepth else maxdepth - 1
         if not out:
             raise FileNotFoundError(path)
         return list(sorted(out))
diff --git a/fsspec/spec.py b/fsspec/spec.py
index 53e87e4d4..b3918722b 100644
--- a/fsspec/spec.py
+++ b/fsspec/spec.py
@@ -976,10 +976,10 @@ def put(
             trailing_sep_maybe_asterisk,
         )
 
-        if isinstance(lpath, str):
+        source_is_str = isinstance(lpath, str)
+        if source_is_str:
             lpath = make_path_posix(lpath)
         fs = LocalFileSystem()
-        source_is_str = isinstance(lpath, str)
         lpaths = fs.expand_path(lpath, recursive=recursive, maxdepth=maxdepth)
         if source_is_str and (not recursive or maxdepth is not None):
             # Non-recursive glob does not copy directories
diff --git a/fsspec/tests/abstract/copy.py b/fsspec/tests/abstract/copy.py
index eb283649b..c39a96a85 100644
--- a/fsspec/tests/abstract/copy.py
+++ b/fsspec/tests/abstract/copy.py
@@ -8,6 +8,7 @@ def test_copy_file_to_existing_directory(
         target = fs_join(fs_path, "target")
         fs.mkdir(target)
         if not self.supports_empty_directories():
+            # Force target directory to exist by adding a dummy file
             fs.touch(fs_join(target, "dummy"))
         assert fs.isdir(target)
 
@@ -84,6 +85,11 @@ def test_copy_directory_to_existing_directory(
 
         target = fs_join(fs_path, "target")
         fs.mkdir(target)
+        if not self.supports_empty_directories():
+            # Force target directory to exist by adding a dummy file
+            dummy = fs_join(target, "dummy")
+            fs.touch(dummy)
+        assert fs.isdir(target)
 
         for source_slash, target_slash in zip([False, True], [False, True]):
             s = fs_join(source, "subdir")
@@ -93,7 +99,7 @@ def test_copy_directory_to_existing_directory(
 
             # Without recursive does nothing
             fs.cp(s, t)
-            assert fs.ls(target) == []
+            assert fs.ls(target) == [] if self.supports_empty_directories() else [dummy]
 
             # With recursive
             fs.cp(s, t, recursive=True)
@@ -113,7 +119,7 @@ def test_copy_directory_to_existing_directory(
                 assert fs.isfile(fs_join(target, "subdir", "nesteddir", "nestedfile"))
 
                 fs.rm(fs_join(target, "subdir"), recursive=True)
-            assert fs.ls(target) == []
+            assert fs.ls(target) == [] if self.supports_empty_directories() else [dummy]
 
             # Limit recursive by maxdepth
             fs.cp(s, t, recursive=True, maxdepth=1)
@@ -131,7 +137,7 @@ def test_copy_directory_to_existing_directory(
                 assert not fs.exists(fs_join(target, "subdir", "nesteddir"))
 
                 fs.rm(fs_join(target, "subdir"), recursive=True)
-            assert fs.ls(target) == []
+            assert fs.ls(target) == [] if self.supports_empty_directories() else [dummy]
 
     def test_copy_directory_to_new_directory(
         self, fs, fs_join, fs_path, fs_scenario_cp
diff --git a/fsspec/tests/abstract/put.py b/fsspec/tests/abstract/put.py
index abc88a830..6b730ddbb 100644
--- a/fsspec/tests/abstract/put.py
+++ b/fsspec/tests/abstract/put.py
@@ -13,6 +13,7 @@ def test_put_file_to_existing_directory(
         target = fs_join(fs_path, "target")
         fs.mkdir(target)
         if not self.supports_empty_directories():
+            # Force target directory to exist by adding a dummy file
             fs.touch(fs_join(target, "dummy"))
         assert fs.isdir(target)
 
@@ -92,6 +93,11 @@ def test_put_directory_to_existing_directory(
 
         target = fs_join(fs_path, "target")
         fs.mkdir(target)
+        if not self.supports_empty_directories():
+            # Force target directory to exist by adding a dummy file
+            dummy = fs_join(target, "dummy")
+            fs.touch(dummy)
+        assert fs.isdir(target)
 
         for source_slash, target_slash in zip([False, True], [False, True]):
             s = fs_join(source, "subdir")
@@ -101,7 +107,7 @@ def test_put_directory_to_existing_directory(
 
             # Without recursive does nothing
             fs.put(s, t)
-            assert fs.ls(target) == []
+            assert fs.ls(target) == [] if self.supports_empty_directories() else [dummy]
 
             # With recursive
             fs.put(s, t, recursive=True)
@@ -121,7 +127,7 @@ def test_put_directory_to_existing_directory(
                 assert fs.isfile(fs_join(target, "subdir", "nesteddir", "nestedfile"))
 
                 fs.rm(fs_join(target, "subdir"), recursive=True)
-            assert fs.ls(target) == []
+            assert fs.ls(target) == [] if self.supports_empty_directories() else [dummy]
 
             # Limit recursive by maxdepth
             fs.put(s, t, recursive=True, maxdepth=1)
@@ -139,7 +145,7 @@ def test_put_directory_to_existing_directory(
                 assert not fs.exists(fs_join(target, "subdir", "nesteddir"))
 
                 fs.rm(fs_join(target, "subdir"), recursive=True)
-            assert fs.ls(target) == []
+            assert fs.ls(target) == [] if self.supports_empty_directories() else [dummy]
 
     def test_put_directory_to_new_directory(
         self, fs, fs_join, fs_path, local_scenario_cp

From 9a4a579ba3d407444579193582c92f0ce2381095 Mon Sep 17 00:00:00 2001
From: Ian Thomas <ianthomas23@gmail.com>
Date: Mon, 15 May 2023 15:35:29 +0100
Subject: [PATCH 5/8] Test filesystems have class scope and tests clean up
 after themselves

---
 .../tests/local/local_fixtures.py             |  2 +-
 .../tests/memory/memory_fixtures.py           |  2 +-
 fsspec/registry.py                            |  2 +-
 fsspec/tests/abstract/__init__.py             | 46 ++++++++--
 fsspec/tests/abstract/copy.py                 | 61 +++++++------
 fsspec/tests/abstract/get.py                  | 44 +++++-----
 fsspec/tests/abstract/put.py                  | 88 ++++++++++++-------
 7 files changed, 153 insertions(+), 92 deletions(-)

diff --git a/fsspec/implementations/tests/local/local_fixtures.py b/fsspec/implementations/tests/local/local_fixtures.py
index c0f07fc43..d850fcf5f 100644
--- a/fsspec/implementations/tests/local/local_fixtures.py
+++ b/fsspec/implementations/tests/local/local_fixtures.py
@@ -5,7 +5,7 @@
 
 
 class LocalFixtures(AbstractFixtures):
-    @pytest.fixture
+    @pytest.fixture(scope="class")
     def fs(self):
         return LocalFileSystem(auto_mkdir=True)
 
diff --git a/fsspec/implementations/tests/memory/memory_fixtures.py b/fsspec/implementations/tests/memory/memory_fixtures.py
index 57c3c0868..27d025267 100644
--- a/fsspec/implementations/tests/memory/memory_fixtures.py
+++ b/fsspec/implementations/tests/memory/memory_fixtures.py
@@ -5,7 +5,7 @@
 
 
 class MemoryFixtures(AbstractFixtures):
-    @pytest.fixture
+    @pytest.fixture(scope="class")
     def fs(self):
         m = filesystem("memory")
         m.store.clear()
diff --git a/fsspec/registry.py b/fsspec/registry.py
index 7527f0a85..c6ef1b08f 100644
--- a/fsspec/registry.py
+++ b/fsspec/registry.py
@@ -197,7 +197,7 @@ def register_implementation(name, cls, clobber=False, errtxt=None):
     "box": {
         "class": "boxfs.BoxFileSystem",
         "err": "Please install boxfs to access BoxFileSystem",
-    }
+    },
 }
 
 
diff --git a/fsspec/tests/abstract/__init__.py b/fsspec/tests/abstract/__init__.py
index 95dbe060d..0d648048f 100644
--- a/fsspec/tests/abstract/__init__.py
+++ b/fsspec/tests/abstract/__init__.py
@@ -21,10 +21,28 @@ def fs_join(self):
 
     @pytest.fixture
     def fs_scenario_cp(self, fs, fs_join, fs_path):
-        """Scenario on remote filesystem that is used for many cp/get/put tests."""
-        return self.scenario_cp(fs, fs_join, fs_path)
+        """
+        Scenario on remote filesystem that is used for many cp/get/put tests.
+
+        Cleans up at the end of each test it which it is used.
+        """
+        source = self._scenario_cp(fs, fs_join, fs_path)
+        yield source
+        fs.rm(source, recursive=True)
 
     @pytest.fixture
+    def fs_target(self, fs, fs_join, fs_path):
+        """
+        Return name of remote directory that does not yet exist to copy into.
+
+        Cleans up at the end of each test it which it is used.
+        """
+        target = fs_join(fs_path, "target")
+        yield target
+        if fs.exists(target):
+            fs.rm(target, recursive=True)
+
+    @pytest.fixture(scope="class")
     def local_fs(self):
         # Maybe need an option for auto_mkdir=False?  This is only relevant
         # for certain implementations.
@@ -42,6 +60,18 @@ def local_join(self):
     def local_path(self, tmpdir):
         return tmpdir
 
+    @pytest.fixture
+    def local_target(self, local_fs, local_join, local_path):
+        """
+        Return name of local directory that does not yet exist to copy into.
+
+        Cleans up at the end of each test it which it is used.
+        """
+        target = local_join(local_path, "target")
+        yield target
+        if local_fs.exists(target):
+            local_fs.rm(target, recursive=True)
+
     def supports_empty_directories(self):
         """
         Return whether this implementation supports empty directories.
@@ -50,10 +80,16 @@ def supports_empty_directories(self):
 
     @pytest.fixture
     def local_scenario_cp(self, local_fs, local_join, local_path):
-        """Scenario on local filesystem that is used for many cp/get/put tests."""
-        return self.scenario_cp(local_fs, local_join, local_path)
+        """
+        Scenario on local filesystem that is used for many cp/get/put tests.
+
+        Cleans up at the end of each test it which it is used.
+        """
+        source = self._scenario_cp(local_fs, local_join, local_path)
+        yield source
+        local_fs.rm(source, recursive=True)
 
-    def scenario_cp(self, some_fs, some_join, some_path):
+    def _scenario_cp(self, some_fs, some_join, some_path):
         """
         Scenario that is used for many cp/get/put tests. Creates the following
         directory and file structure:
diff --git a/fsspec/tests/abstract/copy.py b/fsspec/tests/abstract/copy.py
index c39a96a85..7220fc9be 100644
--- a/fsspec/tests/abstract/copy.py
+++ b/fsspec/tests/abstract/copy.py
@@ -1,11 +1,11 @@
 class AbstractCopyTests:
     def test_copy_file_to_existing_directory(
-        self, fs, fs_join, fs_path, fs_scenario_cp
+        self, fs, fs_join, fs_scenario_cp, fs_target
     ):
         # Copy scenario 1a
         source = fs_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         fs.mkdir(target)
         if not self.supports_empty_directories():
             # Force target directory to exist by adding a dummy file
@@ -36,11 +36,11 @@ def test_copy_file_to_existing_directory(
         fs.cp(fs_join(source, "subdir", "subfile1"), target + "/")
         assert fs.isfile(target_subfile1)
 
-    def test_copy_file_to_new_directory(self, fs, fs_join, fs_path, fs_scenario_cp):
+    def test_copy_file_to_new_directory(self, fs, fs_join, fs_scenario_cp, fs_target):
         # Copy scenario 1b
         source = fs_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         fs.mkdir(target)
 
         fs.cp(
@@ -51,24 +51,24 @@ def test_copy_file_to_new_directory(self, fs, fs_join, fs_path, fs_scenario_cp):
         assert fs.isfile(fs_join(target, "newdir", "subfile1"))
 
     def test_copy_file_to_file_in_existing_directory(
-        self, fs, fs_join, fs_path, fs_scenario_cp
+        self, fs, fs_join, fs_scenario_cp, fs_target
     ):
         # Copy scenario 1c
         source = fs_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         fs.mkdir(target)
 
         fs.cp(fs_join(source, "subdir", "subfile1"), fs_join(target, "newfile"))
         assert fs.isfile(fs_join(target, "newfile"))
 
     def test_copy_file_to_file_in_new_directory(
-        self, fs, fs_join, fs_path, fs_scenario_cp
+        self, fs, fs_join, fs_scenario_cp, fs_target
     ):
         # Copy scenario 1d
         source = fs_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         fs.mkdir(target)
 
         fs.cp(
@@ -78,12 +78,12 @@ def test_copy_file_to_file_in_new_directory(
         assert fs.isfile(fs_join(target, "newdir", "newfile"))
 
     def test_copy_directory_to_existing_directory(
-        self, fs, fs_join, fs_path, fs_scenario_cp
+        self, fs, fs_join, fs_scenario_cp, fs_target
     ):
         # Copy scenario 1e
         source = fs_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         fs.mkdir(target)
         if not self.supports_empty_directories():
             # Force target directory to exist by adding a dummy file
@@ -140,12 +140,12 @@ def test_copy_directory_to_existing_directory(
             assert fs.ls(target) == [] if self.supports_empty_directories() else [dummy]
 
     def test_copy_directory_to_new_directory(
-        self, fs, fs_join, fs_path, fs_scenario_cp
+        self, fs, fs_join, fs_scenario_cp, fs_target
     ):
         # Copy scenario 1f
         source = fs_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         fs.mkdir(target)
 
         for source_slash, target_slash in zip([False, True], [False, True]):
@@ -170,7 +170,7 @@ def test_copy_directory_to_new_directory(
             assert not fs.exists(fs_join(target, "subdir"))
 
             fs.rm(fs_join(target, "newdir"), recursive=True)
-            assert fs.ls(target) == []
+            assert not fs.exists(fs_join(target, "newdir"))
 
             # Limit recursive by maxdepth
             fs.cp(s, t, recursive=True, maxdepth=1)
@@ -181,15 +181,15 @@ def test_copy_directory_to_new_directory(
             assert not fs.exists(fs_join(target, "subdir"))
 
             fs.rm(fs_join(target, "newdir"), recursive=True)
-            assert fs.ls(target) == []
+            assert not fs.exists(fs_join(target, "newdir"))
 
     def test_copy_glob_to_existing_directory(
-        self, fs, fs_join, fs_path, fs_scenario_cp
+        self, fs, fs_join, fs_scenario_cp, fs_target
     ):
         # Copy scenario 1g
         source = fs_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         fs.mkdir(target)
 
         for target_slash in [False, True]:
@@ -227,11 +227,11 @@ def test_copy_glob_to_existing_directory(
             fs.rm(fs.ls(target, detail=False), recursive=True)
             assert fs.ls(target) == []
 
-    def test_copy_glob_to_new_directory(self, fs, fs_join, fs_path, fs_scenario_cp):
+    def test_copy_glob_to_new_directory(self, fs, fs_join, fs_scenario_cp, fs_target):
         # Copy scenario 1h
         source = fs_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         fs.mkdir(target)
 
         for target_slash in [False, True]:
@@ -250,7 +250,7 @@ def test_copy_glob_to_new_directory(self, fs, fs_join, fs_path, fs_scenario_cp):
             assert not fs.exists(fs_join(target, "newdir", "subdir"))
 
             fs.rm(fs_join(target, "newdir"), recursive=True)
-            assert fs.ls(target) == []
+            assert not fs.exists(fs_join(target, "newdir"))
 
             # With recursive
             fs.cp(fs_join(source, "subdir", "*"), t, recursive=True)
@@ -263,7 +263,7 @@ def test_copy_glob_to_new_directory(self, fs, fs_join, fs_path, fs_scenario_cp):
             assert not fs.exists(fs_join(target, "newdir", "subdir"))
 
             fs.rm(fs_join(target, "newdir"), recursive=True)
-            assert fs.ls(target) == []
+            assert not fs.exists(fs_join(target, "newdir"))
 
             # Limit recursive by maxdepth
             fs.cp(fs_join(source, "subdir", "*"), t, recursive=True, maxdepth=1)
@@ -275,16 +275,21 @@ def test_copy_glob_to_new_directory(self, fs, fs_join, fs_path, fs_scenario_cp):
             assert not fs.exists(fs_join(target, "newdir", "subdir"))
 
             fs.rm(fs.ls(target, detail=False), recursive=True)
-            assert fs.ls(target) == []
+            assert not fs.exists(fs_join(target, "newdir"))
 
     def test_copy_list_of_files_to_existing_directory(
-        self, fs, fs_join, fs_path, fs_scenario_cp
+        self, fs, fs_join, fs_scenario_cp, fs_target
     ):
         # Copy scenario 2a
         source = fs_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         fs.mkdir(target)
+        if not self.supports_empty_directories():
+            # Force target directory to exist by adding a dummy file
+            dummy = fs_join(target, "dummy")
+            fs.touch(dummy)
+        assert fs.isdir(target)
 
         source_files = [
             fs_join(source, "file1"),
@@ -301,15 +306,15 @@ def test_copy_list_of_files_to_existing_directory(
             assert fs.isfile(fs_join(target, "subfile1"))
 
             fs.rm(fs.find(target))
-            assert fs.ls(target) == []
+            assert fs.ls(target) == [] if self.supports_empty_directories() else [dummy]
 
     def test_copy_list_of_files_to_new_directory(
-        self, fs, fs_join, fs_path, fs_scenario_cp
+        self, fs, fs_join, fs_scenario_cp, fs_target
     ):
         # Copy scenario 2b
         source = fs_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         fs.mkdir(target)
 
         source_files = [
@@ -324,12 +329,12 @@ def test_copy_list_of_files_to_new_directory(
         assert fs.isfile(fs_join(target, "newdir", "file2"))
         assert fs.isfile(fs_join(target, "newdir", "subfile1"))
 
-    def test_copy_two_files_new_directory(self, fs, fs_join, fs_path, fs_scenario_cp):
+    def test_copy_two_files_new_directory(self, fs, fs_join, fs_scenario_cp, fs_target):
         # This is a duplicate of test_copy_list_of_files_to_new_directory and
         # can eventually be removed.
         source = fs_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         assert not fs.exists(target)
         fs.cp([fs_join(source, "file1"), fs_join(source, "file2")], target)
 
diff --git a/fsspec/tests/abstract/get.py b/fsspec/tests/abstract/get.py
index f3ab98879..92e8fb96c 100644
--- a/fsspec/tests/abstract/get.py
+++ b/fsspec/tests/abstract/get.py
@@ -1,11 +1,11 @@
 class AbstractGetTests:
     def test_get_file_to_existing_directory(
-        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_path
+        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_target
     ):
         # Copy scenario 1a
         source = fs_scenario_cp
 
-        target = local_join(local_path, "target")
+        target = local_target
         local_fs.mkdir(target)
         assert local_fs.isdir(target)
 
@@ -34,12 +34,12 @@ def test_get_file_to_existing_directory(
         assert local_fs.isfile(target_subfile1)
 
     def test_get_file_to_new_directory(
-        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_path
+        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_target
     ):
         # Copy scenario 1b
         source = fs_scenario_cp
 
-        target = local_join(local_path, "target")
+        target = local_target
         local_fs.mkdir(target)
 
         fs.get(
@@ -51,24 +51,24 @@ def test_get_file_to_new_directory(
         assert local_fs.isfile(local_join(target, "newdir", "subfile1"))
 
     def test_get_file_to_file_in_existing_directory(
-        self, fs, fs_join, fs_path, fs_scenario_cp, local_fs, local_join, local_path
+        self, fs, fs_join, fs_path, fs_scenario_cp, local_fs, local_join, local_target
     ):
         # Copy scenario 1c
         source = fs_scenario_cp
 
-        target = local_join(local_path, "target")
+        target = local_target
         local_fs.mkdir(target)
 
         fs.get(fs_join(source, "subdir", "subfile1"), local_join(target, "newfile"))
         assert local_fs.isfile(local_join(target, "newfile"))
 
     def test_get_file_to_file_in_new_directory(
-        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_path
+        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_target
     ):
         # Copy scenario 1d
         source = fs_scenario_cp
 
-        target = local_join(local_path, "target")
+        target = local_target
         local_fs.mkdir(target)
 
         fs.get(
@@ -79,12 +79,12 @@ def test_get_file_to_file_in_new_directory(
         assert local_fs.isfile(local_join(target, "newdir", "newfile"))
 
     def test_get_directory_to_existing_directory(
-        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_path
+        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_target
     ):
         # Copy scenario 1e
         source = fs_scenario_cp
 
-        target = local_join(local_path, "target")
+        target = local_target
         local_fs.mkdir(target)
 
         for source_slash, target_slash in zip([False, True], [False, True]):
@@ -130,12 +130,12 @@ def test_get_directory_to_existing_directory(
             # ERROR: maxdepth ignored here
 
     def test_get_directory_to_new_directory(
-        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_path
+        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_target
     ):
         # Copy scenario 1f
         source = fs_scenario_cp
 
-        target = local_join(local_path, "target")
+        target = local_target
         local_fs.mkdir(target)
 
         for source_slash, target_slash in zip([False, True], [False, True]):
@@ -168,12 +168,12 @@ def test_get_directory_to_new_directory(
             # ERROR: maxdepth ignored here
 
     def test_get_glob_to_existing_directory(
-        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_path
+        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_target
     ):
         # Copy scenario 1g
         source = fs_scenario_cp
 
-        target = local_join(local_path, "target")
+        target = local_target
         local_fs.mkdir(target)
 
         # for target_slash in [False, True]:
@@ -192,12 +192,12 @@ def test_get_glob_to_existing_directory(
             # Limit by maxdepth
 
     def test_get_glob_to_new_directory(
-        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_path
+        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_target
     ):
         # Copy scenario 1h
         source = fs_scenario_cp
 
-        target = local_join(local_path, "target")
+        target = local_target
         local_fs.mkdir(target)
 
         for target_slash in [False, True]:
@@ -233,12 +233,12 @@ def test_get_glob_to_new_directory(
             # ERROR: this is not correct
 
     def test_get_list_of_files_to_existing_directory(
-        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_path
+        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_target
     ):
         # Copy scenario 2a
         source = fs_scenario_cp
 
-        target = local_join(local_path, "target")
+        target = local_target
         local_fs.mkdir(target)
 
         source_files = [
@@ -259,12 +259,12 @@ def test_get_list_of_files_to_existing_directory(
             assert local_fs.ls(target) == []
 
     def test_get_list_of_files_to_new_directory(
-        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_path
+        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_target
     ):
         # Copy scenario 2b
         source = fs_scenario_cp
 
-        target = local_join(local_path, "target")
+        target = local_target
         local_fs.mkdir(target)
 
         source_files = [
@@ -280,7 +280,7 @@ def test_get_list_of_files_to_new_directory(
         assert local_fs.isfile(local_join(target, "newdir", "subfile1"))
 
     def test_get_directory_recursive(
-        self, fs, fs_join, fs_path, local_fs, local_join, local_path
+        self, fs, fs_join, fs_path, local_fs, local_join, local_target
     ):
         # https://github.com/fsspec/filesystem_spec/issues/1062
         # Recursive cp/get/put of source directory into non-existent target directory.
@@ -289,7 +289,7 @@ def test_get_directory_recursive(
         fs.mkdir(src)
         fs.touch(src_file)
 
-        target = local_join(local_path, "target")
+        target = local_target
 
         # get without slash
         assert not local_fs.exists(target)
diff --git a/fsspec/tests/abstract/put.py b/fsspec/tests/abstract/put.py
index 6b730ddbb..b9e5d6278 100644
--- a/fsspec/tests/abstract/put.py
+++ b/fsspec/tests/abstract/put.py
@@ -3,14 +3,14 @@ def test_put_file_to_existing_directory(
         self,
         fs,
         fs_join,
-        fs_path,
-        local_scenario_cp,
+        fs_target,
         local_join,
+        local_scenario_cp,
     ):
         # Copy scenario 1a
         source = local_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         fs.mkdir(target)
         if not self.supports_empty_directories():
             # Force target directory to exist by adding a dummy file
@@ -42,12 +42,12 @@ def test_put_file_to_existing_directory(
         assert fs.isfile(target_subfile1)
 
     def test_put_file_to_new_directory(
-        self, fs, fs_join, fs_path, local_scenario_cp, local_join
+        self, fs, fs_join, fs_target, local_join, local_scenario_cp
     ):
         # Copy scenario 1b
         source = local_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         fs.mkdir(target)
 
         fs.put(
@@ -58,24 +58,24 @@ def test_put_file_to_new_directory(
         assert fs.isfile(fs_join(target, "newdir", "subfile1"))
 
     def test_put_file_to_file_in_existing_directory(
-        self, fs, fs_join, fs_path, local_scenario_cp, local_join
+        self, fs, fs_join, fs_target, local_join, local_scenario_cp
     ):
         # Copy scenario 1c
         source = local_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         fs.mkdir(target)
 
         fs.put(local_join(source, "subdir", "subfile1"), fs_join(target, "newfile"))
         assert fs.isfile(fs_join(target, "newfile"))
 
     def test_put_file_to_file_in_new_directory(
-        self, fs, fs_join, fs_path, local_scenario_cp, local_join
+        self, fs, fs_join, fs_target, local_join, local_scenario_cp
     ):
         # Copy scenario 1d
         source = local_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         fs.mkdir(target)
 
         fs.put(
@@ -86,12 +86,12 @@ def test_put_file_to_file_in_new_directory(
         assert fs.isfile(fs_join(target, "newdir", "newfile"))
 
     def test_put_directory_to_existing_directory(
-        self, fs, fs_join, fs_path, local_scenario_cp
+        self, fs, fs_join, fs_target, local_scenario_cp
     ):
         # Copy scenario 1e
         source = local_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         fs.mkdir(target)
         if not self.supports_empty_directories():
             # Force target directory to exist by adding a dummy file
@@ -148,13 +148,18 @@ def test_put_directory_to_existing_directory(
             assert fs.ls(target) == [] if self.supports_empty_directories() else [dummy]
 
     def test_put_directory_to_new_directory(
-        self, fs, fs_join, fs_path, local_scenario_cp
+        self, fs, fs_join, fs_target, local_scenario_cp
     ):
         # Copy scenario 1f
         source = local_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         fs.mkdir(target)
+        if not self.supports_empty_directories():
+            # Force target directory to exist by adding a dummy file
+            dummy = fs_join(target, "dummy")
+            fs.touch(dummy)
+        assert fs.isdir(target)
 
         for source_slash, target_slash in zip([False, True], [False, True]):
             s = fs_join(source, "subdir")
@@ -166,7 +171,7 @@ def test_put_directory_to_new_directory(
 
             # Without recursive does nothing
             fs.put(s, t)
-            assert fs.ls(target) == []
+            assert fs.ls(target) == [] if self.supports_empty_directories() else [dummy]
 
             # With recursive
             fs.put(s, t, recursive=True)
@@ -178,7 +183,7 @@ def test_put_directory_to_new_directory(
             assert not fs.exists(fs_join(target, "subdir"))
 
             fs.rm(fs_join(target, "newdir"), recursive=True)
-            assert fs.ls(target) == []
+            assert not fs.exists(fs_join(target, "newdir"))
 
             # Limit recursive by maxdepth
             fs.put(s, t, recursive=True, maxdepth=1)
@@ -189,16 +194,21 @@ def test_put_directory_to_new_directory(
             assert not fs.exists(fs_join(target, "subdir"))
 
             fs.rm(fs_join(target, "newdir"), recursive=True)
-            assert fs.ls(target) == []
+            assert not fs.exists(fs_join(target, "newdir"))
 
     def test_put_glob_to_existing_directory(
-        self, fs, fs_join, fs_path, local_scenario_cp, local_join
+        self, fs, fs_join, fs_target, local_join, local_scenario_cp
     ):
         # Copy scenario 1g
         source = local_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         fs.mkdir(target)
+        if not self.supports_empty_directories():
+            # Force target directory to exist by adding a dummy file
+            dummy = fs_join(target, "dummy")
+            fs.touch(dummy)
+        assert fs.isdir(target)
 
         for target_slash in [False, True]:
             t = target + "/" if target_slash else target
@@ -212,7 +222,7 @@ def test_put_glob_to_existing_directory(
             assert not fs.exists(fs_join(target, "subdir"))
 
             fs.rm(fs.ls(target, detail=False), recursive=True)
-            assert fs.ls(target) == []
+            assert fs.ls(target) == [] if self.supports_empty_directories() else [dummy]
 
             # With recursive
             fs.put(local_join(source, "subdir", "*"), t, recursive=True)
@@ -223,7 +233,7 @@ def test_put_glob_to_existing_directory(
             assert not fs.exists(fs_join(target, "subdir"))
 
             fs.rm(fs.ls(target, detail=False), recursive=True)
-            assert fs.ls(target) == []
+            assert fs.ls(target) == [] if self.supports_empty_directories() else [dummy]
 
             # Limit recursive by maxdepth
             fs.put(local_join(source, "subdir", "*"), t, recursive=True, maxdepth=1)
@@ -233,16 +243,21 @@ def test_put_glob_to_existing_directory(
             assert not fs.exists(fs_join(target, "subdir"))
 
             fs.rm(fs.ls(target, detail=False), recursive=True)
-            assert fs.ls(target) == []
+            assert fs.ls(target) == [] if self.supports_empty_directories() else [dummy]
 
     def test_put_glob_to_new_directory(
-        self, fs, fs_join, fs_path, local_scenario_cp, local_join
+        self, fs, fs_join, fs_target, local_join, local_scenario_cp
     ):
         # Copy scenario 1h
         source = local_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         fs.mkdir(target)
+        if not self.supports_empty_directories():
+            # Force target directory to exist by adding a dummy file
+            dummy = fs_join(target, "dummy")
+            fs.touch(dummy)
+        assert fs.isdir(target)
 
         for target_slash in [False, True]:
             t = fs_join(target, "newdir")
@@ -260,7 +275,7 @@ def test_put_glob_to_new_directory(
             assert not fs.exists(fs_join(target, "newdir", "subdir"))
 
             fs.rm(fs_join(target, "newdir"), recursive=True)
-            assert fs.ls(target) == []
+            assert not fs.exists(fs_join(target, "newdir"))
 
             # With recursive
             fs.put(local_join(source, "subdir", "*"), t, recursive=True)
@@ -273,7 +288,7 @@ def test_put_glob_to_new_directory(
             assert not fs.exists(fs_join(target, "newdir", "subdir"))
 
             fs.rm(fs_join(target, "newdir"), recursive=True)
-            assert fs.ls(target) == []
+            assert not fs.exists(fs_join(target, "newdir"))
 
             # Limit recursive by maxdepth
             fs.put(local_join(source, "subdir", "*"), t, recursive=True, maxdepth=1)
@@ -284,17 +299,22 @@ def test_put_glob_to_new_directory(
             assert not fs.exists(fs_join(target, "subdir"))
             assert not fs.exists(fs_join(target, "newdir", "subdir"))
 
-            fs.rm(fs.ls(target, detail=False), recursive=True)
-            assert fs.ls(target) == []
+            fs.rm(fs_join(target, "newdir"), recursive=True)
+            assert not fs.exists(fs_join(target, "newdir"))
 
     def test_put_list_of_files_to_existing_directory(
-        self, fs, fs_join, fs_path, local_scenario_cp, local_join
+        self, fs, fs_join, fs_target, local_join, local_scenario_cp, fs_path
     ):
         # Copy scenario 2a
         source = local_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         fs.mkdir(target)
+        if not self.supports_empty_directories():
+            # Force target directory to exist by adding a dummy file
+            dummy = fs_join(target, "dummy")
+            fs.touch(dummy)
+        assert fs.isdir(target)
 
         source_files = [
             local_join(source, "file1"),
@@ -311,15 +331,15 @@ def test_put_list_of_files_to_existing_directory(
             assert fs.isfile(fs_join(target, "subfile1"))
 
             fs.rm(fs.find(target))
-            assert fs.ls(target) == []
+            assert fs.ls(target) == [] if self.supports_empty_directories() else [dummy]
 
     def test_put_list_of_files_to_new_directory(
-        self, fs, fs_join, fs_path, local_scenario_cp, local_join
+        self, fs, fs_join, fs_target, local_join, local_scenario_cp
     ):
         # Copy scenario 2b
         source = local_scenario_cp
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
         fs.mkdir(target)
 
         source_files = [
@@ -335,7 +355,7 @@ def test_put_list_of_files_to_new_directory(
         assert fs.isfile(fs_join(target, "newdir", "subfile1"))
 
     def test_put_directory_recursive(
-        self, fs, fs_join, fs_path, local_fs, local_join, local_path
+        self, fs, fs_join, fs_target, local_fs, local_join, local_path
     ):
         # https://github.com/fsspec/filesystem_spec/issues/1062
         # Recursive cp/get/put of source directory into non-existent target directory.
@@ -344,7 +364,7 @@ def test_put_directory_recursive(
         local_fs.mkdir(src)
         local_fs.touch(src_file)
 
-        target = fs_join(fs_path, "target")
+        target = fs_target
 
         # put without slash
         assert not fs.exists(target)

From d35b1408bff9ffc8794585ed918f907693a42280 Mon Sep 17 00:00:00 2001
From: Ian Thomas <ianthomas23@gmail.com>
Date: Mon, 22 May 2023 09:29:39 +0100
Subject: [PATCH 6/8] Move imports to top of asyn.py

---
 fsspec/asyn.py | 22 ++++++----------------
 1 file changed, 6 insertions(+), 16 deletions(-)

diff --git a/fsspec/asyn.py b/fsspec/asyn.py
index 6505de33b..4fce103b0 100644
--- a/fsspec/asyn.py
+++ b/fsspec/asyn.py
@@ -13,6 +13,12 @@
 
 from .callbacks import _DEFAULT_CALLBACK
 from .exceptions import FSTimeoutError
+from .implementations.local import (
+    LocalFileSystem,
+    make_path_posix,
+    trailing_sep,
+    trailing_sep_maybe_asterisk,
+)
 from .spec import AbstractBufferedFile, AbstractFileSystem
 from .utils import is_exception, other_paths
 
@@ -331,8 +337,6 @@ async def _copy(
         batch_size=None,
         **kwargs,
     ):
-        from .implementations.local import trailing_sep, trailing_sep_maybe_asterisk
-
         if on_error is None and recursive:
             on_error = "ignore"
         elif on_error is None:
@@ -492,13 +496,6 @@ async def _put(
         constructor, or for all instances by setting the "gather_batch_size" key
         in ``fsspec.config.conf``, falling back to 1/8th of the system limit .
         """
-        from .implementations.local import (
-            LocalFileSystem,
-            make_path_posix,
-            trailing_sep,
-            trailing_sep_maybe_asterisk,
-        )
-
         source_is_str = isinstance(lpath, str)
         if source_is_str:
             lpath = make_path_posix(lpath)
@@ -565,13 +562,6 @@ async def _get(
         constructor, or for all instances by setting the "gather_batch_size" key
         in ``fsspec.config.conf``, falling back to 1/8th of the system limit .
         """
-        from .implementations.local import (
-            LocalFileSystem,
-            make_path_posix,
-            trailing_sep,
-            trailing_sep_maybe_asterisk,
-        )
-
         source_is_str = isinstance(rpath, str)
         # First check for rpath trailing slash as _strip_protocol removes it.
         source_not_trailing_sep = source_is_str and not trailing_sep_maybe_asterisk(

From 190c0b3060063caeb13fef6f44e9cf51664458b2 Mon Sep 17 00:00:00 2001
From: Ian Thomas <ianthomas23@gmail.com>
Date: Wed, 24 May 2023 11:06:56 +0100
Subject: [PATCH 7/8] Use bulk_operations_scenario name

---
 fsspec/tests/abstract/__init__.py |  10 +--
 fsspec/tests/abstract/copy.py     |  50 ++++++++-------
 fsspec/tests/abstract/get.py      | 101 ++++++++++++++++++++++++------
 fsspec/tests/abstract/put.py      |  46 ++++++++------
 4 files changed, 140 insertions(+), 67 deletions(-)

diff --git a/fsspec/tests/abstract/__init__.py b/fsspec/tests/abstract/__init__.py
index 0d648048f..5895f7e68 100644
--- a/fsspec/tests/abstract/__init__.py
+++ b/fsspec/tests/abstract/__init__.py
@@ -20,13 +20,13 @@ def fs_join(self):
         return os.path.join
 
     @pytest.fixture
-    def fs_scenario_cp(self, fs, fs_join, fs_path):
+    def fs_bulk_operations_scenario_0(self, fs, fs_join, fs_path):
         """
         Scenario on remote filesystem that is used for many cp/get/put tests.
 
         Cleans up at the end of each test it which it is used.
         """
-        source = self._scenario_cp(fs, fs_join, fs_path)
+        source = self._bulk_operations_scenario_0(fs, fs_join, fs_path)
         yield source
         fs.rm(source, recursive=True)
 
@@ -79,17 +79,17 @@ def supports_empty_directories(self):
         return True
 
     @pytest.fixture
-    def local_scenario_cp(self, local_fs, local_join, local_path):
+    def local_bulk_operations_scenario_0(self, local_fs, local_join, local_path):
         """
         Scenario on local filesystem that is used for many cp/get/put tests.
 
         Cleans up at the end of each test it which it is used.
         """
-        source = self._scenario_cp(local_fs, local_join, local_path)
+        source = self._bulk_operations_scenario_0(local_fs, local_join, local_path)
         yield source
         local_fs.rm(source, recursive=True)
 
-    def _scenario_cp(self, some_fs, some_join, some_path):
+    def _bulk_operations_scenario_0(self, some_fs, some_join, some_path):
         """
         Scenario that is used for many cp/get/put tests. Creates the following
         directory and file structure:
diff --git a/fsspec/tests/abstract/copy.py b/fsspec/tests/abstract/copy.py
index 7220fc9be..6498fd215 100644
--- a/fsspec/tests/abstract/copy.py
+++ b/fsspec/tests/abstract/copy.py
@@ -1,9 +1,9 @@
 class AbstractCopyTests:
     def test_copy_file_to_existing_directory(
-        self, fs, fs_join, fs_scenario_cp, fs_target
+        self, fs, fs_join, fs_bulk_operations_scenario_0, fs_target
     ):
         # Copy scenario 1a
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = fs_target
         fs.mkdir(target)
@@ -36,9 +36,11 @@ def test_copy_file_to_existing_directory(
         fs.cp(fs_join(source, "subdir", "subfile1"), target + "/")
         assert fs.isfile(target_subfile1)
 
-    def test_copy_file_to_new_directory(self, fs, fs_join, fs_scenario_cp, fs_target):
+    def test_copy_file_to_new_directory(
+        self, fs, fs_join, fs_bulk_operations_scenario_0, fs_target
+    ):
         # Copy scenario 1b
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = fs_target
         fs.mkdir(target)
@@ -51,10 +53,10 @@ def test_copy_file_to_new_directory(self, fs, fs_join, fs_scenario_cp, fs_target
         assert fs.isfile(fs_join(target, "newdir", "subfile1"))
 
     def test_copy_file_to_file_in_existing_directory(
-        self, fs, fs_join, fs_scenario_cp, fs_target
+        self, fs, fs_join, fs_bulk_operations_scenario_0, fs_target
     ):
         # Copy scenario 1c
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = fs_target
         fs.mkdir(target)
@@ -63,10 +65,10 @@ def test_copy_file_to_file_in_existing_directory(
         assert fs.isfile(fs_join(target, "newfile"))
 
     def test_copy_file_to_file_in_new_directory(
-        self, fs, fs_join, fs_scenario_cp, fs_target
+        self, fs, fs_join, fs_bulk_operations_scenario_0, fs_target
     ):
         # Copy scenario 1d
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = fs_target
         fs.mkdir(target)
@@ -78,10 +80,10 @@ def test_copy_file_to_file_in_new_directory(
         assert fs.isfile(fs_join(target, "newdir", "newfile"))
 
     def test_copy_directory_to_existing_directory(
-        self, fs, fs_join, fs_scenario_cp, fs_target
+        self, fs, fs_join, fs_bulk_operations_scenario_0, fs_target
     ):
         # Copy scenario 1e
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = fs_target
         fs.mkdir(target)
@@ -140,10 +142,10 @@ def test_copy_directory_to_existing_directory(
             assert fs.ls(target) == [] if self.supports_empty_directories() else [dummy]
 
     def test_copy_directory_to_new_directory(
-        self, fs, fs_join, fs_scenario_cp, fs_target
+        self, fs, fs_join, fs_bulk_operations_scenario_0, fs_target
     ):
         # Copy scenario 1f
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = fs_target
         fs.mkdir(target)
@@ -184,10 +186,10 @@ def test_copy_directory_to_new_directory(
             assert not fs.exists(fs_join(target, "newdir"))
 
     def test_copy_glob_to_existing_directory(
-        self, fs, fs_join, fs_scenario_cp, fs_target
+        self, fs, fs_join, fs_bulk_operations_scenario_0, fs_target
     ):
         # Copy scenario 1g
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = fs_target
         fs.mkdir(target)
@@ -227,9 +229,11 @@ def test_copy_glob_to_existing_directory(
             fs.rm(fs.ls(target, detail=False), recursive=True)
             assert fs.ls(target) == []
 
-    def test_copy_glob_to_new_directory(self, fs, fs_join, fs_scenario_cp, fs_target):
+    def test_copy_glob_to_new_directory(
+        self, fs, fs_join, fs_bulk_operations_scenario_0, fs_target
+    ):
         # Copy scenario 1h
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = fs_target
         fs.mkdir(target)
@@ -278,10 +282,10 @@ def test_copy_glob_to_new_directory(self, fs, fs_join, fs_scenario_cp, fs_target
             assert not fs.exists(fs_join(target, "newdir"))
 
     def test_copy_list_of_files_to_existing_directory(
-        self, fs, fs_join, fs_scenario_cp, fs_target
+        self, fs, fs_join, fs_bulk_operations_scenario_0, fs_target
     ):
         # Copy scenario 2a
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = fs_target
         fs.mkdir(target)
@@ -309,10 +313,10 @@ def test_copy_list_of_files_to_existing_directory(
             assert fs.ls(target) == [] if self.supports_empty_directories() else [dummy]
 
     def test_copy_list_of_files_to_new_directory(
-        self, fs, fs_join, fs_scenario_cp, fs_target
+        self, fs, fs_join, fs_bulk_operations_scenario_0, fs_target
     ):
         # Copy scenario 2b
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = fs_target
         fs.mkdir(target)
@@ -329,10 +333,12 @@ def test_copy_list_of_files_to_new_directory(
         assert fs.isfile(fs_join(target, "newdir", "file2"))
         assert fs.isfile(fs_join(target, "newdir", "subfile1"))
 
-    def test_copy_two_files_new_directory(self, fs, fs_join, fs_scenario_cp, fs_target):
+    def test_copy_two_files_new_directory(
+        self, fs, fs_join, fs_bulk_operations_scenario_0, fs_target
+    ):
         # This is a duplicate of test_copy_list_of_files_to_new_directory and
         # can eventually be removed.
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = fs_target
         assert not fs.exists(target)
diff --git a/fsspec/tests/abstract/get.py b/fsspec/tests/abstract/get.py
index 92e8fb96c..baa9aa4a9 100644
--- a/fsspec/tests/abstract/get.py
+++ b/fsspec/tests/abstract/get.py
@@ -1,9 +1,15 @@
 class AbstractGetTests:
     def test_get_file_to_existing_directory(
-        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_target
+        self,
+        fs,
+        fs_join,
+        fs_bulk_operations_scenario_0,
+        local_fs,
+        local_join,
+        local_target,
     ):
         # Copy scenario 1a
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = local_target
         local_fs.mkdir(target)
@@ -34,10 +40,16 @@ def test_get_file_to_existing_directory(
         assert local_fs.isfile(target_subfile1)
 
     def test_get_file_to_new_directory(
-        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_target
+        self,
+        fs,
+        fs_join,
+        fs_bulk_operations_scenario_0,
+        local_fs,
+        local_join,
+        local_target,
     ):
         # Copy scenario 1b
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = local_target
         local_fs.mkdir(target)
@@ -51,10 +63,17 @@ def test_get_file_to_new_directory(
         assert local_fs.isfile(local_join(target, "newdir", "subfile1"))
 
     def test_get_file_to_file_in_existing_directory(
-        self, fs, fs_join, fs_path, fs_scenario_cp, local_fs, local_join, local_target
+        self,
+        fs,
+        fs_join,
+        fs_path,
+        fs_bulk_operations_scenario_0,
+        local_fs,
+        local_join,
+        local_target,
     ):
         # Copy scenario 1c
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = local_target
         local_fs.mkdir(target)
@@ -63,10 +82,16 @@ def test_get_file_to_file_in_existing_directory(
         assert local_fs.isfile(local_join(target, "newfile"))
 
     def test_get_file_to_file_in_new_directory(
-        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_target
+        self,
+        fs,
+        fs_join,
+        fs_bulk_operations_scenario_0,
+        local_fs,
+        local_join,
+        local_target,
     ):
         # Copy scenario 1d
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = local_target
         local_fs.mkdir(target)
@@ -79,10 +104,16 @@ def test_get_file_to_file_in_new_directory(
         assert local_fs.isfile(local_join(target, "newdir", "newfile"))
 
     def test_get_directory_to_existing_directory(
-        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_target
+        self,
+        fs,
+        fs_join,
+        fs_bulk_operations_scenario_0,
+        local_fs,
+        local_join,
+        local_target,
     ):
         # Copy scenario 1e
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = local_target
         local_fs.mkdir(target)
@@ -130,10 +161,16 @@ def test_get_directory_to_existing_directory(
             # ERROR: maxdepth ignored here
 
     def test_get_directory_to_new_directory(
-        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_target
+        self,
+        fs,
+        fs_join,
+        fs_bulk_operations_scenario_0,
+        local_fs,
+        local_join,
+        local_target,
     ):
         # Copy scenario 1f
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = local_target
         local_fs.mkdir(target)
@@ -168,10 +205,16 @@ def test_get_directory_to_new_directory(
             # ERROR: maxdepth ignored here
 
     def test_get_glob_to_existing_directory(
-        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_target
+        self,
+        fs,
+        fs_join,
+        fs_bulk_operations_scenario_0,
+        local_fs,
+        local_join,
+        local_target,
     ):
         # Copy scenario 1g
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = local_target
         local_fs.mkdir(target)
@@ -192,10 +235,16 @@ def test_get_glob_to_existing_directory(
             # Limit by maxdepth
 
     def test_get_glob_to_new_directory(
-        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_target
+        self,
+        fs,
+        fs_join,
+        fs_bulk_operations_scenario_0,
+        local_fs,
+        local_join,
+        local_target,
     ):
         # Copy scenario 1h
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = local_target
         local_fs.mkdir(target)
@@ -233,10 +282,16 @@ def test_get_glob_to_new_directory(
             # ERROR: this is not correct
 
     def test_get_list_of_files_to_existing_directory(
-        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_target
+        self,
+        fs,
+        fs_join,
+        fs_bulk_operations_scenario_0,
+        local_fs,
+        local_join,
+        local_target,
     ):
         # Copy scenario 2a
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = local_target
         local_fs.mkdir(target)
@@ -259,10 +314,16 @@ def test_get_list_of_files_to_existing_directory(
             assert local_fs.ls(target) == []
 
     def test_get_list_of_files_to_new_directory(
-        self, fs, fs_join, fs_scenario_cp, local_fs, local_join, local_target
+        self,
+        fs,
+        fs_join,
+        fs_bulk_operations_scenario_0,
+        local_fs,
+        local_join,
+        local_target,
     ):
         # Copy scenario 2b
-        source = fs_scenario_cp
+        source = fs_bulk_operations_scenario_0
 
         target = local_target
         local_fs.mkdir(target)
diff --git a/fsspec/tests/abstract/put.py b/fsspec/tests/abstract/put.py
index b9e5d6278..d06f9d9b5 100644
--- a/fsspec/tests/abstract/put.py
+++ b/fsspec/tests/abstract/put.py
@@ -5,10 +5,10 @@ def test_put_file_to_existing_directory(
         fs_join,
         fs_target,
         local_join,
-        local_scenario_cp,
+        local_bulk_operations_scenario_0,
     ):
         # Copy scenario 1a
-        source = local_scenario_cp
+        source = local_bulk_operations_scenario_0
 
         target = fs_target
         fs.mkdir(target)
@@ -42,10 +42,10 @@ def test_put_file_to_existing_directory(
         assert fs.isfile(target_subfile1)
 
     def test_put_file_to_new_directory(
-        self, fs, fs_join, fs_target, local_join, local_scenario_cp
+        self, fs, fs_join, fs_target, local_join, local_bulk_operations_scenario_0
     ):
         # Copy scenario 1b
-        source = local_scenario_cp
+        source = local_bulk_operations_scenario_0
 
         target = fs_target
         fs.mkdir(target)
@@ -58,10 +58,10 @@ def test_put_file_to_new_directory(
         assert fs.isfile(fs_join(target, "newdir", "subfile1"))
 
     def test_put_file_to_file_in_existing_directory(
-        self, fs, fs_join, fs_target, local_join, local_scenario_cp
+        self, fs, fs_join, fs_target, local_join, local_bulk_operations_scenario_0
     ):
         # Copy scenario 1c
-        source = local_scenario_cp
+        source = local_bulk_operations_scenario_0
 
         target = fs_target
         fs.mkdir(target)
@@ -70,10 +70,10 @@ def test_put_file_to_file_in_existing_directory(
         assert fs.isfile(fs_join(target, "newfile"))
 
     def test_put_file_to_file_in_new_directory(
-        self, fs, fs_join, fs_target, local_join, local_scenario_cp
+        self, fs, fs_join, fs_target, local_join, local_bulk_operations_scenario_0
     ):
         # Copy scenario 1d
-        source = local_scenario_cp
+        source = local_bulk_operations_scenario_0
 
         target = fs_target
         fs.mkdir(target)
@@ -86,10 +86,10 @@ def test_put_file_to_file_in_new_directory(
         assert fs.isfile(fs_join(target, "newdir", "newfile"))
 
     def test_put_directory_to_existing_directory(
-        self, fs, fs_join, fs_target, local_scenario_cp
+        self, fs, fs_join, fs_target, local_bulk_operations_scenario_0
     ):
         # Copy scenario 1e
-        source = local_scenario_cp
+        source = local_bulk_operations_scenario_0
 
         target = fs_target
         fs.mkdir(target)
@@ -148,10 +148,10 @@ def test_put_directory_to_existing_directory(
             assert fs.ls(target) == [] if self.supports_empty_directories() else [dummy]
 
     def test_put_directory_to_new_directory(
-        self, fs, fs_join, fs_target, local_scenario_cp
+        self, fs, fs_join, fs_target, local_bulk_operations_scenario_0
     ):
         # Copy scenario 1f
-        source = local_scenario_cp
+        source = local_bulk_operations_scenario_0
 
         target = fs_target
         fs.mkdir(target)
@@ -197,10 +197,10 @@ def test_put_directory_to_new_directory(
             assert not fs.exists(fs_join(target, "newdir"))
 
     def test_put_glob_to_existing_directory(
-        self, fs, fs_join, fs_target, local_join, local_scenario_cp
+        self, fs, fs_join, fs_target, local_join, local_bulk_operations_scenario_0
     ):
         # Copy scenario 1g
-        source = local_scenario_cp
+        source = local_bulk_operations_scenario_0
 
         target = fs_target
         fs.mkdir(target)
@@ -246,10 +246,10 @@ def test_put_glob_to_existing_directory(
             assert fs.ls(target) == [] if self.supports_empty_directories() else [dummy]
 
     def test_put_glob_to_new_directory(
-        self, fs, fs_join, fs_target, local_join, local_scenario_cp
+        self, fs, fs_join, fs_target, local_join, local_bulk_operations_scenario_0
     ):
         # Copy scenario 1h
-        source = local_scenario_cp
+        source = local_bulk_operations_scenario_0
 
         target = fs_target
         fs.mkdir(target)
@@ -303,10 +303,16 @@ def test_put_glob_to_new_directory(
             assert not fs.exists(fs_join(target, "newdir"))
 
     def test_put_list_of_files_to_existing_directory(
-        self, fs, fs_join, fs_target, local_join, local_scenario_cp, fs_path
+        self,
+        fs,
+        fs_join,
+        fs_target,
+        local_join,
+        local_bulk_operations_scenario_0,
+        fs_path,
     ):
         # Copy scenario 2a
-        source = local_scenario_cp
+        source = local_bulk_operations_scenario_0
 
         target = fs_target
         fs.mkdir(target)
@@ -334,10 +340,10 @@ def test_put_list_of_files_to_existing_directory(
             assert fs.ls(target) == [] if self.supports_empty_directories() else [dummy]
 
     def test_put_list_of_files_to_new_directory(
-        self, fs, fs_join, fs_target, local_join, local_scenario_cp
+        self, fs, fs_join, fs_target, local_join, local_bulk_operations_scenario_0
     ):
         # Copy scenario 2b
-        source = local_scenario_cp
+        source = local_bulk_operations_scenario_0
 
         target = fs_target
         fs.mkdir(target)

From 31bf56cd9972cde7825f2bddaff0e085fb4c3234 Mon Sep 17 00:00:00 2001
From: Ian Thomas <ianthomas23@gmail.com>
Date: Wed, 24 May 2023 11:39:29 +0100
Subject: [PATCH 8/8] Separate class for functions that may/may not be
 overridden

---
 fsspec/tests/abstract/__init__.py | 107 ++++++++++++++++++------------
 1 file changed, 66 insertions(+), 41 deletions(-)

diff --git a/fsspec/tests/abstract/__init__.py b/fsspec/tests/abstract/__init__.py
index 5895f7e68..d2bc1627d 100644
--- a/fsspec/tests/abstract/__init__.py
+++ b/fsspec/tests/abstract/__init__.py
@@ -8,16 +8,12 @@
 from fsspec.tests.abstract.put import AbstractPutTests  # noqa
 
 
-class AbstractFixtures:
-    @pytest.fixture
-    def fs_join(self):
-        """
-        Return a function that joins its arguments together into a path.
-
-        Most fsspec implementations join paths in a platform-dependent way,
-        but some will override this to always use a forward slash.
-        """
-        return os.path.join
+class BaseAbstractFixtures:
+    """
+    Abstract base class containing fixtures that are used by but never need to
+    be overridden in derived filesystem-specific classes to run the abstract
+    tests on such filesystems.
+    """
 
     @pytest.fixture
     def fs_bulk_operations_scenario_0(self, fs, fs_join, fs_path):
@@ -42,23 +38,16 @@ def fs_target(self, fs, fs_join, fs_path):
         if fs.exists(target):
             fs.rm(target, recursive=True)
 
-    @pytest.fixture(scope="class")
-    def local_fs(self):
-        # Maybe need an option for auto_mkdir=False?  This is only relevant
-        # for certain implementations.
-        return LocalFileSystem(auto_mkdir=True)
-
     @pytest.fixture
-    def local_join(self):
-        """
-        Return a function that joins its arguments together into a path, on
-        the local filesystem.
+    def local_bulk_operations_scenario_0(self, local_fs, local_join, local_path):
         """
-        return os.path.join
+        Scenario on local filesystem that is used for many cp/get/put tests.
 
-    @pytest.fixture
-    def local_path(self, tmpdir):
-        return tmpdir
+        Cleans up at the end of each test it which it is used.
+        """
+        source = self._bulk_operations_scenario_0(local_fs, local_join, local_path)
+        yield source
+        local_fs.rm(source, recursive=True)
 
     @pytest.fixture
     def local_target(self, local_fs, local_join, local_path):
@@ -72,23 +61,6 @@ def local_target(self, local_fs, local_join, local_path):
         if local_fs.exists(target):
             local_fs.rm(target, recursive=True)
 
-    def supports_empty_directories(self):
-        """
-        Return whether this implementation supports empty directories.
-        """
-        return True
-
-    @pytest.fixture
-    def local_bulk_operations_scenario_0(self, local_fs, local_join, local_path):
-        """
-        Scenario on local filesystem that is used for many cp/get/put tests.
-
-        Cleans up at the end of each test it which it is used.
-        """
-        source = self._bulk_operations_scenario_0(local_fs, local_join, local_path)
-        yield source
-        local_fs.rm(source, recursive=True)
-
     def _bulk_operations_scenario_0(self, some_fs, some_join, some_path):
         """
         Scenario that is used for many cp/get/put tests. Creates the following
@@ -113,3 +85,56 @@ def _bulk_operations_scenario_0(self, some_fs, some_join, some_path):
         some_fs.touch(some_join(subdir, "subfile2"))
         some_fs.touch(some_join(nesteddir, "nestedfile"))
         return source
+
+
+class AbstractFixtures(BaseAbstractFixtures):
+    """
+    Abstract base class containing fixtures that may be overridden in derived
+    filesystem-specific classes to run the abstract tests on such filesystems.
+
+    For any particular filesystem some of these fixtures must be overridden,
+    such as ``fs`` and ``fs_path``, and others may be overridden if the
+    default functions here are not appropriate, such as ``fs_join``.
+    """
+
+    @pytest.fixture
+    def fs(self):
+        raise NotImplementedError("This function must be overridden in derived classes")
+
+    @pytest.fixture
+    def fs_join(self):
+        """
+        Return a function that joins its arguments together into a path.
+
+        Most fsspec implementations join paths in a platform-dependent way,
+        but some will override this to always use a forward slash.
+        """
+        return os.path.join
+
+    @pytest.fixture
+    def fs_path(self):
+        raise NotImplementedError("This function must be overridden in derived classes")
+
+    @pytest.fixture(scope="class")
+    def local_fs(self):
+        # Maybe need an option for auto_mkdir=False?  This is only relevant
+        # for certain implementations.
+        return LocalFileSystem(auto_mkdir=True)
+
+    @pytest.fixture
+    def local_join(self):
+        """
+        Return a function that joins its arguments together into a path, on
+        the local filesystem.
+        """
+        return os.path.join
+
+    @pytest.fixture
+    def local_path(self, tmpdir):
+        return tmpdir
+
+    def supports_empty_directories(self):
+        """
+        Return whether this implementation supports empty directories.
+        """
+        return True