@@ -401,6 +401,9 @@ upload_url = core.generate_presigned_url(path, "put_object", ContentType="text/c
 `create_bucket()` and `delete_bucket()` create and delete a bucket when called. The
 `allow_bucket_creation` and `allow_bucket_deletion` options apply only to the
 filesystem's `mkdir`/`makedirs` and `rmdir`, not to calls through the core.
+`get_bucket_versioning()` returns the versioning state of a bucket, `Enabled` or
+`Suspended`, or `None` if versioning has never been enabled on it.
+`S3Path.is_directory_bucket` tells whether the bucket of a path is a directory bucket.
 
 `create_bucket()` sends a `LocationConstraint` for the `region_name` argument, or for
 the client's region by default, except in `us-east-1`. The fields of a
 
@@ -1438,15 +1438,10 @@ def _move_pairs(
         buckets = {
             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)
+            if p.version_id == "null" and p.name in unversioned and not p.is_directory_bucket
         }
         versioned_buckets = {
-            bucket
-            for bucket in buckets
-            if self._call(self._client.get_bucket_versioning, Bucket=bucket).get("Status")
-            == "Enabled"
+            bucket for bucket in buckets if self.core.get_bucket_versioning(bucket) == "Enabled"
         }
         missing = {
             source
 
@@ -835,6 +835,23 @@ def delete_bucket(self, bucket: str, **params) -> None:
         _logger.debug(f"Delete bucket: s3://{bucket}")
         self.call(self._client.delete_bucket, Bucket=bucket, **params)
 
+    def get_bucket_versioning(self, bucket: str, **params) -> str | None:
+        """Get the versioning state of a bucket with GetBucketVersioning.
+
+        Args:
+            bucket: The name of the bucket.
+            **params: Additional request parameters, sent as given.
+
+        Returns:
+            ``Enabled`` or ``Suspended``, or None if versioning has never
+            been enabled on the bucket.
+
+        Raises:
+            FileNotFoundError: If the bucket does not exist.
+        """
+        response = self.call(self._client.get_bucket_versioning, Bucket=bucket, **params)
+        return cast(str | None, response.get("Status"))
+
     def delete_object(self, path: S3Path, **params) -> None:
         """Delete an object, or a version of it, with DeleteObject.
 
@@ -1266,7 +1283,7 @@ def plan_multipart_copy(
             request.pop("Tagging", None)
             # Directory buckets do not support GetObjectTagging, and their
             # objects have no tags.
-            if not self._is_directory_bucket(source.bucket):
+            if not source.is_directory_bucket:
                 tags = self.get_object_tagging(
                     source, **self.operation_params("get_object_tagging", source_params)
                 )
@@ -1290,7 +1307,7 @@ def plan_multipart_copy(
             )
             if annotation_directive == "COPY"
             and "CopySourceSSECustomerAlgorithm" not in params
-            and not self._is_directory_bucket(source.bucket)
+            and not source.is_directory_bucket
             else ()
         )
         return S3MultipartCopyPlan(
@@ -1604,20 +1621,6 @@ def generate_presigned_url(
             ),
         )
 
-    @staticmethod
-    def _is_directory_bucket(bucket: str) -> bool:
-        """Return whether the bucket is a directory bucket (S3 Express One Zone).
-
-        Directory bucket names end with ``--x-s3``.
-
-        Args:
-            bucket: S3 bucket name.
-
-        Returns:
-            True if the bucket is a directory bucket.
-        """
-        return bucket.endswith("--x-s3")
-
     @staticmethod
     def _copy_source_params(params: Mapping[str, Any]) -> dict[str, Any]:
         """Map the parameters of a copy to those of the requests that read its source.
 
@@ -123,6 +123,14 @@ def is_bucket(self) -> bool:
         """Whether the path names the bucket: it has no key, or a key of only slashes."""
         return not self.key or not self.key.strip("/")
 
+    @property
+    def is_directory_bucket(self) -> bool:
+        """Whether the bucket is a directory bucket (S3 Express One Zone).
+
+        The names of directory buckets end with ``--x-s3``.
+        """
+        return self.bucket.endswith("--x-s3")
+
     @property
     def name(self) -> str:
         """The path without a scheme or version, in ``bucket/key`` form, or the bucket."""
 
@@ -1036,7 +1036,7 @@ async def copy_file(path1, path2, **kwargs):
     @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(
+        fs._sync_fs._call = fs._sync_fs._core.call = mock.MagicMock(
             return_value={} if status is None else {"Status": status}
         )
         fs._copy_file = mock.AsyncMock(return_value=True)
@@ -1063,7 +1063,7 @@ async def test_mv_null_version_onto_key(self, status):
     @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(
+        fs._sync_fs._call = fs._sync_fs._core.call = mock.MagicMock(
             return_value={} if status is None else {"Status": status}
         )
         fs._copy_file = mock.AsyncMock(return_value=True)
@@ -1090,7 +1090,7 @@ async def test_mv_null_version_conflicts_depend_on_bucket_state(self, status):
     @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._sync_fs._call = fs._sync_fs._core.call = mock.MagicMock()
         fs._copy_file = mock.AsyncMock(return_value=True)
         fs._delete_objects = mock.AsyncMock()
         key = "s3://example--usw2-az1--x-s3/key"
@@ -1105,7 +1105,9 @@ async def test_mv_null_version_directory_bucket_does_not_read_bucket_state(self)
     @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._sync_fs._call = fs._sync_fs._core.call = mock.MagicMock(
+            return_value={"Status": "Enabled"}
+        )
         fs._copy_file = mock.AsyncMock(return_value=True)
         fs._delete_objects = mock.AsyncMock()
         if stage == "lookup":
 
@@ -425,6 +425,36 @@ def test_delete_bucket(self):
                 core.delete_bucket("bucket")
         stubber.assert_no_pending_responses()
 
+    @pytest.mark.parametrize(
+        ("response", "expected"),
+        [
+            ({"Status": "Enabled"}, "Enabled"),
+            ({"Status": "Suspended", "MFADelete": "Disabled"}, "Suspended"),
+            # A bucket whose versioning has never been enabled.
+            ({}, None),
+        ],
+    )
+    def test_get_bucket_versioning(self, response, expected):
+        core, stubber = _make_core(request_kwargs={"RequestPayer": "requester"})
+        # RequestPayer, which GetBucketVersioning does not accept, is not sent.
+        stubber.add_response(
+            "get_bucket_versioning",
+            response,
+            {"Bucket": "bucket", "ExpectedBucketOwner": "123456789012"},
+        )
+        with stubber:
+            status = core.get_bucket_versioning("bucket", ExpectedBucketOwner="123456789012")
+        stubber.assert_no_pending_responses()
+        assert status == expected
+
+    def test_get_bucket_versioning_translates_errors(self):
+        core, stubber = _make_core()
+        stubber.add_client_error(
+            "get_bucket_versioning", service_error_code="NoSuchBucket", http_status_code=404
+        )
+        with stubber, pytest.raises(FileNotFoundError):
+            core.get_bucket_versioning("bucket")
+
     def test_list_objects(self):
         core, stubber = _make_core()
         stubber.add_response(
 
@@ -65,6 +65,18 @@ def test_parse_invalid(self, path):
     def test_is_bucket(self, path, expected):
         assert path.is_bucket is expected
 
+    @pytest.mark.parametrize(
+        ("path", "expected"),
+        [
+            (S3Path("bucket--usw2-az1--x-s3", "key"), True),
+            (S3Path("bucket--usw2-az1--x-s3"), True),
+            (S3Path("bucket", "key--x-s3"), False),
+            (S3Path("bucket--x-s3-other"), False),
+        ],
+    )
+    def test_is_directory_bucket(self, path, expected):
+        assert path.is_directory_bucket is expected
+
     @pytest.mark.parametrize(
         ("path", "name", "string", "uri"),
         [