Add 'BucketNotification.delete' API wrapper.

storage/google/cloud/storage/notification.py

@@ -141,6 +141,12 @@ def client(self):
         """The client bound to this notfication."""
         return self.bucket.client
 
+    @property
+    def path(self):
+        """The URL path for this notification."""
+        return '/b/{}/notificationConfigs/{}'.format(
+            self.bucket.name, self.notification_id)
+
     def _require_client(self, client):
         """Check client or verify over-ride.
 
@@ -180,3 +186,23 @@ def create(self, client=None):
             path=path,
             data=properties,
         )
+
+    def delete(self, client=None):
+        """Delete this notification.
+
+        See:
+        https://cloud.google.com/storage/docs/json_api/v1/notifications/delete
+
+        :type client: :class:`~google.cloud.storage.client.Client` or
+                      ``NoneType``
+        :param client: Optional. The client to use.  If not passed, falls back
+                       to the ``client`` stored on the current bucket.
+
+        :raises: :class:`google.api.core.exceptions.NotFound`:
+            if the notification does not exist.
+        """
+        if self.notification_id is None:
+            raise ValueError("Notification not intialized by server")
+
+        client = self._require_client(client)
+        client._connection.api_request(method='DELETE', path=self.path)

storage/tests/unit/test_notification.py

@@ -243,3 +243,52 @@ def test_create_w_explicit_client(self):
             path=path,
             data=data,
         )
+
+    def test_delete_wo_notification_id(self):
+        client = self._make_client()
+        bucket = self._make_bucket(client)
+        notification = self._make_one(
+            bucket, self.TOPIC_NAME)
+
+        with self.assertRaises(ValueError):
+            notification.delete()
+
+    def test_delete_miss(self):
+        from google.cloud.exceptions import NotFound
+
+        NOTIFICATION_ID = '123'
+        client = self._make_client()
+        bucket = self._make_bucket(client)
+        notification = self._make_one(bucket, self.TOPIC_NAME)
+        notification._properties['id'] = NOTIFICATION_ID
+        api_request = client._connection.api_request
+        api_request.side_effect = NotFound('testing')
+
+        with self.assertRaises(NotFound):
+            notification.delete()
+
+        path = '/b/{}/notificationConfigs/{}'.format(
+            self.BUCKET_NAME, NOTIFICATION_ID)
+        api_request.assert_called_once_with(
+            method='DELETE',
+            path=path,
+        )
+
+    def test_delete_hit(self):
+        NOTIFICATION_ID = '123'
+        client = self._make_client()
+        bucket = self._make_bucket(client)
+        alt_client = self._make_client()
+        notification = self._make_one(bucket, self.TOPIC_NAME)
+        notification._properties['id'] = NOTIFICATION_ID
+        api_request = client._connection.api_request
+        api_request.return_value = None
+
+        notification.delete(client=client)
+
+        path = '/b/{}/notificationConfigs/{}'.format(
+            self.BUCKET_NAME, NOTIFICATION_ID)
+        api_request.assert_called_once_with(
+            method='DELETE',
+            path=path,
+        )