From 371ebd6bcde4681071235f5ad580aea6995f5005 Mon Sep 17 00:00:00 2001
From: Aza Tulepbergenov <atulep@google.com>
Date: Wed, 22 Dec 2021 13:40:23 -0800
Subject: [PATCH 01/12] feat: adds template for server side streaming.

---
 .../services/%service/transports/rest.py.j2   | 118 +++++++++++++++++-
 1 file changed, 115 insertions(+), 3 deletions(-)

diff --git a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2 b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2
index 0d0998cf43..4386d731d4 100644
--- a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2
+++ b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2
@@ -8,6 +8,116 @@ from google.api_core import retry as retries
 from google.api_core import rest_helpers
 from google.api_core import path_template
 from google.api_core import gapic_v1
+
+# TODO: Remove once my PR gets merged and released.
+# Begin of ResponseIterator depedencies.
+from collections import deque
+import string
+from typing import Deque
+import requests
+
+class ResponseIterator:
+    """Iterator over REST API responses.
+
+    Args:
+        response (requests.Response): An API response object.
+        response_message_cls (Callable[proto.Message]): A proto
+        class expected to be returned from an API.
+    """
+
+    def __init__(self, response: requests.Response, response_message_cls):
+        self._response = response
+        self._response_message_cls = response_message_cls
+        # Inner iterator over HTTP response's content.
+        self._response_itr = self._response.iter_content(decode_unicode=True)
+        # Contains a list of JSON responses ready to be sent to user.
+        self._ready_objs: Deque[str] = deque()
+        # Current JSON response being built.
+        self._obj = ""
+        # Keeps track of the nesting level within a JSON object.
+        self._level = 0
+        # Keeps track whether HTTP response is currently sending values
+        # inside of a string value.
+        self._in_string = False
+        # Whether an escape symbol "\" was encountered.
+        self._next_should_be_escaped = False
+
+    def cancel(self):
+        """Cancel existing streaming operation.
+        """
+        self._response.close()
+
+    def _process_chunk(self, chunk: str):
+        if self._level == 0:
+            if chunk[0] != "[":
+                raise ValueError(
+                    "Can only parse array of JSON objects, instead got %s" % chunk
+                )
+        for char in chunk:
+            if char == "{":
+                if self._level == 1:
+                    # Level 1 corresponds to the outermost JSON object
+                    # (i.e. the one we care about).
+                    self._obj = ""
+                if not self._in_string:
+                    self._level += 1
+                self._obj += char
+            elif char == "}":
+                self._obj += char
+                if not self._in_string:
+                    self._level -= 1
+                if not self._in_string and self._level == 1:
+                    self._ready_objs.append(self._obj)
+            elif char == '"':
+                # Helps to deal with an escaped quotes inside of a string.
+                if not self._next_should_be_escaped:
+                    self._in_string = not self._in_string
+                self._obj += char
+            elif char in string.whitespace:
+                if self._in_string:
+                    self._obj += char
+            elif char == "[":
+                if self._level == 0:
+                    self._level += 1
+                else:
+                    self._obj += char
+            elif char == "]":
+                if self._level == 1:
+                    self._level -= 1
+                else:
+                    self._obj += char
+            else:
+                self._obj += char
+
+            if char == "\\":
+                # Escaping the "\".
+                if self._next_should_be_escaped:
+                    self._next_should_be_escaped = False
+                else:
+                    self._next_should_be_escaped = True
+            else:
+                self._next_should_be_escaped = False
+
+    def __next__(self):
+        while not self._ready_objs:
+            try:
+                chunk = next(self._response_itr)
+                self._process_chunk(chunk)
+            except StopIteration as e:
+                if self._level > 0:
+                    raise ValueError("Unfinished stream: %s" % self._obj)
+                raise e
+        return self._grab()
+
+    def _grab(self):
+        # Add extra quotes to make json.loads happy.
+        return self._response_message_cls.from_json(self._ready_objs.popleft())
+
+    def __iter__(self):
+        return self
+
+# End of ResponseIterator dependencies.
+
 {% if service.has_lro %}
 from google.api_core import operations_v1
 from google.protobuf import json_format
@@ -165,7 +275,7 @@ class {{service.name}}RestTransport({{service.name}}Transport):
 
     {% endif %}{# service.has_lro #}
     {% for method in service.methods.values() %}
-    {%- if method.http_options and not (method.server_streaming or method.client_streaming) %}
+    {%- if method.http_options and not method.client_streaming %}
 
     {% if method.input.required_fields %}
     __{{ method.name | snake_case }}_required_fields_default_values =  {
@@ -283,6 +393,8 @@ class {{service.name}}RestTransport({{service.name}}Transport):
         return_op = operations_pb2.Operation()
         json_format.Parse(response.content, return_op, ignore_unknown_fields=True)
         return return_op
+        {% elif method.server_streaming %}
+        return ResponseIterator(response, {{method.output.ident}})
         {% else %}
         return {{method.output.ident}}.from_json(
             response.content,
@@ -302,10 +414,10 @@ class {{service.name}}RestTransport({{service.name}}Transport):
 
         raise RuntimeError(
             "Cannot define a method without a valid 'google.api.http' annotation.")
-        {%- elif method.server_streaming or method.client_streaming %}
+        {%- elif method.client_streaming %}
 
         raise NotImplementedError(
-            "Streaming over REST is not yet defined for python client")
+            "Client streaming over REST is not yet defined for python client")
         {%- else %}
 
         raise NotImplementedError()

From 9c3034fe102e2dc6ae7ac2fe3b5fbfafbae5a8b2 Mon Sep 17 00:00:00 2001
From: Aza Tulepbergenov <atulep@google.com>
Date: Tue, 4 Jan 2022 15:36:27 -0800
Subject: [PATCH 02/12] feat: adds tests for rest streaming.

---
 gapic/templates/setup.py.j2                   |  2 +-
 .../%name_%version/%sub/test_%service.py.j2   | 82 +++++++++++--------
 noxfile.py                                    |  3 +-
 3 files changed, 50 insertions(+), 37 deletions(-)

diff --git a/gapic/templates/setup.py.j2 b/gapic/templates/setup.py.j2
index 22eef15fd5..1bdbfffe90 100644
--- a/gapic/templates/setup.py.j2
+++ b/gapic/templates/setup.py.j2
@@ -29,7 +29,7 @@ setuptools.setup(
     install_requires=(
         {# TODO(dovs): remove when 1.x deprecation is complete #}
         {% if 'rest' in opts.transport %}
-        'google-api-core[grpc] >= 2.2.0, < 3.0.0dev',
+        'google-api-core[grpc] >= 2.3.2, < 3.0.0dev',
         {% else %}
         'google-api-core[grpc] >= 1.28.0, < 3.0.0dev',
         {% endif %}
diff --git a/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2 b/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
index 4cd8c37bf3..de4bfc2db4 100644
--- a/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
+++ b/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
@@ -8,6 +8,7 @@ import mock
 import grpc
 from grpc.experimental import aio
 {% if "rest" in opts.transport %}
+from collections.abc import Iterable
 import json
 {% endif %}
 import math
@@ -1125,8 +1126,8 @@ def test_{{ method_name }}_raw_page_lro():
 
 {% for method in service.methods.values() if 'rest' in opts.transport and
     method.http_options %}{% with method_name = method.name|snake_case + "_unary" if method.operation_service else method.name|snake_case %}
-{# TODO(kbandes): remove this if condition when streaming is supported in rest. #}
-{% if not (method.server_streaming or method.client_streaming) %}
+{# TODO(kbandes): remove this if condition when client streaming is supported in rest. #}
+{% if not method.client_streaming %}
 def test_{{ method_name }}_rest(transport: str = 'rest', request_type={{ method.input.ident }}):
     client = {{ service.client_name }}(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -1153,8 +1154,6 @@ def test_{{ method_name }}_rest(transport: str = 'rest', request_type={{ method.
         return_value = None
         {% elif method.lro %}
         return_value = operations_pb2.Operation(name='operations/spam')
-        {% elif method.server_streaming %}
-        return_value = iter([{{ method.output.ident }}()])
         {% else %}
         return_value = {{ method.output.ident }}(
             {% for field in method.output.fields.values() | rejectattr('message')%}
@@ -1177,13 +1176,20 @@ def test_{{ method_name }}_rest(transport: str = 'rest', request_type={{ method.
         json_return_value = ''
         {% elif method.lro %}
         json_return_value = json_format.MessageToJson(return_value)
+        {% elif method.server_streaming %}
+        json_return_value = "[{}]".format({{ method.output.ident }}.to_json(return_value))
         {% else %}
         json_return_value = {{ method.output.ident }}.to_json(return_value)
         {% endif %}
         response_value._content = json_return_value.encode('UTF-8')
         req.return_value = response_value
+
         {% if method.client_streaming %}
         response = client.{{ method.name|snake_case }}(iter(requests))
+        {% elif method.server_streaming %}
+        with mock.patch.object(response_value, 'iter_content') as iter_content:
+            iter_content.return_value = iter(json_return_value)
+            response = client.{{ method_name }}(request)
         {% else %}
         response = client.{{ method_name }}(request)
         {% endif %}
@@ -1194,6 +1200,11 @@ def test_{{ method_name }}_rest(transport: str = 'rest', request_type={{ method.
 
     {% endif %}
 
+    {% if method.server_streaming %}
+    assert isinstance(response, Iterable)
+    response = next(response)
+    {% endif %}
+
     # Establish that the response is the type that we expect.
     {% if method.void %}
     assert response is None
@@ -1279,8 +1290,6 @@ def test_{{ method_name }}_rest_required_fields(request_type={{ method.input.ide
     return_value = None
     {% elif method.lro %}
     return_value = operations_pb2.Operation(name='operations/spam')
-    {% elif method.server_streaming %}
-    return_value = iter([{{ method.output.ident }}()])
     {% else %}
     return_value = {{ method.output.ident }}()
     {% endif %}
@@ -1308,6 +1317,8 @@ def test_{{ method_name }}_rest_required_fields(request_type={{ method.input.ide
             json_return_value = ''
             {% elif method.lro %}
             json_return_value = json_format.MessageToJson(return_value)
+            {% elif method.server_streaming %}
+            json_return_value = "[{}]".format({{ method.output.ident }}.to_json(return_value))
             {% else %}
             json_return_value = {{ method.output.ident }}.to_json(return_value)
             {% endif %}
@@ -1316,6 +1327,10 @@ def test_{{ method_name }}_rest_required_fields(request_type={{ method.input.ide
 
             {% if method.client_streaming %}
             response = client.{{ method.name|snake_case }}(iter(requests))
+            {% elif method.server_streaming %}
+            with mock.patch.object(response_value, 'iter_content') as iter_content:
+                iter_content.return_value = iter(json_return_value)
+                response = client.{{ method_name }}(request)
             {% else %}
             response = client.{{ method_name }}(request)
             {% endif %}
@@ -1390,12 +1405,24 @@ def test_{{ method_name }}_rest_flattened(transport: str = 'rest'):
         return_value = None
         {% elif method.lro %}
         return_value = operations_pb2.Operation(name='operations/spam')
-        {% elif method.server_streaming %}
-        return_value = iter([{{ method.output.ident }}()])
         {% else %}
         return_value = {{ method.output.ident }}()
         {% endif %}
 
+        # get arguments that satisfy an http rule for this method
+        sample_request = {{ method.http_options[0].sample_request(method) }}
+
+        # get truthy value for each flattened field
+        mock_args = dict(
+            {% for field in method.flattened_fields.values() %}
+            {% if not field.oneof or field.proto3_optional %}
+            {# ignore oneof fields that might conflict with sample_request #}
+            {{ field.name }}={{ field.mock_value }},
+            {% endif %}
+            {% endfor %}
+        )
+        mock_args.update(sample_request)
+
         # Wrap the value into a proper Response obj
         response_value = Response()
         response_value.status_code = 200
@@ -1403,6 +1430,8 @@ def test_{{ method_name }}_rest_flattened(transport: str = 'rest'):
         json_return_value = ''
         {% elif method.lro %}
         json_return_value = json_format.MessageToJson(return_value)
+        {% elif method.server_streaming %}
+        json_return_value = "[{}]".format({{ method.output.ident }}.to_json(return_value))
         {% else %}
         json_return_value = {{ method.output.ident }}.to_json(return_value)
         {% endif %}
@@ -1410,20 +1439,13 @@ def test_{{ method_name }}_rest_flattened(transport: str = 'rest'):
         response_value._content = json_return_value.encode('UTF-8')
         req.return_value = response_value
 
-        # get arguments that satisfy an http rule for this method
-        sample_request = {{ method.http_options[0].sample_request(method) }}
-
-        # get truthy value for each flattened field
-        mock_args = dict(
-            {% for field in method.flattened_fields.values() %}
-            {% if not field.oneof or field.proto3_optional %}
-            {# ignore oneof fields that might conflict with sample_request #}
-            {{ field.name }}={{ field.mock_value }},
-            {% endif %}
-            {% endfor %}
-        )
-        mock_args.update(sample_request)
+        {% if method.server_streaming %}
+        with mock.patch.object(response_value, 'iter_content') as iter_content:
+            iter_content.return_value = iter(json_return_value)
+            client.{{ method_name }}(**mock_args)
+        {% else %}
         client.{{ method_name }}(**mock_args)
+        {% endif %}
 
         # Establish that the underlying call was made with the expected
         # request object values.
@@ -1527,6 +1549,9 @@ def test_{{ method_name }}_rest_pager(transport: str = 'rest'):
             response = tuple({{ method.output.ident }}.to_json(x) for x in response)
             return_values = tuple(Response() for i in response)
             for return_val, response_val in zip(return_values, response):
+                {% if method.server_streaming %}
+                response_val = "[{}]".format({{ method.output.ident }}.to_json(response_val))
+                {% endif %}
                 return_val._content = response_val.encode('UTF-8')
                 return_val.status_code = 200
             req.side_effect = return_values
@@ -1569,21 +1594,8 @@ def test_{{ method_name }}_rest_pager(transport: str = 'rest'):
 
 
 {% endif %} {# paged methods #}
-{%- else %}
-
-def test_{{ method_name }}_rest_error():
-    client = {{ service.client_name }}(
-        credentials=ga_credentials.AnonymousCredentials(),
-        transport='rest'
-    )
-
-    # TODO(yon-mg): Remove when this method has a working implementation
-    #               or testing straegy
-    with pytest.raises(NotImplementedError):
-        client.{{ method_name }}({})
-
 
-{% endif %}{# not streaming #}{% endwith %}{# method_name #}
+{% endif %}{# not client streaming #}{% endwith %}{# method_name #}
 
 {% endfor -%} {#- method in methods for rest #}
 
diff --git a/noxfile.py b/noxfile.py
index a656b2943d..45f08aeded 100644
--- a/noxfile.py
+++ b/noxfile.py
@@ -189,6 +189,7 @@ def showcase_library(
         # TODO(yon-mg): add "transports=grpc+rest" when all rest features required for
         #               Showcase are implemented i.e. (grpc transcoding, LROs, etc)
         opts = "--python_gapic_opt="
+        #other_opts = ("transport=rest")
         opts += ",".join(other_opts + (f"{template_opt}",))
         cmd_tup = (
             "python",
@@ -296,11 +297,11 @@ def run_showcase_unit_tests(session, fail_under=100):
 
 
 @nox.session(python=ALL_PYTHON)
+@nox.parametrize("other_opts", [("transport=rest",), ("transport=grpc",)])
 def showcase_unit(
     session, templates="DEFAULT", other_opts: typing.Iterable[str] = (),
 ):
     """Run the generated unit tests against the Showcase library."""
-
     with showcase_library(session, templates=templates, other_opts=other_opts) as lib:
         session.chdir(lib)
 

From 733be32ad96eae6368888893cd005c6dab42dcc5 Mon Sep 17 00:00:00 2001
From: Aza Tulepbergenov <atulep@google.com>
Date: Thu, 6 Jan 2022 11:34:22 -0800
Subject: [PATCH 03/12] feat: adds fragment test.

---
 tests/fragments/test_rest_streaming.proto | 43 +++++++++++++++++++++++
 1 file changed, 43 insertions(+)
 create mode 100644 tests/fragments/test_rest_streaming.proto

diff --git a/tests/fragments/test_rest_streaming.proto b/tests/fragments/test_rest_streaming.proto
new file mode 100644
index 0000000000..b47d2030b9
--- /dev/null
+++ b/tests/fragments/test_rest_streaming.proto
@@ -0,0 +1,43 @@
+// Copyright (C) 2022  Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+syntax = "proto3";
+
+package google.fragment;
+
+import "google/api/client.proto";
+
+service MyService {
+  option (google.api.default_host) = "my.example.com";
+
+  rpc MyMethod(MethodRequest) returns (stream MethodResponse) {
+    option (google.api.method_signature) = "from,class,import,any,license,type";
+  }
+}
+
+message MethodRequest {
+  string from = 1;
+  string class = 2;
+  string import = 3;
+  string any = 4;
+  string license = 5;
+  string type = 6;
+  int32 page_size = 7;
+  string page_token = 8;
+}
+
+message MethodResponse {
+  string result = 1;
+  string next_page_token = 2;
+}

From b6eedd3cc9e3c6f76c1225ee92dc5ad7ea81ed3f Mon Sep 17 00:00:00 2001
From: Aza Tulepbergenov <atulep@google.com>
Date: Thu, 6 Jan 2022 12:22:18 -0800
Subject: [PATCH 04/12] feat: adds streaming to ads templates.

---
 .../services/%service/transports/rest.py.j2   | 122 +++++++++++++++++-
 .../%name_%version/%sub/test_%service.py.j2   |  35 +++--
 2 files changed, 144 insertions(+), 13 deletions(-)

diff --git a/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/rest.py.j2 b/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/rest.py.j2
index 3a47d1e363..57c198d69a 100644
--- a/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/rest.py.j2
+++ b/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/rest.py.j2
@@ -13,6 +13,116 @@ from google.api_core import retry as retries
 from google.api_core import rest_helpers
 from google.api_core import path_template
 from google.api_core import gapic_v1
+
+# TODO: Remove once my PR gets merged and released.
+# Begin of ResponseIterator depedencies.
+from collections import deque
+import string
+from typing import Deque
+import requests
+
+class ResponseIterator:
+    """Iterator over REST API responses.
+
+    Args:
+        response (requests.Response): An API response object.
+        response_message_cls (Callable[proto.Message]): A proto
+        class expected to be returned from an API.
+    """
+
+    def __init__(self, response: requests.Response, response_message_cls):
+        self._response = response
+        self._response_message_cls = response_message_cls
+        # Inner iterator over HTTP response's content.
+        self._response_itr = self._response.iter_content(decode_unicode=True)
+        # Contains a list of JSON responses ready to be sent to user.
+        self._ready_objs: Deque[str] = deque()
+        # Current JSON response being built.
+        self._obj = ""
+        # Keeps track of the nesting level within a JSON object.
+        self._level = 0
+        # Keeps track whether HTTP response is currently sending values
+        # inside of a string value.
+        self._in_string = False
+        # Whether an escape symbol "\" was encountered.
+        self._next_should_be_escaped = False
+
+    def cancel(self):
+        """Cancel existing streaming operation.
+        """
+        self._response.close()
+
+    def _process_chunk(self, chunk: str):
+        if self._level == 0:
+            if chunk[0] != "[":
+                raise ValueError(
+                    "Can only parse array of JSON objects, instead got %s" % chunk
+                )
+        for char in chunk:
+            if char == "{":
+                if self._level == 1:
+                    # Level 1 corresponds to the outermost JSON object
+                    # (i.e. the one we care about).
+                    self._obj = ""
+                if not self._in_string:
+                    self._level += 1
+                self._obj += char
+            elif char == "}":
+                self._obj += char
+                if not self._in_string:
+                    self._level -= 1
+                if not self._in_string and self._level == 1:
+                    self._ready_objs.append(self._obj)
+            elif char == '"':
+                # Helps to deal with an escaped quotes inside of a string.
+                if not self._next_should_be_escaped:
+                    self._in_string = not self._in_string
+                self._obj += char
+            elif char in string.whitespace:
+                if self._in_string:
+                    self._obj += char
+            elif char == "[":
+                if self._level == 0:
+                    self._level += 1
+                else:
+                    self._obj += char
+            elif char == "]":
+                if self._level == 1:
+                    self._level -= 1
+                else:
+                    self._obj += char
+            else:
+                self._obj += char
+
+            if char == "\\":
+                # Escaping the "\".
+                if self._next_should_be_escaped:
+                    self._next_should_be_escaped = False
+                else:
+                    self._next_should_be_escaped = True
+            else:
+                self._next_should_be_escaped = False
+
+    def __next__(self):
+        while not self._ready_objs:
+            try:
+                chunk = next(self._response_itr)
+                self._process_chunk(chunk)
+            except StopIteration as e:
+                if self._level > 0:
+                    raise ValueError("Unfinished stream: %s" % self._obj)
+                raise e
+        return self._grab()
+
+    def _grab(self):
+        # Add extra quotes to make json.loads happy.
+        return self._response_message_cls.from_json(self._ready_objs.popleft())
+
+    def __iter__(self):
+        return self
+
+# End of ResponseIterator dependencies.
+
 {% if service.has_lro %}
 from google.api_core import operations_v1
 from google.protobuf import json_format
@@ -179,7 +289,7 @@ class {{service.name}}RestTransport({{service.name}}Transport):
         def __hash__(self):
             return hash("{{method.name}}")
 
-        {% if not (method.server_streaming or method.client_streaming) %}
+        {% if not method.client_streaming %}
         {% if method.input.required_fields %}
         __REQUIRED_FIELDS_DEFAULT_VALUES =  {
         {% for req_field in method.input.required_fields if req_field.is_primitive %}
@@ -200,7 +310,7 @@ class {{service.name}}RestTransport({{service.name}}Transport):
                 timeout: float=None,
                 metadata: Sequence[Tuple[str, str]]=(),
                 ) -> {{method.output.ident}}:
-    {% if method.http_options and not (method.server_streaming or method.client_streaming) %}
+    {% if method.http_options and not method.client_streaming %}
             r"""Call the {{- ' ' -}}
             {{ (method.name|snake_case).replace('_',' ')|wrap(
                     width=70, offset=45, indent=8) }}
@@ -291,6 +401,8 @@ class {{service.name}}RestTransport({{service.name}}Transport):
             return_op = operations_pb2.Operation()
             json_format.Parse(response.content, return_op, ignore_unknown_fields=True)
             return return_op
+            {% elif method.server_streaming %}
+            return ResponseIterator(response, {{method.output.ident}})
             {% else %}
             return {{method.output.ident}}.from_json(
                 response.content,
@@ -299,14 +411,14 @@ class {{service.name}}RestTransport({{service.name}}Transport):
 
             {% endif %}{# method.lro #}
             {% endif %}{# method.void #}
-     {% else %}{# method.http_options and not (method.server_streaming or method.client_streaming) #}
+     {% else %}{# method.http_options and not method.client_streaming #}
             {% if not method.http_options %}
             raise RuntimeError(
                 "Cannot define a method without a valid 'google.api.http' annotation.")
 
-            {% elif method.server_streaming or method.client_streaming %}
+            {% elif method.client_streaming %}
             raise NotImplementedError(
-                "Streaming over REST is not yet defined for python client")
+                "Client streaming over REST is not yet defined for python client")
 
             {% else %}
             raise NotImplementedError()
diff --git a/gapic/ads-templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2 b/gapic/ads-templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
index 38d91f3ae9..48e77c96dc 100644
--- a/gapic/ads-templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
+++ b/gapic/ads-templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
@@ -8,6 +8,7 @@ import mock
 import grpc
 from grpc.experimental import aio
 {% if "rest" in opts.transport %}
+from collections.abc import Iterable
 import json
 {% endif %}
 import math
@@ -823,8 +824,8 @@ def test_{{ method_name }}_raw_page_lro():
 {% endfor %} {# method in methods for grpc #}
 
 {% for method in service.methods.values() if 'rest' in opts.transport %}{% with method_name = method.name|snake_case + "_unary" if method.operation_service else method.name|snake_case %}{% if method.http_options %}
-{# TODO(kbandes): remove this if condition when streaming are supported. #}
-{% if not (method.server_streaming or method.client_streaming) %}
+{# TODO(kbandes): remove this if condition when client streaming are supported. #}
+{% if not method.client_streaming %}
 @pytest.mark.parametrize("request_type", [
   {{ method.input.ident }},
   dict,
@@ -846,8 +847,6 @@ def test_{{ method_name }}_rest(request_type, transport: str = 'rest'):
         return_value = None
         {% elif method.lro %}
         return_value = operations_pb2.Operation(name='operations/spam')
-        {% elif method.server_streaming %}
-        return_value = iter([{{ method.output.ident }}()])
         {% else %}
         return_value = {{ method.output.ident }}(
             {% for field in method.output.fields.values() | rejectattr('message')%}
@@ -867,6 +866,8 @@ def test_{{ method_name }}_rest(request_type, transport: str = 'rest'):
         req.return_value.request = PreparedRequest()
         {% if method.void %}
         json_return_value = ''
+        {% elif method.server_streaming %}
+        json_return_value = "[{}]".format({{ method.output.ident }}.to_json(return_value))
         {% else %}
         json_return_value = {{ method.output.ident }}.to_json(return_value)
         {% endif %}
@@ -876,6 +877,10 @@ def test_{{ method_name }}_rest(request_type, transport: str = 'rest'):
             # the request over the wire, so an empty request is fine.
             {% if method.client_streaming %}
             client.{{ method_name }}(iter([requests]))
+            {% elif method.server_streaming %}
+            with mock.patch.object(response_value, 'iter_content') as iter_content:
+                iter_content.return_value = iter(json_return_value)
+                response = client.{{ method_name }}(request)
             {% else %}
             client.{{ method_name }}(request)
             {% endif %}
@@ -1038,8 +1043,6 @@ def test_{{ method_name }}_rest_required_fields(request_type={{ method.input.ide
     return_value = None
     {% elif method.lro %}
     return_value = operations_pb2.Operation(name='operations/spam')
-    {% elif method.server_streaming %}
-    return_value = iter([{{ method.output.ident }}()])
     {% else %}
     return_value = {{ method.output.ident }}()
     {% endif %}
@@ -1067,6 +1070,8 @@ def test_{{ method_name }}_rest_required_fields(request_type={{ method.input.ide
             json_return_value = ''
             {% elif method.lro %}
             json_return_value = json_format.MessageToJson(return_value)
+            {% elif method.server_streaming %}
+            json_return_value = "[{}]".format({{ method.output.ident }}.to_json(return_value))
             {% else %}
             json_return_value = {{ method.output.ident }}.to_json(return_value)
             {% endif %}
@@ -1075,6 +1080,10 @@ def test_{{ method_name }}_rest_required_fields(request_type={{ method.input.ide
 
             {% if method.client_streaming %}
             response = client.{{ method.name|snake_case }}(iter(requests))
+            {% elif method.server_streaming %}
+            with mock.patch.object(response_value, 'iter_content') as iter_content:
+                iter_content.return_value = iter(json_return_value)
+                response = client.{{ method_name }}(request)
             {% else %}
             response = client.{{ method_name }}(request)
             {% endif %}
@@ -1145,8 +1154,6 @@ def test_{{ method.name|snake_case }}_rest_flattened():
         return_value = None
         {% elif method.lro %}
         return_value = operations_pb2.Operation(name='operations/spam')
-        {% elif method.server_streaming %}
-        return_value = iter([{{ method.output.ident }}()])
         {% else %}
         return_value = {{ method.output.ident }}()
         {% endif %}
@@ -1158,6 +1165,8 @@ def test_{{ method.name|snake_case }}_rest_flattened():
         json_return_value = ''
         {% elif method.lro %}
         json_return_value = json_format.MessageToJson(return_value)
+    {% elif method.server_streaming %}
+        json_return_value = "[{}]".format({{ method.output.ident }}.to_json(return_value))
     {% else %}
         json_return_value = {{ method.output.ident }}.to_json(return_value)
     {% endif %}
@@ -1178,7 +1187,14 @@ def test_{{ method.name|snake_case }}_rest_flattened():
             {% endfor %}
         )
         mock_args.update(sample_request)
+
+        {% if method.server_streaming %}
+        with mock.patch.object(response_value, 'iter_content') as iter_content:
+            iter_content.return_value = iter(json_return_value)
+            client.{{ method_name }}(**mock_args)
+        {% else %}
         client.{{ method_name }}(**mock_args)
+        {% endif %}
 
         # Establish that the underlying call was made with the expected
         # request object values.
@@ -1282,6 +1298,9 @@ def test_{{ method_name }}_rest_pager(transport: str = 'rest'):
         response = tuple({{ method.output.ident }}.to_json(x) for x in response)
         return_values = tuple(Response() for i in response)
         for return_val, response_val in zip(return_values, response):
+            {% if method.server_streaming %}
+            response_val = "[{}]".format({{ method.output.ident }}.to_json(response_val))
+            {% endif %}
             return_val._content = response_val.encode('UTF-8')
             return_val.status_code = 200
         req.side_effect = return_values

From 4eaab2b48596d5a3ebf2f8f632818b191090ad50 Mon Sep 17 00:00:00 2001
From: Aza Tulepbergenov <atulep@google.com>
Date: Mon, 10 Jan 2022 11:19:46 -0800
Subject: [PATCH 05/12] feat: adds showcase test for streaming.

---
 tests/system/conftest.py     | 2 ++
 tests/system/test_streams.py | 8 ++------
 2 files changed, 4 insertions(+), 6 deletions(-)

diff --git a/tests/system/conftest.py b/tests/system/conftest.py
index fd92e72bc8..8201b29704 100644
--- a/tests/system/conftest.py
+++ b/tests/system/conftest.py
@@ -111,6 +111,8 @@ def construct_client(
                 channel=channel_creator("localhost:7469"),
             )
         elif transport_name == "rest":
+            # TODO(atulep): The template code hardcodes https://, so
+            # the below line doesn't work.
             # The custom host explicitly bypasses https.
             transport = transport_cls(
                 host="http://localhost:7469",
diff --git a/tests/system/test_streams.py b/tests/system/test_streams.py
index 685f7300d8..8294e67958 100644
--- a/tests/system/test_streams.py
+++ b/tests/system/test_streams.py
@@ -23,10 +23,6 @@
 
 
 def test_unary_stream(echo):
-    if isinstance(echo.transport, type(echo).get_transport_class("rest")):
-        # (TODO: dovs) Temporarily disabling rest
-        return
-
     content = 'The hail in Wales falls mainly on the snails.'
     responses = echo.expand({
         'content': content,
@@ -37,8 +33,8 @@ def test_unary_stream(echo):
     for ground_truth, response in zip(content.split(' '), responses):
         assert response.content == ground_truth
     assert ground_truth == 'snails.'
-
-    assert responses.trailing_metadata() == metadata
+    if isinstance(echo.transport, type(echo).get_transport_class("grpc")):
+        assert responses.trailing_metadata() == metadata
 
 
 def test_stream_unary(echo):

From 9b4f68edff18668cebe6d3e2eec06b8ffe3d0843 Mon Sep 17 00:00:00 2001
From: Aza Tulepbergenov <atulep@google.com>
Date: Tue, 11 Jan 2022 20:41:41 -0800
Subject: [PATCH 06/12] chore: removes temporary code.

---
 .../services/%service/transports/rest.py.j2   | 112 +-----------------
 gapic/templates/setup.py.j2                   |   2 +-
 2 files changed, 3 insertions(+), 111 deletions(-)

diff --git a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2 b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2
index a1df6af4f1..239c7f6474 100644
--- a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2
+++ b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2
@@ -11,118 +11,10 @@ from google.auth import credentials as ga_credentials  # type: ignore
 from google.api_core import exceptions as core_exceptions
 from google.api_core import retry as retries
 from google.api_core import rest_helpers
+from google.api_core import rest_streaming
 from google.api_core import path_template
 from google.api_core import gapic_v1
 
-# TODO: Remove once my PR gets merged and released.
-# Begin of ResponseIterator depedencies.
-from collections import deque
-import string
-from typing import Deque
-import requests
-
-class ResponseIterator:
-    """Iterator over REST API responses.
-
-    Args:
-        response (requests.Response): An API response object.
-        response_message_cls (Callable[proto.Message]): A proto
-        class expected to be returned from an API.
-    """
-
-    def __init__(self, response: requests.Response, response_message_cls):
-        self._response = response
-        self._response_message_cls = response_message_cls
-        # Inner iterator over HTTP response's content.
-        self._response_itr = self._response.iter_content(decode_unicode=True)
-        # Contains a list of JSON responses ready to be sent to user.
-        self._ready_objs: Deque[str] = deque()
-        # Current JSON response being built.
-        self._obj = ""
-        # Keeps track of the nesting level within a JSON object.
-        self._level = 0
-        # Keeps track whether HTTP response is currently sending values
-        # inside of a string value.
-        self._in_string = False
-        # Whether an escape symbol "\" was encountered.
-        self._next_should_be_escaped = False
-
-    def cancel(self):
-        """Cancel existing streaming operation.
-        """
-        self._response.close()
-
-    def _process_chunk(self, chunk: str):
-        if self._level == 0:
-            if chunk[0] != "[":
-                raise ValueError(
-                    "Can only parse array of JSON objects, instead got %s" % chunk
-                )
-        for char in chunk:
-            if char == "{":
-                if self._level == 1:
-                    # Level 1 corresponds to the outermost JSON object
-                    # (i.e. the one we care about).
-                    self._obj = ""
-                if not self._in_string:
-                    self._level += 1
-                self._obj += char
-            elif char == "}":
-                self._obj += char
-                if not self._in_string:
-                    self._level -= 1
-                if not self._in_string and self._level == 1:
-                    self._ready_objs.append(self._obj)
-            elif char == '"':
-                # Helps to deal with an escaped quotes inside of a string.
-                if not self._next_should_be_escaped:
-                    self._in_string = not self._in_string
-                self._obj += char
-            elif char in string.whitespace:
-                if self._in_string:
-                    self._obj += char
-            elif char == "[":
-                if self._level == 0:
-                    self._level += 1
-                else:
-                    self._obj += char
-            elif char == "]":
-                if self._level == 1:
-                    self._level -= 1
-                else:
-                    self._obj += char
-            else:
-                self._obj += char
-
-            if char == "\\":
-                # Escaping the "\".
-                if self._next_should_be_escaped:
-                    self._next_should_be_escaped = False
-                else:
-                    self._next_should_be_escaped = True
-            else:
-                self._next_should_be_escaped = False
-
-    def __next__(self):
-        while not self._ready_objs:
-            try:
-                chunk = next(self._response_itr)
-                self._process_chunk(chunk)
-            except StopIteration as e:
-                if self._level > 0:
-                    raise ValueError("Unfinished stream: %s" % self._obj)
-                raise e
-        return self._grab()
-
-    def _grab(self):
-        # Add extra quotes to make json.loads happy.
-        return self._response_message_cls.from_json(self._ready_objs.popleft())
-
-    def __iter__(self):
-        return self
-
-# End of ResponseIterator dependencies.
-
 {% if service.has_lro %}
 from google.api_core import operations_v1
 from google.protobuf import json_format
@@ -402,7 +294,7 @@ class {{service.name}}RestTransport({{service.name}}Transport):
             json_format.Parse(response.content, return_op, ignore_unknown_fields=True)
             return return_op
             {% elif method.server_streaming %}
-            return ResponseIterator(response, {{method.output.ident}})
+            return rest_streaming.ResponseIterator(response, {{method.output.ident}})
             {% else %}
             return {{method.output.ident}}.from_json(
                 response.content,
diff --git a/gapic/templates/setup.py.j2 b/gapic/templates/setup.py.j2
index 1bdbfffe90..575539a365 100644
--- a/gapic/templates/setup.py.j2
+++ b/gapic/templates/setup.py.j2
@@ -29,7 +29,7 @@ setuptools.setup(
     install_requires=(
         {# TODO(dovs): remove when 1.x deprecation is complete #}
         {% if 'rest' in opts.transport %}
-        'google-api-core[grpc] >= 2.3.2, < 3.0.0dev',
+        'google-api-core[grpc] >= 2.4.0, < 3.0.0dev',
         {% else %}
         'google-api-core[grpc] >= 1.28.0, < 3.0.0dev',
         {% endif %}

From b338973d43d3d0bd486f9fd0dccdcd5176db26d5 Mon Sep 17 00:00:00 2001
From: Aza Tulepbergenov <atulep@google.com>
Date: Tue, 11 Jan 2022 20:42:54 -0800
Subject: [PATCH 07/12] chore: removes temporary code from ads templates.

---
 .../services/%service/transports/rest.py.j2   | 112 +-----------------
 1 file changed, 2 insertions(+), 110 deletions(-)

diff --git a/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/rest.py.j2 b/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/rest.py.j2
index 57c198d69a..a45f9378ce 100644
--- a/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/rest.py.j2
+++ b/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/rest.py.j2
@@ -11,118 +11,10 @@ from google.auth import credentials as ga_credentials  # type: ignore
 from google.api_core import exceptions as core_exceptions
 from google.api_core import retry as retries
 from google.api_core import rest_helpers
+from google.api_core import rest_streaming
 from google.api_core import path_template
 from google.api_core import gapic_v1
 
-# TODO: Remove once my PR gets merged and released.
-# Begin of ResponseIterator depedencies.
-from collections import deque
-import string
-from typing import Deque
-import requests
-
-class ResponseIterator:
-    """Iterator over REST API responses.
-
-    Args:
-        response (requests.Response): An API response object.
-        response_message_cls (Callable[proto.Message]): A proto
-        class expected to be returned from an API.
-    """
-
-    def __init__(self, response: requests.Response, response_message_cls):
-        self._response = response
-        self._response_message_cls = response_message_cls
-        # Inner iterator over HTTP response's content.
-        self._response_itr = self._response.iter_content(decode_unicode=True)
-        # Contains a list of JSON responses ready to be sent to user.
-        self._ready_objs: Deque[str] = deque()
-        # Current JSON response being built.
-        self._obj = ""
-        # Keeps track of the nesting level within a JSON object.
-        self._level = 0
-        # Keeps track whether HTTP response is currently sending values
-        # inside of a string value.
-        self._in_string = False
-        # Whether an escape symbol "\" was encountered.
-        self._next_should_be_escaped = False
-
-    def cancel(self):
-        """Cancel existing streaming operation.
-        """
-        self._response.close()
-
-    def _process_chunk(self, chunk: str):
-        if self._level == 0:
-            if chunk[0] != "[":
-                raise ValueError(
-                    "Can only parse array of JSON objects, instead got %s" % chunk
-                )
-        for char in chunk:
-            if char == "{":
-                if self._level == 1:
-                    # Level 1 corresponds to the outermost JSON object
-                    # (i.e. the one we care about).
-                    self._obj = ""
-                if not self._in_string:
-                    self._level += 1
-                self._obj += char
-            elif char == "}":
-                self._obj += char
-                if not self._in_string:
-                    self._level -= 1
-                if not self._in_string and self._level == 1:
-                    self._ready_objs.append(self._obj)
-            elif char == '"':
-                # Helps to deal with an escaped quotes inside of a string.
-                if not self._next_should_be_escaped:
-                    self._in_string = not self._in_string
-                self._obj += char
-            elif char in string.whitespace:
-                if self._in_string:
-                    self._obj += char
-            elif char == "[":
-                if self._level == 0:
-                    self._level += 1
-                else:
-                    self._obj += char
-            elif char == "]":
-                if self._level == 1:
-                    self._level -= 1
-                else:
-                    self._obj += char
-            else:
-                self._obj += char
-
-            if char == "\\":
-                # Escaping the "\".
-                if self._next_should_be_escaped:
-                    self._next_should_be_escaped = False
-                else:
-                    self._next_should_be_escaped = True
-            else:
-                self._next_should_be_escaped = False
-
-    def __next__(self):
-        while not self._ready_objs:
-            try:
-                chunk = next(self._response_itr)
-                self._process_chunk(chunk)
-            except StopIteration as e:
-                if self._level > 0:
-                    raise ValueError("Unfinished stream: %s" % self._obj)
-                raise e
-        return self._grab()
-
-    def _grab(self):
-        # Add extra quotes to make json.loads happy.
-        return self._response_message_cls.from_json(self._ready_objs.popleft())
-
-    def __iter__(self):
-        return self
-
-# End of ResponseIterator dependencies.
-
 {% if service.has_lro %}
 from google.api_core import operations_v1
 from google.protobuf import json_format
@@ -402,7 +294,7 @@ class {{service.name}}RestTransport({{service.name}}Transport):
             json_format.Parse(response.content, return_op, ignore_unknown_fields=True)
             return return_op
             {% elif method.server_streaming %}
-            return ResponseIterator(response, {{method.output.ident}})
+            return rest_streaming.ResponseIterator(response, {{method.output.ident}})
             {% else %}
             return {{method.output.ident}}.from_json(
                 response.content,

From f78ee1a864c332a01b815e752d79191ebe68e5cd Mon Sep 17 00:00:00 2001
From: Aza Tulepbergenov <atulep@google.com>
Date: Wed, 12 Jan 2022 10:09:09 -0800
Subject: [PATCH 08/12] chore: fixes ads templates.

---
 .../gapic/%name_%version/%sub/test_%service.py.j2   | 13 +++++++++++--
 1 file changed, 11 insertions(+), 2 deletions(-)

diff --git a/gapic/ads-templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2 b/gapic/ads-templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
index 48e77c96dc..f1a22abf4d 100644
--- a/gapic/ads-templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
+++ b/gapic/ads-templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
@@ -917,8 +917,6 @@ def test_{{ method.name|snake_case }}_rest(request_type):
         return_value = None
         {% elif method.lro %}
         return_value = operations_pb2.Operation(name='operations/spam')
-        {% elif method.server_streaming %}
-        return_value = iter([{{ method.output.ident }}()])
         {% else %}
         return_value = {{ method.output.ident }}(
             {% for field in method.output.fields.values() | rejectattr('message')%}
@@ -941,6 +939,8 @@ def test_{{ method.name|snake_case }}_rest(request_type):
         json_return_value = ''
         {% elif method.lro %}
         json_return_value = json_format.MessageToJson(return_value)
+    {% elif method.server_streaming %}
+        json_return_value = "[{}]".format({{ method.output.ident }}.to_json(return_value))
     {% else %}
         json_return_value = {{ method.output.ident }}.to_json(return_value)
     {% endif %}
@@ -948,6 +948,10 @@ def test_{{ method.name|snake_case }}_rest(request_type):
         req.return_value = response_value
         {% if method.client_streaming %}
         response = client.{{ method.name|snake_case }}(iter(requests))
+        {% elif method.server_streaming %}
+        with mock.patch.object(response_value, 'iter_content') as iter_content:
+            iter_content.return_value = iter(json_return_value)
+            response = client.{{ method_name }}(request)
         {% else %}
         response = client.{{ method_name }}(request)
         {% endif %}
@@ -958,6 +962,11 @@ def test_{{ method.name|snake_case }}_rest(request_type):
 
     {% endif %}
 
+    {% if method.server_streaming %}
+    assert isinstance(response, Iterable)
+    response = next(response)
+    {% endif %}
+
     # Establish that the response is the type that we expect.
     {% if method.void %}
     assert response is None

From 9867d00f35e9073c65023c4f5a896317ca1bd383 Mon Sep 17 00:00:00 2001
From: Aza Tulepbergenov <atulep@google.com>
Date: Thu, 27 Jan 2022 16:48:21 -0800
Subject: [PATCH 09/12] chore: fix mypy

---
 .../%version/%sub/services/%service/transports/rest.py.j2     | 4 ++--
 .../%sub/services/%service/transports/rest.py.j2              | 4 ++--
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/rest.py.j2 b/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/rest.py.j2
index 66b03643d6..2d90ea31dd 100644
--- a/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/rest.py.j2
+++ b/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/rest.py.j2
@@ -94,7 +94,7 @@ class {{ service.name }}RestInterceptor:
         return request, metadata
 
     {% if not method.void %}
-    def post_{{ method.name|snake_case }}(self, response: {{method.output.ident}}) -> {{method.output.ident}}:
+    def post_{{ method.name|snake_case }}(self, response: Union[{{method.output.ident}}, rest_streaming.ResponseIterator]) -> Union[{{method.output.ident}}, rest_streaming.ResponseIterator]:
         """Post-rpc interceptor for {{ method.name|snake_case }}
 
         Override in a subclass to manipulate the response
@@ -269,7 +269,7 @@ class {{service.name}}RestTransport({{service.name}}Transport):
                 retry: OptionalRetry=gapic_v1.method.DEFAULT,
                 timeout: float=None,
                 metadata: Sequence[Tuple[str, str]]=(),
-                ){% if not method.void %} -> {{method.output.ident}}{% endif %}:
+                ){% if not method.void %} -> Union[{{method.output.ident}}, rest_streaming.ResponseIterator]{% endif %}:
     {% if method.http_options and not method.client_streaming %}
             r"""Call the {{- ' ' -}}
             {{ (method.name|snake_case).replace('_',' ')|wrap(
diff --git a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2 b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2
index 66b03643d6..2d90ea31dd 100644
--- a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2
+++ b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2
@@ -94,7 +94,7 @@ class {{ service.name }}RestInterceptor:
         return request, metadata
 
     {% if not method.void %}
-    def post_{{ method.name|snake_case }}(self, response: {{method.output.ident}}) -> {{method.output.ident}}:
+    def post_{{ method.name|snake_case }}(self, response: Union[{{method.output.ident}}, rest_streaming.ResponseIterator]) -> Union[{{method.output.ident}}, rest_streaming.ResponseIterator]:
         """Post-rpc interceptor for {{ method.name|snake_case }}
 
         Override in a subclass to manipulate the response
@@ -269,7 +269,7 @@ class {{service.name}}RestTransport({{service.name}}Transport):
                 retry: OptionalRetry=gapic_v1.method.DEFAULT,
                 timeout: float=None,
                 metadata: Sequence[Tuple[str, str]]=(),
-                ){% if not method.void %} -> {{method.output.ident}}{% endif %}:
+                ){% if not method.void %} -> Union[{{method.output.ident}}, rest_streaming.ResponseIterator]{% endif %}:
     {% if method.http_options and not method.client_streaming %}
             r"""Call the {{- ' ' -}}
             {{ (method.name|snake_case).replace('_',' ')|wrap(

From b7aa3094e70b4a3fa462837442fbbb76b683af8f Mon Sep 17 00:00:00 2001
From: Aza Tulepbergenov <atulep@google.com>
Date: Thu, 27 Jan 2022 17:04:01 -0800
Subject: [PATCH 10/12] chore: fix null interceptor check.

---
 .../unit/gapic/%name_%version/%sub/test_%service.py.j2     | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2 b/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
index a5dd84f64b..258d598e02 100644
--- a/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
+++ b/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
@@ -1530,7 +1530,7 @@ def test_{{ method_name }}_rest_unset_required_fields():
     {% endif %}{# required_fields #}
 
 
-{% if not (method.server_streaming or method.client_streaming) %}
+{% if not method.client_streaming %}
 @pytest.mark.parametrize("null_interceptor", [True, False])
 def test_{{ method_name }}_rest_interceptors(null_interceptor):
     transport = transports.{{ service.name }}RestTransport(
@@ -1559,6 +1559,11 @@ def test_{{ method_name }}_rest_interceptors(null_interceptor):
         req.return_value.request = PreparedRequest()
         {% if not method.void %}
         req.return_value._content = {% if method.output.ident.package == method.ident.package %}{{ method.output.ident }}.to_json({{ method.output.ident }}()){% else %}json_format.MessageToJson({{ method.output.ident }}()){% endif %}
+        
+        {% if method.server_streaming %}
+        req.return_value._content = "[{}]".format(req.return_value._content)
+        {% endif %}
+
         {% endif %}
 
         request = {{ method.input.ident }}()

From 5b2f7e702cd946526d718a109605e9f98cb0b982 Mon Sep 17 00:00:00 2001
From: Aza Tulepbergenov <atulep@google.com>
Date: Thu, 27 Jan 2022 17:36:10 -0800
Subject: [PATCH 11/12] chore: fix type annotation.

---
 .../%version/%sub/services/%service/transports/rest.py.j2 | 8 ++++++--
 .../%sub/services/%service/transports/rest.py.j2          | 8 ++++++--
 2 files changed, 12 insertions(+), 4 deletions(-)

diff --git a/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/rest.py.j2 b/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/rest.py.j2
index 2d90ea31dd..3abf640810 100644
--- a/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/rest.py.j2
+++ b/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/rest.py.j2
@@ -94,7 +94,11 @@ class {{ service.name }}RestInterceptor:
         return request, metadata
 
     {% if not method.void %}
-    def post_{{ method.name|snake_case }}(self, response: Union[{{method.output.ident}}, rest_streaming.ResponseIterator]) -> Union[{{method.output.ident}}, rest_streaming.ResponseIterator]:
+    {% if not method.server_streaming %}
+    def post_{{ method.name|snake_case }}(self, response: {{method.output.ident}}) -> {{method.output.ident}}:
+    {% else %}
+    def post_{{ method.name|snake_case }}(self, response: rest_streaming.ResponseIterator) -> rest_streaming.ResponseIterator:
+    {% endif %}
         """Post-rpc interceptor for {{ method.name|snake_case }}
 
         Override in a subclass to manipulate the response
@@ -269,7 +273,7 @@ class {{service.name}}RestTransport({{service.name}}Transport):
                 retry: OptionalRetry=gapic_v1.method.DEFAULT,
                 timeout: float=None,
                 metadata: Sequence[Tuple[str, str]]=(),
-                ){% if not method.void %} -> Union[{{method.output.ident}}, rest_streaming.ResponseIterator]{% endif %}:
+                ){% if not method.void %} -> {% if not method.server_streaming %}{{method.output.ident}}{% else %}rest_streaming.ResponseIterator{% endif %}{% endif %}:
     {% if method.http_options and not method.client_streaming %}
             r"""Call the {{- ' ' -}}
             {{ (method.name|snake_case).replace('_',' ')|wrap(
diff --git a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2 b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2
index 2d90ea31dd..3abf640810 100644
--- a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2
+++ b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2
@@ -94,7 +94,11 @@ class {{ service.name }}RestInterceptor:
         return request, metadata
 
     {% if not method.void %}
-    def post_{{ method.name|snake_case }}(self, response: Union[{{method.output.ident}}, rest_streaming.ResponseIterator]) -> Union[{{method.output.ident}}, rest_streaming.ResponseIterator]:
+    {% if not method.server_streaming %}
+    def post_{{ method.name|snake_case }}(self, response: {{method.output.ident}}) -> {{method.output.ident}}:
+    {% else %}
+    def post_{{ method.name|snake_case }}(self, response: rest_streaming.ResponseIterator) -> rest_streaming.ResponseIterator:
+    {% endif %}
         """Post-rpc interceptor for {{ method.name|snake_case }}
 
         Override in a subclass to manipulate the response
@@ -269,7 +273,7 @@ class {{service.name}}RestTransport({{service.name}}Transport):
                 retry: OptionalRetry=gapic_v1.method.DEFAULT,
                 timeout: float=None,
                 metadata: Sequence[Tuple[str, str]]=(),
-                ){% if not method.void %} -> Union[{{method.output.ident}}, rest_streaming.ResponseIterator]{% endif %}:
+                ){% if not method.void %} -> {% if not method.server_streaming %}{{method.output.ident}}{% else %}rest_streaming.ResponseIterator{% endif %}{% endif %}:
     {% if method.http_options and not method.client_streaming %}
             r"""Call the {{- ' ' -}}
             {{ (method.name|snake_case).replace('_',' ')|wrap(

From 95a0861900975c53c7d800637a246ba48d2ae50a Mon Sep 17 00:00:00 2001
From: Aza Tulepbergenov <atulep@google.com>
Date: Thu, 27 Jan 2022 17:36:59 -0800
Subject: [PATCH 12/12] chore: fix todo.

---
 tests/system/conftest.py | 2 --
 1 file changed, 2 deletions(-)

diff --git a/tests/system/conftest.py b/tests/system/conftest.py
index 8cf4556fa5..86935c7742 100644
--- a/tests/system/conftest.py
+++ b/tests/system/conftest.py
@@ -112,8 +112,6 @@ def construct_client(
                 channel=channel_creator("localhost:7469"),
             )
         elif transport_name == "rest":
-            # TODO(atulep): The template code hardcodes https://, so
-            # the below line doesn't work.
             # The custom host explicitly bypasses https.
             transport = transport_cls(
                 credentials=credentials.AnonymousCredentials(),