add tests

Author: Yonghui-Lee

gcsfs/tests/test_credentials.py

@@ -45,6 +45,43 @@ def test_connect_google_default_uses_request():
         assert isinstance(kwargs["request"], Request)
 
 
+def test_connect_cloud_success():
+    with patch("google.auth.compute_engine.Credentials") as mock_creds_class:
+        mock_creds = Mock()
+        mock_creds_class.return_value = mock_creds
+
+        cred = GoogleCredentials(
+            project="my-project", access="read_only", token="cloud", on_google=True
+        )
+
+        assert cred.credentials == mock_creds
+        assert mock_creds.refresh.called
+        assert cred.method == "cloud"
+
+
+def test_connect_cloud_failure():
+    import google.auth.exceptions
+
+    with patch("google.auth.compute_engine.Credentials") as mock_creds_class:
+        mock_creds = Mock()
+        mock_creds_class.return_value = mock_creds
+        mock_creds.refresh.side_effect = google.auth.exceptions.RefreshError(
+            "mock error"
+        )
+
+        with pytest.raises(ValueError, match="Invalid gcloud credentials"):
+            GoogleCredentials(
+                project="my-project", access="read_only", token="cloud", on_google=True
+            )
+
+
+def test_connect_cloud_not_on_google():
+    with pytest.raises(ValueError):
+        GoogleCredentials(
+            project="my-project", access="read_only", token="cloud", on_google=False
+        )
+
+
 @pytest.mark.parametrize("token", ["", "incorrect.token", "x" * 100])
 def test_credentials_from_raw_token(token):
     with patch.dict(os.environ, {"FETCH_RAW_TOKEN_EXPIRY": "false"}):

gcsfs/tests/test_extended_gcsfs.py

@@ -1615,3 +1615,55 @@ async def test_cat_file_non_zonal_fallback(extended_gcsfs):
         mock_super_cat.assert_awaited_once_with(
             "standard_bucket/obj", start=10, end=20, concurrency=2, custom_arg="val"
         )
+
+
+@pytest.mark.asyncio
+async def test_lookup_bucket_type_not_cached_unknown(extended_gcsfs):
+    """Test that BucketType.UNKNOWN is not cached when _cache_unknown_buckets is False."""
+    fs = extended_gcsfs
+    fs._cache_unknown_buckets = False
+
+    # Clear cache just in case
+    fs._storage_layout_cache.clear()
+
+    # Mock _get_bucket_type to return UNKNOWN
+    with mock.patch.object(
+        fs, "_get_bucket_type", new_callable=mock.AsyncMock
+    ) as mock_get_type:
+        mock_get_type.return_value = BucketType.UNKNOWN
+
+        # First lookup
+        type1 = await fs._lookup_bucket_type("my-bucket")
+        assert type1 == BucketType.UNKNOWN
+        assert mock_get_type.call_count == 1
+
+        # Second lookup should call _get_bucket_type again because it's not cached
+        type2 = await fs._lookup_bucket_type("my-bucket")
+        assert type2 == BucketType.UNKNOWN
+        assert mock_get_type.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_lookup_bucket_type_cached_unknown(extended_gcsfs):
+    """Test that BucketType.UNKNOWN is cached when _cache_unknown_buckets is True."""
+    fs = extended_gcsfs
+    fs._cache_unknown_buckets = True
+
+    # Clear cache just in case
+    fs._storage_layout_cache.clear()
+
+    # Mock _get_bucket_type to return UNKNOWN
+    with mock.patch.object(
+        fs, "_get_bucket_type", new_callable=mock.AsyncMock
+    ) as mock_get_type:
+        mock_get_type.return_value = BucketType.UNKNOWN
+
+        # First lookup
+        type1 = await fs._lookup_bucket_type("my-bucket")
+        assert type1 == BucketType.UNKNOWN
+        assert mock_get_type.call_count == 1
+
+        # Second lookup should NOT call _get_bucket_type again because it's cached
+        type2 = await fs._lookup_bucket_type("my-bucket")
+        assert type2 == BucketType.UNKNOWN
+        assert mock_get_type.call_count == 1