diff --git a/.circleci/config.yml b/.circleci/config.yml
index 5d00282c83..9ae961fcee 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -38,12 +38,6 @@ workflows:
           filters:
             tags:
               only: /^v\d+\.\d+\.\d+$/
-      - showcase-unit-alternative-templates-3.6:
-          requires:
-            - unit-3.6
-          filters:
-            tags:
-              only: /^v\d+\.\d+\.\d+$/
       - showcase-unit-alternative-templates-3.7:
           requires:
             - unit-3.7
@@ -100,7 +94,6 @@ workflows:
           requires:
             - docs
             - mypy
-            - showcase-unit-alternative-templates-3.6
             - showcase-unit-alternative-templates-3.7
             - showcase-unit-alternative-templates-3.8
             - showcase-mypy-alternative-templates
@@ -111,7 +104,6 @@ workflows:
           requires:
             - docs
             - mypy
-            - showcase-unit-alternative-templates-3.6
             - showcase-unit-alternative-templates-3.7
             - showcase-unit-alternative-templates-3.8
             - showcase-mypy-alternative-templates
@@ -424,30 +416,6 @@ jobs:
       - run:
           name: Run unit tests.
           command: nox -s showcase_unit-3.8
-  showcase-unit-alternative-templates-3.6:
-    docker:
-      - image: python:3.6-slim
-    steps:
-      - checkout
-      - run:
-          name: Install system dependencies.
-          command: |
-            apt-get update
-            apt-get install -y curl pandoc unzip gcc
-      - run:
-          name: Install protoc 3.12.1.
-          command: |
-            mkdir -p /usr/src/protoc/
-            curl --location https://github.com/google/protobuf/releases/download/v3.12.1/protoc-3.12.1-linux-x86_64.zip --output /usr/src/protoc/protoc-3.12.1.zip
-            cd /usr/src/protoc/
-            unzip protoc-3.12.1.zip
-            ln -s /usr/src/protoc/bin/protoc /usr/local/bin/protoc
-      - run:
-          name: Install nox.
-          command: pip install nox
-      - run:
-          name: Run unit tests.
-          command: nox -s showcase_unit_alternative_templates-3.6
   showcase-unit-alternative-templates-3.7:
     docker:
       - image: python:3.7-slim
diff --git a/.github/sync-repo-settings.yaml b/.github/sync-repo-settings.yaml
index d0e56dcae8..86677eec40 100644
--- a/.github/sync-repo-settings.yaml
+++ b/.github/sync-repo-settings.yaml
@@ -17,7 +17,6 @@ branchProtectionRules:
     - 'ci/circleci: showcase-unit-3.7'
     - 'ci/circleci: showcase-unit-3.8'
     - 'ci/circleci: showcase-unit-add-iam-methods'
-    - 'ci/circleci: showcase-unit-alternative-templates-3.6'
     - 'ci/circleci: showcase-unit-alternative-templates-3.7'
     - 'ci/circleci: showcase-unit-alternative-templates-3.8'
     - 'ci/circleci: style-check'
diff --git a/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/base.py.j2 b/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/base.py.j2
index 2053e9fe4f..f25ba96a1a 100644
--- a/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/base.py.j2
+++ b/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/base.py.j2
@@ -55,10 +55,10 @@ class {{ service.name }}Transport(metaclass=abc.ABCMeta):
                 credentials identify the application to the service; if none
                 are specified, the client will attempt to ascertain the
                 credentials from the environment.
-            client_info (google.api_core.gapic_v1.client_info.ClientInfo):	
-                The client info used to send a user-agent string along with	
-                API requests. If ``None``, then default info will be used.	
-                Generally, you only need to set this if you're developing	
+            client_info (google.api_core.gapic_v1.client_info.ClientInfo):
+                The client info used to send a user-agent string along with
+                API requests. If ``None``, then default info will be used.
+                Generally, you only need to set this if you're developing
                 your own client library.
         """
         # Save the hostname. Default to port 443 (HTTPS) if none is specified.
@@ -89,7 +89,7 @@ class {{ service.name }}Transport(metaclass=abc.ABCMeta):
                     {% if method.retry.max_backoff %}maximum={{ method.retry.max_backoff }},{% endif %}
                     {% if method.retry.backoff_multiplier %}multiplier={{ method.retry.backoff_multiplier }},{% endif %}
                     predicate=retries.if_exception_type(
-                        {%- for ex in method.retry.retryable_exceptions|sort(attribute='__name__) %}
+                        {%- for ex in method.retry.retryable_exceptions|sort(attribute='__name__') %}
                         exceptions.{{ ex.__name__ }},
                         {%- endfor %}
                     ),
diff --git a/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/grpc.py.j2 b/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/grpc.py.j2
index 7fbf53e78c..6995549378 100644
--- a/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/grpc.py.j2
+++ b/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/grpc.py.j2
@@ -1,7 +1,8 @@
 {% extends '_base.py.j2' %}
 
 {% block content %}
-from typing import Callable, Dict, Tuple
+import warnings
+from typing import Callable, Dict, Optional, Sequence, Tuple
 
 from google.api_core import grpc_helpers   # type: ignore
 {%- if service.has_lro %}
@@ -10,7 +11,7 @@ from google.api_core import operations_v1  # type: ignore
 from google.api_core import gapic_v1       # type: ignore
 from google import auth                    # type: ignore
 from google.auth import credentials        # type: ignore
-
+from google.auth.transport.grpc import SslCredentials  # type: ignore
 
 import grpc  # type: ignore
 
@@ -38,8 +39,13 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
     def __init__(self, *,
             host: str{% if service.host %} = '{{ service.host }}'{% endif %},
             credentials: credentials.Credentials = None,
+            credentials_file: str = None,
+            scopes: Sequence[str] = None,
             channel: grpc.Channel = None,
+            api_mtls_endpoint: str = None,
+            client_cert_source: Callable[[], Tuple[bytes, bytes]] = None,
             ssl_channel_credentials: grpc.ChannelCredentials = None,
+            quota_project_id: Optional[str] = None,
             client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
             ) -> None:
         """Instantiate the transport.
@@ -53,14 +59,29 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
                 are specified, the client will attempt to ascertain the
                 credentials from the environment.
                 This argument is ignored if ``channel`` is provided.
+            credentials_file (Optional[str]): A file with credentials that can
+                be loaded with :func:`google.auth.load_credentials_from_file`.
+                This argument is ignored if ``channel`` is provided.
+            scopes (Optional(Sequence[str])): A list of scopes. This argument is
+                ignored if ``channel`` is provided.
             channel (Optional[grpc.Channel]): A ``Channel`` instance through
                 which to make calls.
+            api_mtls_endpoint (Optional[str]): Deprecated. The mutual TLS endpoint.
+                If provided, it overrides the ``host`` argument and tries to create
+                a mutual TLS channel with client SSL credentials from
+                ``client_cert_source`` or applicatin default SSL credentials.
+            client_cert_source (Optional[Callable[[], Tuple[bytes, bytes]]]):
+                Deprecated. A callback to provide client SSL certificate bytes and
+                private key bytes, both in PEM format. It is ignored if
+                ``api_mtls_endpoint`` is None.
             ssl_channel_credentials (grpc.ChannelCredentials): SSL credentials
                 for grpc channel. It is ignored if ``channel`` is provided.
-            client_info (google.api_core.gapic_v1.client_info.ClientInfo):	
-                The client info used to send a user-agent string along with	
-                API requests. If ``None``, then default info will be used.	
-                Generally, you only need to set this if you're developing	
+            quota_project_id (Optional[str]): An optional project to use for billing
+                and quota.
+            client_info (google.api_core.gapic_v1.client_info.ClientInfo):
+                The client info used to send a user-agent string along with
+                API requests. If ``None``, then default info will be used.
+                Generally, you only need to set this if you're developing
                 your own client library.
 
         Raises:
@@ -74,6 +95,33 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
 
             # If a channel was explicitly provided, set it.
             self._grpc_channel = channel
+        elif api_mtls_endpoint:
+            warnings.warn("api_mtls_endpoint and client_cert_source are deprecated", DeprecationWarning)
+
+            host = api_mtls_endpoint if ":" in api_mtls_endpoint else api_mtls_endpoint + ":443"
+
+            if credentials is None:
+                credentials, _ = auth.default(scopes=self.AUTH_SCOPES, quota_project_id=quota_project_id)
+
+            # Create SSL credentials with client_cert_source or application
+            # default SSL credentials.
+            if client_cert_source:
+                cert, key = client_cert_source()
+                ssl_credentials = grpc.ssl_channel_credentials(
+                    certificate_chain=cert, private_key=key
+                )
+            else:
+                ssl_credentials = SslCredentials().ssl_credentials
+
+            # create a new channel. The provided one is ignored.
+            self._grpc_channel = type(self).create_channel(
+                host,
+                credentials=credentials,
+                credentials_file=credentials_file,
+                ssl_credentials=ssl_credentials,
+                scopes=scopes or self.AUTH_SCOPES,
+                quota_project_id=quota_project_id,
+            )
         else:
             host = host if ":" in host else host + ":443"
 
@@ -81,7 +129,7 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
                 credentials, _ = auth.default(scopes=self.AUTH_SCOPES)
 
             # create a new channel. The provided one is ignored.
-            self._grpc_channel = grpc_helpers.create_channel(
+            self._grpc_channel = type(self).create_channel(
                 host,
                 credentials=credentials,
                 ssl_credentials=ssl_channel_credentials,
@@ -102,6 +150,7 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
     def create_channel(cls,
                        host: str{% if service.host %} = '{{ service.host }}'{% endif %},
                        credentials: credentials.Credentials = None,
+                       scopes: Optional[Sequence[str]] = None,
                        **kwargs) -> grpc.Channel:
         """Create and return a gRPC channel object.
         Args:
@@ -111,6 +160,9 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
                 credentials identify this application to the service. If
                 none are specified, the client will attempt to ascertain
                 the credentials from the environment.
+            scopes (Optional[Sequence[str]]): A optional list of scopes needed for this
+                service. These are only used when credentials are not specified and
+                are passed to :func:`google.auth.default`.
             kwargs (Optional[dict]): Keyword arguments, which are passed to the
                 channel creation.
         Returns:
@@ -119,26 +171,14 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
         return grpc_helpers.create_channel(
             host,
             credentials=credentials,
-            scopes=cls.AUTH_SCOPES,
+            scopes=scopes or cls.AUTH_SCOPES,
             **kwargs
         )
 
     @property
     def grpc_channel(self) -> grpc.Channel:
-        """Create the channel designed to connect to this service.
-
-        This property caches on the instance; repeated calls return
-        the same channel.
+        """Return the channel designed to connect to this service.
         """
-        # Sanity check: Only create a new channel if we do not already
-        # have one.
-        if not hasattr(self, '_grpc_channel'):
-            self._grpc_channel = self.create_channel(
-                self._host,
-                credentials=self._credentials,
-            )
-
-        # Return the channel from cache.
         return self._grpc_channel
     {%- if service.has_lro %}
 
diff --git a/gapic/ads-templates/noxfile.py.j2 b/gapic/ads-templates/noxfile.py.j2
index 71f99a4144..4760bc548b 100644
--- a/gapic/ads-templates/noxfile.py.j2
+++ b/gapic/ads-templates/noxfile.py.j2
@@ -20,7 +20,7 @@ def unit(session):
         '--cov-config=.coveragerc',
         '--cov-report=term',
         '--cov-report=html',
-        os.path.join('tests', 'unit', '{{ api.naming.versioned_module_name }}'),
+        os.path.join('tests', 'unit', 'gapic', '{{ api.naming.versioned_module_name }}'),
     )
 
 
diff --git a/gapic/ads-templates/setup.py.j2 b/gapic/ads-templates/setup.py.j2
index 6587ebd21e..92ae4ea7af 100644
--- a/gapic/ads-templates/setup.py.j2
+++ b/gapic/ads-templates/setup.py.j2
@@ -24,7 +24,7 @@ setuptools.setup(
         'grpc-google-iam-v1',
     {%- endif %}
     ),
-    python_requires='>={% if opts.lazy_import %}3.7{% else %}3.6{% endif %}',{# Lazy import requires module-level getattr #}
+    python_requires='>=3.7',{# Lazy import requires module-level getattr #}
     setup_requires=[
         'libcst >= 0.2.5',
     ],
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 55768e7cd7..810a6f2e92 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
@@ -144,7 +144,7 @@ def test_{{ service.client_name|snake_case }}_client_options():
     # Check the case GOOGLE_API_USE_CLIENT_CERTIFICATE has unsupported value.
     with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "Unsupported"}):
         with pytest.raises(ValueError):
-            client = client_class()
+            client = {{ service.client_name }}()
 
 
 @mock.patch.object({{ service.client_name }}, "DEFAULT_ENDPOINT", modify_default_endpoint({{ service.client_name }}))
@@ -222,7 +222,7 @@ def test_{{ service.client_name|snake_case }}_mtls_env_auto(use_client_cert_env)
 
 
 def test_{{ service.client_name|snake_case }}_client_options_from_dict():
-    with mock.patch('{{ (api.naming.module_namespace + (api.naming.versioned_module_name,) + service.meta.address.subpackage)|join(".") }}.services.{{ service.name|snake_case }}.transports.{{ service.name }}Transport.__init__') as grpc_transport:
+    with mock.patch('{{ (api.naming.module_namespace + (api.naming.versioned_module_name,) + service.meta.address.subpackage)|join(".") }}.services.{{ service.name|snake_case }}.transports.{{ service.name }}GrpcTransport.__init__') as grpc_transport:
         grpc_transport.return_value = None
         client = {{ service.client_name }}(
             client_options={'api_endpoint': 'squid.clam.whelk'}
@@ -583,6 +583,15 @@ def test_transport_instance():
     assert client.transport is transport
 
 
+def test_transport_get_channel():
+    # A client may be instantiated with a custom transport instance.
+    transport = transports.{{ service.name }}GrpcTransport(
+        credentials=credentials.AnonymousCredentials(),
+    )
+    channel = transport.grpc_channel
+    assert channel
+
+
 def test_transport_grpc_default():
     # A client should use the gRPC transport by default.
     client = {{ service.client_name }}(
@@ -593,18 +602,20 @@ def test_transport_grpc_default():
         transports.{{ service.name }}GrpcTransport,
     )
 
-
-def test_transport_adc():
+@pytest.mark.parametrize("transport_class", [
+    transports.{{ service.grpc_transport_name }},
+])
+def test_transport_adc(transport_class):
     # Test default credentials are used if not provided.
     with mock.patch.object(auth, 'default') as adc:
         adc.return_value = (credentials.AnonymousCredentials(), None)
-        transports.{{ service.name }}Transport()
+        transport_class()
         adc.assert_called_once()
 
 
 def test_{{ service.name|snake_case }}_base_transport():
     # Instantiate the base transport.
-    with mock.patch('{{ (api.naming.module_namespace + (api.naming.versioned_module_name,) + service.meta.address.subpackage)|join(".") }}.services.{{ service.name|snake_case }}.transports.{{ service.name }}GrpcTransport.__init__') as Transport:
+    with mock.patch('{{ (api.naming.module_namespace + (api.naming.versioned_module_name,) + service.meta.address.subpackage)|join(".") }}.services.{{ service.name|snake_case }}.transports.{{ service.name }}Transport.__init__') as Transport:
         Transport.return_value = None
         transport = transports.{{ service.name }}Transport(
             credentials=credentials.AnonymousCredentials(),
@@ -695,6 +706,85 @@ def test_{{ service.name|snake_case }}_grpc_transport_channel():
     assert transport._host == "squid.clam.whelk:443"
 
 
+@pytest.mark.parametrize("transport_class", [transports.{{ service.grpc_transport_name }}])
+def test_{{ service.name|snake_case }}_transport_channel_mtls_with_client_cert_source(
+    transport_class
+):
+    with mock.patch("grpc.ssl_channel_credentials", autospec=True) as grpc_ssl_channel_cred:
+        with mock.patch.object(transport_class, "create_channel", autospec=True) as grpc_create_channel:
+            mock_ssl_cred = mock.Mock()
+            grpc_ssl_channel_cred.return_value = mock_ssl_cred
+
+            mock_grpc_channel = mock.Mock()
+            grpc_create_channel.return_value = mock_grpc_channel
+
+            cred = credentials.AnonymousCredentials()
+            with pytest.warns(DeprecationWarning):
+                with mock.patch.object(auth, 'default') as adc:
+                    adc.return_value = (cred, None)
+                    transport = transport_class(
+                        host="squid.clam.whelk",
+                        api_mtls_endpoint="mtls.squid.clam.whelk",
+                        client_cert_source=client_cert_source_callback,
+                    )
+                    adc.assert_called_once()
+
+            grpc_ssl_channel_cred.assert_called_once_with(
+                certificate_chain=b"cert bytes", private_key=b"key bytes"
+            )
+            grpc_create_channel.assert_called_once_with(
+                "mtls.squid.clam.whelk:443",
+                credentials=cred,
+                credentials_file=None,
+                scopes=(
+                    {%- for scope in service.oauth_scopes %}
+                    '{{ scope }}',
+                    {%- endfor %}
+                ),
+                ssl_credentials=mock_ssl_cred,
+                quota_project_id=None,
+            )
+            assert transport.grpc_channel == mock_grpc_channel
+
+
+@pytest.mark.parametrize("transport_class", [transports.{{ service.grpc_transport_name }},])
+def test_{{ service.name|snake_case }}_transport_channel_mtls_with_adc(
+    transport_class
+):
+    mock_ssl_cred = mock.Mock()
+    with mock.patch.multiple(
+        "google.auth.transport.grpc.SslCredentials",
+        __init__=mock.Mock(return_value=None),
+        ssl_credentials=mock.PropertyMock(return_value=mock_ssl_cred),
+    ):
+        with mock.patch.object(transport_class, "create_channel", autospec=True) as grpc_create_channel:
+            mock_grpc_channel = mock.Mock()
+            grpc_create_channel.return_value = mock_grpc_channel
+            mock_cred = mock.Mock()
+
+            with pytest.warns(DeprecationWarning):
+                transport = transport_class(
+                    host="squid.clam.whelk",
+                    credentials=mock_cred,
+                    api_mtls_endpoint="mtls.squid.clam.whelk",
+                    client_cert_source=None,
+                )
+
+            grpc_create_channel.assert_called_once_with(
+                "mtls.squid.clam.whelk:443",
+                credentials=mock_cred,
+                credentials_file=None,
+                scopes=(
+                    {%- for scope in service.oauth_scopes %}
+                    '{{ scope }}',
+                    {%- endfor %}
+                ),
+                ssl_credentials=mock_ssl_cred,
+                quota_project_id=None,
+            )
+            assert transport.grpc_channel == mock_grpc_channel
+
+
 {% if service.has_lro -%}
 def test_{{ service.name|snake_case }}_grpc_lro_client():
     client = {{ service.client_name }}(
diff --git a/gapic/templates/%namespace/%name_%version/%sub/services/%service/async_client.py.j2 b/gapic/templates/%namespace/%name_%version/%sub/services/%service/async_client.py.j2
index 275c7c9e8b..0f2e88700f 100644
--- a/gapic/templates/%namespace/%name_%version/%sub/services/%service/async_client.py.j2
+++ b/gapic/templates/%namespace/%name_%version/%sub/services/%service/async_client.py.j2
@@ -50,6 +50,16 @@ class {{ service.async_client_name }}:
     from_service_account_file = {{ service.client_name }}.from_service_account_file
     from_service_account_json = from_service_account_file
 
+    @property
+    def transport(self) -> {{ service.name }}Transport:
+        """Return the transport used by the client instance.
+
+        Returns:
+            {{ service.name }}Transport: The transport used by the client instance.
+        """
+        return self._client.transport
+
+
     get_transport_class = functools.partial(type({{ service.client_name }}).get_transport_class, type({{ service.client_name }}))
 
     def __init__(self, *,
diff --git a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc.py.j2 b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc.py.j2
index 4283b12ad6..47eaffeb19 100644
--- a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc.py.j2
+++ b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc.py.j2
@@ -76,7 +76,7 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
                 If provided, it overrides the ``host`` argument and tries to create
                 a mutual TLS channel with client SSL credentials from
                 ``client_cert_source`` or applicatin default SSL credentials.
-            client_cert_source (Optional[Callable[[], Tuple[bytes, bytes]]]): 
+            client_cert_source (Optional[Callable[[], Tuple[bytes, bytes]]]):
                 Deprecated. A callback to provide client SSL certificate bytes and
                 private key bytes, both in PEM format. It is ignored if
                 ``api_mtls_endpoint`` is None.
@@ -84,10 +84,10 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
                 for grpc channel. It is ignored if ``channel`` is provided.
             quota_project_id (Optional[str]): An optional project to use for billing
                 and quota.
-            client_info (google.api_core.gapic_v1.client_info.ClientInfo):	
-                The client info used to send a user-agent string along with	
-                API requests. If ``None``, then default info will be used.	
-                Generally, you only need to set this if you're developing	
+            client_info (google.api_core.gapic_v1.client_info.ClientInfo):
+                The client info used to send a user-agent string along with
+                API requests. If ``None``, then default info will be used.
+                Generally, you only need to set this if you're developing
                 your own client library.
 
         Raises:
@@ -105,7 +105,7 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
             self._grpc_channel = channel
         elif api_mtls_endpoint:
             warnings.warn("api_mtls_endpoint and client_cert_source are deprecated", DeprecationWarning)
-            
+
             host = api_mtls_endpoint if ":" in api_mtls_endpoint else api_mtls_endpoint + ":443"
 
             if credentials is None:
@@ -203,12 +203,8 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
 
     @property
     def grpc_channel(self) -> grpc.Channel:
-        """Create the channel designed to connect to this service.
-
-        This property caches on the instance; repeated calls return
-        the same channel.
+        """Return the channel designed to connect to this service.
         """
-        # Return the channel from cache.
         return self._grpc_channel
     {%- if service.has_lro %}
 
@@ -339,7 +335,7 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
                 response_deserializer=iam_policy.TestIamPermissionsResponse.FromString,
             )
         return self._stubs["test_iam_permissions"]
-    {% endif %} 
+    {% endif %}
 
 __all__ = (
     '{{ service.name }}GrpcTransport',
diff --git a/gapic/templates/setup.py.j2 b/gapic/templates/setup.py.j2
index e2d1ad6597..fdfbf7a637 100644
--- a/gapic/templates/setup.py.j2
+++ b/gapic/templates/setup.py.j2
@@ -23,7 +23,7 @@ setuptools.setup(
         'grpc-google-iam-v1',
     {%- endif %}
     ),
-    python_requires='>={% if opts.lazy_import %}3.7{% else %}3.6{% endif %}',{# Lazy import requires module-level getattr #}
+    python_requires='>=3.6',
     scripts=[
         'scripts/fixup_{{ api.naming.versioned_module_name }}_keywords.py',
     ],
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 f9365635d7..359b548a0b 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
@@ -432,7 +432,7 @@ async def test_{{ method.name|snake_case }}_async(transport: str = 'grpc_asyncio
 
     # Mock the actual call within the gRPC stub, and fake the request.
     with mock.patch.object(
-            type(client._client.transport.{{ method.name|snake_case }}),
+            type(client.transport.{{ method.name|snake_case }}),
             '__call__') as call:
         # Designate an appropriate return value for the call.
         {% if method.void -%}
@@ -561,7 +561,7 @@ async def test_{{ method.name|snake_case }}_field_headers_async():
 
     # Mock the actual call within the gRPC stub, and fake the request.
     with mock.patch.object(
-            type(client._client.transport.{{ method.name|snake_case }}),
+            type(client.transport.{{ method.name|snake_case }}),
             '__call__') as call:
         {% if method.void -%}
         call.return_value = grpc_helpers_async.FakeUnaryUnaryCall(None)
@@ -693,7 +693,7 @@ async def test_{{ method.name|snake_case }}_flattened_async():
 
     # Mock the actual call within the gRPC stub, and fake the request.
     with mock.patch.object(
-            type(client._client.transport.{{ method.name|snake_case }}),
+            type(client.transport.{{ method.name|snake_case }}),
             '__call__') as call:
         # Designate an appropriate return value for the call.
         {% if method.void -%}
@@ -880,7 +880,7 @@ async def test_{{ method.name|snake_case }}_async_pager():
 
     # Mock the actual call within the gRPC stub, and fake the request.
     with mock.patch.object(
-            type(client._client.transport.{{ method.name|snake_case }}),
+            type(client.transport.{{ method.name|snake_case }}),
             '__call__', new_callable=mock.AsyncMock) as call:
         # Set the response to a series of pages.
         call.side_effect = (
@@ -928,7 +928,7 @@ async def test_{{ method.name|snake_case }}_async_pages():
 
     # Mock the actual call within the gRPC stub, and fake the request.
     with mock.patch.object(
-            type(client._client.transport.{{ method.name|snake_case }}),
+            type(client.transport.{{ method.name|snake_case }}),
             '__call__', new_callable=mock.AsyncMock) as call:
         # Set the response to a series of pages.
         call.side_effect = (
@@ -1291,7 +1291,7 @@ def test_{{ service.name|snake_case }}_grpc_lro_async_client():
         credentials=credentials.AnonymousCredentials(),
         transport='grpc_asyncio',
     )
-    transport = client._client.transport
+    transport = client.transport
 
     # Ensure that we have a api-core operations client.
     assert isinstance(
@@ -1416,7 +1416,7 @@ async def test_set_iam_policy_async(transport: str = "grpc_asyncio"):
 
     # Mock the actual call within the gRPC stub, and fake the request.
     with mock.patch.object(
-        type(client._client.transport.set_iam_policy), "__call__"
+        type(client.transport.set_iam_policy), "__call__"
     ) as call:
         # Designate an appropriate return value for the call.
         call.return_value = grpc_helpers_async.FakeUnaryUnaryCall(
@@ -1478,7 +1478,7 @@ async def test_set_iam_policy_field_headers_async():
 
     # Mock the actual call within the gRPC stub, and fake the request.
     with mock.patch.object(
-        type(client._client.transport.set_iam_policy), "__call__"
+        type(client.transport.set_iam_policy), "__call__"
     ) as call:
         call.return_value = grpc_helpers_async.FakeUnaryUnaryCall(policy.Policy())
 
@@ -1512,6 +1512,27 @@ def test_set_iam_policy_from_dict():
         call.assert_called()
 
 
+@pytest.mark.asyncio
+async def test_set_iam_policy_from_dict_async():
+    client = {{ service.async_client_name }}(
+        credentials=credentials.AnonymousCredentials(),
+    )
+    # Mock the actual call within the gRPC stub, and fake the request.
+    with mock.patch.object(type(client.transport.set_iam_policy), "__call__") as call:
+        # Designate an appropriate return value for the call.
+        call.return_value = grpc_helpers_async.FakeUnaryUnaryCall(
+            policy.Policy()
+        )
+
+        response = await client.set_iam_policy(
+            request={
+                "resource": "resource_value",
+                "policy": policy.Policy(version=774),
+            }
+        )
+        call.assert_called()
+
+
 def test_get_iam_policy(transport: str = "grpc"):
     client = {{ service.client_name }}(
         credentials=credentials.AnonymousCredentials(), transport=transport,
@@ -1554,7 +1575,7 @@ async def test_get_iam_policy_async(transport: str = "grpc_asyncio"):
 
     # Mock the actual call within the gRPC stub, and fake the request.
     with mock.patch.object(
-        type(client._client.transport.get_iam_policy), "__call__"
+        type(client.transport.get_iam_policy), "__call__"
     ) as call:
         # Designate an appropriate return value for the call.
         call.return_value = grpc_helpers_async.FakeUnaryUnaryCall(
@@ -1616,7 +1637,7 @@ async def test_get_iam_policy_field_headers_async():
 
     # Mock the actual call within the gRPC stub, and fake the request.
     with mock.patch.object(
-        type(client._client.transport.get_iam_policy), "__call__"
+        type(client.transport.get_iam_policy), "__call__"
     ) as call:
         call.return_value = grpc_helpers_async.FakeUnaryUnaryCall(policy.Policy())
 
@@ -1649,6 +1670,26 @@ def test_get_iam_policy_from_dict():
         )
         call.assert_called()
 
+@pytest.mark.asyncio
+async def test_get_iam_policy_from_dict_async():
+    client = {{ service.async_client_name }}(
+        credentials=credentials.AnonymousCredentials(),
+    )
+    # Mock the actual call within the gRPC stub, and fake the request.
+    with mock.patch.object(type(client.transport.get_iam_policy), "__call__") as call:
+        # Designate an appropriate return value for the call.
+        call.return_value = grpc_helpers_async.FakeUnaryUnaryCall(
+            policy.Policy()
+        )
+
+        response = await client.get_iam_policy(
+            request={
+                "resource": "resource_value",
+                "options": options.GetPolicyOptions(requested_policy_version=2598),
+            }
+        )
+        call.assert_called()
+
 
 def test_test_iam_permissions(transport: str = "grpc"):
     client = {{ service.client_name }}(
@@ -1694,7 +1735,7 @@ async def test_test_iam_permissions_async(transport: str = "grpc_asyncio"):
 
     # Mock the actual call within the gRPC stub, and fake the request.
     with mock.patch.object(
-        type(client._client.transport.test_iam_permissions), "__call__"
+        type(client.transport.test_iam_permissions), "__call__"
     ) as call:
         # Designate an appropriate return value for the call.
         call.return_value = grpc_helpers_async.FakeUnaryUnaryCall(
@@ -1756,7 +1797,7 @@ async def test_test_iam_permissions_field_headers_async():
 
     # Mock the actual call within the gRPC stub, and fake the request.
     with mock.patch.object(
-        type(client._client.transport.test_iam_permissions), "__call__"
+        type(client.transport.test_iam_permissions), "__call__"
     ) as call:
         call.return_value = grpc_helpers_async.FakeUnaryUnaryCall(
             iam_policy.TestIamPermissionsResponse()
@@ -1792,6 +1833,29 @@ def test_test_iam_permissions_from_dict():
             }
         )
         call.assert_called()
+
+@pytest.mark.asyncio
+async def test_test_iam_permissions_from_dict_async():
+    client = {{ service.async_client_name }}(
+        credentials=credentials.AnonymousCredentials(),
+    )
+    # Mock the actual call within the gRPC stub, and fake the request.
+    with mock.patch.object(
+        type(client.transport.test_iam_permissions), "__call__"
+    ) as call:
+        # Designate an appropriate return value for the call.
+        call.return_value = grpc_helpers_async.FakeUnaryUnaryCall(
+            iam_policy.TestIamPermissionsResponse()
+        )
+
+        response = await client.test_iam_permissions(
+            request={
+                "resource": "resource_value",
+                "permissions": ["permissions_value"],
+            }
+        )
+        call.assert_called()
+
 {% endif %}
 
 {% endblock %}
diff --git a/noxfile.py b/noxfile.py
index 47880e3d1c..ba283480e1 100644
--- a/noxfile.py
+++ b/noxfile.py
@@ -87,16 +87,20 @@ def showcase_library(
 
         # Write out a client library for Showcase.
         template_opt = f"python-gapic-templates={templates}"
-        opts = f"--python_gapic_opt={template_opt}"
-        opts += ",".join(other_opts + ("lazy-import",))
-        session.run(
-            "protoc",
-            "--experimental_allow_proto3_optional",
+        opts = "--python_gapic_opt="
+        opts += ",".join(other_opts + (f"{template_opt}",))
+        cmd_tup = (
+            f"protoc",
+            f"--experimental_allow_proto3_optional",
             f"--descriptor_set_in={tmp_dir}{path.sep}showcase.desc",
+            opts,
             f"--python_gapic_out={tmp_dir}",
-            "google/showcase/v1beta1/echo.proto",
-            "google/showcase/v1beta1/identity.proto",
-            "google/showcase/v1beta1/messaging.proto",
+            f"google/showcase/v1beta1/echo.proto",
+            f"google/showcase/v1beta1/identity.proto",
+            f"google/showcase/v1beta1/messaging.proto",
+        )
+        session.run(
+            *cmd_tup,
             external=True,
         )
 
@@ -108,20 +112,27 @@ def showcase_library(
 
 @nox.session(python="3.8")
 def showcase(
-    session, templates="DEFAULT", other_opts: typing.Iterable[str] = (),
+    session,
+    templates="DEFAULT",
+    other_opts: typing.Iterable[str] = (),
+    env: typing.Optional[typing.Dict[str, str]] = None,
 ):
     """Run the Showcase test suite."""
 
     with showcase_library(session, templates=templates, other_opts=other_opts):
         session.install("mock", "pytest", "pytest-asyncio")
         session.run(
-            "py.test", "--quiet", *(session.posargs or [path.join("tests", "system")])
+            "py.test", "--quiet", *(session.posargs or [path.join("tests", "system")]),
+            env=env,
         )
 
 
 @nox.session(python="3.8")
 def showcase_mtls(
-    session, templates="DEFAULT", other_opts: typing.Iterable[str] = (),
+    session,
+    templates="DEFAULT",
+    other_opts: typing.Iterable[str] = (),
+    env: typing.Optional[typing.Dict[str, str]] = None,
 ):
     """Run the Showcase mtls test suite."""
 
@@ -132,19 +143,30 @@ def showcase_mtls(
             "--quiet",
             "--mtls",
             *(session.posargs or [path.join("tests", "system")]),
+            env=env,
         )
 
 
 @nox.session(python="3.8")
 def showcase_alternative_templates(session):
     templates = path.join(path.dirname(__file__), "gapic", "ads-templates")
-    showcase(session, templates=templates, other_opts=("old-naming",))
+    showcase(
+        session,
+        templates=templates,
+        other_opts=("old-naming",),
+        env={"GAPIC_PYTHON_ASYNC": "False"},
+    )
 
 
 @nox.session(python="3.8")
 def showcase_mtls_alternative_templates(session):
     templates = path.join(path.dirname(__file__), "gapic", "ads-templates")
-    showcase_mtls(session, templates=templates, other_opts=("old-naming",))
+    showcase_mtls(
+        session,
+        templates=templates,
+        other_opts=("old-naming",),
+        env={"GAPIC_PYTHON_ASYNC": "False"},
+    )
 
 
 @nox.session(python=["3.6", "3.7", "3.8"])
@@ -176,7 +198,7 @@ def showcase_unit(
         )
 
 
-@nox.session(python=["3.6", "3.7", "3.8"])
+@nox.session(python=["3.7", "3.8"])
 def showcase_unit_alternative_templates(session):
     showcase_unit(session, templates=ADS_TEMPLATES, other_opts=("old-naming",))
 
diff --git a/tests/system/conftest.py b/tests/system/conftest.py
index db3742aa04..e01f596b2d 100644
--- a/tests/system/conftest.py
+++ b/tests/system/conftest.py
@@ -13,32 +13,53 @@
 # limitations under the License.
 
 import collections
+import distutils
+import grpc
 import mock
 import os
 import pytest
-import asyncio
 
 import google.api_core.client_options as ClientOptions
 from google.auth import credentials
-from google.showcase import EchoClient, EchoAsyncClient
-from google.showcase import IdentityClient, IdentityAsyncClient
+from google.showcase import EchoClient
+from google.showcase import IdentityClient
 from google.showcase import MessagingClient
 
-import grpc
-from grpc.experimental import aio
+if distutils.util.strtobool(os.environ.get("GAPIC_PYTHON_ASYNC", "true")):
+    from grpc.experimental import aio
+    import asyncio
+    from google.showcase import EchoAsyncClient
+    from google.showcase import IdentityAsyncClient
+
+    @pytest.fixture
+    def async_echo(use_mtls, event_loop):
+        return construct_client(
+            EchoAsyncClient,
+            use_mtls,
+            transport="grpc_asyncio",
+            channel_creator=aio.insecure_channel
+        )
 
-_test_event_loop = asyncio.new_event_loop()
+    @pytest.fixture
+    def async_identity(use_mtls, event_loop):
+        return construct_client(
+            IdentityAsyncClient,
+            use_mtls,
+            transport="grpc_asyncio",
+            channel_creator=aio.insecure_channel
+        )
 
-# NOTE(lidiz) We must override the default event_loop fixture from
-# pytest-asyncio. pytest fixture frees resources once there isn't any reference
-# to it. So, the event loop might close before tests finishes. In the
-# customized version, we don't close the event loop.
+    _test_event_loop = asyncio.new_event_loop()
 
+    # NOTE(lidiz) We must override the default event_loop fixture from
+    # pytest-asyncio. pytest fixture frees resources once there isn't any reference
+    # to it. So, the event loop might close before tests finishes. In the
+    # customized version, we don't close the event loop.
 
-@pytest.fixture
-def event_loop():
-    asyncio.set_event_loop(_test_event_loop)
-    return asyncio.get_event_loop()
+    @pytest.fixture
+    def event_loop():
+        asyncio.set_event_loop(_test_event_loop)
+        return asyncio.get_event_loop()
 
 
 dir = os.path.dirname(__file__)
@@ -99,16 +120,6 @@ def echo(use_mtls):
     return construct_client(EchoClient, use_mtls)
 
 
-@pytest.fixture
-def async_echo(use_mtls, event_loop):
-    return construct_client(
-        EchoAsyncClient,
-        use_mtls,
-        transport="grpc_asyncio",
-        channel_creator=aio.insecure_channel
-    )
-
-
 @pytest.fixture
 def identity():
     transport = IdentityClient.get_transport_class('grpc')(
@@ -117,16 +128,6 @@ def identity():
     return IdentityClient(transport=transport)
 
 
-@pytest.fixture
-def async_identity(use_mtls, event_loop):
-    return construct_client(
-        IdentityAsyncClient,
-        use_mtls,
-        transport="grpc_asyncio",
-        channel_creator=aio.insecure_channel
-    )
-
-
 @pytest.fixture
 def identity(use_mtls):
     return construct_client(IdentityClient, use_mtls)
diff --git a/tests/system/test_grpc_lro.py b/tests/system/test_grpc_lro.py
index a4578a168a..6dbd3af396 100644
--- a/tests/system/test_grpc_lro.py
+++ b/tests/system/test_grpc_lro.py
@@ -12,10 +12,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import distutils
+import os
 import pytest
 from datetime import datetime, timedelta, timezone
 
-from google import showcase_v1beta1
+from google import showcase
 
 
 def test_lro(echo):
@@ -26,18 +28,20 @@ def test_lro(echo):
         }}
     )
     response = future.result()
-    assert isinstance(response, showcase_v1beta1.WaitResponse)
+    assert isinstance(response, showcase.WaitResponse)
     assert response.content.endswith('the snails...eventually.')
 
 
-@pytest.mark.asyncio
-async def test_lro_async(async_echo):
-    future = await async_echo.wait({
-        'end_time': datetime.now(tz=timezone.utc) + timedelta(seconds=1),
-        'success': {
-            'content': 'The hail in Wales falls mainly on the snails...eventually.'
-        }}
-    )
-    response = await future.result()
-    assert isinstance(response, showcase_v1beta1.WaitResponse)
-    assert response.content.endswith('the snails...eventually.')
+if distutils.util.strtobool(os.environ.get("GAPIC_PYTHON_ASYNC", "true")):
+
+    @pytest.mark.asyncio
+    async def test_lro_async(async_echo):
+        future = await async_echo.wait({
+            'end_time': datetime.now(tz=timezone.utc) + timedelta(seconds=1),
+            'success': {
+                'content': 'The hail in Wales falls mainly on the snails...eventually.'
+            }}
+        )
+        response = await future.result()
+        assert isinstance(response, showcase.WaitResponse)
+        assert response.content.endswith('the snails...eventually.')
diff --git a/tests/system/test_grpc_streams.py b/tests/system/test_grpc_streams.py
index f77e819986..f9da07948c 100644
--- a/tests/system/test_grpc_streams.py
+++ b/tests/system/test_grpc_streams.py
@@ -12,9 +12,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import distutils
 import logging
+import os
 import pytest
-import asyncio
 import threading
 from google import showcase
 
@@ -77,129 +78,124 @@ def test_stream_stream_passing_dict(echo):
     assert responses.trailing_metadata() == metadata
 
 
-@pytest.mark.asyncio
-async def test_async_unary_stream_reader(async_echo):
-    content = 'The hail in Wales falls mainly on the snails.'
-    call = await async_echo.expand({
-        'content': content,
-    }, metadata=metadata)
-
-    # Consume the response and ensure it matches what we expect.
-    # with pytest.raises(exceptions.NotFound) as exc:
-    for ground_truth in content.split(' '):
-        response = await call.read()
-        assert response.content == ground_truth
-    assert ground_truth == 'snails.'
-
-    trailing_metadata = await call.trailing_metadata()
-    assert trailing_metadata == metadata
-
-
-@pytest.mark.asyncio
-async def test_async_unary_stream_async_generator(async_echo):
-    content = 'The hail in Wales falls mainly on the snails.'
-    call = await async_echo.expand({
-        'content': content,
-    }, metadata=metadata)
-
-    # Consume the response and ensure it matches what we expect.
-    # with pytest.raises(exceptions.NotFound) as exc:
-    tokens = iter(content.split(' '))
-    async for response in call:
-        ground_truth = next(tokens)
-        assert response.content == ground_truth
-    assert ground_truth == 'snails.'
-
-    trailing_metadata = await call.trailing_metadata()
-    assert trailing_metadata == metadata
-
-
-@pytest.mark.asyncio
-async def test_async_stream_unary_iterable(async_echo):
-    requests = []
-    requests.append(showcase.EchoRequest(content="hello"))
-    requests.append(showcase.EchoRequest(content="world!"))
-
-    call = await async_echo.collect(requests)
-    response = await call
-    assert response.content == 'hello world!'
-
-
-@pytest.mark.asyncio
-async def test_async_stream_unary_async_generator(async_echo):
-
-    async def async_generator():
-        yield showcase.EchoRequest(content="hello")
-        yield showcase.EchoRequest(content="world!")
-
-    call = await async_echo.collect(async_generator())
-    response = await call
-    assert response.content == 'hello world!'
-
-
-@pytest.mark.asyncio
-async def test_async_stream_unary_writer(async_echo):
-    call = await async_echo.collect()
-    await call.write(showcase.EchoRequest(content="hello"))
-    await call.write(showcase.EchoRequest(content="world!"))
-    await call.done_writing()
-
-    response = await call
-    assert response.content == 'hello world!'
-
-
-@pytest.mark.asyncio
-async def test_async_stream_unary_passing_dict(async_echo):
-    requests = [{'content': 'hello'}, {'content': 'world!'}]
-    call = await async_echo.collect(iter(requests))
-    response = await call
-    assert response.content == 'hello world!'
-
-
-@pytest.mark.asyncio
-async def test_async_stream_stream_reader_writier(async_echo):
-    call = await async_echo.chat(metadata=metadata)
-    await call.write(showcase.EchoRequest(content="hello"))
-    await call.write(showcase.EchoRequest(content="world!"))
-    await call.done_writing()
-
-    contents = [
-        (await call.read()).content,
-        (await call.read()).content
-    ]
-    assert contents == ['hello', 'world!']
-
-    trailing_metadata = await call.trailing_metadata()
-    assert trailing_metadata == metadata
-
-
-@pytest.mark.asyncio
-async def test_async_stream_stream_async_generator(async_echo):
-
-    async def async_generator():
-        yield showcase.EchoRequest(content="hello")
-        yield showcase.EchoRequest(content="world!")
-
-    call = await async_echo.chat(async_generator(), metadata=metadata)
-
-    contents = []
-    async for response in call:
-        contents.append(response.content)
-    assert contents == ['hello', 'world!']
-
-    trailing_metadata = await call.trailing_metadata()
-    assert trailing_metadata == metadata
-
-
-@pytest.mark.asyncio
-async def test_async_stream_stream_passing_dict(async_echo):
-    requests = [{'content': 'hello'}, {'content': 'world!'}]
-    call = await async_echo.chat(iter(requests), metadata=metadata)
-
-    contents = []
-    async for response in call:
-        contents.append(response.content)
-    assert contents == ['hello', 'world!']
-
-    trailing_metadata = await call.trailing_metadata()
-    assert trailing_metadata == metadata
+if distutils.util.strtobool(os.environ.get("GAPIC_PYTHON_ASYNC", "true")):
+    import asyncio
+
+    @pytest.mark.asyncio
+    async def test_async_unary_stream_reader(async_echo):
+        content = 'The hail in Wales falls mainly on the snails.'
+        call = await async_echo.expand({
+            'content': content,
+        }, metadata=metadata)
+
+        # Consume the response and ensure it matches what we expect.
+        # with pytest.raises(exceptions.NotFound) as exc:
+        for ground_truth in content.split(' '):
+            response = await call.read()
+            assert response.content == ground_truth
+        assert ground_truth == 'snails.'
+
+        trailing_metadata = await call.trailing_metadata()
+        assert trailing_metadata == metadata
+
+    @pytest.mark.asyncio
+    async def test_async_unary_stream_async_generator(async_echo):
+        content = 'The hail in Wales falls mainly on the snails.'
+        call = await async_echo.expand({
+            'content': content,
+        }, metadata=metadata)
+
+        # Consume the response and ensure it matches what we expect.
+        # with pytest.raises(exceptions.NotFound) as exc:
+        tokens = iter(content.split(' '))
+        async for response in call:
+            ground_truth = next(tokens)
+            assert response.content == ground_truth
+        assert ground_truth == 'snails.'
+
+        trailing_metadata = await call.trailing_metadata()
+        assert trailing_metadata == metadata
+
+    @pytest.mark.asyncio
+    async def test_async_stream_unary_iterable(async_echo):
+        requests = []
+        requests.append(showcase.EchoRequest(content="hello"))
+        requests.append(showcase.EchoRequest(content="world!"))
+
+        call = await async_echo.collect(requests)
+        response = await call
+        assert response.content == 'hello world!'
+
+    @pytest.mark.asyncio
+    async def test_async_stream_unary_async_generator(async_echo):
+
+        async def async_generator():
+            yield showcase.EchoRequest(content="hello")
+            yield showcase.EchoRequest(content="world!")
+
+        call = await async_echo.collect(async_generator())
+        response = await call
+        assert response.content == 'hello world!'
+
+    @pytest.mark.asyncio
+    async def test_async_stream_unary_writer(async_echo):
+        call = await async_echo.collect()
+        await call.write(showcase.EchoRequest(content="hello"))
+        await call.write(showcase.EchoRequest(content="world!"))
+        await call.done_writing()
+
+        response = await call
+        assert response.content == 'hello world!'
+
+    @pytest.mark.asyncio
+    async def test_async_stream_unary_passing_dict(async_echo):
+        requests = [{'content': 'hello'}, {'content': 'world!'}]
+        call = await async_echo.collect(iter(requests))
+        response = await call
+        assert response.content == 'hello world!'
+
+    @pytest.mark.asyncio
+    async def test_async_stream_stream_reader_writier(async_echo):
+        call = await async_echo.chat(metadata=metadata)
+        await call.write(showcase.EchoRequest(content="hello"))
+        await call.write(showcase.EchoRequest(content="world!"))
+        await call.done_writing()
+
+        contents = [
+            (await call.read()).content,
+            (await call.read()).content
+        ]
+        assert contents == ['hello', 'world!']
+
+        trailing_metadata = await call.trailing_metadata()
+        assert trailing_metadata == metadata
+
+    @pytest.mark.asyncio
+    async def test_async_stream_stream_async_generator(async_echo):
+
+        async def async_generator():
+            yield showcase.EchoRequest(content="hello")
+            yield showcase.EchoRequest(content="world!")
+
+        call = await async_echo.chat(async_generator(), metadata=metadata)
+
+        contents = []
+        async for response in call:
+            contents.append(response.content)
+        assert contents == ['hello', 'world!']
+
+        trailing_metadata = await call.trailing_metadata()
+        assert trailing_metadata == metadata
+
+    @pytest.mark.asyncio
+    async def test_async_stream_stream_passing_dict(async_echo):
+        requests = [{'content': 'hello'}, {'content': 'world!'}]
+        call = await async_echo.chat(iter(requests), metadata=metadata)
+
+        contents = []
+        async for response in call:
+            contents.append(response.content)
+        assert contents == ['hello', 'world!']
+
+        trailing_metadata = await call.trailing_metadata()
+        assert trailing_metadata == metadata
diff --git a/tests/system/test_grpc_unary.py b/tests/system/test_grpc_unary.py
index c1694d975e..f8ca2f6527 100644
--- a/tests/system/test_grpc_unary.py
+++ b/tests/system/test_grpc_unary.py
@@ -12,8 +12,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import distutils
+import os
 import pytest
-import asyncio
 
 from google.api_core import exceptions
 from google.rpc import code_pb2
@@ -48,31 +49,32 @@ def test_unary_error(echo):
         assert exc.value.message == message
 
 
-@pytest.mark.asyncio
-async def test_async_unary_with_request_object(async_echo):
-    response = await async_echo.echo(showcase.EchoRequest(
-        content='The hail in Wales falls mainly on the snails.',
-    ), timeout=1)
-    assert response.content == 'The hail in Wales falls mainly on the snails.'
-
-
-@pytest.mark.asyncio
-async def test_async_unary_with_dict(async_echo):
-    response = await async_echo.echo({
-        'content': 'The hail in Wales falls mainly on the snails.',
-    })
-    assert response.content == 'The hail in Wales falls mainly on the snails.'
+if distutils.util.strtobool(os.environ.get("GAPIC_PYTHON_ASYNC", "true")):
+    import asyncio
 
+    @pytest.mark.asyncio
+    async def test_async_unary_with_request_object(async_echo):
+        response = await async_echo.echo(showcase.EchoRequest(
+            content='The hail in Wales falls mainly on the snails.',
+        ), timeout=1)
+        assert response.content == 'The hail in Wales falls mainly on the snails.'
 
-@pytest.mark.asyncio
-async def test_async_unary_error(async_echo):
-    message = 'Bad things! Bad things!'
-    with pytest.raises(exceptions.InvalidArgument) as exc:
-        await async_echo.echo({
-            'error': {
-                'code': code_pb2.Code.Value('INVALID_ARGUMENT'),
-                'message': message,
-            },
+    @pytest.mark.asyncio
+    async def test_async_unary_with_dict(async_echo):
+        response = await async_echo.echo({
+            'content': 'The hail in Wales falls mainly on the snails.',
         })
-        assert exc.value.code == 400
-        assert exc.value.message == message
+        assert response.content == 'The hail in Wales falls mainly on the snails.'
+
+    @pytest.mark.asyncio
+    async def test_async_unary_error(async_echo):
+        message = 'Bad things! Bad things!'
+        with pytest.raises(exceptions.InvalidArgument) as exc:
+            await async_echo.echo({
+                'error': {
+                    'code': code_pb2.Code.Value('INVALID_ARGUMENT'),
+                    'message': message,
+                },
+            })
+            assert exc.value.code == 400
+            assert exc.value.message == message
diff --git a/tests/system/test_pagination.py b/tests/system/test_pagination.py
index 8f53a6c01d..eb195ea128 100644
--- a/tests/system/test_pagination.py
+++ b/tests/system/test_pagination.py
@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import distutils
+import os
 import pytest
 from google import showcase
 
@@ -47,39 +49,39 @@ def test_pagination_pages(echo):
                        for i in text.split(' ')]
 
 
-@pytest.mark.asyncio
-async def test_pagination_async(async_echo):
-    text = 'The hail in Wales falls mainly on the snails.'
-    results = []
-    async for i in await async_echo.paged_expand({
-        'content': text,
-        'page_size': 3,
-    }):
-        results.append(i)
-
-    assert len(results) == 9
-    assert results == [showcase.EchoResponse(content=i)
-                       for i in text.split(' ')]
-
-
-@pytest.mark.asyncio
-async def test_pagination_pages_async(async_echo):
-    text = "The hail in Wales falls mainly on the snails."
-    page_results = []
-    async for page in (await async_echo.paged_expand({
-        'content': text,
-        'page_size': 3,
-    })).pages:
-        page_results.append(page)
-
-    assert len(page_results) == 3
-    assert not page_results[-1].next_page_token
-
-    # The monolithic surface uses a wrapper type that needs an explicit property
-    # for a 'raw_page': we need to duplicate that interface, even though the
-    # architecture is different.
-    assert page_results[0].raw_page is page_results[0]
-
-    results = [r for p in page_results for r in p.responses]
-    assert results == [showcase.EchoResponse(content=i)
-                       for i in text.split(' ')]
+if distutils.util.strtobool(os.environ.get("GAPIC_PYTHON_ASYNC", "true")):
+    @pytest.mark.asyncio
+    async def test_pagination_async(async_echo):
+        text = 'The hail in Wales falls mainly on the snails.'
+        results = []
+        async for i in await async_echo.paged_expand({
+            'content': text,
+            'page_size': 3,
+        }):
+            results.append(i)
+
+        assert len(results) == 9
+        assert results == [showcase.EchoResponse(content=i)
+                           for i in text.split(' ')]
+
+    @pytest.mark.asyncio
+    async def test_pagination_pages_async(async_echo):
+        text = "The hail in Wales falls mainly on the snails."
+        page_results = []
+        async for page in (await async_echo.paged_expand({
+            'content': text,
+            'page_size': 3,
+        })).pages:
+            page_results.append(page)
+
+        assert len(page_results) == 3
+        assert not page_results[-1].next_page_token
+
+        # The monolithic surface uses a wrapper type that needs an explicit property
+        # for a 'raw_page': we need to duplicate that interface, even though the
+        # architecture is different.
+        assert page_results[0].raw_page is page_results[0]
+
+        results = [r for p in page_results for r in p.responses]
+        assert results == [showcase.EchoResponse(content=i)
+                           for i in text.split(' ')]
diff --git a/tests/system/test_resource_crud.py b/tests/system/test_resource_crud.py
index 5372da4b6b..85bafe561a 100644
--- a/tests/system/test_resource_crud.py
+++ b/tests/system/test_resource_crud.py
@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import distutils
+import os
 import pytest
 
 
@@ -76,43 +78,43 @@ def test_path_parsing(messaging):
     assert expected == actual
 
 
-@pytest.mark.asyncio
-async def test_crud_with_request_async(async_identity):
-    pager = await async_identity.list_users()
-    count = len(pager.users)
-    user = await async_identity.create_user(request={'user': {
-        'display_name': 'Guido van Rossum',
-        'email': 'guido@guido.fake',
-    }})
-    try:
-        assert user.display_name == 'Guido van Rossum'
-        assert user.email == 'guido@guido.fake'
-        pager = (await async_identity.list_users())
-        assert len(pager.users) == count + 1
-        assert (await async_identity.get_user({
-            'name': user.name
-        })).display_name == 'Guido van Rossum'
-    finally:
-        await async_identity.delete_user({'name': user.name})
-
-
-@pytest.mark.asyncio
-async def test_crud_flattened_async(async_identity):
-    count = len((await async_identity.list_users()).users)
-    user = await async_identity.create_user(
-        display_name='Monty Python',
-        email='monty@python.org',
-    )
-    try:
-        assert user.display_name == 'Monty Python'
-        assert user.email == 'monty@python.org'
-        assert len((await async_identity.list_users()).users) == count + 1
-        assert (await async_identity.get_user(name=user.name)).display_name == 'Monty Python'
-    finally:
-        await async_identity.delete_user(name=user.name)
-
-
-def test_path_methods_async(async_identity):
-    expected = "users/bdfl"
-    actual = async_identity.user_path("bdfl")
-    assert expected == actual
+if distutils.util.strtobool(os.environ.get("GAPIC_PYTHON_ASYNC", "true")):
+
+    @pytest.mark.asyncio
+    async def test_crud_with_request_async(async_identity):
+        pager = await async_identity.list_users()
+        count = len(pager.users)
+        user = await async_identity.create_user(request={'user': {
+            'display_name': 'Guido van Rossum',
+            'email': 'guido@guido.fake',
+        }})
+        try:
+            assert user.display_name == 'Guido van Rossum'
+            assert user.email == 'guido@guido.fake'
+            pager = (await async_identity.list_users())
+            assert len(pager.users) == count + 1
+            assert (await async_identity.get_user({
+                'name': user.name
+            })).display_name == 'Guido van Rossum'
+        finally:
+            await async_identity.delete_user({'name': user.name})
+
+    @pytest.mark.asyncio
+    async def test_crud_flattened_async(async_identity):
+        count = len((await async_identity.list_users()).users)
+        user = await async_identity.create_user(
+            display_name='Monty Python',
+            email='monty@python.org',
+        )
+        try:
+            assert user.display_name == 'Monty Python'
+            assert user.email == 'monty@python.org'
+            assert len((await async_identity.list_users()).users) == count + 1
+            assert (await async_identity.get_user(name=user.name)).display_name == 'Monty Python'
+        finally:
+            await async_identity.delete_user(name=user.name)
+
+    def test_path_methods_async(async_identity):
+        expected = "users/bdfl"
+        actual = async_identity.user_path("bdfl")
+        assert expected == actual
diff --git a/tests/system/test_retry.py b/tests/system/test_retry.py
index 0bc70f9f8e..97e0c60beb 100644
--- a/tests/system/test_retry.py
+++ b/tests/system/test_retry.py
@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import distutils
+import os
 import pytest
 
 from google.api_core import exceptions
@@ -28,12 +30,14 @@ def test_retry_bubble(echo):
         })
 
 
-@pytest.mark.asyncio
-async def test_retry_bubble_async(async_echo):
-    with pytest.raises(exceptions.DeadlineExceeded):
-        await async_echo.echo({
-            'error': {
-                'code': code_pb2.Code.Value('DEADLINE_EXCEEDED'),
-                'message': 'This took longer than you said it should.',
-            },
-        })
+if distutils.util.strtobool(os.environ.get("GAPIC_PYTHON_ASYNC", "true")):
+
+    @pytest.mark.asyncio
+    async def test_retry_bubble_async(async_echo):
+        with pytest.raises(exceptions.DeadlineExceeded):
+            await async_echo.echo({
+                'error': {
+                    'code': code_pb2.Code.Value('DEADLINE_EXCEEDED'),
+                    'message': 'This took longer than you said it should.',
+                },
+            })