Add **kwargs parameters to _mkdir, _cat_file, adhere to fsspec API

gcsfs/core.py

@@ -605,17 +605,23 @@ def invalidate_cache(self, path=None):
 
     async def _mkdir(
         self,
-        bucket,
+        path,
         acl="projectPrivate",
         default_acl="bucketOwnerFullControl",
         location=None,
+        create_parents=True,
+        **kwargs,
     ):
         """
         New bucket
 
+        If path is more than just a bucket, will create bucket if create_parents=True;
+        otherwise is a noop. If create_parents is False and bucket does not exist,
+        will produce FileNotFFoundError.
+
         Parameters
         ----------
-        bucket: str
+        path: str
             bucket name. If contains '/' (i.e., looks like subdir), will
             have no effect because GCS doesn't have real directories.
         acl: string, one of bACLs
@@ -627,11 +633,18 @@ async def _mkdir(
             If not provided, defaults to `self.default_location`.
             You can find a list of all available locations here:
             https://cloud.google.com/storage/docs/locations#available-locations
+        create_parents: bool
+            If True, creates the bucket in question, if it doesn't already exist
         """
+        bucket, object = self.split_path(path)
         if bucket in ["", "/"]:
             raise ValueError("Cannot create root bucket")
-        if "/" in bucket:
+        if "/" in path and create_parents and await self._exists(bucket):
+            # nothing to do
             return
+        if "/" in path and not create_parents and not await self._exists(bucket):
+            raise FileNotFoundError(bucket)
+
         json_data = {"name": bucket}
         location = location or self.default_location
         if location:
@@ -758,7 +771,7 @@ def url(self, path):
         object = quote_plus(object)
         return u.format(self._location, bucket, object)
 
-    async def _cat_file(self, path, start=None, end=None):
+    async def _cat_file(self, path, start=None, end=None, **kwargs):
         """Simple one-shot get of file data"""
         u2 = self.url(path)
         if start or end:

gcsfs/tests/test_core.py

@@ -1035,3 +1035,15 @@ def test_dir_marker(gcs):
     out3 = gcs.info(f"{TEST_BUCKET}/placeholder/")
     assert out2 == out3
     assert out2["type"] == "directory"
+
+
+def test_mkdir_with_path(gcs):
+    with pytest.raises(FileNotFoundError):
+        gcs.mkdir("new/path", create_parents=False)
+    assert not gcs.exists("new")
+    gcs.mkdir("new/path", create_parents=True)
+    assert gcs.exists("new")
+
+    # these lines do nothing, but should not fail
+    gcs.mkdir("new/path", create_parents=False)
+    gcs.mkdir("new/path", create_parents=True)