feat: add --resource-name-alias flag to resolve namespace collisions (#16769)

Introduces the `--resource-name-alias` CLI flag to allow API owners
to safely disambiguate fully qualified resource paths that flatten
to identical method names in the generated client. 

The flag accepts mappings in the format `resource.path/Name:AliasName`. 
The configuration is parsed natively into the compiler Options and 
injected into the schema models during construction, avoiding global 
state mutations.

Example usage in BUILD.bazel:
```
--resource-name-alias=ces.googleapis.com/Tool:CesTool
```

Towards: b/505425328

Author: ohmayr

packages/gapic-generator/gapic/schema/api.py

@@ -1689,6 +1689,7 @@ def _load_message(
                 documentation=self.docs.get(path, self.EMPTY),
             ),
             oneofs=oneofs,
+            resource_name_aliases=self.opts.resource_name_aliases,
         )
         return self.proto_messages[address.proto]
 

packages/gapic-generator/gapic/schema/wrappers.py

@@ -520,6 +520,7 @@ class MessageType:
         default_factory=metadata.Metadata,
     )
     oneofs: Optional[Mapping[str, "Oneof"]] = None
+    resource_name_aliases: Mapping[str, str] = dataclasses.field(default_factory=dict)
 
     def __getattr__(self, name):
         return getattr(self.message_pb, name)
@@ -688,7 +689,12 @@ def resource_path(self) -> Optional[str]:
     @property
     def resource_type(self) -> Optional[str]:
         resource = self.options.Extensions[resource_pb2.resource]
-        return resource.type[resource.type.find("/") + 1 :] if resource else None
+        if not resource.type:
+            return None
+            
+        default_type = resource.type[resource.type.find("/") + 1 :]
+
+        return self.resource_name_aliases.get(resource.type, default_type)
 
     @property
     def resource_type_full_path(self) -> Optional[str]:
@@ -2323,6 +2329,29 @@ def gen_indirect_resources_used(message):
             key=lambda m: m.resource_type_full_path or m.name
         )
 
+        # Fail-fast collision detection
+        seen_types: Dict[str, "MessageType"] = {}
+        for msg in sorted_messages:
+            res_type = msg.resource_type
+            if not res_type:
+                # Fail fast if a resource is missing type
+                raise ValueError(
+                    f"\n\nFatal: Message '{msg.name}' defines a resource pattern but is missing a resource type. "
+                    f"This violates AIP-123 (https://google.aip.dev/123). Please define a 'type' in the google.api.resource option."
+                )
+                
+            if res_type in seen_types:
+                incumbent = seen_types[res_type]
+                raise ValueError(
+                    f"\n\nFatal: Namespace collision detected for resource type '{res_type}'.\n"
+                    f"Resources '{incumbent.resource_type_full_path}' and '{msg.resource_type_full_path}' "
+                    f"both flatten to the exact same method name.\n"
+                    f"To protect backward compatibility, explicitly alias one of these using "
+                    f"the `--resource-name-alias` CLI parameter.\n"
+                    f"Example: --resource-name-alias={msg.resource_type_full_path}:CustomName\n"
+                )
+            seen_types[res_type] = msg
+
         return tuple(sorted_messages)
 
     @utils.cached_property

packages/gapic-generator/gapic/utils/options.py

@@ -51,6 +51,7 @@ class Options:
     rest_numeric_enums: bool = False
     proto_plus_deps: Tuple[str, ...] = dataclasses.field(default=("",))
     gapic_version: str = "0.0.0"
+    resource_name_aliases: Dict[str, str] = dataclasses.field(default_factory=dict)
 
     # Class constants
     PYTHON_GAPIC_PREFIX: str = "python-gapic-"
@@ -72,7 +73,11 @@ class Options:
             # proto plus dependencies delineated by '+'
             # For example, 'google.cloud.api.v1+google.cloud.anotherapi.v2'
             "proto-plus-deps",
-            "gapic-version",  # A version string following https://peps.python.org/pep-0440
+            "gapic-version",  # A version string following https://peps.python.org/pep-0440,
+            # Resolves method name collisions by mapping a fully qualified 
+            # resource path to a custom TitleCase alias.
+            # Format: resource.path/Name:AliasName
+            "resource-name-alias",
         )
     )
 
@@ -188,6 +193,35 @@ def tweak_path(p):
         if len(proto_plus_deps):
             proto_plus_deps = tuple(proto_plus_deps[0].split("+"))
 
+        # Parse the resource name aliases dictionary (Format: "path/to/Resource:AliasName")
+        resource_name_aliases = {}
+        raw_aliases = opts.pop("resource-name-alias", [])
+            
+        # Parse explicitly and safely
+        for mapping in raw_aliases:
+            if not mapping:
+                # We only need to check `not mapping` because the top-level 
+                # opt_string parser already stripped trailing whitespaces
+                continue
+                
+            try:
+                # split(":", 1) ensures we only split on the FIRST colon 
+                res_path, alias_name = mapping.split(":", 1)
+                
+                clean_path = res_path.strip()
+                clean_alias = alias_name.strip()
+                
+                if not clean_path or not clean_alias:
+                    raise ValueError()
+                    
+                resource_name_aliases[clean_path] = clean_alias
+                
+            except ValueError:
+                warnings.warn(
+                    f"Ignored malformed resource-name-alias: '{mapping}'. "
+                    "Expected format is 'resource.path/Name:AliasName'."
+                )
+
         answer = Options(
             name=opts.pop("name", [""]).pop(),
             namespace=tuple(opts.pop("namespace", [])),
@@ -210,6 +244,7 @@ def tweak_path(p):
             rest_numeric_enums=rest_numeric_enums,
             proto_plus_deps=proto_plus_deps,
             gapic_version=opts.pop("gapic-version", ["0.0.0"]).pop(),
+            resource_name_aliases=resource_name_aliases,
         )
 
         # Note: if we ever need to recursively check directories for sample

packages/gapic-generator/tests/unit/generator/test_options.py

@@ -254,3 +254,29 @@ def test_options_proto_plus_deps():
         "google.apps.script.type.slides",
         "google.apps.script.type",
     )
+
+
+def test_options_resource_name_aliases():
+    with mock.patch.object(warnings, "warn") as warn:
+        opts = Options.build(
+            "resource-name-alias=ces.googleapis.com/Tool:CesTool,"
+            "resource-name-alias=workspace.googleapis.com/Tool:WorkspaceTool,"
+            "resource-name-alias=bad_string_without_colon,"
+            "resource-name-alias=:MissingPath,"
+            "resource-name-alias=MissingAlias:,"
+            "resource-name-alias=" # Covers empty string ("")
+        )
+
+        # Verify the dictionary is perfectly formed
+        expected = {
+            "ces.googleapis.com/Tool": "CesTool",
+            "workspace.googleapis.com/Tool": "WorkspaceTool",
+        }
+        assert opts.resource_name_aliases == expected
+        
+        # We expect exactly 3 warnings: 
+        # 1. bad_string
+        # 2. :MissingPath 
+        # 3. MissingAlias: 
+        # (The empty ' ' string safely 'continues' without warning, as intended)
+        assert warn.call_count == 3

packages/gapic-generator/tests/unit/schema/wrappers/test_message.py

@@ -473,3 +473,44 @@ def test_extended_operation_request_response_fields():
 
     actual = poll_request.extended_operation_response_fields
     assert actual == expected
+
+
+def test_message_type_resource_type_with_alias():
+    # 1. Cleanly mock the options
+    opts = descriptor_pb2.MessageOptions()
+    opts.Extensions[resource_pb2.resource].type = "ces.googleapis.com/Tool"
+    
+    msg_pb = descriptor_pb2.DescriptorProto(name="MyMessage", options=opts)
+    
+    # 2. Explicitly instantiate MessageType here instead
+    # of using `make_message` to inject 'resource_name_aliases' field.
+    message = wrappers.MessageType(
+        message_pb=msg_pb,
+        fields={},
+        nested_enums={},
+        nested_messages={},
+        resource_name_aliases={"ces.googleapis.com/Tool": "CesTool"},
+    )
+    
+    # 3. Verify it intercepted the default Protobuf logic
+    # and returned the alias
+    assert message.resource_type == "CesTool"
+
+
+def test_message_type_resource_type_without_alias():
+    # 1. Cleanly mock the options
+    opts = descriptor_pb2.MessageOptions()
+    opts.Extensions[resource_pb2.resource].type = "ces.googleapis.com/Tool"
+    msg_pb = descriptor_pb2.DescriptorProto(name="MyMessage", options=opts)
+    
+    # 2. Instantiate WITHOUT any aliases matching this resource
+    message = wrappers.MessageType(
+        message_pb=msg_pb,
+        fields={},
+        nested_enums={},
+        nested_messages={},
+        resource_name_aliases={"some.other/Resource": "OtherAlias"},
+    )
+    
+    # 3. Verify it falls back to the default type ("Tool")
+    assert message.resource_type == "Tool"

packages/gapic-generator/tests/unit/schema/wrappers/test_service.py

@@ -14,14 +14,16 @@
 
 import collections
 import itertools
+import pytest
 import typing
 
 from google.api import resource_pb2
 from google.cloud import extended_operations_pb2 as ex_ops_pb2
 from google.protobuf import descriptor_pb2
 
 from gapic.schema import imp
-from gapic.schema.wrappers import CommonResource
+from gapic.schema.wrappers import CommonResource, MessageType
+from gapic.schema import wrappers
 
 from test_utils.test_utils import (
     get_method,
@@ -693,3 +695,57 @@ def test_extended_operations_lro_detection():
     # because Service objects can't perform the lookup.
     # Instead we kick that can to the API object and make it do the lookup and verification.
     assert lro.operation_service == "CustomOperations"
+
+
+def test_resource_messages_throws_informative_collision_error():
+    # 1. Create options with explicit colliding resource types AND valid patterns
+    opts_1 = descriptor_pb2.MessageOptions()
+    opts_1.Extensions[resource_pb2.resource].type = "ces.googleapis.com/Tool"
+    opts_1.Extensions[resource_pb2.resource].pattern.append("ces/{ces}/tool")
+    
+    opts_2 = descriptor_pb2.MessageOptions()
+    opts_2.Extensions[resource_pb2.resource].type = "workspace.googleapis.com/Tool"
+    opts_2.Extensions[resource_pb2.resource].pattern.append("workspaces/{workspace}/tool")
+    
+    # 2. Use the test helpers to build the messages
+    msg_1 = make_message("MessageOne", options=opts_1)
+    msg_2 = make_message("MessageTwo", options=opts_2)
+
+    # 3. Build the service with BOTH methods (so the property loops) 
+    # AND visible_resources (so the internal lookups succeed).
+    service = make_service(
+        name="MyService",
+        methods=(
+            make_method("GetToolOne", input_message=msg_1),
+            make_method("GetToolTwo", input_message=msg_2),
+        ),
+        visible_resources={
+            "ces.googleapis.com/Tool": msg_1,
+            "workspace.googleapis.com/Tool": msg_2,
+        }
+    )
+
+    # 4. Assert that asking the service for its resources throws the custom error
+    expected_error_regex = r"(?s)Namespace collision detected.*--resource-name-alias"
+    
+    with pytest.raises(ValueError, match=expected_error_regex):
+        _ = service.resource_messages
+
+
+def test_resource_messages_raises_on_malformed_typeless_resource():
+    # 1. Create a malformed resource: it has a pattern, but no type!
+    opts = descriptor_pb2.MessageOptions()
+    opts.Extensions[resource_pb2.resource].pattern.append("workspaces/{workspace}")
+    
+    malformed_msg = make_message("MalformedMessage", options=opts)
+    
+    service = make_service(
+        name="MyService",
+        methods=(
+            make_method("DoThing", input_message=malformed_msg),
+        )
+    )
+    
+    # 2. Trigger the property and expect it to fail fast with the AIP-123 URL
+    with pytest.raises(ValueError, match="https://google.aip.dev/123"):
+        _ = service.resource_messages