fix: raise a ValueError in BucketNotification.create() if a topic nam…

…e 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

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