test: add unit tests for URL query params encoding to cover urlencode

Add test_*_rest_url_query_params_encoding() tests that verify the '$'
character is preserved (not URL-encoded as '%24') when building URL
query strings. This allows removing the # pragma: NO COVER markers
from the urlencode calls in _shared_macros.j2 templates.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: taimo3810

gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/_shared_macros.j2

@@ -183,7 +183,7 @@ def _get_http_options():
     _query_params = rest_helpers.flatten_query_params(query_params, strict=True)
     _request_url = "{host}{uri}".format(host=host, uri=uri)
     if _query_params:
-        _request_url = "{}?{}".format(_request_url, urlencode(_query_params, safe="$"))  # pragma: NO COVER
+        _request_url = "{}?{}".format(_request_url, urlencode(_query_params, safe="$"))
     response = {{ await_prefix }}getattr(session, method)(
         _request_url,
         timeout=timeout,

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

@@ -47,6 +47,7 @@ from google.api_core import client_options
 from google.api_core import exceptions as core_exceptions
 from google.api_core import grpc_helpers
 from google.api_core import path_template
+from google.api_core import rest_helpers
 from google.api_core import retry as retries
 {% if service.has_lro %}
 from google.api_core import future
@@ -1467,9 +1468,55 @@ def test_{{ method_name }}_rest_unset_required_fields():
     unset_fields = transport.{{ method.transport_safe_name|snake_case }}._get_unset_required_fields({})
     assert set(unset_fields) == (set(({% for param in method.query_params|sort %}"{{ param|camel_case }}", {% endfor %})) & set(({% for param in method.input.required_fields %}"{{param.name|camel_case}}", {% endfor %})))
 
-
     {% endif %}{# required_fields #}
 
+
+def test_{{ method_name }}_rest_url_query_params_encoding():
+    # Verify that special characters like '$' are correctly preserved (not URL-encoded)
+    # when building the URL query string. This tests the urlencode call with safe="$".
+    transport = transports.{{ service.rest_transport_name }}(credentials=ga_credentials.AnonymousCredentials)
+    method_class = transport.{{ method.transport_safe_name|snake_case }}.__class__
+    # Get the _get_response method from the method class
+    get_response_fn = method_class._get_response.__func__
+
+    mock_session = mock.Mock()
+    mock_response = mock.Mock()
+    mock_response.status_code = 200
+    mock_session.get.return_value = mock_response
+    mock_session.post.return_value = mock_response
+    mock_session.put.return_value = mock_response
+    mock_session.patch.return_value = mock_response
+    mock_session.delete.return_value = mock_response
+
+    # Mock flatten_query_params to return query params that include '$' character
+    with mock.patch.object(rest_helpers, 'flatten_query_params') as mock_flatten:
+        mock_flatten.return_value = [('$alt', 'json;enum-encoding=int'), ('foo', 'bar')]
+
+        transcoded_request = {
+            'uri': '/v1/test',
+            'method': '{{ method.http_options[0].method }}',
+        }
+
+        get_response_fn(
+            host='https://example.com',
+            metadata=[],
+            query_params={},
+            session=mock_session,
+            timeout=None,
+            transcoded_request=transcoded_request,
+        )
+
+        # Verify the session method was called with the URL containing query params
+        session_method = getattr(mock_session, '{{ method.http_options[0].method }}')
+        assert session_method.called
+
+        # The URL should contain '$alt' (not '%24alt') because safe="$" is used
+        call_url = session_method.call_args.args[0]
+        assert '$alt=json' in call_url
+        assert '%24alt' not in call_url
+        assert 'foo=bar' in call_url
+
+
 {% if not method.client_streaming %}
 @pytest.mark.parametrize("null_interceptor", [True, False])
 def test_{{ method_name }}_rest_interceptors(null_interceptor):

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

@@ -169,15 +169,14 @@ def _get_http_options():
     method = transcoded_request['method']
     headers = dict(metadata)
     headers['Content-Type'] = 'application/json'
-    # Coverage is insufficient for the `urlencode` call below. Use pragma to skip it.
     # Build query string manually to avoid URL-encoding special characters like '$'.
     # The `requests` library encodes '$' as '%24' when using the `params` argument,
     # which causes API errors for parameters like '$alt'. See:
     # https://github.com/googleapis/gapic-generator-python/issues/2514
     _query_params = rest_helpers.flatten_query_params(query_params, strict=True)
     _request_url = "{host}{uri}".format(host=host, uri=uri)
     if _query_params:
-        _request_url = "{}?{}".format(_request_url, urlencode(_query_params, safe="$"))  # pragma: NO COVER
+        _request_url = "{}?{}".format(_request_url, urlencode(_query_params, safe="$"))
     response = {{ await_prefix }}getattr(session, method)(
         _request_url,
         timeout=timeout,

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

@@ -74,6 +74,7 @@ from google.api_core import exceptions as core_exceptions
 from google.api_core import grpc_helpers
 from google.api_core import grpc_helpers_async
 from google.api_core import path_template
+from google.api_core import rest_helpers
 from google.api_core import retry as retries
 {% if service.has_lro or service.has_extended_lro %}
 from google.api_core import future

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

@@ -1217,6 +1217,52 @@ def test_{{ method_name }}_rest_unset_required_fields():
     {% endif %}{# required_fields #}
 
 
+def test_{{ method_name }}_rest_url_query_params_encoding():
+    # Verify that special characters like '$' are correctly preserved (not URL-encoded)
+    # when building the URL query string. This tests the urlencode call with safe="$".
+    transport = transports.{{ service.rest_transport_name }}(credentials=ga_credentials.AnonymousCredentials)
+    method_class = transport.{{ method.transport_safe_name|snake_case }}.__class__
+    # Get the _get_response method from the method class
+    get_response_fn = method_class._get_response.__func__
+
+    mock_session = mock.Mock()
+    mock_response = mock.Mock()
+    mock_response.status_code = 200
+    mock_session.get.return_value = mock_response
+    mock_session.post.return_value = mock_response
+    mock_session.put.return_value = mock_response
+    mock_session.patch.return_value = mock_response
+    mock_session.delete.return_value = mock_response
+
+    # Mock flatten_query_params to return query params that include '$' character
+    with mock.patch.object(rest_helpers, 'flatten_query_params') as mock_flatten:
+        mock_flatten.return_value = [('$alt', 'json;enum-encoding=int'), ('foo', 'bar')]
+
+        transcoded_request = {
+            'uri': '/v1/test',
+            'method': '{{ method.http_options[0].method }}',
+        }
+
+        get_response_fn(
+            host='https://example.com',
+            metadata=[],
+            query_params={},
+            session=mock_session,
+            timeout=None,
+            transcoded_request=transcoded_request,
+        )
+
+        # Verify the session method was called with the URL containing query params
+        session_method = getattr(mock_session, '{{ method.http_options[0].method }}')
+        assert session_method.called
+
+        # The URL should contain '$alt' (not '%24alt') because safe="$" is used
+        call_url = session_method.call_args.args[0]
+        assert '$alt=json' in call_url
+        assert '%24alt' not in call_url
+        assert 'foo=bar' in call_url
+
+
 {% if method.flattened_fields and not method.client_streaming %}
 def test_{{ method_name }}_rest_flattened():
     client = {{ service.client_name }}(