Test and impl for resource path parsing methods in generated clients

gapic/schema/wrappers.py

@@ -209,6 +209,10 @@ def with_context(self, *, collisions: FrozenSet[str]) -> 'Field':
 @dataclasses.dataclass(frozen=True)
 class MessageType:
     """Description of a message (defined with the ``message`` keyword)."""
+    # Class attributes
+    PATH_ARG_RE = re.compile(r'\{([a-zA-Z0-9_-]+)\}')
+
+    # Instance attributes
     message_pb: descriptor_pb2.DescriptorProto
     fields: Mapping[str, Field]
     nested_enums: Mapping[str, 'EnumType']
@@ -278,8 +282,34 @@ def resource_type(self) -> Optional[str]:
 
     @property
     def resource_path_args(self) -> Sequence[str]:
-        path_arg_re = re.compile(r'\{([a-zA-Z0-9_-]+)\}')
-        return path_arg_re.findall(self.resource_path or '')
+        return self.PATH_ARG_RE.findall(self.resource_path or '')
+
+    @utils.cached_property
+    def path_regex_str(self) -> str:
+        def gen_component_re(m: re.Match) -> str:
+            # We can't just use (?P<name>[^/]+) because segments may be
+            # separated by delimiters other than '/'.
+            # Multiple delimiter characters within one schema are allowed, e.g.
+            # as/{a}-{b}/cs/{c}%{d}_{e}
+            # This is discouraged but permitted by AIP4231
+            return "(?P<{name}>.+?)".format(name=m.groups()[0])
+
+        # The indirection here is a little confusing:
+        # we're using the resource path template as the base of a regex,
+        # with each resource ID segment being captured by a regex.
+        # E.g., the path schema
+        # kingdoms/{kingdom}/phyla/{phylum}
+        # becomes the regex
+        # ^kingdoms/(?P<kingdom>.+?)/phyla/(?P<phylum>.+?)$
+        parsing_regex_str = (
+            "^" +
+            self.PATH_ARG_RE.sub(
+                gen_component_re,
+                self.resource_path or ''
+            ) +
+            "$"
+        )
+        return parsing_regex_str
 
     def get_field(self, *field_path: str,
                   collisions: FrozenSet[str] = frozenset()) -> Field:

gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2

@@ -120,6 +120,11 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
         """Return a fully-qualified {{ message.resource_type|snake_case }} string."""
         return "{{ message.resource_path }}".format({% for arg in message.resource_path_args %}{{ arg }}={{ arg }}, {% endfor %})
 
+
+    def parse_{{ message.resource_type|snake_case }}_path(path: str) -> Dict[str,str]:
+        """Parse a {{ message.resource_type|snake_case }} path into its component segments."""
+        m = re.match(r"{{ message.path_regex_str }}", path)
+        return m.groupdict() if m else {}
     {% endfor %}
 
     def __init__(self, *,

gapic/templates/tests/unit/%name_%version/%sub/test_%service.py.j2

@@ -692,14 +692,27 @@ def test_{{ service.name|snake_case }}_grpc_lro_client():
 {% endif -%}
 
 {% for message in service.resource_messages -%}
-{% with molluscs = cycler("squid", "clam", "whelk", "octopus", "oyster", "nudibranch", "cuttlefish", "mussel", "winkle") -%}
+{% with molluscs = cycler("squid", "clam", "whelk", "octopus", "oyster", "nudibranch", "cuttlefish", "mussel", "winkle", "nautilus", "scallop", "abalone") -%}
 def test_{{ message.resource_type|snake_case }}_path():
-  {% for arg in message.resource_path_args -%}
-  {{ arg }} = "{{ molluscs.next() }}"
-  {% endfor %}
-  expected = "{{ message.resource_path }}".format({% for arg in message.resource_path_args %}{{ arg }}={{ arg }}, {% endfor %})
-  actual = {{ service.client_name }}.{{ message.resource_type|snake_case }}_path({{message.resource_path_args|join(", ") }})
-  assert expected == actual
+    {% for arg in message.resource_path_args -%}
+    {{ arg }} = "{{ molluscs.next() }}"
+    {% endfor %}
+    expected = "{{ message.resource_path }}".format({% for arg in message.resource_path_args %}{{ arg }}={{ arg }}, {% endfor %})
+    actual = {{ service.client_name }}.{{ message.resource_type|snake_case }}_path({{message.resource_path_args|join(", ") }})
+    assert expected == actual
+
+
+def test_parse_{{ message.resource_type|snake_case }}_path():
+    expected = {
+    {% for arg in message.resource_path_args -%}
+        "{{ arg }}": "{{ molluscs.next() }}",
+    {% endfor %}
+    }
+    path = {{ service.client_name }}.{{ message.resource_type|snake_case }}_path(**expected)
+
+    # Check that the path construction is reversible.
+    actual = {{ service.client_name }}.parse_{{ message.resource_type|snake_case }}_path(path)
+    assert expected == actual
 
 {% endwith -%}
 {% endfor -%}

tests/unit/schema/wrappers/test_message.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import collections
+import re
 from typing import Sequence, Tuple
 
 import pytest
@@ -181,6 +182,46 @@ def test_resource_path():
     assert message.resource_type == "Class"
 
 
+def test_parse_resource_path():
+    options = descriptor_pb2.MessageOptions()
+    resource = options.Extensions[resource_pb2.resource]
+    resource.pattern.append(
+        "kingdoms/{kingdom}/phyla/{phylum}/classes/{klass}"
+    )
+    resource.type = "taxonomy.biology.com/Klass"
+    message = make_message('Klass', options=options)
+
+    # Plausible resource ID path
+    path = "kingdoms/animalia/phyla/mollusca/classes/cephalopoda"
+    expected = {
+        'kingdom': 'animalia',
+        'phylum': 'mollusca',
+        'klass': 'cephalopoda',
+    }
+    actual = re.match(message.path_regex_str, path).groupdict()
+
+    assert expected == actual
+
+    options2 = descriptor_pb2.MessageOptions()
+    resource2 = options2.Extensions[resource_pb2.resource]
+    resource2.pattern.append(
+        "kingdoms-{kingdom}_{phylum}#classes%{klass}"
+    )
+    resource2.type = "taxonomy.biology.com/Klass"
+    message2 = make_message('Klass', options=options2)
+
+    # Plausible resource ID path from a non-standard schema
+    path2 = "kingdoms-Animalia/_Mollusca~#classes%Cephalopoda"
+    expected2 = {
+        'kingdom': 'Animalia/',
+        'phylum': 'Mollusca~',
+        'klass': 'Cephalopoda',
+    }
+    actual2 = re.match(message2.path_regex_str, path2).groupdict()
+
+    assert expected2 == actual2
+
+
 def test_field_map():
     # Create an Entry message.
     entry_msg = make_message(