diff --git a/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/client.py.j2 b/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/client.py.j2
index 90e5002b68..3d14c341a5 100644
--- a/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/client.py.j2
+++ b/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/client.py.j2
@@ -148,6 +148,18 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
         """
         return self._transport
 
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        """Releases underlying transport's resources.
+
+        .. warning::
+            ONLY use as a context manager if the transport is NOT shared
+            with other clients! Exiting the with block will CLOSE the transport
+            and may cause errors in other clients!
+        """
+        self.transport.close()
 
     {% for message in service.resource_messages|sort(attribute="resource_type") %}
     @staticmethod
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 82ab146438..a87e3b6717 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
@@ -103,6 +103,14 @@ class {{ service.name }}Transport(metaclass=abc.ABCMeta):
             {% endfor %} {# precomputed wrappers loop #}
         }
 
+    def close(self):
+        """Closes resources associated with the transport.
+
+        .. warning::
+            Only call this method if the transport is NOT shared
+            with other clients - this may cause errors in other clients!
+        """
+        raise NotImplementedError()
 
     {% if service.has_lro %}
 
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 1562b6f88c..59a2219ca9 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
@@ -188,6 +188,9 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
             **kwargs
         )
 
+    def close(self):
+        self.grpc_channel.close()
+
     @property
     def grpc_channel(self) -> grpc.Channel:
         """Return the channel designed to connect to this service.
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 8d9af17ae6..0392324b06 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
@@ -654,6 +654,9 @@ def test_{{ service.name|snake_case }}_base_transport():
         with pytest.raises(NotImplementedError):
             getattr(transport, method)(request=object())
 
+    with pytest.raises(NotImplementedError):
+        transport.close()
+
     {% if service.has_lro %}
     # Additionally, the LRO client (a property) should
     # also raise NotImplementedError
@@ -903,5 +906,26 @@ def test_client_withDEFAULT_CLIENT_INFO():
         )
         prep.assert_called_once_with(client_info)
 
+def test_grpc_transport_close():
+    client = {{ service.client_name }}(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport='grpc',
+    )
+    with mock.patch.object(type(client.transport._grpc_channel), 'close') as chan_close:
+        with client as _:
+            chan_close.assert_not_called()
+        chan_close.assert_called_once()
+
+def test_grpc_client_ctx():
+    client = {{ service.client_name }}(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport='grpc',
+    )
+    # Test client calls underlying transport.
+    with mock.patch.object(type(client.transport), "close") as close:
+        close.assert_not_called()
+        with client as _:
+            pass
+        close.assert_called()
 
 {% endblock %}
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 ea40ea1c25..3df81e9b05 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
@@ -599,6 +599,12 @@ class {{ service.async_client_name }}:
         return response
     {% endif %}
 
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        await self.transport.close()
+
 try:
     DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
         gapic_version=pkg_resources.get_distribution(
diff --git a/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2 b/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2
index 3bc7f20c5f..f0248d53a5 100644
--- a/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2
+++ b/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2
@@ -477,6 +477,19 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
         {{ "\n" }}
     {% endfor %}
 
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        """Releases underlying transport's resources.
+
+        .. warning::
+            ONLY use as a context manager if the transport is NOT shared
+            with other clients! Exiting the with block will CLOSE the transport
+            and may cause errors in other clients!
+        """
+        self.transport.close()
+
     {% if opts.add_iam_methods %}
     def set_iam_policy(
         self,
diff --git a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/base.py.j2 b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/base.py.j2
index b5b7bb2ac4..fb937ae192 100644
--- a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/base.py.j2
+++ b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/base.py.j2
@@ -174,6 +174,14 @@ class {{ service.name }}Transport(abc.ABC):
             {% endfor %} {# precomputed wrappers loop #}
         }
 
+    def close(self):
+        """Closes resources associated with the transport.
+
+       .. warning::
+            Only call this method if the transport is NOT shared
+            with other clients - this may cause errors in other clients!
+        """
+        raise NotImplementedError()
 
     {% if service.has_lro %}
 
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 f2f0da6ea3..894f4470d5 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
@@ -224,6 +224,7 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
         """Return the channel designed to connect to this service.
         """
         return self._grpc_channel
+
     {% if service.has_lro %}
 
     @property
@@ -355,6 +356,9 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
         return self._stubs["test_iam_permissions"]
     {% endif %}
 
+    def close(self):
+        self.grpc_channel.close()
+
 __all__ = (
     '{{ service.name }}GrpcTransport',
 )
diff --git a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc_asyncio.py.j2 b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc_asyncio.py.j2
index 934a931aff..5c4f8d251f 100644
--- a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc_asyncio.py.j2
+++ b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc_asyncio.py.j2
@@ -359,6 +359,10 @@ class {{ service.grpc_asyncio_transport_name }}({{ service.name }}Transport):
         return self._stubs["test_iam_permissions"]
     {% endif %}
 
+    def close(self):
+        return self.grpc_channel.close()
+
+
 __all__ = (
     '{{ service.name }}GrpcAsyncIOTransport',
 )
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 87eef7d497..8072567cb6 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
@@ -283,6 +283,9 @@ class {{service.name}}RestTransport({{service.name}}Transport):
         return self._{{method.name | snake_case}}
     {%- endfor %}
 
+    def close(self):
+        self._session.close()
+
 
 __all__=(
     '{{ service.name }}RestTransport',
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 d76414d34d..165e12a8b4 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
@@ -1516,6 +1516,9 @@ def test_{{ service.name|snake_case }}_base_transport():
         with pytest.raises(NotImplementedError):
             getattr(transport, method)(request=object())
 
+    with pytest.raises(NotImplementedError):
+        transport.close()
+
     {% if service.has_lro %}
     # Additionally, the LRO client (a property) should
     # also raise NotImplementedError
@@ -2487,4 +2490,56 @@ async def test_test_iam_permissions_from_dict_async():
 
 {% endif %}
 
+@pytest.mark.asyncio
+async def test_transport_close_async():
+    client = {{ service.async_client_name }}(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="grpc_asyncio",
+    )
+    with mock.patch.object(type(getattr(client.transport, "grpc_channel")), "close") as close:
+        async with client:
+            close.assert_not_called()
+        close.assert_called_once()
+
+def test_transport_close():
+    transports = {
+        {% if 'rest' in opts.transport %}
+        "rest": "_session",
+        {% endif %}
+        {% if 'grpc' in opts.transport %}
+        "grpc": "_grpc_channel",
+        {% endif %}
+    }
+
+    for transport, close_name in transports.items():
+        client = {{ service.client_name }}(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport=transport
+        )
+        with mock.patch.object(type(getattr(client.transport, close_name)), "close") as close:
+            with client:
+                close.assert_not_called()
+            close.assert_called_once()
+
+def test_client_ctx():
+    transports = [
+        {% if 'rest' in opts.transport %}
+        'rest',
+        {% endif %}
+        {% if 'grpc' in opts.transport %}
+        'grpc',
+        {% endif %}
+    ]
+    for transport in transports:
+        client = {{ service.client_name }}(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport=transport
+        )
+        # Test client calls underlying transport.
+        with mock.patch.object(type(client.transport), "close") as close:
+            close.assert_not_called()
+            with client:
+                pass
+            close.assert_called()
+
 {% endblock %}
diff --git a/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/async_client.py b/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/async_client.py
index 054bfe5565..a6327c26df 100644
--- a/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/async_client.py
+++ b/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/async_client.py
@@ -1292,9 +1292,11 @@ async def analyze_iam_policy_longrunning(self,
         # Done; return the response.
         return response
 
+    async def __aenter__(self):
+        return self
 
-
-
+    async def __aexit__(self, exc_type, exc, tb):
+        await self.transport.close()
 
 try:
     DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
diff --git a/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py b/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py
index 0d7bac0e5e..0462c07e42 100644
--- a/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py
+++ b/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py
@@ -1445,7 +1445,18 @@ def analyze_iam_policy_longrunning(self,
         # Done; return the response.
         return response
 
+    def __enter__(self):
+        return self
 
+    def __exit__(self, type, value, traceback):
+        """Releases underlying transport's resources.
+
+        .. warning::
+            ONLY use as a context manager if the transport is NOT shared
+            with other clients! Exiting the with block will CLOSE the transport
+            and may cause errors in other clients!
+        """
+        self.transport.close()
 
 
 
diff --git a/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/transports/base.py b/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/transports/base.py
index 45083d6687..1da08824f8 100644
--- a/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/transports/base.py
+++ b/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/transports/base.py
@@ -259,6 +259,15 @@ def _prep_wrapped_messages(self, client_info):
             ),
          }
 
+    def close(self):
+        """Closes resources associated with the transport.
+
+       .. warning::
+            Only call this method if the transport is NOT shared
+            with other clients - this may cause errors in other clients!
+        """
+        raise NotImplementedError()
+
     @property
     def operations_client(self) -> operations_v1.OperationsClient:
         """Return the client designed to process long-running operations."""
diff --git a/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/transports/grpc.py b/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/transports/grpc.py
index c95a90bff1..7ed6de3dc2 100644
--- a/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/transports/grpc.py
+++ b/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/transports/grpc.py
@@ -592,6 +592,8 @@ def analyze_iam_policy_longrunning(self) -> Callable[
             )
         return self._stubs['analyze_iam_policy_longrunning']
 
+    def close(self):
+        self.grpc_channel.close()
 
 __all__ = (
     'AssetServiceGrpcTransport',
diff --git a/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/transports/grpc_asyncio.py b/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/transports/grpc_asyncio.py
index 8559391cf0..cc021dfbb0 100644
--- a/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/transports/grpc_asyncio.py
+++ b/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/transports/grpc_asyncio.py
@@ -596,6 +596,9 @@ def analyze_iam_policy_longrunning(self) -> Callable[
             )
         return self._stubs['analyze_iam_policy_longrunning']
 
+    def close(self):
+        return self.grpc_channel.close()
+
 
 __all__ = (
     'AssetServiceGrpcAsyncIOTransport',
diff --git a/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py b/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py
index 21d14df35c..f864b50348 100644
--- a/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py
+++ b/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py
@@ -3551,6 +3551,9 @@ def test_asset_service_base_transport():
         with pytest.raises(NotImplementedError):
             getattr(transport, method)(request=object())
 
+    with pytest.raises(NotImplementedError):
+        transport.close()
+
     # Additionally, the LRO client (a property) should
     # also raise NotImplementedError
     with pytest.raises(NotImplementedError):
@@ -4049,3 +4052,46 @@ def test_client_withDEFAULT_CLIENT_INFO():
             client_info=client_info,
         )
         prep.assert_called_once_with(client_info)
+
+
+@pytest.mark.asyncio
+async def test_transport_close_async():
+    client = AssetServiceAsyncClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="grpc_asyncio",
+    )
+    with mock.patch.object(type(getattr(client.transport, "grpc_channel")), "close") as close:
+        async with client:
+            close.assert_not_called()
+        close.assert_called_once()
+
+def test_transport_close():
+    transports = {
+        "grpc": "_grpc_channel",
+    }
+
+    for transport, close_name in transports.items():
+        client = AssetServiceClient(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport=transport
+        )
+        with mock.patch.object(type(getattr(client.transport, close_name)), "close") as close:
+            with client:
+                close.assert_not_called()
+            close.assert_called_once()
+
+def test_client_ctx():
+    transports = [
+        'grpc',
+    ]
+    for transport in transports:
+        client = AssetServiceClient(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport=transport
+        )
+        # Test client calls underlying transport.
+        with mock.patch.object(type(client.transport), "close") as close:
+            close.assert_not_called()
+            with client:
+                pass
+            close.assert_called()
diff --git a/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/async_client.py b/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/async_client.py
index a6390ec743..c7ae75924a 100644
--- a/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/async_client.py
+++ b/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/async_client.py
@@ -644,9 +644,11 @@ async def sign_jwt(self,
         # Done; return the response.
         return response
 
+    async def __aenter__(self):
+        return self
 
-
-
+    async def __aexit__(self, exc_type, exc, tb):
+        await self.transport.close()
 
 try:
     DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
diff --git a/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py b/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py
index caa32cecbf..f6345dabc9 100644
--- a/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py
+++ b/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py
@@ -804,7 +804,18 @@ def sign_jwt(self,
         # Done; return the response.
         return response
 
+    def __enter__(self):
+        return self
 
+    def __exit__(self, type, value, traceback):
+        """Releases underlying transport's resources.
+
+        .. warning::
+            ONLY use as a context manager if the transport is NOT shared
+            with other clients! Exiting the with block will CLOSE the transport
+            and may cause errors in other clients!
+        """
+        self.transport.close()
 
 
 
diff --git a/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/transports/base.py b/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/transports/base.py
index 99c1477502..5a0c487eb6 100644
--- a/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/transports/base.py
+++ b/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/transports/base.py
@@ -196,6 +196,15 @@ def _prep_wrapped_messages(self, client_info):
             ),
          }
 
+    def close(self):
+        """Closes resources associated with the transport.
+
+       .. warning::
+            Only call this method if the transport is NOT shared
+            with other clients - this may cause errors in other clients!
+        """
+        raise NotImplementedError()
+
     @property
     def generate_access_token(self) -> Callable[
             [common.GenerateAccessTokenRequest],
diff --git a/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/transports/grpc.py b/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/transports/grpc.py
index 95c0f5f975..4c5159d800 100644
--- a/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/transports/grpc.py
+++ b/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/transports/grpc.py
@@ -337,6 +337,8 @@ def sign_jwt(self) -> Callable[
             )
         return self._stubs['sign_jwt']
 
+    def close(self):
+        self.grpc_channel.close()
 
 __all__ = (
     'IAMCredentialsGrpcTransport',
diff --git a/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/transports/grpc_asyncio.py b/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/transports/grpc_asyncio.py
index c944ba08b8..cebb4866b0 100644
--- a/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/transports/grpc_asyncio.py
+++ b/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/transports/grpc_asyncio.py
@@ -341,6 +341,9 @@ def sign_jwt(self) -> Callable[
             )
         return self._stubs['sign_jwt']
 
+    def close(self):
+        return self.grpc_channel.close()
+
 
 __all__ = (
     'IAMCredentialsGrpcAsyncIOTransport',
diff --git a/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_iam_credentials.py b/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_iam_credentials.py
index 6945e557a0..80c276662e 100644
--- a/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_iam_credentials.py
+++ b/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_iam_credentials.py
@@ -1485,6 +1485,9 @@ def test_iam_credentials_base_transport():
         with pytest.raises(NotImplementedError):
             getattr(transport, method)(request=object())
 
+    with pytest.raises(NotImplementedError):
+        transport.close()
+
 
 @requires_google_auth_gte_1_25_0
 def test_iam_credentials_base_transport_with_credentials_file():
@@ -1929,3 +1932,46 @@ def test_client_withDEFAULT_CLIENT_INFO():
             client_info=client_info,
         )
         prep.assert_called_once_with(client_info)
+
+
+@pytest.mark.asyncio
+async def test_transport_close_async():
+    client = IAMCredentialsAsyncClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="grpc_asyncio",
+    )
+    with mock.patch.object(type(getattr(client.transport, "grpc_channel")), "close") as close:
+        async with client:
+            close.assert_not_called()
+        close.assert_called_once()
+
+def test_transport_close():
+    transports = {
+        "grpc": "_grpc_channel",
+    }
+
+    for transport, close_name in transports.items():
+        client = IAMCredentialsClient(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport=transport
+        )
+        with mock.patch.object(type(getattr(client.transport, close_name)), "close") as close:
+            with client:
+                close.assert_not_called()
+            close.assert_called_once()
+
+def test_client_ctx():
+    transports = [
+        'grpc',
+    ]
+    for transport in transports:
+        client = IAMCredentialsClient(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport=transport
+        )
+        # Test client calls underlying transport.
+        with mock.patch.object(type(client.transport), "close") as close:
+            close.assert_not_called()
+            with client:
+                pass
+            close.assert_called()
diff --git a/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/async_client.py b/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/async_client.py
index 82e84aab81..5d322cb502 100644
--- a/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/async_client.py
+++ b/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/async_client.py
@@ -1997,9 +1997,11 @@ async def update_cmek_settings(self,
         # Done; return the response.
         return response
 
+    async def __aenter__(self):
+        return self
 
-
-
+    async def __aexit__(self, exc_type, exc, tb):
+        await self.transport.close()
 
 try:
     DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
diff --git a/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/client.py b/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/client.py
index af174010f6..1deb0ddaff 100644
--- a/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/client.py
+++ b/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/client.py
@@ -2176,7 +2176,18 @@ def update_cmek_settings(self,
         # Done; return the response.
         return response
 
+    def __enter__(self):
+        return self
 
+    def __exit__(self, type, value, traceback):
+        """Releases underlying transport's resources.
+
+        .. warning::
+            ONLY use as a context manager if the transport is NOT shared
+            with other clients! Exiting the with block will CLOSE the transport
+            and may cause errors in other clients!
+        """
+        self.transport.close()
 
 
 
diff --git a/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/transports/base.py b/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/transports/base.py
index e7f0db9d40..92a840f2a3 100644
--- a/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/transports/base.py
+++ b/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/transports/base.py
@@ -323,6 +323,15 @@ def _prep_wrapped_messages(self, client_info):
             ),
          }
 
+    def close(self):
+        """Closes resources associated with the transport.
+
+       .. warning::
+            Only call this method if the transport is NOT shared
+            with other clients - this may cause errors in other clients!
+        """
+        raise NotImplementedError()
+
     @property
     def list_buckets(self) -> Callable[
             [logging_config.ListBucketsRequest],
diff --git a/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/transports/grpc.py b/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/transports/grpc.py
index 94e4af6833..3a640eaf0b 100644
--- a/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/transports/grpc.py
+++ b/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/transports/grpc.py
@@ -872,6 +872,8 @@ def update_cmek_settings(self) -> Callable[
             )
         return self._stubs['update_cmek_settings']
 
+    def close(self):
+        self.grpc_channel.close()
 
 __all__ = (
     'ConfigServiceV2GrpcTransport',
diff --git a/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/transports/grpc_asyncio.py b/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/transports/grpc_asyncio.py
index 78e442d8a4..2f18d861c1 100644
--- a/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/transports/grpc_asyncio.py
+++ b/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/transports/grpc_asyncio.py
@@ -876,6 +876,9 @@ def update_cmek_settings(self) -> Callable[
             )
         return self._stubs['update_cmek_settings']
 
+    def close(self):
+        return self.grpc_channel.close()
+
 
 __all__ = (
     'ConfigServiceV2GrpcAsyncIOTransport',
diff --git a/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/async_client.py b/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/async_client.py
index dd9cbb78dd..f03a8d0c04 100644
--- a/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/async_client.py
+++ b/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/async_client.py
@@ -762,9 +762,11 @@ def tail_log_entries(self,
         # Done; return the response.
         return response
 
+    async def __aenter__(self):
+        return self
 
-
-
+    async def __aexit__(self, exc_type, exc, tb):
+        await self.transport.close()
 
 try:
     DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
diff --git a/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/client.py b/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/client.py
index 4a73c23181..db8650fb87 100644
--- a/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/client.py
+++ b/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/client.py
@@ -898,7 +898,18 @@ def tail_log_entries(self,
         # Done; return the response.
         return response
 
+    def __enter__(self):
+        return self
 
+    def __exit__(self, type, value, traceback):
+        """Releases underlying transport's resources.
+
+        .. warning::
+            ONLY use as a context manager if the transport is NOT shared
+            with other clients! Exiting the with block will CLOSE the transport
+            and may cause errors in other clients!
+        """
+        self.transport.close()
 
 
 
diff --git a/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/transports/base.py b/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/transports/base.py
index 222ed3c1f9..195d115497 100644
--- a/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/transports/base.py
+++ b/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/transports/base.py
@@ -231,6 +231,15 @@ def _prep_wrapped_messages(self, client_info):
             ),
          }
 
+    def close(self):
+        """Closes resources associated with the transport.
+
+       .. warning::
+            Only call this method if the transport is NOT shared
+            with other clients - this may cause errors in other clients!
+        """
+        raise NotImplementedError()
+
     @property
     def delete_log(self) -> Callable[
             [logging.DeleteLogRequest],
diff --git a/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/transports/grpc.py b/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/transports/grpc.py
index f66cb54a21..4ec732a268 100644
--- a/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/transports/grpc.py
+++ b/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/transports/grpc.py
@@ -396,6 +396,8 @@ def tail_log_entries(self) -> Callable[
             )
         return self._stubs['tail_log_entries']
 
+    def close(self):
+        self.grpc_channel.close()
 
 __all__ = (
     'LoggingServiceV2GrpcTransport',
diff --git a/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/transports/grpc_asyncio.py b/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/transports/grpc_asyncio.py
index a19007ab65..9978fe57f7 100644
--- a/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/transports/grpc_asyncio.py
+++ b/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/transports/grpc_asyncio.py
@@ -400,6 +400,9 @@ def tail_log_entries(self) -> Callable[
             )
         return self._stubs['tail_log_entries']
 
+    def close(self):
+        return self.grpc_channel.close()
+
 
 __all__ = (
     'LoggingServiceV2GrpcAsyncIOTransport',
diff --git a/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/async_client.py b/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/async_client.py
index 764f44f666..d3b11a198e 100644
--- a/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/async_client.py
+++ b/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/async_client.py
@@ -621,9 +621,11 @@ async def delete_log_metric(self,
             metadata=metadata,
         )
 
+    async def __aenter__(self):
+        return self
 
-
-
+    async def __aexit__(self, exc_type, exc, tb):
+        await self.transport.close()
 
 try:
     DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
diff --git a/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/client.py b/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/client.py
index 616d0a69e8..e31a068c67 100644
--- a/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/client.py
+++ b/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/client.py
@@ -777,7 +777,18 @@ def delete_log_metric(self,
             metadata=metadata,
         )
 
+    def __enter__(self):
+        return self
 
+    def __exit__(self, type, value, traceback):
+        """Releases underlying transport's resources.
+
+        .. warning::
+            ONLY use as a context manager if the transport is NOT shared
+            with other clients! Exiting the with block will CLOSE the transport
+            and may cause errors in other clients!
+        """
+        self.transport.close()
 
 
 
diff --git a/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/transports/base.py b/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/transports/base.py
index b9170bf568..2a4ff25e1a 100644
--- a/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/transports/base.py
+++ b/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/transports/base.py
@@ -210,6 +210,15 @@ def _prep_wrapped_messages(self, client_info):
             ),
          }
 
+    def close(self):
+        """Closes resources associated with the transport.
+
+       .. warning::
+            Only call this method if the transport is NOT shared
+            with other clients - this may cause errors in other clients!
+        """
+        raise NotImplementedError()
+
     @property
     def list_log_metrics(self) -> Callable[
             [logging_metrics.ListLogMetricsRequest],
diff --git a/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/transports/grpc.py b/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/transports/grpc.py
index e300d9f532..8aced9e6d3 100644
--- a/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/transports/grpc.py
+++ b/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/transports/grpc.py
@@ -351,6 +351,8 @@ def delete_log_metric(self) -> Callable[
             )
         return self._stubs['delete_log_metric']
 
+    def close(self):
+        self.grpc_channel.close()
 
 __all__ = (
     'MetricsServiceV2GrpcTransport',
diff --git a/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/transports/grpc_asyncio.py b/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/transports/grpc_asyncio.py
index 7da832822e..9c07c1172b 100644
--- a/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/transports/grpc_asyncio.py
+++ b/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/transports/grpc_asyncio.py
@@ -355,6 +355,9 @@ def delete_log_metric(self) -> Callable[
             )
         return self._stubs['delete_log_metric']
 
+    def close(self):
+        return self.grpc_channel.close()
+
 
 __all__ = (
     'MetricsServiceV2GrpcAsyncIOTransport',
diff --git a/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_config_service_v2.py b/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_config_service_v2.py
index 288323a4bc..8a1d5e3189 100644
--- a/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_config_service_v2.py
+++ b/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_config_service_v2.py
@@ -5907,6 +5907,9 @@ def test_config_service_v2_base_transport():
         with pytest.raises(NotImplementedError):
             getattr(transport, method)(request=object())
 
+    with pytest.raises(NotImplementedError):
+        transport.close()
+
 
 @requires_google_auth_gte_1_25_0
 def test_config_service_v2_base_transport_with_credentials_file():
@@ -6446,3 +6449,46 @@ def test_client_withDEFAULT_CLIENT_INFO():
             client_info=client_info,
         )
         prep.assert_called_once_with(client_info)
+
+
+@pytest.mark.asyncio
+async def test_transport_close_async():
+    client = ConfigServiceV2AsyncClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="grpc_asyncio",
+    )
+    with mock.patch.object(type(getattr(client.transport, "grpc_channel")), "close") as close:
+        async with client:
+            close.assert_not_called()
+        close.assert_called_once()
+
+def test_transport_close():
+    transports = {
+        "grpc": "_grpc_channel",
+    }
+
+    for transport, close_name in transports.items():
+        client = ConfigServiceV2Client(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport=transport
+        )
+        with mock.patch.object(type(getattr(client.transport, close_name)), "close") as close:
+            with client:
+                close.assert_not_called()
+            close.assert_called_once()
+
+def test_client_ctx():
+    transports = [
+        'grpc',
+    ]
+    for transport in transports:
+        client = ConfigServiceV2Client(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport=transport
+        )
+        # Test client calls underlying transport.
+        with mock.patch.object(type(client.transport), "close") as close:
+            close.assert_not_called()
+            with client:
+                pass
+            close.assert_called()
diff --git a/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_logging_service_v2.py b/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_logging_service_v2.py
index 31f15bca04..f372ad41b9 100644
--- a/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_logging_service_v2.py
+++ b/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_logging_service_v2.py
@@ -2029,6 +2029,9 @@ def test_logging_service_v2_base_transport():
         with pytest.raises(NotImplementedError):
             getattr(transport, method)(request=object())
 
+    with pytest.raises(NotImplementedError):
+        transport.close()
+
 
 @requires_google_auth_gte_1_25_0
 def test_logging_service_v2_base_transport_with_credentials_file():
@@ -2493,3 +2496,46 @@ def test_client_withDEFAULT_CLIENT_INFO():
             client_info=client_info,
         )
         prep.assert_called_once_with(client_info)
+
+
+@pytest.mark.asyncio
+async def test_transport_close_async():
+    client = LoggingServiceV2AsyncClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="grpc_asyncio",
+    )
+    with mock.patch.object(type(getattr(client.transport, "grpc_channel")), "close") as close:
+        async with client:
+            close.assert_not_called()
+        close.assert_called_once()
+
+def test_transport_close():
+    transports = {
+        "grpc": "_grpc_channel",
+    }
+
+    for transport, close_name in transports.items():
+        client = LoggingServiceV2Client(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport=transport
+        )
+        with mock.patch.object(type(getattr(client.transport, close_name)), "close") as close:
+            with client:
+                close.assert_not_called()
+            close.assert_called_once()
+
+def test_client_ctx():
+    transports = [
+        'grpc',
+    ]
+    for transport in transports:
+        client = LoggingServiceV2Client(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport=transport
+        )
+        # Test client calls underlying transport.
+        with mock.patch.object(type(client.transport), "close") as close:
+            close.assert_not_called()
+            with client:
+                pass
+            close.assert_called()
diff --git a/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_metrics_service_v2.py b/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_metrics_service_v2.py
index e3eb2f3aca..a42c0e2fdd 100644
--- a/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_metrics_service_v2.py
+++ b/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_metrics_service_v2.py
@@ -1894,6 +1894,9 @@ def test_metrics_service_v2_base_transport():
         with pytest.raises(NotImplementedError):
             getattr(transport, method)(request=object())
 
+    with pytest.raises(NotImplementedError):
+        transport.close()
+
 
 @requires_google_auth_gte_1_25_0
 def test_metrics_service_v2_base_transport_with_credentials_file():
@@ -2358,3 +2361,46 @@ def test_client_withDEFAULT_CLIENT_INFO():
             client_info=client_info,
         )
         prep.assert_called_once_with(client_info)
+
+
+@pytest.mark.asyncio
+async def test_transport_close_async():
+    client = MetricsServiceV2AsyncClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="grpc_asyncio",
+    )
+    with mock.patch.object(type(getattr(client.transport, "grpc_channel")), "close") as close:
+        async with client:
+            close.assert_not_called()
+        close.assert_called_once()
+
+def test_transport_close():
+    transports = {
+        "grpc": "_grpc_channel",
+    }
+
+    for transport, close_name in transports.items():
+        client = MetricsServiceV2Client(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport=transport
+        )
+        with mock.patch.object(type(getattr(client.transport, close_name)), "close") as close:
+            with client:
+                close.assert_not_called()
+            close.assert_called_once()
+
+def test_client_ctx():
+    transports = [
+        'grpc',
+    ]
+    for transport in transports:
+        client = MetricsServiceV2Client(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport=transport
+        )
+        # Test client calls underlying transport.
+        with mock.patch.object(type(client.transport), "close") as close:
+            close.assert_not_called()
+            with client:
+                pass
+            close.assert_called()
diff --git a/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/async_client.py b/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/async_client.py
index 512bab3f12..8d22164a88 100644
--- a/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/async_client.py
+++ b/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/async_client.py
@@ -1078,9 +1078,11 @@ async def delete_instance(self,
         # Done; return the response.
         return response
 
+    async def __aenter__(self):
+        return self
 
-
-
+    async def __aexit__(self, exc_type, exc, tb):
+        await self.transport.close()
 
 try:
     DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
diff --git a/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/client.py b/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/client.py
index 9afc35601a..7be2d6781b 100644
--- a/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/client.py
+++ b/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/client.py
@@ -1266,7 +1266,18 @@ def delete_instance(self,
         # Done; return the response.
         return response
 
+    def __enter__(self):
+        return self
 
+    def __exit__(self, type, value, traceback):
+        """Releases underlying transport's resources.
+
+        .. warning::
+            ONLY use as a context manager if the transport is NOT shared
+            with other clients! Exiting the with block will CLOSE the transport
+            and may cause errors in other clients!
+        """
+        self.transport.close()
 
 
 
diff --git a/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/base.py b/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/base.py
index d8161b1e78..240133be80 100644
--- a/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/base.py
+++ b/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/base.py
@@ -195,6 +195,15 @@ def _prep_wrapped_messages(self, client_info):
             ),
          }
 
+    def close(self):
+        """Closes resources associated with the transport.
+
+       .. warning::
+            Only call this method if the transport is NOT shared
+            with other clients - this may cause errors in other clients!
+        """
+        raise NotImplementedError()
+
     @property
     def operations_client(self) -> operations_v1.OperationsClient:
         """Return the client designed to process long-running operations."""
diff --git a/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/grpc.py b/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/grpc.py
index 5b15755d04..82eeca6212 100644
--- a/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/grpc.py
+++ b/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/grpc.py
@@ -536,6 +536,8 @@ def delete_instance(self) -> Callable[
             )
         return self._stubs['delete_instance']
 
+    def close(self):
+        self.grpc_channel.close()
 
 __all__ = (
     'CloudRedisGrpcTransport',
diff --git a/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/grpc_asyncio.py b/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/grpc_asyncio.py
index 2a2184de64..d89351dcb1 100644
--- a/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/grpc_asyncio.py
+++ b/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/grpc_asyncio.py
@@ -540,6 +540,9 @@ def delete_instance(self) -> Callable[
             )
         return self._stubs['delete_instance']
 
+    def close(self):
+        return self.grpc_channel.close()
+
 
 __all__ = (
     'CloudRedisGrpcAsyncIOTransport',
diff --git a/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py b/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py
index 1b71940694..54f12b158e 100644
--- a/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py
+++ b/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py
@@ -2860,6 +2860,9 @@ def test_cloud_redis_base_transport():
         with pytest.raises(NotImplementedError):
             getattr(transport, method)(request=object())
 
+    with pytest.raises(NotImplementedError):
+        transport.close()
+
     # Additionally, the LRO client (a property) should
     # also raise NotImplementedError
     with pytest.raises(NotImplementedError):
@@ -3345,3 +3348,46 @@ def test_client_withDEFAULT_CLIENT_INFO():
             client_info=client_info,
         )
         prep.assert_called_once_with(client_info)
+
+
+@pytest.mark.asyncio
+async def test_transport_close_async():
+    client = CloudRedisAsyncClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="grpc_asyncio",
+    )
+    with mock.patch.object(type(getattr(client.transport, "grpc_channel")), "close") as close:
+        async with client:
+            close.assert_not_called()
+        close.assert_called_once()
+
+def test_transport_close():
+    transports = {
+        "grpc": "_grpc_channel",
+    }
+
+    for transport, close_name in transports.items():
+        client = CloudRedisClient(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport=transport
+        )
+        with mock.patch.object(type(getattr(client.transport, close_name)), "close") as close:
+            with client:
+                close.assert_not_called()
+            close.assert_called_once()
+
+def test_client_ctx():
+    transports = [
+        'grpc',
+    ]
+    for transport in transports:
+        client = CloudRedisClient(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport=transport
+        )
+        # Test client calls underlying transport.
+        with mock.patch.object(type(client.transport), "close") as close:
+            close.assert_not_called()
+            with client:
+                pass
+            close.assert_called()
diff --git a/tests/system/test_client_context_manager.py b/tests/system/test_client_context_manager.py
new file mode 100644
index 0000000000..88ced947ed
--- /dev/null
+++ b/tests/system/test_client_context_manager.py
@@ -0,0 +1,53 @@
+# Copyright 2021 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
+#
+#     https://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.
+
+import os
+import pytest
+import grpc
+import distutils
+
+
+def test_client(echo):
+    with echo as c:
+        resp = c.echo({
+            'content': 'hello'
+        })
+        assert resp.content == 'hello'
+
+
+def test_client_destroyed(echo):
+    echo.__exit__(None, None, None)
+    with pytest.raises(ValueError):
+        echo.echo({
+            'content': 'hello'
+        })
+
+
+if distutils.util.strtobool(os.environ.get("GAPIC_PYTHON_ASYNC", "true")):
+
+    @pytest.mark.asyncio
+    async def test_client_async(async_echo):
+        async with async_echo:
+            response = await async_echo.echo({
+                'content': 'hello'
+            })
+            assert response.content == 'hello'
+
+    @pytest.mark.asyncio
+    async def test_client_destroyed_async(async_echo):
+        await async_echo.__aexit__(None, None, None)
+        with pytest.raises(grpc._cython.cygrpc.UsageError):
+            await async_echo.echo({
+                'content': 'hello'
+            })