diff --git a/storage/google/cloud/storage/notification.py b/storage/google/cloud/storage/notification.py index a4c5a66a4d5a..7ac6c9aaf1af 100644 --- a/storage/google/cloud/storage/notification.py +++ b/storage/google/cloud/storage/notification.py @@ -28,11 +28,15 @@ NONE_PAYLOAD_FORMAT = 'NONE' _TOPIC_REF_FMT = '//pubsub.googleapis.com/projects/{}/topics/{}' -_PROJECT_PATTERN = r'(?P[a-z]+-[a-z]+-\d+)' +_PROJECT_PATTERN = r'(?P[a-z][a-z0-9-]{4,28}[a-z0-9])' _TOPIC_NAME_PATTERN = r'(?P[A-Za-z](\w|[-_.~+%])+)' _TOPIC_REF_PATTERN = _TOPIC_REF_FMT.format( _PROJECT_PATTERN, _TOPIC_NAME_PATTERN) _TOPIC_REF_RE = re.compile(_TOPIC_REF_PATTERN) +_BAD_TOPIC = ( + 'Resource has invalid topic: {}; see ' + 'https://cloud.google.com/storage/docs/json_api/v1/' + 'notifications/insert#topic') class BucketNotification(object): @@ -110,25 +114,12 @@ def from_api_repr(cls, resource, bucket): :rtype: :class:`BucketNotification` :returns: the new notification instance - :raises ValueError: - if resource is missing 'topic' key, or if it is not formatted - per the spec documented in - https://cloud.google.com/storage/docs/json_api/v1/notifications/insert#topic """ topic_path = resource.get('topic') if topic_path is None: raise ValueError('Resource has no topic') - match = _TOPIC_REF_RE.match(topic_path) - if match is None: - raise ValueError( - 'Resource has invalid topic: {}; see {}'.format( - topic_path, - 'https://cloud.google.com/storage/docs/json_api/v1/' - 'notifications/insert#topic')) - - name = match.group('name') - project = match.group('project') + name, project = _parse_topic_path(topic_path) instance = cls(bucket, name, topic_project=project) instance._properties = resource @@ -360,3 +351,40 @@ def delete(self, client=None): path=self.path, query_params=query_params, ) + + +def _parse_topic_path(topic_path): + """Verify that a topic path is in the correct format. + + .. _resource manager docs: https://cloud.google.com/resource-manager/\ + reference/rest/v1beta1/projects#\ + Project.FIELDS.project_id + .. _topic spec: https://cloud.google.com/storage/docs/json_api/v1/\ + notifications/insert#topic + + Expected to be of the form: + + //pubsub.googleapis.com/projects/{project}/topics/{topic} + + where the ``project`` value must be "6 to 30 lowercase letters, digits, + or hyphens. It must start with a letter. Trailing hyphens are prohibited." + (see `resource manager docs`_) and ``topic`` must have length at least two, + must start with a letter and may only contain alphanumeric characters or + ``-``, ``_``, ``.``, ``~``, ``+`` or ``%`` (i.e characters used for URL + encoding, see `topic spec`_). + + Args: + topic_path (str): The topic path to be verified. + + Returns: + Tuple[str, str]: The ``project`` and ``topic`` parsed from the + ``topic_path``. + + Raises: + ValueError: If the topic path is invalid. + """ + match = _TOPIC_REF_RE.match(topic_path) + if match is None: + raise ValueError(_BAD_TOPIC.format(topic_path)) + + return match.group('name'), match.group('project') diff --git a/storage/tests/unit/test_notification.py b/storage/tests/unit/test_notification.py index ffd33280c4bc..623bc7013067 100644 --- a/storage/tests/unit/test_notification.py +++ b/storage/tests/unit/test_notification.py @@ -487,3 +487,67 @@ def test_delete_hit(self): path=self.NOTIFICATION_PATH, query_params={'userProject': USER_PROJECT}, ) + + +class Test__parse_topic_path(unittest.TestCase): + + @staticmethod + def _call_fut(*args, **kwargs): + from google.cloud.storage import notification + + return notification._parse_topic_path(*args, **kwargs) + + @staticmethod + def _make_topic_path(project, topic_name): + from google.cloud.storage import notification + + return notification._TOPIC_REF_FMT.format(project, topic_name) + + def test_project_name_too_long(self): + project = 'a' * 31 + topic_path = self._make_topic_path(project, 'topic-name') + with self.assertRaises(ValueError): + self._call_fut(topic_path) + + def test_project_name_uppercase(self): + project = 'aaaAaa' + topic_path = self._make_topic_path(project, 'topic-name') + with self.assertRaises(ValueError): + self._call_fut(topic_path) + + def test_leading_digit(self): + project = '1aaaaa' + topic_path = self._make_topic_path(project, 'topic-name') + with self.assertRaises(ValueError): + self._call_fut(topic_path) + + def test_leading_hyphen(self): + project = '-aaaaa' + topic_path = self._make_topic_path(project, 'topic-name') + with self.assertRaises(ValueError): + self._call_fut(topic_path) + + def test_trailing_hyphen(self): + project = 'aaaaa-' + topic_path = self._make_topic_path(project, 'topic-name') + with self.assertRaises(ValueError): + self._call_fut(topic_path) + + def test_invalid_format(self): + topic_path = '@#$%' + with self.assertRaises(ValueError): + self._call_fut(topic_path) + + def test_success(self): + topic_name = 'tah-pic-nehm' + project_choices = ( + 'a' * 30, # Max length. + 'a-b--c---d', # Valid hyphen usage. + 'abcdefghijklmnopqrstuvwxyz', # Valid letters. + 'z0123456789', # Valid digits (non-leading). + 'a-bcdefghijklmn12opqrstuv0wxyz', + ) + for project in project_choices: + topic_path = self._make_topic_path(project, topic_name) + result = self._call_fut(topic_path) + self.assertEqual(result, (topic_name, project))