cleanup(storage): remove unused code in emulator (#7048)

Since we are no longer sending gRPC requests for Bucket metadata
operations, we do not need to emulate them either. This is needed
because the v2/ protos do not expose these RPCs (yet).

google/cloud/storage/emulator/database.py

@@ -100,12 +100,9 @@ def insert_test_bucket(self, context):
             "GOOGLE_CLOUD_CPP_STORAGE_TEST_BUCKET_NAME", "bucket"
         )
         if self.buckets.get(bucket_name) is None:
-            if context is not None:
-                request = storage_pb2.InsertBucketRequest(bucket={"name": bucket_name})
-            else:
-                request = utils.common.FakeRequest(
-                    args={}, data=json.dumps({"name": bucket_name})
-                )
+            request = utils.common.FakeRequest(
+                args={}, data=json.dumps({"name": bucket_name})
+            )
             bucket_test, _ = gcs.bucket.Bucket.init(request, context)
             self.insert_bucket(request, bucket_test, context)
             bucket_test.metadata.metageneration = 4

google/cloud/storage/emulator/emulator.py

@@ -363,32 +363,21 @@ def bucket_default_object_acl_delete(bucket_name, entity):
 @retry_test(method="storage.notifications.list")
 def bucket_notification_list(bucket_name):
     bucket = db.get_bucket(flask.request, bucket_name, None)
-    response = {"kind": "storage#notifications", "items": []}
-    for notification in bucket.notifications.values():
-        response["items"].append(
-            json_format.MessageToDict(notification, preserving_proto_field_name=True)
-        )
-    return response
+    return bucket.list_notifications(None)
 
 
 @gcs.route("/b/<bucket_name>/notificationConfigs", methods=["POST"])
 @retry_test(method="storage.notifications.insert")
 def bucket_notification_insert(bucket_name):
     bucket = db.get_bucket(flask.request, bucket_name, None)
-    notification = bucket.insert_notification(flask.request, None)
-    response = json_format.MessageToDict(notification, preserving_proto_field_name=True)
-    response["kind"] = "storage#notification"
-    return response
+    return bucket.insert_notification(flask.request, None)
 
 
 @gcs.route("/b/<bucket_name>/notificationConfigs/<notification_id>")
 @retry_test(method="storage.notifications.get")
 def bucket_notification_get(bucket_name, notification_id):
     bucket = db.get_bucket(flask.request, bucket_name, None)
-    notification = bucket.get_notification(notification_id, None)
-    response = json_format.MessageToDict(notification, preserving_proto_field_name=True)
-    response["kind"] = "storage#notification"
-    return response
+    return bucket.get_notification(notification_id, None)
 
 
 @gcs.route("/b/<bucket_name>/notificationConfigs/<notification_id>", methods=["DELETE"])

google/cloud/storage/emulator/gcs/bucket.py

@@ -460,15 +460,24 @@ def delete_default_object_acl(self, entity, context):
     # === NOTIFICATIONS === #
 
     def insert_notification(self, request, context):
-        notification = None
-        if context is not None:
-            notification = request.notification
-        else:
-            notification = json_format.ParseDict(
-                json.loads(request.data), resources_pb2.Notification()
-            )
-        notification.id = "notification-%d" % random.getrandbits(16)
-        self.notifications[notification.id] = notification
+        notification = {
+            "kind": "storage#notification",
+            "id": "notification-%d" % random.getrandbits(16),
+        }
+        data = json.loads(request.data)
+        for required_key in {"topic", "payload_format"}:
+            value = data.pop(required_key, None)
+            if value is not None:
+                notification[required_key] = value
+            else:
+                utils.error.invalid(
+                    "Missing field in notification %s" % required_key, context
+                )
+        for key in {"event_types", "custom_attributes", "object_name_prefix"}:
+            value = data.pop(key, None)
+            if value is not None:
+                notification[key] = value
+        self.notifications[notification["id"]] = notification
         return notification
 
     def get_notification(self, notification_id, context):
@@ -477,6 +486,12 @@ def get_notification(self, notification_id, context):
     def delete_notification(self, notification_id, context):
         del self.notifications[notification_id]
 
+    def list_notifications(self, context):
+        response = {"kind": "storage#notifications", "items": []}
+        for notification in self.notifications.values():
+            response["items"].append(notification)
+        return response
+
     # === RESPONSE === #
 
     def rest(self):

google/cloud/storage/emulator/grpc_server.py

@@ -30,136 +30,6 @@
 
 class StorageServicer(storage_pb2_grpc.StorageServicer):
 
-    # === BUCKET ===#
-
-    def ListBuckets(self, request, context):
-        db.insert_test_bucket(context)
-        result = resources_pb2.ListBucketsResponse(next_page_token="", items=[])
-        for bucket in db.list_bucket(request, request.project, context):
-            result.items.append(bucket.metadata)
-        return result
-
-    def InsertBucket(self, request, context):
-        db.insert_test_bucket(context)
-        bucket, projection = gcs_type.bucket.Bucket.init(request, context)
-        db.insert_bucket(request, bucket, context)
-        return bucket.metadata
-
-    def GetBucket(self, request, context):
-        bucket_name = request.bucket
-        bucket = db.get_bucket(request, bucket_name, context)
-        return bucket.metadata
-
-    def UpdateBucket(self, request, context):
-        bucket_name = request.bucket
-        bucket = db.get_bucket(request, bucket_name, context)
-        bucket.update(request, context)
-        return bucket.metadata
-
-    def DeleteBucket(self, request, context):
-        bucket_name = request.bucket
-        db.delete_bucket(request, bucket_name, context)
-        return Empty()
-
-    def ListBucketAccessControls(self, request, context):
-        bucket_name = request.bucket
-        bucket = db.get_bucket(request, bucket_name, context)
-        result = resources_pb2.ListBucketAccessControlsResponse(
-            items=bucket.metadata.acl
-        )
-        return result
-
-    def InsertBucketAccessControl(self, request, context):
-        bucket_name = request.bucket
-        bucket = db.get_bucket(request, bucket_name, context)
-        return bucket.insert_acl(request, context)
-
-    def GetBucketAccessControl(self, request, context):
-        bucket_name = request.bucket
-        bucket = db.get_bucket(request, bucket_name, context)
-        return bucket.get_acl(request.entity, context)
-
-    def UpdateBucketAccessControl(self, request, context):
-        bucket_name = request.bucket
-        bucket = db.get_bucket(request, bucket_name, context)
-        return bucket.update_acl(request, request.entity, context)
-
-    def DeleteBucketAccessControl(self, request, context):
-        bucket_name = request.bucket
-        bucket = db.get_bucket(request, bucket_name, context)
-        bucket.delete_acl(request.entity, context)
-        return Empty()
-
-    def ListDefaultObjectAccessControls(self, request, context):
-        bucket_name = request.bucket
-        bucket = db.get_bucket(request, bucket_name, context)
-        result = resources_pb2.ListObjectAccessControlsResponse(
-            items=bucket.metadata.default_object_acl
-        )
-        return result
-
-    def InsertDefaultObjectAccessControl(self, request, context):
-        bucket_name = request.bucket
-        bucket = db.get_bucket(request, bucket_name, context)
-        return bucket.insert_default_object_acl(request, context)
-
-    def GetDefaultObjectAccessControl(self, request, context):
-        bucket_name = request.bucket
-        bucket = db.get_bucket(request, bucket_name, context)
-        return bucket.get_default_object_acl(request.entity, context)
-
-    def UpdateDefaultObjectAccessControl(self, request, context):
-        bucket_name = request.bucket
-        bucket = db.get_bucket(request, bucket_name, context)
-        return bucket.update_default_object_acl(request, request.entity, context)
-
-    def DeleteDefaultObjectAccessControl(self, request, context):
-        bucket_name = request.bucket
-        bucket = db.get_bucket(request, bucket_name, context)
-        bucket.delete_default_object_acl(request.entity, context)
-        return Empty()
-
-    def InsertNotification(self, request, context):
-        bucket_name = request.bucket
-        bucket = db.get_bucket(request, bucket_name, context)
-        return bucket.insert_notification(request, context)
-
-    def ListNotifications(self, request, context):
-        bucket_name = request.bucket
-        bucket = db.get_bucket(request, bucket_name, context)
-        result = resources_pb2.ListNotificationsResponse(
-            items=bucket.notifications.values()
-        )
-        return result
-
-    def GetNotification(self, request, context):
-        bucket_name = request.bucket
-        bucket = db.get_bucket(request, bucket_name, context)
-        notification_id = request.notification
-        return bucket.get_notification(notification_id, context)
-
-    def DeleteNotification(self, request, context):
-        bucket_name = request.bucket
-        bucket = db.get_bucket(request, bucket_name, context)
-        notification_id = request.notification
-        bucket.delete_notification(notification_id, context)
-        return Empty()
-
-    def GetBucketIamPolicy(self, request, context):
-        bucket_name = request.iam_request.resource
-        bucket = db.get_bucket(request, bucket_name, context)
-        return bucket.get_iam_policy(request, context)
-
-    def SetBucketIamPolicy(self, request, context):
-        bucket_name = request.iam_request.resource
-        bucket = db.get_bucket(request, bucket_name, context)
-        return bucket.set_iam_policy(request, context)
-
-    def TestBucketIamPermissions(self, request, context):
-        return iam_policy_pb2.TestIamPermissionsResponse(
-            permissions=request.iam_request.permissions
-        )
-
     # === OBJECT === #
 
     def handle_insert_object_streaming_rpc(self, request_iterator, context):