chore: allow custom routing header metadata

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

@@ -368,15 +368,18 @@ class {{ service.async_client_name }}:
         {% if method.field_headers %}
         # Certain fields should be provided within the metadata header;
         # add these here.
-        metadata = tuple(metadata) + (
-            gapic_v1.routing_header.to_grpc_metadata((
-                {% for field_header in method.field_headers %}
-                {% if not method.client_streaming %}
-                ("{{ field_header.raw }}", request.{{ field_header.disambiguated }}),
-                {% endif %}
-                {% endfor %}
-            )),
-        )
+        metadata = tuple(metadata)
+        if all(m[0] != gapic_v1.routing_header.ROUTING_METADATA_KEY for m in metadata):
+            metadata = (
+                *metadata,
+                gapic_v1.routing_header.to_grpc_metadata((
+                    {% for field_header in method.field_headers %}
+                    {% if not method.client_streaming %}
+                    ("{{ field_header.raw }}", request.{{ field_header.disambiguated }}),
+                    {% endif %}
+                    {% endfor %}
+                ))
+            )
         {% endif %}
 
 {{ shared_macros.add_api_version_header_to_metadata(service.version) }}

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

@@ -698,10 +698,14 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
 
         # Certain fields should be provided within the metadata header;
         # add these here.
-        metadata = tuple(metadata) + (
-            gapic_v1.routing_header.to_grpc_metadata(
-                (("resource", request.resource),)),
-        )
+        metadata = tuple(metadata)
+        if all(m[0] != gapic_v1.routing_header.ROUTING_METADATA_KEY for m in metadata):
+            metadata = (
+                *metadata,
+                gapic_v1.routing_header.to_grpc_metadata(
+                    (("resource", request.resource),)
+                )
+            )
 
         # Validate the universe domain.
         self._validate_universe_domain()

gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2

@@ -510,6 +510,44 @@ def test_{{ method.name|snake_case }}_routing_parameters():
     {% endfor %}
 {% endif %}
 
+{% if method.field_headers %}
+def test_{{ method.name|snake_case }}_routing_header_override():
+    client = {{ service.client_name }}(
+        credentials=ga_credentials.AnonymousCredentials(),
+    )
+
+    # Any value that is part of the HTTP/1.1 URI should be sent as
+    # a field header. Set these to a non-empty value.
+    request = {{ method.input.ident }}()
+
+    # Mock the actual call within the gRPC stub, and fake the request.
+    with mock.patch.object(
+            type(client.transport.{{ method.transport_safe_name|snake_case }}),
+            '__call__') as call:
+        {% if method.void %}
+        call.return_value = None
+        {% elif method.lro %}
+        call.return_value = operations_pb2.Operation(name='operations/op')
+        {% elif method.server_streaming %}
+        call.return_value = iter([{{ method.output.ident }}()])
+        {% else %}
+        call.return_value = {{ method.output.ident }}()
+        {% endif %}
+        custom_val = "key=custom"
+        override = [('x-goog-request-params', custom_val)]
+        client.{{ method.safe_name|snake_case }}(request, metadata=override)
+
+        # Establish that the underlying gRPC stub method was called.
+        assert len(call.mock_calls) == 1
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == request
+
+    _, _, kw = call.mock_calls[0]
+    # enure there is just one x-goog-request-params header
+    assert len([x for x in kw["metadata"] if x[0] == "x-goog-request-params"]) == 1
+    # ensure that the custom header is the only one
+    assert [x for x in kw["metadata"] if x[0] == "x-goog-request-params"][0][1] == custom_val
+{% endif %}
 
 {% if method.field_headers and not method.client_streaming and not method.explicit_routing %}
 def test_{{ method_name }}_field_headers():