From 354dcadded0bd4aae16b26df2094e269e35f2fa4 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sun, 4 Oct 2026 18:58:25 +0900
Subject: [PATCH 1/5] Fix null-version moves in versioning-enabled S3 buckets

---
 docs/filesystem.md                            |  15 ++
 docs/testing.md                               |  23 +++
 pyathena/filesystem/s3.py                     |  33 +++-
 pyathena/filesystem/s3_path_pairing.py        |  39 +++-
 tests/pyathena/filesystem/test_s3.py          | 169 +++++++++++++++++-
 tests/pyathena/filesystem/test_s3_async.py    |  76 ++++++++
 .../pyathena/filesystem/test_s3_versioning.py | 131 ++++++++++++++
 7 files changed, 471 insertions(+), 15 deletions(-)
 create mode 100644 tests/pyathena/filesystem/test_s3_versioning.py

diff --git a/docs/filesystem.md b/docs/filesystem.md
index 181b09b3..5c6d79b9 100644
--- a/docs/filesystem.md
+++ b/docs/filesystem.md
@@ -261,6 +261,21 @@ for version in versions:
 Version-aware operations require the `s3:GetObjectVersion` and
 `s3:ListBucketVersions` permissions.
 
+Moving a `?versionId=null` source onto its own key copies that version and
+deletes the source version when bucket versioning is enabled. If versioning has
+never been enabled or is suspended, the move leaves the source in place: copying
+onto the key would replace the null version that the subsequent deletion removes.
+The same distinction applies to move conflict checks.
+
+A move compares a null version with its unversioned key using `GetBucketVersioning`,
+which requires `s3:GetBucketVersioning`. Each relevant bucket is looked up once
+during planning; other moves do not make this request. The result is not cached
+between moves. A failed lookup stops the move before any copy or deletion.
+
+The opt-in S3 versioning integration tests create dedicated temporary buckets and
+check unversioned, enabled, and suspended states. See [Testing](testing.md) for the
+command and required test permissions.
+
 ## Bucket lifecycle
 
 Bucket creation and deletion are infrastructure-level changes and are disabled by
diff --git a/docs/testing.md b/docs/testing.md
index db559fcc..df3cc2f0 100644
--- a/docs/testing.md
+++ b/docs/testing.md
@@ -158,6 +158,29 @@ The offline-only `.env` example disables EC2 metadata to prevent implicit creden
 `--noconftest` excludes the AWS session hooks and fixtures; disabling the rerun plugin also avoids its local socket setup in restricted environments.
 Use this invocation only for self-contained modules; integration tests need their normal fixtures and a real AWS environment.
 
+### S3 null-version moves
+
+The null-version move regression tests create three dedicated temporary buckets:
+unversioned, versioning-enabled, and versioning-suspended. They run only when
+explicitly enabled, because provisioning requires additional permissions and a
+15-minute wait after first enabling versioning. Run them serially:
+
+```bash
+AWS_ATHENA_S3_VERSIONING_TESTS=1 uv run --env-file .env pytest -n 1 \
+  tests/pyathena/filesystem/test_s3_versioning.py -v
+```
+
+The test identity needs `s3:CreateBucket`, `s3:GetBucketVersioning`,
+`s3:PutBucketVersioning`, `s3:PutObject`, `s3:GetObject`, `s3:GetObjectVersion`,
+`s3:ListBucketVersions`, `s3:DeleteObjectVersion`, and `s3:DeleteBucket` on the
+temporary `pyathena-mv-null-*` buckets and their objects, in addition to the normal
+test session permissions. The fixture deletes all versions and delete markers,
+then removes its buckets, including when a test fails. An interrupted process may
+leave buckets behind; remove only the buckets created by that run.
+
+Ordinary CI does not enable these tests. Its state-specific mock regression tests
+and existing AWS suite still run; report the opt-in results separately.
+
 ### SQLAlchemy suites
 
 The SQLAlchemy compliance suites under `tests/sqlalchemy/` run with different configurations: use `sqla` for synchronous dialects and `sqla-async` for native asyncio dialects.
diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index 3e2f8a01..b868f55c 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -1414,9 +1414,17 @@ def mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs) -> None:
         ``AbstractFileSystem.mv()`` instead removes ``path1`` by expanding it
         again, which also deletes copies placed where ``path1`` matches them
         and files that ``maxdepth`` kept from being copied. A file whose
-        destination is the file itself, or the ``null`` version of a file
-        moved to the file, is left in place, and directories, which S3 does
-        not store as objects, are not copied.
+        destination is the file itself is left in place, and directories,
+        which S3 does not store as objects, are not copied. A ``null`` version
+        moved onto its key is copied and deleted only when bucket versioning
+        is enabled. Without versioning, or with versioning suspended, that
+        move is left in place because the copy would replace the version
+        that the deletion removes.
+
+        Comparing a ``null`` version with its unversioned key calls
+        GetBucketVersioning once per bucket during the move's planning and
+        requires ``s3:GetBucketVersioning``. The result is not cached across
+        moves.
 
         Args:
             path1: Source S3 path, glob pattern, or list of paths.
@@ -1465,12 +1473,27 @@ def _move_pairs(
         """
         pairing = S3PathPairing(path1, path2, recursive=recursive, maxdepth=maxdepth)
         pairs = self._copy_pairs(pairing)
+        paths = {p: S3Path.parse(p) for pair in pairs for p in pair}
+        unversioned = {p.name for p in paths.values() if not p.version_id}
+        buckets = {
+            p.bucket for p in paths.values() if p.version_id == "null" and p.name in unversioned
+        }
+        versioned_buckets = {
+            bucket
+            for bucket in buckets
+            if self._call(self._client.get_bucket_versioning, Bucket=bucket).get("Status")
+            == "Enabled"
+        }
         missing = {
             source
-            for source in pairing.conflict_candidates(pairs)
+            for source in pairing.conflict_candidates(
+                pairs, versioning_enabled_buckets=versioned_buckets
+            )
             if self._head_object(source) is None
         }
-        return pairing.move_pairs(pairs, missing=missing)
+        return pairing.move_pairs(
+            pairs, missing=missing, versioning_enabled_buckets=versioned_buckets
+        )
 
     def _copy_pairs(
         self, pairing: S3PathPairing, isdir: Callable[[str], bool] | None = None
diff --git a/pyathena/filesystem/s3_path_pairing.py b/pyathena/filesystem/s3_path_pairing.py
index d4233416..4a3fd832 100644
--- a/pyathena/filesystem/s3_path_pairing.py
+++ b/pyathena/filesystem/s3_path_pairing.py
@@ -138,7 +138,12 @@ def copy_pairs(
         destinations = other_paths(names, path2, exists=exists, flatten=not source_is_str)
         return list(zip(sources, destinations, strict=True))
 
-    def conflict_candidates(self, pairs: Sequence[tuple[str, str]]) -> list[str]:
+    def conflict_candidates(
+        self,
+        pairs: Sequence[tuple[str, str]],
+        *,
+        versioning_enabled_buckets: Collection[str] = (),
+    ) -> list[str]:
         """Return the sources of a move whose conflicts depend on an object at their key.
 
         A source with another source below it may be a directory. If no object
@@ -149,6 +154,9 @@ def conflict_candidates(self, pairs: Sequence[tuple[str, str]]) -> list[str]:
         Args:
             pairs: The sources and destinations of the move, as
                 :meth:`copy_pairs` of this pairing returns them.
+            versioning_enabled_buckets: Buckets whose versioning is enabled.
+                Their ``null`` versions are distinct from their unversioned
+                keys. Use the same collection for :meth:`move_pairs`.
 
         Returns:
             The sources, in ``bucket/key`` form (their
@@ -157,10 +165,14 @@ def conflict_candidates(self, pairs: Sequence[tuple[str, str]]) -> list[str]:
             per pair in the order of the pairs; empty if nothing needs to be
             looked up.
         """
-        return self._moves(pairs)[2]
+        return self._moves(pairs, versioning_enabled_buckets)[2]
 
     def move_pairs(
-        self, pairs: Sequence[tuple[str, str]], missing: Collection[str] | None = None
+        self,
+        pairs: Sequence[tuple[str, str]],
+        missing: Collection[str] | None = None,
+        *,
+        versioning_enabled_buckets: Collection[str] = (),
     ) -> list[tuple[str, str]]:
         """Check the pairs of a move and leave out the sources that stay in place.
 
@@ -170,10 +182,14 @@ def move_pairs(
             missing: The :meth:`conflict_candidates` without an object at their
                 key, in any form that names them; needed when there are
                 candidates.
+            versioning_enabled_buckets: Buckets whose versioning is enabled.
+                Their ``null`` versions are distinct from their unversioned
+                keys. Use the same collection for :meth:`conflict_candidates`.
 
         Returns:
             The pairs, except those whose destination is the source itself or,
-            for a ``null`` version, the key of the source.
+            for a ``null`` version in a bucket without enabled versioning,
+            the key of the source.
 
         Raises:
             ValueError: If two sources have the same destination, or a
@@ -184,12 +200,12 @@ def move_pairs(
         """
         if isinstance(missing, str):
             raise TypeError("missing is a collection of paths, not a path.")
-        named, sources, candidates = self._moves(pairs)
+        named, sources, candidates = self._moves(pairs, versioning_enabled_buckets)
         if missing is None and candidates:
             raise ValueError("missing is needed to check the pairs.")
         # A directory without an object at its key writes no destination.
         skipped = set(candidates).intersection(
-            str(S3Path.parse(path).target) for path in missing or ()
+            self._target(S3Path.parse(path), versioning_enabled_buckets) for path in missing or ()
         )
         # A path with a version always names an object, so it writes its
         # destination even when the key has no current object.
@@ -244,14 +260,21 @@ def _is_glob(path: str) -> bool:
         """
         return has_magic(path) and not S3Path.has_version_id(path)
 
+    @staticmethod
+    def _target(path: S3Path, versioning_enabled_buckets: Collection[str]) -> str:
+        """Return a move target using the caller's bucket versioning state."""
+        return str(path if path.bucket in versioning_enabled_buckets else path.target)
+
     @staticmethod
     def _moves(
         pairs: Sequence[tuple[str, str]],
+        versioning_enabled_buckets: Collection[str] = (),
     ) -> tuple[list[tuple[str, str, bool, str, str]], set[str], list[str]]:
         """Compare the paths of a move by what they name.
 
         Args:
             pairs: The sources and destinations of the move.
+            versioning_enabled_buckets: Buckets whose versioning is enabled.
 
         Returns:
             Each pair with whether its source has a version and the targets
@@ -267,8 +290,8 @@ def _moves(
                     p1,
                     p2,
                     bool(source_path.version_id),
-                    str(source_path.target),
-                    str(S3Path.parse(p2).target),
+                    S3PathPairing._target(source_path, versioning_enabled_buckets),
+                    S3PathPairing._target(S3Path.parse(p2), versioning_enabled_buckets),
                 )
             )
         # The sources left in place count too; a copy onto one overwrites it.
diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index 0e34f755..4809936d 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -795,6 +795,8 @@ def _serve_keys(fs, keys):
         keys = set(keys)
 
         def call(method, **kwargs):
+            if method is fs._client.get_bucket_versioning:
+                return {}
             if method is fs._client.head_object:
                 if kwargs["Key"] not in keys:
                     raise FileNotFoundError(kwargs["Key"])
@@ -1542,11 +1544,15 @@ def test_cp_file_directory(self):
     def test_mv_conflicting_destinations(self, path1, path2):
         fs = self._make_fs()
         fs.info = mock.MagicMock()
+        fs._call.return_value = {}
 
         with pytest.raises(ValueError, match="Cannot move"):
             fs.mv(path1, path2, recursive=True)
         fs.info.assert_not_called()
-        fs._call.assert_not_called()
+        if any(S3Path.parse(p).version_id == "null" for p in path1):
+            fs._call.assert_called_once_with(fs._client.get_bucket_versioning, Bucket="bucket")
+        else:
+            fs._call.assert_not_called()
 
     def test_mv_conflicting_destinations_of_listed_sources(self):
         # A list of sources is mapped without looking up the destination.
@@ -1644,7 +1650,7 @@ def test_mv_versions_onto_their_key(self):
 
         # The "null" version is the object at the key, so it stays in place.
         fs.mv(["s3://bucket/b?versionId=null"], ["s3://bucket/b"])
-        fs._call.assert_not_called()
+        fs._call.assert_called_once_with(fs._client.get_bucket_versioning, Bucket="bucket")
 
         # Another version is copied onto the key, and then deleted.
         fs.mv(["s3://bucket/b?versionId=v1"], ["s3://bucket/b"])
@@ -1657,6 +1663,165 @@ def test_mv_versions_onto_their_key(self):
         ]
         assert deletes == [[{"Key": "b", "VersionId": "v1"}]]
 
+    @pytest.mark.parametrize("status", [None, "Suspended", "Enabled"])
+    def test_mv_null_version_onto_key(self, status):
+        fs = self._make_fs()
+        fs._call.return_value = {} if status is None else {"Status": status}
+        fs._copy_file = mock.MagicMock(return_value=True)
+        fs._delete_objects = mock.MagicMock()
+        sources = ["s3://bucket/a?versionId=null", "s3a://bucket/b?version_id=null"]
+        destinations = ["s3://bucket/a", "s3://bucket/b"]
+
+        fs.mv(sources, destinations, MetadataDirective="COPY")
+
+        fs._call.assert_called_once_with(fs._client.get_bucket_versioning, Bucket="bucket")
+        if status == "Enabled":
+            assert fs._copy_file.call_args_list == [
+                mock.call(source, dest, MetadataDirective="COPY")
+                for source, dest in zip(sources, destinations, strict=True)
+            ]
+            fs._delete_objects.assert_called_once_with(sources)
+        else:
+            fs._copy_file.assert_not_called()
+            fs._delete_objects.assert_called_once_with([])
+
+    def test_mv_null_version_request_versions(self):
+        fs = S3FileSystem(
+            key="dummy", secret="dummy", region_name="us-east-1", skip_instance_cache=True
+        )
+        with Stubber(fs._client) as stubber:
+            stubber.add_response(
+                "get_bucket_versioning", {"Status": "Enabled"}, {"Bucket": "bucket"}
+            )
+            stubber.add_response(
+                "head_object",
+                {"ContentLength": 3, "VersionId": "null"},
+                {"Bucket": "bucket", "Key": "key", "VersionId": "null"},
+            )
+            stubber.add_response(
+                "copy_object",
+                {"CopyObjectResult": {"ETag": '"copied"'}, "VersionId": "new"},
+                {
+                    "Bucket": "bucket",
+                    "Key": "key",
+                    "CopySource": {"Bucket": "bucket", "Key": "key", "VersionId": "null"},
+                },
+            )
+            stubber.add_response(
+                "delete_objects",
+                {},
+                {
+                    "Bucket": "bucket",
+                    "Delete": {"Objects": [{"Key": "key", "VersionId": "null"}], "Quiet": True},
+                },
+            )
+
+            fs.mv(["s3://bucket/key?versionId=null"], ["s3://bucket/key"])
+            stubber.assert_no_pending_responses()
+
+    @pytest.mark.parametrize("status", [None, "Suspended", "Enabled"])
+    def test_mv_null_version_conflicts_depend_on_bucket_state(self, status):
+        fs = self._make_fs()
+        fs._call.return_value = {} if status is None else {"Status": status}
+        fs._copy_file = mock.MagicMock(return_value=True)
+        fs._delete_objects = mock.MagicMock()
+        sources = ["s3://bucket/a", "s3://bucket/b?versionId=null"]
+        destinations = ["s3a://bucket/b", "s3://bucket/out"]
+
+        with (
+            contextlib.nullcontext()
+            if status == "Enabled"
+            else pytest.raises(ValueError, match="another path that is moved")
+        ):
+            fs.mv(sources, destinations)
+
+        fs._call.assert_called_once_with(fs._client.get_bucket_versioning, Bucket="bucket")
+        if status == "Enabled":
+            assert fs._copy_file.call_args_list == [
+                mock.call(source, dest) for source, dest in zip(sources, destinations, strict=True)
+            ]
+            fs._delete_objects.assert_called_once_with(sources)
+        else:
+            fs._copy_file.assert_not_called()
+            fs._delete_objects.assert_not_called()
+
+    @pytest.mark.parametrize(
+        ("source", "dest"),
+        [
+            ("s3://bucket/a", "s3://bucket/b"),
+            ("s3://bucket/a?versionId=v1", "s3://bucket/a"),
+            ("s3://bucket/a?versionId=null", "s3://bucket/b"),
+            ("s3://bucket/a?versionId=null", "s3a://bucket/a?version_id=null"),
+        ],
+    )
+    def test_mv_without_null_key_comparison_does_not_read_bucket_state(self, source, dest):
+        fs = self._make_fs()
+        fs._copy_file = mock.MagicMock(return_value=True)
+        fs._delete_objects = mock.MagicMock()
+
+        fs.mv([source], [dest])
+        fs._call.assert_not_called()
+
+    def test_mv_null_version_reads_each_bucket(self):
+        fs = self._make_fs()
+        fs._call.side_effect = lambda method, **request: (
+            {"Status": "Enabled"} if request["Bucket"] == "enabled" else {"Status": "Suspended"}
+        )
+        fs._copy_file = mock.MagicMock(return_value=True)
+        fs._delete_objects = mock.MagicMock()
+
+        fs.mv(
+            ["s3://enabled/key?versionId=null", "s3://suspended/key?versionId=null"],
+            ["s3://enabled/key", "s3://suspended/key"],
+        )
+        assert fs._call.call_count == 2
+        fs._call.assert_has_calls(
+            [
+                mock.call(fs._client.get_bucket_versioning, Bucket="enabled"),
+                mock.call(fs._client.get_bucket_versioning, Bucket="suspended"),
+            ],
+            any_order=True,
+        )
+        fs._copy_file.assert_called_once_with("s3://enabled/key?versionId=null", "s3://enabled/key")
+        fs._delete_objects.assert_called_once_with(["s3://enabled/key?versionId=null"])
+
+    def test_mv_null_version_does_not_cache_bucket_state(self):
+        fs = self._make_fs()
+        fs._call.side_effect = [{}, {"Status": "Enabled"}, {"Status": "Suspended"}]
+        fs._copy_file = mock.MagicMock(return_value=True)
+        fs._delete_objects = mock.MagicMock()
+        source = "s3://bucket/key?versionId=null"
+
+        for _ in range(3):
+            fs.mv([source], ["s3://bucket/key"])
+        assert fs._call.call_count == 3
+        fs._copy_file.assert_called_once_with(source, "s3://bucket/key")
+        assert fs._delete_objects.call_args_list == [
+            mock.call([]),
+            mock.call([source]),
+            mock.call([]),
+        ]
+
+    @pytest.mark.parametrize("stage", ["lookup", "copy"])
+    def test_mv_null_version_failure_does_not_delete(self, stage):
+        fs = self._make_fs()
+        fs._call.return_value = {"Status": "Enabled"}
+        fs._copy_file = mock.MagicMock(return_value=True)
+        fs._delete_objects = mock.MagicMock()
+        if stage == "lookup":
+            fs._call.side_effect = PermissionError("Access Denied")
+        else:
+            fs._copy_file.side_effect = PermissionError("Access Denied")
+
+        with pytest.raises(PermissionError, match="Access Denied"):
+            fs.mv(
+                ["s3://bucket/other", "s3://bucket/key?versionId=null"],
+                ["s3://bucket/out", "s3://bucket/key"],
+            )
+        fs._delete_objects.assert_not_called()
+        if stage == "lookup":
+            fs._copy_file.assert_not_called()
+
     def test_question_mark_keys(self):
         # GH-979: keys containing "?" are keys, not version ID queries.
         fs = self._make_fs()
diff --git a/tests/pyathena/filesystem/test_s3_async.py b/tests/pyathena/filesystem/test_s3_async.py
index d77b3fff..a402d1cf 100644
--- a/tests/pyathena/filesystem/test_s3_async.py
+++ b/tests/pyathena/filesystem/test_s3_async.py
@@ -1036,6 +1036,82 @@ async def copy_file(path1, path2, **kwargs):
         assert finished == ["s3://bucket/b"]
         fs._sync_fs._call.assert_not_called()
 
+    @pytest.mark.parametrize("status", [None, "Suspended", "Enabled"])
+    @pytest.mark.asyncio
+    async def test_mv_null_version_onto_key(self, status):
+        fs = AioS3FileSystem(connection=mock.MagicMock(), skip_instance_cache=True)
+        fs._sync_fs._call = mock.MagicMock(
+            return_value={} if status is None else {"Status": status}
+        )
+        fs._copy_file = mock.AsyncMock(return_value=True)
+        fs._delete_objects = mock.AsyncMock()
+        sources = ["s3://bucket/a?versionId=null", "s3a://bucket/b?version_id=null"]
+        destinations = ["s3://bucket/a", "s3://bucket/b"]
+
+        await fs._mv(sources, destinations, MetadataDirective="COPY")
+
+        fs._sync_fs._call.assert_called_once_with(
+            fs._sync_fs._client.get_bucket_versioning, Bucket="bucket"
+        )
+        if status == "Enabled":
+            assert fs._copy_file.await_args_list == [
+                mock.call(source, dest, MetadataDirective="COPY")
+                for source, dest in zip(sources, destinations, strict=True)
+            ]
+            fs._delete_objects.assert_awaited_once_with(sources)
+        else:
+            fs._copy_file.assert_not_awaited()
+            fs._delete_objects.assert_awaited_once_with([])
+
+    @pytest.mark.parametrize("status", [None, "Suspended", "Enabled"])
+    @pytest.mark.asyncio
+    async def test_mv_null_version_conflicts_depend_on_bucket_state(self, status):
+        fs = AioS3FileSystem(connection=mock.MagicMock(), skip_instance_cache=True)
+        fs._sync_fs._call = mock.MagicMock(
+            return_value={} if status is None else {"Status": status}
+        )
+        fs._copy_file = mock.AsyncMock(return_value=True)
+        fs._delete_objects = mock.AsyncMock()
+        sources = ["s3://bucket/a", "s3://bucket/b?versionId=null"]
+        destinations = ["s3a://bucket/b", "s3://bucket/out"]
+
+        with (
+            contextlib.nullcontext()
+            if status == "Enabled"
+            else pytest.raises(ValueError, match="another path that is moved")
+        ):
+            await fs._mv(sources, destinations)
+
+        if status == "Enabled":
+            assert fs._copy_file.await_args_list == [
+                mock.call(source, dest) for source, dest in zip(sources, destinations, strict=True)
+            ]
+            fs._delete_objects.assert_awaited_once_with(sources)
+        else:
+            fs._copy_file.assert_not_awaited()
+            fs._delete_objects.assert_not_awaited()
+
+    @pytest.mark.parametrize("stage", ["lookup", "copy"])
+    @pytest.mark.asyncio
+    async def test_mv_null_version_failure_does_not_delete(self, stage):
+        fs = AioS3FileSystem(connection=mock.MagicMock(), skip_instance_cache=True)
+        fs._sync_fs._call = mock.MagicMock(return_value={"Status": "Enabled"})
+        fs._copy_file = mock.AsyncMock(return_value=True)
+        fs._delete_objects = mock.AsyncMock()
+        if stage == "lookup":
+            fs._sync_fs._call.side_effect = PermissionError("Access Denied")
+        else:
+            fs._copy_file.side_effect = PermissionError("Access Denied")
+
+        with pytest.raises(PermissionError, match="Access Denied"):
+            await fs._mv(
+                ["s3://bucket/other", "s3://bucket/key?versionId=null"],
+                ["s3://bucket/out", "s3://bucket/key"],
+            )
+        fs._delete_objects.assert_not_awaited()
+        if stage == "lookup":
+            fs._copy_file.assert_not_awaited()
+
     @pytest.mark.parametrize("size", [10, 5 * 2**30 + 1])
     @pytest.mark.asyncio
     async def test_cp_file_multipart_parameters(self, size):
diff --git a/tests/pyathena/filesystem/test_s3_versioning.py b/tests/pyathena/filesystem/test_s3_versioning.py
new file mode 100644
index 00000000..74195aec
--- /dev/null
+++ b/tests/pyathena/filesystem/test_s3_versioning.py
@@ -0,0 +1,131 @@
+# Copyright 2026 The PyAthena authors
+#
+# Licensed under the MIT License.
+# See LICENSE or https://opensource.org/licenses/MIT.
+#
+# SPDX-License-Identifier: MIT
+
+"""Opt-in real S3 tests of null-version moves using temporary buckets."""
+
+import os
+import time
+import uuid
+
+import boto3
+import pytest
+
+from pyathena.filesystem.s3 import S3FileSystem
+from pyathena.filesystem.s3_async import AioS3FileSystem
+from tests import ENV
+
+pytestmark = pytest.mark.skipif(
+    os.getenv("AWS_ATHENA_S3_VERSIONING_TESTS") != "1",
+    reason="Set AWS_ATHENA_S3_VERSIONING_TESTS=1 to create temporary versioning test buckets.",
+)
+
+BACKENDS = ["sync", "async", "async-wrapper"]
+
+
+@pytest.fixture(scope="module")
+def versioning_buckets():
+    client = boto3.client("s3", region_name=ENV.region_name)
+    buckets = {}
+    try:
+        for status in (None, "Enabled", "Suspended"):
+            bucket = f"pyathena-mv-null-{uuid.uuid4().hex}"
+            params = (
+                {"CreateBucketConfiguration": {"LocationConstraint": ENV.region_name}}
+                if ENV.region_name != "us-east-1"
+                else {}
+            )
+            client.create_bucket(Bucket=bucket, **params)
+            buckets[status] = bucket
+            for key in BACKENDS:
+                client.put_object(Bucket=bucket, Key=key, Body=b"original")
+            if status:
+                client.put_bucket_versioning(
+                    Bucket=bucket, VersioningConfiguration={"Status": "Enabled"}
+                )
+
+        # AWS recommends waiting 15 minutes after first enabling versioning
+        # before writing. Both versioned buckets propagate during this wait.
+        deadline = time.monotonic() + 15 * 60
+        while (remaining := deadline - time.monotonic()) > 0:
+            time.sleep(min(30, remaining))
+        for status in ("Enabled", "Suspended"):
+            bucket = buckets[status]
+            for key in BACKENDS:
+                response = client.put_object(Bucket=bucket, Key=key, Body=b"current")
+                assert response["VersionId"] != "null"
+            if status == "Suspended":
+                client.put_bucket_versioning(
+                    Bucket=bucket, VersioningConfiguration={"Status": "Suspended"}
+                )
+        yield client, buckets
+    finally:
+        # A failure to clean one bucket must not strand the other buckets.
+        errors = []
+        for bucket in buckets.values():
+            try:
+                objects = [
+                    {"Key": version["Key"], "VersionId": version["VersionId"]}
+                    for page in client.get_paginator("list_object_versions").paginate(Bucket=bucket)
+                    for version in page.get("Versions", []) + page.get("DeleteMarkers", [])
+                ]
+                for start in range(0, len(objects), 1000):
+                    response = client.delete_objects(
+                        Bucket=bucket,
+                        Delete={"Objects": objects[start : start + 1000], "Quiet": True},
+                    )
+                    if response.get("Errors"):
+                        raise OSError(f"Failed to clean {bucket}: {response['Errors']}")
+                client.delete_bucket(Bucket=bucket)
+            except Exception as error:
+                errors.append(error)
+        if errors:
+            raise ExceptionGroup("Failed to remove versioning test buckets", errors)
+
+
+@pytest.fixture(params=BACKENDS)
+def fs(request):
+    backend = request.param
+    cls = S3FileSystem if backend == "sync" else AioS3FileSystem
+    return backend, cls(region_name=ENV.region_name, skip_instance_cache=True)
+
+
+class TestS3NullVersionMove:
+    @pytest.mark.parametrize("status", [None, "Enabled", "Suspended"])
+    @pytest.mark.asyncio
+    async def test_mv_null_version_onto_key(self, fs, versioning_buckets, status):
+        backend, filesystem = fs
+        client, buckets = versioning_buckets
+        bucket = buckets[status]
+        path = f"s3://{bucket}/{backend}"
+        before = [
+            v
+            for v in client.list_object_versions(Bucket=bucket, Prefix=backend)["Versions"]
+            if v["Key"] == backend
+        ]
+        assert any(v["VersionId"] == "null" for v in before)
+        if status:
+            assert not next(v for v in before if v["VersionId"] == "null")["IsLatest"]
+
+        if backend == "async":
+            await filesystem._mv(f"{path}?versionId=null", path)
+        else:
+            filesystem.mv(f"{path}?versionId=null", path)
+
+        with client.get_object(Bucket=bucket, Key=backend)["Body"] as body:
+            assert body.read() == (b"original" if status != "Suspended" else b"current")
+        after = [
+            v
+            for v in client.list_object_versions(Bucket=bucket, Prefix=backend)["Versions"]
+            if v["Key"] == backend
+        ]
+        if status == "Enabled":
+            assert not any(v["VersionId"] == "null" for v in after)
+            assert len(after) == len(before)
+            latest = next(v for v in after if v["IsLatest"])
+            assert latest["VersionId"] not in {v["VersionId"] for v in before}
+        else:
+            assert after == before

From 421ee3c8196ad8734c84f6b80b732c0826049a27 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sun, 4 Oct 2026 19:15:38 +0900
Subject: [PATCH 2/5] Preserve null-version moves in directory buckets

---
 docs/filesystem.md                         |  2 ++
 pyathena/filesystem/s3.py                  |  8 ++++++--
 tests/pyathena/filesystem/test_s3.py       | 21 +++++++++++++++++++++
 tests/pyathena/filesystem/test_s3_async.py | 14 ++++++++++++++
 4 files changed, 43 insertions(+), 2 deletions(-)

diff --git a/docs/filesystem.md b/docs/filesystem.md
index 5c6d79b9..126d93b0 100644
--- a/docs/filesystem.md
+++ b/docs/filesystem.md
@@ -271,6 +271,8 @@ A move compares a null version with its unversioned key using `GetBucketVersioni
 which requires `s3:GetBucketVersioning`. Each relevant bucket is looked up once
 during planning; other moves do not make this request. The result is not cached
 between moves. A failed lookup stops the move before any copy or deletion.
+Directory buckets do not support versioning, so their null versions are treated
+as the key itself without a versioning lookup.
 
 The opt-in S3 versioning integration tests create dedicated temporary buckets and
 check unversioned, enabled, and suspended states. See [Testing](testing.md) for the
diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index b868f55c..731ca08c 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -1424,7 +1424,7 @@ def mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs) -> None:
         Comparing a ``null`` version with its unversioned key calls
         GetBucketVersioning once per bucket during the move's planning and
         requires ``s3:GetBucketVersioning``. The result is not cached across
-        moves.
+        moves. Directory buckets do not support versioning and need no lookup.
 
         Args:
             path1: Source S3 path, glob pattern, or list of paths.
@@ -1476,7 +1476,11 @@ def _move_pairs(
         paths = {p: S3Path.parse(p) for pair in pairs for p in pair}
         unversioned = {p.name for p in paths.values() if not p.version_id}
         buckets = {
-            p.bucket for p in paths.values() if p.version_id == "null" and p.name in unversioned
+            p.bucket
+            for p in paths.values()
+            if p.version_id == "null"
+            and p.name in unversioned
+            and not self.core._is_directory_bucket(p.bucket)
         }
         versioned_buckets = {
             bucket
diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index 4809936d..9397f933 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -1762,6 +1762,27 @@ def test_mv_without_null_key_comparison_does_not_read_bucket_state(self, source,
         fs.mv([source], [dest])
         fs._call.assert_not_called()
 
+    @pytest.mark.parametrize("same_key", [False, True])
+    def test_mv_null_version_directory_bucket_does_not_read_bucket_state(self, same_key):
+        fs = self._make_fs()
+        fs._copy_file = mock.MagicMock(return_value=True)
+        fs._delete_objects = mock.MagicMock()
+        key = "s3://example--usw2-az1--x-s3/key"
+        sources = [f"{key}?versionId=null"] if same_key else [f"{key}?versionId=null", key]
+        destinations = [key] if same_key else [f"{key}-copy1", f"{key}-copy2"]
+
+        fs.mv(sources, destinations)
+
+        fs._call.assert_not_called()
+        if same_key:
+            fs._copy_file.assert_not_called()
+            fs._delete_objects.assert_called_once_with([])
+        else:
+            assert fs._copy_file.call_args_list == [
+                mock.call(source, dest) for source, dest in zip(sources, destinations, strict=True)
+            ]
+            fs._delete_objects.assert_called_once_with(sources)
+
     def test_mv_null_version_reads_each_bucket(self):
         fs = self._make_fs()
         fs._call.side_effect = lambda method, **request: (
diff --git a/tests/pyathena/filesystem/test_s3_async.py b/tests/pyathena/filesystem/test_s3_async.py
index a402d1cf..f12022d6 100644
--- a/tests/pyathena/filesystem/test_s3_async.py
+++ b/tests/pyathena/filesystem/test_s3_async.py
@@ -1091,6 +1091,20 @@ async def test_mv_null_version_conflicts_depend_on_bucket_state(self, status):
             fs._copy_file.assert_not_awaited()
             fs._delete_objects.assert_not_awaited()
 
+    @pytest.mark.asyncio
+    async def test_mv_null_version_directory_bucket_does_not_read_bucket_state(self):
+        fs = AioS3FileSystem(connection=mock.MagicMock(), skip_instance_cache=True)
+        fs._sync_fs._call = mock.MagicMock()
+        fs._copy_file = mock.AsyncMock(return_value=True)
+        fs._delete_objects = mock.AsyncMock()
+        key = "s3://example--usw2-az1--x-s3/key"
+
+        await fs._mv([f"{key}?versionId=null"], [key])
+
+        fs._sync_fs._call.assert_not_called()
+        fs._copy_file.assert_not_awaited()
+        fs._delete_objects.assert_awaited_once_with([])
+
     @pytest.mark.parametrize("stage", ["lookup", "copy"])
     @pytest.mark.asyncio
     async def test_mv_null_version_failure_does_not_delete(self, stage):

From ca44c77c8f3aed8dd5f9af3d04b8cede280196f9 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sun, 4 Oct 2026 22:47:14 +0900
Subject: [PATCH 3/5] Document and test bucket state in move path pairing

---
 pyathena/filesystem/s3_path.py                |  6 ++--
 pyathena/filesystem/s3_path_pairing.py        |  4 ++-
 .../filesystem/test_s3_path_pairing.py        | 31 +++++++++++++++++++
 3 files changed, 37 insertions(+), 4 deletions(-)

diff --git a/pyathena/filesystem/s3_path.py b/pyathena/filesystem/s3_path.py
index 1f2a4f38..19dbcf84 100644
--- a/pyathena/filesystem/s3_path.py
+++ b/pyathena/filesystem/s3_path.py
@@ -135,13 +135,13 @@ def uri(self) -> str:
 
     @property
     def target(self) -> S3Path:
-        """What the path names when paths are compared: a ``null`` version names its key.
+        """The default move target, assuming bucket versioning is not enabled.
 
         A ``null`` version is taken to name the key itself: in a bucket without
         versioning, or with versioning suspended, a write to the key replaces
         its ``null`` version. With versioning enabled, a write adds a new
-        version instead, but the ``null`` version is still taken to name the
-        key. Any other path is its own target.
+        version instead, so callers with that state must keep the ``null``
+        version distinct from its key. Any other path is its own target.
         """
         return self.with_version_id(None) if self.version_id == "null" else self
 
diff --git a/pyathena/filesystem/s3_path_pairing.py b/pyathena/filesystem/s3_path_pairing.py
index 4a3fd832..fa8cfe37 100644
--- a/pyathena/filesystem/s3_path_pairing.py
+++ b/pyathena/filesystem/s3_path_pairing.py
@@ -28,7 +28,9 @@ class S3PathPairing:
     pairs them, except that a path with a version ID names that version of an
     object: it is not a glob pattern, and its destination is named after its
     key without the version. A move compares the paths by what they name
-    (see :attr:`~pyathena.filesystem.s3_path.S3Path.target`).
+    (see :attr:`~pyathena.filesystem.s3_path.S3Path.target`), keeping ``null``
+    versions distinct from their keys in the caller's versioning-enabled
+    buckets.
     :meth:`delete_paths` splits the paths of an ``rm()``.
 
     The pairing is a frozen dataclass of the paths as given, and holds no
diff --git a/tests/pyathena/filesystem/test_s3_path_pairing.py b/tests/pyathena/filesystem/test_s3_path_pairing.py
index f57fbacf..9212a0f3 100644
--- a/tests/pyathena/filesystem/test_s3_path_pairing.py
+++ b/tests/pyathena/filesystem/test_s3_path_pairing.py
@@ -141,6 +141,37 @@ def test_move_pairs(self):
         assert _pairing(pairs).conflict_candidates(pairs) == []
         assert _pairing(pairs).move_pairs(pairs) == [("bucket/d/a", "s3://bucket/z")]
 
+    @pytest.mark.parametrize(
+        ("enabled_buckets", "expected"),
+        [
+            ((), []),
+            (("enabled",), [("enabled/key?versionId=null", "enabled/key")]),
+            (("other",), [("other/key?versionId=null", "other/key")]),
+        ],
+    )
+    def test_move_pairs_null_version_bucket_state(self, enabled_buckets, expected):
+        pairs = [
+            ("enabled/key?versionId=null", "enabled/key"),
+            ("other/key?versionId=null", "other/key"),
+            ("enabled/unchanged?versionId=null", "s3://enabled/unchanged?versionID=null"),
+        ]
+        pairing = _pairing(pairs)
+
+        assert pairing.conflict_candidates(pairs, versioning_enabled_buckets=enabled_buckets) == []
+        assert pairing.move_pairs(pairs, versioning_enabled_buckets=enabled_buckets) == expected
+        assert pairing.move_pairs(pairs) == []
+
+    def test_move_pairs_distinguishes_null_version_source(self):
+        pairs = [
+            ("bucket/key?versionId=null", "bucket/out"),
+            ("bucket/other", "bucket/key"),
+        ]
+        pairing = _pairing(pairs)
+
+        with pytest.raises(ValueError, match="another path"):
+            pairing.move_pairs(pairs)
+        assert pairing.move_pairs(pairs, versioning_enabled_buckets={"bucket"}) == pairs
+
     @pytest.mark.parametrize(
         ("pairs", "match"),
         [

From 4a3edf616af79ebba843d43b2f779fe92788ef1f Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sun, 4 Oct 2026 23:01:40 +0900
Subject: [PATCH 4/5] Validate move bucket context and document its public
 contract

---
 docs/filesystem.md                            |  7 ++++++
 pyathena/filesystem/s3.py                     |  2 ++
 pyathena/filesystem/s3_async.py               |  5 +++++
 pyathena/filesystem/s3_path_pairing.py        | 22 ++++++++++++++++++-
 tests/pyathena/filesystem/test_s3.py          |  2 +-
 .../filesystem/test_s3_path_pairing.py        |  6 +++++
 6 files changed, 42 insertions(+), 2 deletions(-)

diff --git a/docs/filesystem.md b/docs/filesystem.md
index 126d93b0..a5cf5f99 100644
--- a/docs/filesystem.md
+++ b/docs/filesystem.md
@@ -404,6 +404,13 @@ paths, and pass the results to `copy_pairs()` and `move_pairs()`. A `sources`,
 `destination_is_dir` or `missing` that a rule needs and that is not passed raises
 `ValueError`.
 
+For moves that compare a null version with its unversioned key, the caller also
+looks up bucket versioning and passes the names of the versioning-enabled buckets
+as `versioning_enabled_buckets` to both `conflict_candidates()` and `move_pairs()`.
+Pass a collection of bucket names, such as a set, rather than a single string.
+The default empty collection treats null versions as their keys, as in unversioned
+or suspended buckets. The model makes no AWS requests.
+
 ```python
 from pyathena.filesystem.s3_path_pairing import S3PathPairing
 
diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index 731ca08c..519064c5 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -1439,6 +1439,7 @@ def mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs) -> None:
                 destination is another source, including one left in place,
                 which is checked before anything is copied. A directory with
                 no object at its key, which is not copied, does not conflict.
+            OSError: If bucket versioning cannot be read.
         """
         if path1 == path2:
             return
@@ -1470,6 +1471,7 @@ def _move_pairs(
 
         Raises:
             ValueError: If the move has conflicting paths.
+            OSError: If bucket versioning cannot be read.
         """
         pairing = S3PathPairing(path1, path2, recursive=recursive, maxdepth=maxdepth)
         pairs = self._copy_pairs(pairing)
diff --git a/pyathena/filesystem/s3_async.py b/pyathena/filesystem/s3_async.py
index 59256218..c1def16f 100644
--- a/pyathena/filesystem/s3_async.py
+++ b/pyathena/filesystem/s3_async.py
@@ -369,6 +369,7 @@ async def _mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs) -> N
                 destination is another source, including one left in place,
                 which is checked before anything is copied. A directory with
                 no object at its key, which is not copied, does not conflict.
+            OSError: If bucket versioning cannot be read.
         """
         if path1 == path2:
             return
@@ -400,6 +401,10 @@ def mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs) -> None:
             recursive: Whether to move the directories with their contents.
             maxdepth: Maximum depth of a recursive move.
             **kwargs: Additional S3 copy parameters.
+
+        Raises:
+            ValueError: If the move has conflicting paths.
+            OSError: If bucket versioning cannot be read.
         """
         sync(self.loop, self._mv, path1, path2, recursive=recursive, maxdepth=maxdepth, **kwargs)
 
diff --git a/pyathena/filesystem/s3_path_pairing.py b/pyathena/filesystem/s3_path_pairing.py
index fa8cfe37..5b5a80f9 100644
--- a/pyathena/filesystem/s3_path_pairing.py
+++ b/pyathena/filesystem/s3_path_pairing.py
@@ -166,6 +166,9 @@ def conflict_candidates(
             another source below them and a destination that conflicts, one
             per pair in the order of the pairs; empty if nothing needs to be
             looked up.
+
+        Raises:
+            TypeError: If ``versioning_enabled_buckets`` is a string.
         """
         return self._moves(pairs, versioning_enabled_buckets)[2]
 
@@ -199,6 +202,7 @@ def move_pairs(
                 except for a directory with no object at its key, which is not
                 copied. Also if ``missing`` is needed and None.
             TypeError: If ``missing`` is a string instead of a collection.
+                Also if ``versioning_enabled_buckets`` is a string.
         """
         if isinstance(missing, str):
             raise TypeError("missing is a collection of paths, not a path.")
@@ -264,7 +268,16 @@ def _is_glob(path: str) -> bool:
 
     @staticmethod
     def _target(path: S3Path, versioning_enabled_buckets: Collection[str]) -> str:
-        """Return a move target using the caller's bucket versioning state."""
+        """Return a move target using the caller's bucket versioning state.
+
+        Args:
+            path: The path of the source or destination.
+            versioning_enabled_buckets: Buckets whose versioning is enabled.
+
+        Returns:
+            The normalized path, retaining a ``null`` version only when its
+            bucket has versioning enabled.
+        """
         return str(path if path.bucket in versioning_enabled_buckets else path.target)
 
     @staticmethod
@@ -283,7 +296,14 @@ def _moves(
             of its source and destination; the targets of all sources,
             including those left in place; and the conflict candidates (see
             :meth:`conflict_candidates`).
+
+        Raises:
+            TypeError: If ``versioning_enabled_buckets`` is a string.
         """
+        if isinstance(versioning_enabled_buckets, str):
+            raise TypeError(
+                "versioning_enabled_buckets is a collection of bucket names, not a bucket name."
+            )
         named = []
         for p1, p2 in pairs:
             source_path = S3Path.parse(p1)
diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index 9397f933..4379074f 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -1648,7 +1648,7 @@ def test_mv_versions_onto_their_key(self):
         fs = self._make_fs()
         self._serve_keys(fs, {"b"})
 
-        # The "null" version is the object at the key, so it stays in place.
+        # In this unversioned bucket, the "null" version names the key and stays in place.
         fs.mv(["s3://bucket/b?versionId=null"], ["s3://bucket/b"])
         fs._call.assert_called_once_with(fs._client.get_bucket_versioning, Bucket="bucket")
 
diff --git a/tests/pyathena/filesystem/test_s3_path_pairing.py b/tests/pyathena/filesystem/test_s3_path_pairing.py
index 9212a0f3..4b549352 100644
--- a/tests/pyathena/filesystem/test_s3_path_pairing.py
+++ b/tests/pyathena/filesystem/test_s3_path_pairing.py
@@ -172,6 +172,12 @@ def test_move_pairs_distinguishes_null_version_source(self):
             pairing.move_pairs(pairs)
         assert pairing.move_pairs(pairs, versioning_enabled_buckets={"bucket"}) == pairs
 
+    @pytest.mark.parametrize("method", ["conflict_candidates", "move_pairs"])
+    def test_move_bucket_names_are_a_collection(self, method):
+        pairs = [("logs/key?versionId=null", "logs/key")]
+        with pytest.raises(TypeError, match="collection of bucket names"):
+            getattr(_pairing(pairs), method)(pairs, versioning_enabled_buckets="logs-archive")
+
     @pytest.mark.parametrize(
         ("pairs", "match"),
         [

From d4d7fe5bd3ccb414c8b161c14453a3a1a15e4718 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sun, 4 Oct 2026 23:37:09 +0900
Subject: [PATCH 5/5] Align S3 versioning regressions with filesystem test
 classes

---
 docs/testing.md                               | 10 ++-
 .../{test_s3_versioning.py => conftest.py}    | 65 ++------------
 tests/pyathena/filesystem/test_s3.py          | 43 +++++++++
 tests/pyathena/filesystem/test_s3_async.py    | 87 +++++++++++++++++++
 4 files changed, 147 insertions(+), 58 deletions(-)
 rename tests/pyathena/filesystem/{test_s3_versioning.py => conftest.py} (55%)

diff --git a/docs/testing.md b/docs/testing.md
index df3cc2f0..52b2b541 100644
--- a/docs/testing.md
+++ b/docs/testing.md
@@ -167,9 +167,17 @@ explicitly enabled, because provisioning requires additional permissions and a
 
 ```bash
 AWS_ATHENA_S3_VERSIONING_TESTS=1 uv run --env-file .env pytest -n 1 \
-  tests/pyathena/filesystem/test_s3_versioning.py -v
+  tests/pyathena/filesystem/test_s3.py \
+  tests/pyathena/filesystem/test_s3_async.py -k move_null_version_onto_key -v
 ```
 
+These nine cases belong to `TestS3FileSystem` and `TestAioS3FileSystem` and use
+their existing filesystem fixtures. The shared session-scoped bucket fixture in
+`tests/pyathena/filesystem/conftest.py` provisions the three buckets once per
+worker, so a serial run shares one setup and wait across both test modules.
+The synchronous API, asynchronous API, and asynchronous filesystem's synchronous
+wrapper each have a separate test method.
+
 The test identity needs `s3:CreateBucket`, `s3:GetBucketVersioning`,
 `s3:PutBucketVersioning`, `s3:PutObject`, `s3:GetObject`, `s3:GetObjectVersion`,
 `s3:ListBucketVersions`, `s3:DeleteObjectVersion`, and `s3:DeleteBucket` on the
diff --git a/tests/pyathena/filesystem/test_s3_versioning.py b/tests/pyathena/filesystem/conftest.py
similarity index 55%
rename from tests/pyathena/filesystem/test_s3_versioning.py
rename to tests/pyathena/filesystem/conftest.py
index 74195aec..4f665195 100644
--- a/tests/pyathena/filesystem/test_s3_versioning.py
+++ b/tests/pyathena/filesystem/conftest.py
@@ -5,7 +5,7 @@
 #
 # SPDX-License-Identifier: MIT
 
-"""Opt-in real S3 tests of null-version moves using temporary buckets."""
+"""Shared S3 filesystem integration-test fixtures."""
 
 import os
 import time
@@ -14,20 +14,16 @@
 import boto3
 import pytest
 
-from pyathena.filesystem.s3 import S3FileSystem
-from pyathena.filesystem.s3_async import AioS3FileSystem
 from tests import ENV
 
-pytestmark = pytest.mark.skipif(
-    os.getenv("AWS_ATHENA_S3_VERSIONING_TESTS") != "1",
-    reason="Set AWS_ATHENA_S3_VERSIONING_TESTS=1 to create temporary versioning test buckets.",
-)
+VERSIONING_TEST_KEYS = ("sync", "async", "async-wrapper")
 
-BACKENDS = ["sync", "async", "async-wrapper"]
 
-
-@pytest.fixture(scope="module")
+@pytest.fixture(scope="session")
 def versioning_buckets():
+    """Share temporary buckets across the sync and async opt-in move tests."""
+    if os.getenv("AWS_ATHENA_S3_VERSIONING_TESTS") != "1":
+        pytest.skip("Set AWS_ATHENA_S3_VERSIONING_TESTS=1 to create versioning test buckets.")
     client = boto3.client("s3", region_name=ENV.region_name)
     buckets = {}
     try:
@@ -40,7 +36,7 @@ def versioning_buckets():
             )
             client.create_bucket(Bucket=bucket, **params)
             buckets[status] = bucket
-            for key in BACKENDS:
+            for key in VERSIONING_TEST_KEYS:
                 client.put_object(Bucket=bucket, Key=key, Body=b"original")
             if status:
                 client.put_bucket_versioning(
@@ -54,7 +50,7 @@ def versioning_buckets():
             time.sleep(min(30, remaining))
         for status in ("Enabled", "Suspended"):
             bucket = buckets[status]
-            for key in BACKENDS:
+            for key in VERSIONING_TEST_KEYS:
                 response = client.put_object(Bucket=bucket, Key=key, Body=b"current")
                 assert response["VersionId"] != "null"
             if status == "Suspended":
@@ -84,48 +80,3 @@ def versioning_buckets():
                 errors.append(error)
         if errors:
             raise ExceptionGroup("Failed to remove versioning test buckets", errors)
-
-
-@pytest.fixture(params=BACKENDS)
-def fs(request):
-    backend = request.param
-    cls = S3FileSystem if backend == "sync" else AioS3FileSystem
-    return backend, cls(region_name=ENV.region_name, skip_instance_cache=True)
-
-
-class TestS3NullVersionMove:
-    @pytest.mark.parametrize("status", [None, "Enabled", "Suspended"])
-    @pytest.mark.asyncio
-    async def test_mv_null_version_onto_key(self, fs, versioning_buckets, status):
-        backend, filesystem = fs
-        client, buckets = versioning_buckets
-        bucket = buckets[status]
-        path = f"s3://{bucket}/{backend}"
-        before = [
-            v
-            for v in client.list_object_versions(Bucket=bucket, Prefix=backend)["Versions"]
-            if v["Key"] == backend
-        ]
-        assert any(v["VersionId"] == "null" for v in before)
-        if status:
-            assert not next(v for v in before if v["VersionId"] == "null")["IsLatest"]
-
-        if backend == "async":
-            await filesystem._mv(f"{path}?versionId=null", path)
-        else:
-            filesystem.mv(f"{path}?versionId=null", path)
-
-        with client.get_object(Bucket=bucket, Key=backend)["Body"] as body:
-            assert body.read() == (b"original" if status != "Suspended" else b"current")
-        after = [
-            v
-            for v in client.list_object_versions(Bucket=bucket, Prefix=backend)["Versions"]
-            if v["Key"] == backend
-        ]
-        if status == "Enabled":
-            assert not any(v["VersionId"] == "null" for v in after)
-            assert len(after) == len(before)
-            latest = next(v for v in after if v["IsLatest"])
-            assert latest["VersionId"] not in {v["VersionId"] for v in before}
-        else:
-            assert after == before
diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index 4379074f..c68e5ae4 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -5395,6 +5395,49 @@ def test_upload_cp_file(self, fs, base, exp):
             assert fs.cat(rpath_copy) == tmp.read()
             assert fs.cat(rpath_copy) == fs.cat(rpath)
 
+    @pytest.mark.skipif(
+        os.getenv("AWS_ATHENA_S3_VERSIONING_TESTS") != "1",
+        reason="Set AWS_ATHENA_S3_VERSIONING_TESTS=1 to create versioning test buckets.",
+    )
+    @pytest.mark.parametrize(
+        "status",
+        [
+            pytest.param(None, id="sync-None"),
+            pytest.param("Enabled", id="sync-Enabled"),
+            pytest.param("Suspended", id="sync-Suspended"),
+        ],
+    )
+    def test_move_null_version_onto_key(self, fs, versioning_buckets, status):
+        client, buckets = versioning_buckets
+        bucket = buckets[status]
+        key = "sync"
+        path = f"s3://{bucket}/{key}"
+        before = [
+            v
+            for v in client.list_object_versions(Bucket=bucket, Prefix=key)["Versions"]
+            if v["Key"] == key
+        ]
+        assert any(v["VersionId"] == "null" for v in before)
+        if status:
+            assert not next(v for v in before if v["VersionId"] == "null")["IsLatest"]
+
+        fs.mv(f"{path}?versionId=null", path)
+
+        with client.get_object(Bucket=bucket, Key=key)["Body"] as body:
+            assert body.read() == (b"original" if status != "Suspended" else b"current")
+        after = [
+            v
+            for v in client.list_object_versions(Bucket=bucket, Prefix=key)["Versions"]
+            if v["Key"] == key
+        ]
+        if status == "Enabled":
+            assert not any(v["VersionId"] == "null" for v in after)
+            assert len(after) == len(before)
+            latest = next(v for v in after if v["IsLatest"])
+            assert latest["VersionId"] not in {v["VersionId"] for v in before}
+        else:
+            assert after == before
+
     def test_move(self, fs):
         path1 = (
             f"s3://{ENV.s3_staging_bucket}/{ENV.s3_staging_key}{ENV.schema}/"
diff --git a/tests/pyathena/filesystem/test_s3_async.py b/tests/pyathena/filesystem/test_s3_async.py
index f12022d6..9cf6b7c3 100644
--- a/tests/pyathena/filesystem/test_s3_async.py
+++ b/tests/pyathena/filesystem/test_s3_async.py
@@ -2112,6 +2112,93 @@ async def test_upload_cp_file(self, fs, base, exp):
             assert await fs._cat_file(rpath_copy) == tmp.read()
             assert await fs._cat_file(rpath_copy) == await fs._cat_file(rpath)
 
+    @pytest.mark.skipif(
+        os.getenv("AWS_ATHENA_S3_VERSIONING_TESTS") != "1",
+        reason="Set AWS_ATHENA_S3_VERSIONING_TESTS=1 to create versioning test buckets.",
+    )
+    @pytest.mark.parametrize(
+        "status",
+        [
+            pytest.param(None, id="async-None"),
+            pytest.param("Enabled", id="async-Enabled"),
+            pytest.param("Suspended", id="async-Suspended"),
+        ],
+    )
+    @pytest.mark.asyncio
+    async def test_move_null_version_onto_key(self, fs, versioning_buckets, status):
+        client, buckets = versioning_buckets
+        bucket = buckets[status]
+        key = "async"
+        path = f"s3://{bucket}/{key}"
+        before = [
+            v
+            for v in client.list_object_versions(Bucket=bucket, Prefix=key)["Versions"]
+            if v["Key"] == key
+        ]
+        assert any(v["VersionId"] == "null" for v in before)
+        if status:
+            assert not next(v for v in before if v["VersionId"] == "null")["IsLatest"]
+
+        await fs._mv(f"{path}?versionId=null", path)
+
+        with client.get_object(Bucket=bucket, Key=key)["Body"] as body:
+            assert body.read() == (b"original" if status != "Suspended" else b"current")
+        after = [
+            v
+            for v in client.list_object_versions(Bucket=bucket, Prefix=key)["Versions"]
+            if v["Key"] == key
+        ]
+        if status == "Enabled":
+            assert not any(v["VersionId"] == "null" for v in after)
+            assert len(after) == len(before)
+            latest = next(v for v in after if v["IsLatest"])
+            assert latest["VersionId"] not in {v["VersionId"] for v in before}
+        else:
+            assert after == before
+
+    @pytest.mark.skipif(
+        os.getenv("AWS_ATHENA_S3_VERSIONING_TESTS") != "1",
+        reason="Set AWS_ATHENA_S3_VERSIONING_TESTS=1 to create versioning test buckets.",
+    )
+    @pytest.mark.parametrize(
+        "status",
+        [
+            pytest.param(None, id="async-wrapper-None"),
+            pytest.param("Enabled", id="async-wrapper-Enabled"),
+            pytest.param("Suspended", id="async-wrapper-Suspended"),
+        ],
+    )
+    def test_move_null_version_onto_key_sync_wrapper(self, fs, versioning_buckets, status):
+        client, buckets = versioning_buckets
+        bucket = buckets[status]
+        key = "async-wrapper"
+        path = f"s3://{bucket}/{key}"
+        before = [
+            v
+            for v in client.list_object_versions(Bucket=bucket, Prefix=key)["Versions"]
+            if v["Key"] == key
+        ]
+        assert any(v["VersionId"] == "null" for v in before)
+        if status:
+            assert not next(v for v in before if v["VersionId"] == "null")["IsLatest"]
+
+        fs.mv(f"{path}?versionId=null", path)
+
+        with client.get_object(Bucket=bucket, Key=key)["Body"] as body:
+            assert body.read() == (b"original" if status != "Suspended" else b"current")
+        after = [
+            v
+            for v in client.list_object_versions(Bucket=bucket, Prefix=key)["Versions"]
+            if v["Key"] == key
+        ]
+        if status == "Enabled":
+            assert not any(v["VersionId"] == "null" for v in after)
+            assert len(after) == len(before)
+            latest = next(v for v in after if v["IsLatest"])
+            assert latest["VersionId"] not in {v["VersionId"] for v in before}
+        else:
+            assert after == before
+
     @pytest.mark.asyncio
     async def test_move(self, fs):
         path1 = (