Separate versioned and non-versioned tests to use different bucket

Author: Mahalaxmibejugam

docs/source/developer.rst

@@ -18,9 +18,10 @@ real GCS. A small number of tests run differently or are skipped.
 
 If you want to actually test against real GCS, then you should set
 STORAGE_EMULATOR_HOST to "https://storage.googleapis.com" and also
-provide appropriate GCSFS_TEST_BUCKET and GCSFS_TEST_PROJECT, as well
-as setting your default google credentials (or providing them via the
-fsspec config).
+provide appropriate GCSFS_TEST_BUCKET, GCSFS_TEST_VERSIONED_BUCKET
+(To use for tests that target GCS object versioning, this bucket must have object versioning enabled)
+and GCSFS_TEST_PROJECT, as well as setting your default google
+credentials (or providing them via the fsspec config).
 
 .. _fake-gcs-server: https://github.com/fsouza/fake-gcs-server
 

gcsfs/tests/conftest.py

@@ -9,9 +9,10 @@
 import fsspec
 import pytest
 import requests
+from google.cloud import storage
 
 from gcsfs import GCSFileSystem
-from gcsfs.tests.settings import TEST_BUCKET
+from gcsfs.tests.settings import TEST_BUCKET, TEST_VERSIONED_BUCKET
 
 files = {
     "test/accounts.1.json": (
@@ -176,21 +177,68 @@ def extended_gcsfs(gcs_factory, populate=True):
 def gcs_versioned(gcs_factory):
     gcs = gcs_factory()
     gcs.version_aware = True
-    try:
-        try:
-            gcs.rm(gcs.find(TEST_BUCKET, versions=True))
-        except FileNotFoundError:
-            pass
-
-        try:
-            gcs.mkdir(TEST_BUCKET, enable_versioning=True)
-        except Exception:
-            pass
+    is_real_gcs = (
+        os.environ.get("STORAGE_EMULATOR_HOST") == "https://storage.googleapis.com"
+    )
+    try:  # ensure we're empty.
+        if is_real_gcs:
+            # For real GCS, we assume the bucket exists and only clean its contents.
+            try:
+                cleanup_versioned_bucket(gcs, TEST_VERSIONED_BUCKET)
+            except Exception as e:
+                logging.warning(
+                    f"Failed to empty versioned bucket {TEST_VERSIONED_BUCKET}: {e}"
+                )
+        else:
+            # For emulators, we delete and recreate the bucket for a clean state.
+            try:
+                gcs.rm(TEST_VERSIONED_BUCKET, recursive=True)
+            except FileNotFoundError:
+                pass
+            gcs.mkdir(TEST_VERSIONED_BUCKET, enable_versioning=True)
         gcs.invalidate_cache()
         yield gcs
     finally:
         try:
-            gcs.rm(gcs.find(TEST_BUCKET, versions=True))
-            gcs.rm(TEST_BUCKET)
+            if not is_real_gcs:
+                gcs.rm(gcs.find(TEST_VERSIONED_BUCKET, versions=True))
+                gcs.rm(TEST_VERSIONED_BUCKET)
         except:  # noqa: E722
             pass
+
+
+def cleanup_versioned_bucket(gcs, bucket_name, prefix=None):
+    """
+    Deletes all object versions in a bucket using the google-cloud-storage client,
+    ensuring it uses the same credentials as the gcsfs instance.
+    """
+    # Define a retry policy for API calls to handle rate limiting.
+    # This can retry on 429 Too Many Requests errors, which can happen
+    # when deleting many object versions quickly.
+    from google.api_core.retry import Retry
+
+    retry_policy = Retry(
+        initial=1.0,  # Initial delay in seconds
+        maximum=30.0,  # Maximum delay in seconds
+        multiplier=1.2,  # Backoff factor
+    )
+
+    client = storage.Client(
+        credentials=gcs.credentials.credentials, project=gcs.project
+    )
+
+    # List all blobs, including old versions
+    blobs_to_delete = list(client.list_blobs(bucket_name, versions=True, prefix=prefix))
+
+    if not blobs_to_delete:
+        logging.info("No object versions to delete in %s.", bucket_name)
+        return
+
+    logging.info(
+        "Deleting %d object versions from %s.", len(blobs_to_delete), bucket_name
+    )
+    time.sleep(2)
+    for blob in blobs_to_delete:
+        blob.delete(retry=retry_policy)
+
+    logging.info("Successfully deleted %d object versions.", len(blobs_to_delete))

gcsfs/tests/settings.py

@@ -1,8 +1,9 @@
 import os
 
 TEST_BUCKET = os.getenv("GCSFS_TEST_BUCKET", "gcsfs_test")
+TEST_VERSIONED_BUCKET = os.getenv("GCSFS_TEST_VERSIONED_BUCKET", "gcsfs_test_versioned")
 TEST_PROJECT = os.getenv("GCSFS_TEST_PROJECT", "project")
-TEST_REQUESTER_PAYS_BUCKET = "gcsfs_test_req_pay"
+TEST_REQUESTER_PAYS_BUCKET = f"{TEST_BUCKET}_req_pay"
 TEST_KMS_KEY = os.getenv(
     "GCSFS_TEST_KMS_KEY",
     f"projects/{TEST_PROJECT}/locations/us/keyRings/gcsfs_test/cryptKeys/gcsfs_test_key",

gcsfs/tests/test_core.py

@@ -1392,70 +1392,6 @@ def test_deep_find_wthdirs(gcs):
     ]
 
 
-def test_info_versioned(gcs_versioned):
-    with gcs_versioned.open(a, "wb") as wo:
-        wo.write(b"v1")
-    v1 = gcs_versioned.info(a)["generation"]
-    assert v1 is not None
-    with gcs_versioned.open(a, "wb") as wo:
-        wo.write(b"v2")
-    v2 = gcs_versioned.info(a)["generation"]
-    assert v2 is not None and v1 != v2
-    assert gcs_versioned.info(f"{a}#{v1}")["generation"] == v1
-    assert gcs_versioned.info(f"{a}?generation={v2}")["generation"] == v2
-
-
-def test_cat_versioned(gcs_versioned):
-    with gcs_versioned.open(a, "wb") as wo:
-        wo.write(b"v1")
-    v1 = gcs_versioned.info(a)["generation"]
-    assert v1 is not None
-    with gcs_versioned.open(a, "wb") as wo:
-        wo.write(b"v2")
-    gcs_versioned.cat(f"{a}#{v1}") == b"v1"
-
-
-def test_cp_versioned(gcs_versioned):
-    with gcs_versioned.open(a, "wb") as wo:
-        wo.write(b"v1")
-    v1 = gcs_versioned.info(a)["generation"]
-    assert v1 is not None
-    with gcs_versioned.open(a, "wb") as wo:
-        wo.write(b"v2")
-    gcs_versioned.cp_file(f"{a}#{v1}", b)
-    assert gcs_versioned.cat(b) == b"v1"
-
-
-def test_ls_versioned(gcs_versioned):
-    import posixpath
-
-    with gcs_versioned.open(a, "wb") as wo:
-        wo.write(b"v1")
-    v1 = gcs_versioned.info(a)["generation"]
-    with gcs_versioned.open(a, "wb") as wo:
-        wo.write(b"v2")
-    v2 = gcs_versioned.info(a)["generation"]
-    dpath = posixpath.dirname(a)
-    versions = {f"{a}#{v1}", f"{a}#{v2}"}
-    assert versions == set(gcs_versioned.ls(dpath, versions=True))
-    assert versions == {
-        entry["name"] for entry in gcs_versioned.ls(dpath, detail=True, versions=True)
-    }
-    assert gcs_versioned.ls(TEST_BUCKET, versions=True) == ["gcsfs_test/tmp"]
-
-
-def test_find_versioned(gcs_versioned):
-    with gcs_versioned.open(a, "wb") as wo:
-        wo.write(b"v1")
-    v1 = gcs_versioned.info(a)["generation"]
-    with gcs_versioned.open(a, "wb") as wo:
-        wo.write(b"v2")
-    v2 = gcs_versioned.info(a)["generation"]
-    versions = {f"{a}#{v1}", f"{a}#{v2}"}
-    assert versions == set(gcs_versioned.find(a, versions=True))
-    assert versions == set(gcs_versioned.find(a, detail=True, versions=True))
-
-
 def test_cp_directory_recursive(gcs):
     src = TEST_BUCKET + "/src"
     src_file = src + "/file"

gcsfs/tests/test_core_versioned.py

@@ -0,0 +1,107 @@
+import os
+import posixpath
+
+import pytest
+from google.cloud import storage
+
+from gcsfs import GCSFileSystem
+from gcsfs.tests.settings import TEST_VERSIONED_BUCKET
+
+a = TEST_VERSIONED_BUCKET + "/tmp/test/a"
+b = TEST_VERSIONED_BUCKET + "/tmp/test/b"
+
+
+def is_versioning_enabled():
+    """
+    Helper function to check if the test bucket has versioning enabled.
+    Returns a tuple of (bool, reason_string).
+    """
+    # Don't skip when using an emulator, as we create the versioned bucket ourselves.
+    if os.environ.get("STORAGE_EMULATOR_HOST") != "https://storage.googleapis.com":
+        return True, ""
+    try:
+        gcs = GCSFileSystem(project=os.getenv("GCSFS_TEST_PROJECT", "project"))
+        client = storage.Client(
+            credentials=gcs.credentials.credentials, project=gcs.project
+        )
+        bucket = client.get_bucket(TEST_VERSIONED_BUCKET)
+        if bucket.versioning_enabled:
+            return True, ""
+        return (
+            False,
+            f"Bucket '{TEST_VERSIONED_BUCKET}' does not have versioning enabled.",
+        )
+    except Exception as e:
+        return (
+            False,
+            f"Could not verify versioning status for bucket '{TEST_VERSIONED_BUCKET}': {e}",
+        )
+
+
+pytestmark = pytest.mark.skipif(
+    not is_versioning_enabled()[0], reason=is_versioning_enabled()[1]
+)
+
+
+def test_info_versioned(gcs_versioned):
+    with gcs_versioned.open(a, "wb") as wo:
+        wo.write(b"v1")
+    v1 = gcs_versioned.info(a)["generation"]
+    assert v1 is not None
+    with gcs_versioned.open(a, "wb") as wo:
+        wo.write(b"v2")
+    v2 = gcs_versioned.info(a)["generation"]
+    assert v2 is not None and v1 != v2
+    assert gcs_versioned.info(f"{a}#{v1}")["generation"] == v1
+    assert gcs_versioned.info(f"{a}?generation={v2}")["generation"] == v2
+
+
+def test_cat_versioned(gcs_versioned):
+    with gcs_versioned.open(b, "wb") as wo:
+        wo.write(b"v1")
+    v1 = gcs_versioned.info(b)["generation"]
+    assert v1 is not None
+    with gcs_versioned.open(b, "wb") as wo:
+        wo.write(b"v2")
+    assert gcs_versioned.cat(f"{b}#{v1}") == b"v1"
+
+
+def test_cp_versioned(gcs_versioned):
+    with gcs_versioned.open(a, "wb") as wo:
+        wo.write(b"v1")
+    v1 = gcs_versioned.info(a)["generation"]
+    assert v1 is not None
+    with gcs_versioned.open(a, "wb") as wo:
+        wo.write(b"v2")
+    gcs_versioned.cp_file(f"{a}#{v1}", b)
+    assert gcs_versioned.cat(b) == b"v1"
+
+
+def test_ls_versioned(gcs_versioned):
+    with gcs_versioned.open(b, "wb") as wo:
+        wo.write(b"v1")
+    v1 = gcs_versioned.info(b)["generation"]
+    with gcs_versioned.open(b, "wb") as wo:
+        wo.write(b"v2")
+    v2 = gcs_versioned.info(b)["generation"]
+    dpath = posixpath.dirname(b)
+    versions = {f"{b}#{v1}", f"{b}#{v2}"}
+    assert versions == set(gcs_versioned.ls(dpath, versions=True))
+    assert versions == {
+        entry["name"] for entry in gcs_versioned.ls(dpath, detail=True, versions=True)
+    }
+    assert gcs_versioned.ls(TEST_VERSIONED_BUCKET, versions=True) == [
+        f"{TEST_VERSIONED_BUCKET}/tmp"
+    ]
+
+
+def test_find_versioned(gcs_versioned):
+    with gcs_versioned.open(a, "wb") as wo:
+        wo.write(b"v1")
+    v1 = gcs_versioned.info(a)["generation"]
+    with gcs_versioned.open(a, "wb") as wo:
+        wo.write(b"v2")
+    v2 = gcs_versioned.info(a)["generation"]
+    versions = {f"{a}#{v1}", f"{a}#{v2}"}
+    assert versions == set(gcs_versioned.find(a, versions=True))
+    assert versions == set(gcs_versioned.find(a, detail=True, versions=True))