fix: raise a ValueError in BucketNotification.create() if a topic name is not set (#617)

* fix: topic name is required to create a notification. add check and tests * revise conf tests * fix topic formatting and remove extra validation * blacken lint * update docstrings

Author: cojenco

google/cloud/storage/notification.py

@@ -253,6 +253,8 @@ def create(self, client=None, timeout=_DEFAULT_TIMEOUT, retry=None):
  :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
  :param retry:
  (Optional) How to retry the RPC. See: :ref:`configuring_retries`
+
+ :raises ValueError: if the notification already exists.
  """
  if self.notification_id is not None:
  raise ValueError(
@@ -267,7 +269,14 @@ def create(self, client=None, timeout=_DEFAULT_TIMEOUT, retry=None):
 
  path = "/b/{}/notificationConfigs".format(self.bucket.name)
  properties = self._properties.copy()
- properties["topic"] = _TOPIC_REF_FMT.format(self.topic_project, self.topic_name)
+
+ if self.topic_name is None:
+ properties["topic"] = _TOPIC_REF_FMT.format(self.topic_project, "")
+ else:
+ properties["topic"] = _TOPIC_REF_FMT.format(
+ self.topic_project, self.topic_name
+ )
+
  self._properties = client._post_resource(
  path, properties, query_params=query_params, timeout=timeout, retry=retry,
  )

tests/conformance/test_conformance.py

@@ -52,6 +52,7 @@
 _CONF_TEST_SERVICE_ACCOUNT_EMAIL = (
  "my-service-account@my-project-id.iam.gserviceaccount.com"
 )
+_CONF_TEST_PUBSUB_TOPIC_NAME = "my-topic-name"
 
 _STRING_CONTENT = "hello world"
 _BYTE_CONTENT = b"12345678"
@@ -190,7 +191,7 @@ def client_get_service_account_email(client, _preconditions, **_):
 
 def notification_create(client, _preconditions, **resources):
  bucket = client.bucket(resources.get("bucket").name)
- notification = bucket.notification()
+ notification = bucket.notification(topic_name=_CONF_TEST_PUBSUB_TOPIC_NAME)
  notification.create()
 
 
@@ -761,7 +762,9 @@ def object(client, bucket):
 
 @pytest.fixture
 def notification(client, bucket):
- notification = client.bucket(bucket.name).notification()
+ notification = client.bucket(bucket.name).notification(
+ topic_name=_CONF_TEST_PUBSUB_TOPIC_NAME
+ )
  notification.create()
  notification.reload()
  yield notification

tests/system/test_notification.py

@@ -149,6 +149,34 @@ def test_notification_create_w_user_project(
  notification.delete()
 
 
+def test_notification_create_wo_topic_name(
+ storage_client,
+ buckets_to_delete,
+ topic_name,
+ notification_topic,
+ event_types,
+ payload_format,
+):
+ from google.cloud.exceptions import BadRequest
+
+ bucket_name = _helpers.unique_name("notification-wo-name")
+ bucket = _helpers.retry_429_503(storage_client.create_bucket)(bucket_name)
+ buckets_to_delete.append(bucket)
+
+ assert list(bucket.list_notifications()) == []
+
+ notification = bucket.notification(
+ topic_name=None,
+ custom_attributes=custom_attributes,
+ event_types=event_types,
+ blob_name_prefix=blob_name_prefix,
+ payload_format=payload_format,
+ )
+
+ with pytest.raises(BadRequest):
+ notification.create()
+
+
 def test_bucket_get_notification(
  storage_client,
  buckets_to_delete,

tests/unit/test_notification.py

@@ -242,6 +242,35 @@ def test_create_w_existing_notification_id(self):
 
  client._post_resource.assert_not_called()
 
+ def test_create_wo_topic_name(self):
+ from google.cloud.exceptions import BadRequest
+ from google.cloud.storage.notification import NONE_PAYLOAD_FORMAT
+
+ client = mock.Mock(spec=["_post_resource", "project"])
+ client.project = self.BUCKET_PROJECT
+ client._post_resource.side_effect = BadRequest(
+ "Invalid Google Cloud Pub/Sub topic."
+ )
+ bucket = self._make_bucket(client)
+ notification = self._make_one(bucket, None)
+
+ with self.assertRaises(BadRequest):
+ notification.create()
+
+ expected_topic = self.TOPIC_REF_FMT.format(self.BUCKET_PROJECT, "")
+ expected_data = {
+ "topic": expected_topic,
+ "payload_format": NONE_PAYLOAD_FORMAT,
+ }
+ expected_query_params = {}
+ client._post_resource.assert_called_once_with(
+ self.CREATE_PATH,
+ expected_data,
+ query_params=expected_query_params,
+ timeout=self._get_default_timeout(),
+ retry=None,
+ )
+
  def test_create_w_defaults(self):
  from google.cloud.storage.notification import NONE_PAYLOAD_FORMAT