Storage: ProjectID Regex

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<project>[a-z]+-[a-z]+-\d+)'
+_PROJECT_PATTERN = r'(?P<project>[a-z][a-z0-9-]{4,28}[a-z0-9])'
 _TOPIC_NAME_PATTERN = r'(?P<name>[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')

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))