From 2833a1b2c456268a40e6f94c33196961a81c041e Mon Sep 17 00:00:00 2001
From: Rakshith Bhyravabhotla <rakshith.bhyravabhotla@gmail.com>
Date: Wed, 16 Oct 2019 15:28:05 -0700
Subject: [PATCH 1/6] Initital Commit

---
 .../azure/storage/blob/aio/blob_service_client_async.py       | 4 ++--
 .../azure/storage/blob/aio/container_client_async.py          | 2 +-
 .../azure/storage/blob/blob_service_client.py                 | 4 ++--
 .../azure-storage-blob/azure/storage/blob/container_client.py | 2 +-
 4 files changed, 6 insertions(+), 6 deletions(-)

diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/aio/blob_service_client_async.py b/sdk/storage/azure-storage-blob/azure/storage/blob/aio/blob_service_client_async.py
index ac2942cad0d0..ff398151fa81 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/aio/blob_service_client_async.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/aio/blob_service_client_async.py
@@ -493,7 +493,7 @@ def get_container_client(self, container):
             credential=self.credential, _configuration=self._config,
             _pipeline=self._pipeline, _location_mode=self._location_mode, _hosts=self._hosts,
             require_encryption=self.require_encryption, key_encryption_key=self.key_encryption_key,
-            key_resolver_function=self.key_resolver_function, loop=self._loop)
+            key_resolver_function=self.key_resolver_function, session_owner=False, loop=self._loop)
 
     def get_blob_client(
             self, container,  # type: Union[ContainerProperties, str]
@@ -543,4 +543,4 @@ def get_blob_client(
             credential=self.credential, _configuration=self._config,
             _pipeline=self._pipeline, _location_mode=self._location_mode, _hosts=self._hosts,
             require_encryption=self.require_encryption, key_encryption_key=self.key_encryption_key,
-            key_resolver_function=self.key_resolver_function, loop=self._loop)
+            key_resolver_function=self.key_resolver_function, session_owner=False, loop=self._loop)
diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/aio/container_client_async.py b/sdk/storage/azure-storage-blob/azure/storage/blob/aio/container_client_async.py
index 684034ed93ff..9ce174a0e783 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/aio/container_client_async.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/aio/container_client_async.py
@@ -1006,4 +1006,4 @@ def get_blob_client(
             credential=self.credential, _configuration=self._config,
             _pipeline=self._pipeline, _location_mode=self._location_mode, _hosts=self._hosts,
             require_encryption=self.require_encryption, key_encryption_key=self.key_encryption_key,
-            key_resolver_function=self.key_resolver_function, loop=self._loop)
+            key_resolver_function=self.key_resolver_function, session_owner=False, loop=self._loop)
diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/blob_service_client.py b/sdk/storage/azure-storage-blob/azure/storage/blob/blob_service_client.py
index 2a9d87b22598..c8af2423be90 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/blob_service_client.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/blob_service_client.py
@@ -612,7 +612,7 @@ def get_container_client(self, container):
             credential=self.credential, _configuration=self._config,
             _pipeline=self._pipeline, _location_mode=self._location_mode, _hosts=self._hosts,
             require_encryption=self.require_encryption, key_encryption_key=self.key_encryption_key,
-            key_resolver_function=self.key_resolver_function)
+            key_resolver_function=self.key_resolver_function, session_owner=False)
 
     def get_blob_client(
             self, container,  # type: Union[ContainerProperties, str]
@@ -662,4 +662,4 @@ def get_blob_client(
             credential=self.credential, _configuration=self._config,
             _pipeline=self._pipeline, _location_mode=self._location_mode, _hosts=self._hosts,
             require_encryption=self.require_encryption, key_encryption_key=self.key_encryption_key,
-            key_resolver_function=self.key_resolver_function)
+            key_resolver_function=self.key_resolver_function, session_owner=False)
diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/container_client.py b/sdk/storage/azure-storage-blob/azure/storage/blob/container_client.py
index 15e9012f9b7c..a6bd8564ddeb 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/container_client.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/container_client.py
@@ -1303,4 +1303,4 @@ def get_blob_client(
             credential=self.credential, _configuration=self._config,
             _pipeline=self._pipeline, _location_mode=self._location_mode, _hosts=self._hosts,
             require_encryption=self.require_encryption, key_encryption_key=self.key_encryption_key,
-            key_resolver_function=self.key_resolver_function)
+            key_resolver_function=self.key_resolver_function, session_owner=False)

From 21fcfdb8d00ba4ce39ecc4bb7f1e4f625b389c36 Mon Sep 17 00:00:00 2001
From: Rakshith Bhyravabhotla <rakshith.bhyravabhotla@gmail.com>
Date: Thu, 17 Oct 2019 10:23:32 -0700
Subject: [PATCH 2/6] Pipeline Ownership

---
 .../azure/storage/blob/_shared/base_client.py | 19 ++++++++++++++++-
 .../storage/blob/_shared/base_client_async.py | 19 +++++++++++++++++
 .../blob/aio/blob_service_client_async.py     | 21 ++++++++++++-------
 .../blob/aio/container_client_async.py        | 12 +++++++----
 .../azure/storage/blob/blob_service_client.py | 21 ++++++++++++-------
 .../azure/storage/blob/container_client.py    |  9 ++++++--
 .../tests/test_common_blob.py                 | 12 +++++++++++
 .../tests/test_common_blob_async.py           | 12 ++++++++++-
 8 files changed, 103 insertions(+), 22 deletions(-)

diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client.py b/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client.py
index 91205c8f3fdb..158b7abbf15b 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client.py
@@ -28,7 +28,7 @@
 from azure.core import Configuration
 from azure.core.exceptions import HttpResponseError
 from azure.core.pipeline import Pipeline
-from azure.core.pipeline.transport import RequestsTransport
+from azure.core.pipeline.transport import RequestsTransport, HttpTransport
 from azure.core.pipeline.policies.distributed_tracing import DistributedTracingPolicy
 from azure.core.pipeline.policies import RedirectPolicy, ContentDecodePolicy, BearerTokenCredentialPolicy, ProxyPolicy
 
@@ -216,6 +216,23 @@ def _batch_send(
         except StorageErrorException as error:
             process_storage_error(error)
 
+class TransportWrapper(HttpTransport):
+
+    def __init__(self, transport):
+        self._transport = transport
+
+    def send(self, request, **kwargs):
+        return self._transport.send(request, **kwargs)
+
+    def open(self):
+        pass
+
+    def close(self):
+        pass
+
+    def __exit__(self, *args): # pylint: disable=arguments-differ
+        pass
+
 
 def format_shared_key_credential(account, credential):
     if isinstance(credential, six.string_types):
diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client_async.py b/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client_async.py
index e76fad464878..050eede96660 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client_async.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client_async.py
@@ -17,6 +17,7 @@
     ContentDecodePolicy,
     AsyncBearerTokenCredentialPolicy,
     AsyncRedirectPolicy)
+from azure.core.pipeline.transport import AsyncHttpTransport
 
 from .constants import STORAGE_OAUTH_SCOPE, DEFAULT_SOCKET_TIMEOUT
 from .authentication import SharedKeyCredentialPolicy
@@ -124,3 +125,21 @@ async def _batch_send(
             return response.parts()  # Return an AsyncIterator
         except StorageErrorException as error:
             process_storage_error(error)
+
+
+class AsyncTransportWrapper(AsyncHttpTransport):
+
+    def __init__(self, async_transport):
+        self._transport = async_transport
+
+    async def send(self, request, **kwargs):
+        return await self._transport.send(request, **kwargs)
+
+    async def open(self):
+        pass
+
+    async def close(self):
+        pass
+
+    async def __aexit__(self, *args): # pylint: disable=arguments-differ
+        pass
diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/aio/blob_service_client_async.py b/sdk/storage/azure-storage-blob/azure/storage/blob/aio/blob_service_client_async.py
index ff398151fa81..0c7244f90dcb 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/aio/blob_service_client_async.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/aio/blob_service_client_async.py
@@ -11,12 +11,13 @@
 )
 
 from azure.core.tracing.decorator import distributed_trace
+from azure.core.pipeline import Pipeline
 from azure.core.tracing.decorator_async import distributed_trace_async
 from azure.core.async_paging import AsyncItemPaged
 
 from .._shared.models import LocationMode
 from .._shared.policies_async import ExponentialRetry
-from .._shared.base_client_async import AsyncStorageAccountHostsMixin
+from .._shared.base_client_async import AsyncStorageAccountHostsMixin, AsyncTransportWrapper
 from .._shared.response_handlers import return_response_headers, process_storage_error
 from .._shared.parser import _to_utc_datetime
 from .._shared.response_handlers import parse_to_internal_user_delegation_key
@@ -487,13 +488,16 @@ def get_container_client(self, container):
             container_name = container.name
         except AttributeError:
             container_name = container
-
+        _pipeline = Pipeline(
+            transport=AsyncTransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
+            policies=self._pipeline._impl_policies # pylint: disable = protected-access
+        )
         return ContainerClient(
             self.url, container_name=container_name,
             credential=self.credential, _configuration=self._config,
-            _pipeline=self._pipeline, _location_mode=self._location_mode, _hosts=self._hosts,
+            _pipeline=_pipeline, _location_mode=self._location_mode, _hosts=self._hosts,
             require_encryption=self.require_encryption, key_encryption_key=self.key_encryption_key,
-            key_resolver_function=self.key_resolver_function, session_owner=False, loop=self._loop)
+            key_resolver_function=self.key_resolver_function, loop=self._loop)
 
     def get_blob_client(
             self, container,  # type: Union[ContainerProperties, str]
@@ -537,10 +541,13 @@ def get_blob_client(
             blob_name = blob.name
         except AttributeError:
             blob_name = blob
-
+        _pipeline = Pipeline(
+            transport=AsyncTransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
+            policies=self._pipeline._impl_policies # pylint: disable = protected-access
+        )
         return BlobClient( # type: ignore
             self.url, container_name=container_name, blob_name=blob_name, snapshot=snapshot,
             credential=self.credential, _configuration=self._config,
-            _pipeline=self._pipeline, _location_mode=self._location_mode, _hosts=self._hosts,
+            _pipeline=_pipeline, _location_mode=self._location_mode, _hosts=self._hosts,
             require_encryption=self.require_encryption, key_encryption_key=self.key_encryption_key,
-            key_resolver_function=self.key_resolver_function, session_owner=False, loop=self._loop)
+            key_resolver_function=self.key_resolver_function, loop=self._loop)
diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/aio/container_client_async.py b/sdk/storage/azure-storage-blob/azure/storage/blob/aio/container_client_async.py
index 9ce174a0e783..f676493fee4e 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/aio/container_client_async.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/aio/container_client_async.py
@@ -14,9 +14,10 @@
 from azure.core.tracing.decorator import distributed_trace
 from azure.core.tracing.decorator_async import distributed_trace_async
 from azure.core.async_paging import AsyncItemPaged
+from azure.core.pipeline import Pipeline
 from azure.core.pipeline.transport import HttpRequest, AsyncHttpResponse
 
-from .._shared.base_client_async import AsyncStorageAccountHostsMixin
+from .._shared.base_client_async import AsyncStorageAccountHostsMixin, AsyncTransportWrapper
 from .._shared.policies_async import ExponentialRetry
 from .._shared.request_handlers import add_metadata_headers, serialize_iso
 from .._shared.response_handlers import (
@@ -1000,10 +1001,13 @@ def get_blob_client(
             blob_name = blob.name
         except AttributeError:
             blob_name = blob
-
+        _pipeline = Pipeline(
+            transport=AsyncTransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
+            policies=self._pipeline._impl_policies # pylint: disable = protected-access
+        )
         return BlobClient(
             self.url, container_name=self.container_name, blob_name=blob_name, snapshot=snapshot,
             credential=self.credential, _configuration=self._config,
-            _pipeline=self._pipeline, _location_mode=self._location_mode, _hosts=self._hosts,
+            _pipeline=_pipeline, _location_mode=self._location_mode, _hosts=self._hosts,
             require_encryption=self.require_encryption, key_encryption_key=self.key_encryption_key,
-            key_resolver_function=self.key_resolver_function, session_owner=False, loop=self._loop)
+            key_resolver_function=self.key_resolver_function, loop=self._loop)
diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/blob_service_client.py b/sdk/storage/azure-storage-blob/azure/storage/blob/blob_service_client.py
index c8af2423be90..60182e891072 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/blob_service_client.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/blob_service_client.py
@@ -16,11 +16,12 @@
     from urlparse import urlparse # type: ignore
 
 from azure.core.paging import ItemPaged
+from azure.core.pipeline import Pipeline
 from azure.core.tracing.decorator import distributed_trace
 
 from ._shared.shared_access_signature import SharedAccessSignature
 from ._shared.models import LocationMode, Services, UserDelegationKey
-from ._shared.base_client import StorageAccountHostsMixin, parse_connection_str, parse_query
+from ._shared.base_client import StorageAccountHostsMixin, TransportWrapper, parse_connection_str, parse_query
 from ._shared.parser import _to_utc_datetime
 from ._shared.response_handlers import return_response_headers, process_storage_error, \
     parse_to_internal_user_delegation_key
@@ -606,13 +607,16 @@ def get_container_client(self, container):
             container_name = container.name
         except AttributeError:
             container_name = container
-
+        _pipeline = Pipeline(
+            transport=TransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
+            policies=self._pipeline._impl_policies # pylint: disable = protected-access
+        )
         return ContainerClient(
             self.url, container_name=container_name,
             credential=self.credential, _configuration=self._config,
-            _pipeline=self._pipeline, _location_mode=self._location_mode, _hosts=self._hosts,
+            _pipeline=_pipeline, _location_mode=self._location_mode, _hosts=self._hosts,
             require_encryption=self.require_encryption, key_encryption_key=self.key_encryption_key,
-            key_resolver_function=self.key_resolver_function, session_owner=False)
+            key_resolver_function=self.key_resolver_function)
 
     def get_blob_client(
             self, container,  # type: Union[ContainerProperties, str]
@@ -656,10 +660,13 @@ def get_blob_client(
             blob_name = blob.name
         except AttributeError:
             blob_name = blob
-
+        _pipeline = Pipeline(
+            transport=TransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
+            policies=self._pipeline._impl_policies # pylint: disable = protected-access
+        )
         return BlobClient( # type: ignore
             self.url, container_name=container_name, blob_name=blob_name, snapshot=snapshot,
             credential=self.credential, _configuration=self._config,
-            _pipeline=self._pipeline, _location_mode=self._location_mode, _hosts=self._hosts,
+            _pipeline=_pipeline, _location_mode=self._location_mode, _hosts=self._hosts,
             require_encryption=self.require_encryption, key_encryption_key=self.key_encryption_key,
-            key_resolver_function=self.key_resolver_function, session_owner=False)
+            key_resolver_function=self.key_resolver_function)
diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/container_client.py b/sdk/storage/azure-storage-blob/azure/storage/blob/container_client.py
index a6bd8564ddeb..5abc08eb55c9 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/container_client.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/container_client.py
@@ -21,9 +21,10 @@
 
 from azure.core.paging import ItemPaged
 from azure.core.tracing.decorator import distributed_trace
+from azure.core.pipeline import Pipeline
 from azure.core.pipeline.transport import HttpRequest
 
-from ._shared.base_client import StorageAccountHostsMixin, parse_connection_str, parse_query
+from ._shared.base_client import StorageAccountHostsMixin, TransportWrapper, parse_connection_str, parse_query
 from ._shared.request_handlers import add_metadata_headers, serialize_iso
 from ._shared.response_handlers import (
     process_storage_error,
@@ -1298,9 +1299,13 @@ def get_blob_client(
             blob_name = blob.name
         except AttributeError:
             blob_name = blob
+        _pipeline = Pipeline(
+            transport=TransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
+            policies=self._pipeline._impl_policies # pylint: disable = protected-access
+        )
         return BlobClient(
             self.url, container_name=self.container_name, blob_name=blob_name, snapshot=snapshot,
             credential=self.credential, _configuration=self._config,
             _pipeline=self._pipeline, _location_mode=self._location_mode, _hosts=self._hosts,
             require_encryption=self.require_encryption, key_encryption_key=self.key_encryption_key,
-            key_resolver_function=self.key_resolver_function, session_owner=False)
+            key_resolver_function=self.key_resolver_function)
diff --git a/sdk/storage/azure-storage-blob/tests/test_common_blob.py b/sdk/storage/azure-storage-blob/tests/test_common_blob.py
index 92f66abad986..c7c9034fec3b 100644
--- a/sdk/storage/azure-storage-blob/tests/test_common_blob.py
+++ b/sdk/storage/azure-storage-blob/tests/test_common_blob.py
@@ -17,6 +17,7 @@
     ResourceNotFoundError,
     ResourceExistsError,
     ClientAuthenticationError)
+from azure.core.pipeline.transport import RequestsTransport
 from azure.storage.blob import (
     upload_blob_to_url,
     download_blob_from_url,
@@ -1833,6 +1834,17 @@ def test_set_blob_permission(self):
         self.assertEqual(permission.delete, True)
         self.assertEqual(permission.write, True)
         self.assertEqual(permission._str, 'wrdx')
+    
+    def test_transport_closed_only_once(self):
+        transport = RequestsTransport()
+        url = self._get_account_url()
+        credential = self._get_shared_key_credential()
+        blob_name = self._get_blob_reference()
+        with BlobServiceClient(url, credential=credential, transport=transport) as bsc:
+            assert transport.session is not None
+            with bsc.get_blob_client(self.container_name, blob_name) as bc:
+                assert transport.session is not None
+            assert transport.session is not None  # Right now it's None
 
 #------------------------------------------------------------------------------
 if __name__ == '__main__':
diff --git a/sdk/storage/azure-storage-blob/tests/test_common_blob_async.py b/sdk/storage/azure-storage-blob/tests/test_common_blob_async.py
index dd732030b4de..96ad644394dc 100644
--- a/sdk/storage/azure-storage-blob/tests/test_common_blob_async.py
+++ b/sdk/storage/azure-storage-blob/tests/test_common_blob_async.py
@@ -18,7 +18,7 @@
     ResourceNotFoundError,
     ResourceExistsError,
     ClientAuthenticationError)
-
+from azure.core.pipeline.transport import AsyncioRequestsTransport
 from azure.core.pipeline.transport import AioHttpTransport
 from multidict import CIMultiDict, CIMultiDictProxy
 
@@ -2281,6 +2281,16 @@ def test_upload_to_url_file_with_credential(self):
         loop = asyncio.get_event_loop()
         loop.run_until_complete(self._test_upload_to_url_file_with_credential())
 
+    async def test_transport_closed_only_once(self):
+        transport = AsyncioRequestsTransport()
+        url = self._get_account_url()
+        credential = self._get_shared_key_credential()
+        blob_name = self._get_blob_reference()
+        async with BlobServiceClient(url, credential=credential, transport=transport) as bsc:
+            assert transport.session is not None
+            async with bsc.get_blob_client(self.container_name, blob_name) as bc:
+                assert transport.session is not None
+            assert transport.session is not None  # Right now it's None
 # ------------------------------------------------------------------------------
 if __name__ == '__main__':
     unittest.main()

From 952c1250185494dcb717048641ae3c538b559db3 Mon Sep 17 00:00:00 2001
From: Rakshith Bhyravabhotla <rakshith.bhyravabhotla@gmail.com>
Date: Thu, 17 Oct 2019 10:56:00 -0700
Subject: [PATCH 3/6] slight modifications

---
 .../azure/storage/blob/_shared/base_client.py               | 3 +++
 .../azure/storage/blob/_shared/base_client_async.py         | 3 +++
 .../azure/storage/blob/aio/blob_service_client_async.py     | 6 +++---
 .../azure/storage/blob/aio/container_client_async.py        | 4 ++--
 4 files changed, 11 insertions(+), 5 deletions(-)

diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client.py b/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client.py
index 158b7abbf15b..cd219e469020 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client.py
@@ -230,6 +230,9 @@ def open(self):
     def close(self):
         pass
 
+    def __enter__(self, *args): # pylint: disable=arguments-differ
+        pass
+
     def __exit__(self, *args): # pylint: disable=arguments-differ
         pass
 
diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client_async.py b/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client_async.py
index 050eede96660..a61767878eae 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client_async.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client_async.py
@@ -141,5 +141,8 @@ async def open(self):
     async def close(self):
         pass
 
+    async def __aenter__(self, *args): # pylint: disable=arguments-differ
+        pass
+
     async def __aexit__(self, *args): # pylint: disable=arguments-differ
         pass
diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/aio/blob_service_client_async.py b/sdk/storage/azure-storage-blob/azure/storage/blob/aio/blob_service_client_async.py
index 0c7244f90dcb..c0d18a06787e 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/aio/blob_service_client_async.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/aio/blob_service_client_async.py
@@ -11,7 +11,7 @@
 )
 
 from azure.core.tracing.decorator import distributed_trace
-from azure.core.pipeline import Pipeline
+from azure.core.pipeline import AsyncPipeline
 from azure.core.tracing.decorator_async import distributed_trace_async
 from azure.core.async_paging import AsyncItemPaged
 
@@ -488,7 +488,7 @@ def get_container_client(self, container):
             container_name = container.name
         except AttributeError:
             container_name = container
-        _pipeline = Pipeline(
+        _pipeline = AsyncPipeline(
             transport=AsyncTransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
             policies=self._pipeline._impl_policies # pylint: disable = protected-access
         )
@@ -541,7 +541,7 @@ def get_blob_client(
             blob_name = blob.name
         except AttributeError:
             blob_name = blob
-        _pipeline = Pipeline(
+        _pipeline = AsyncPipeline(
             transport=AsyncTransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
             policies=self._pipeline._impl_policies # pylint: disable = protected-access
         )
diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/aio/container_client_async.py b/sdk/storage/azure-storage-blob/azure/storage/blob/aio/container_client_async.py
index f676493fee4e..d8b20b371631 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/aio/container_client_async.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/aio/container_client_async.py
@@ -14,7 +14,7 @@
 from azure.core.tracing.decorator import distributed_trace
 from azure.core.tracing.decorator_async import distributed_trace_async
 from azure.core.async_paging import AsyncItemPaged
-from azure.core.pipeline import Pipeline
+from azure.core.pipeline import AsyncPipeline
 from azure.core.pipeline.transport import HttpRequest, AsyncHttpResponse
 
 from .._shared.base_client_async import AsyncStorageAccountHostsMixin, AsyncTransportWrapper
@@ -1001,7 +1001,7 @@ def get_blob_client(
             blob_name = blob.name
         except AttributeError:
             blob_name = blob
-        _pipeline = Pipeline(
+        _pipeline = AsyncPipeline(
             transport=AsyncTransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
             policies=self._pipeline._impl_policies # pylint: disable = protected-access
         )

From 6c73352c81fd5ca64e8eff17134a43cc41d90058 Mon Sep 17 00:00:00 2001
From: Krista Pratico <krpratic@microsoft.com>
Date: Thu, 17 Oct 2019 13:18:54 -0700
Subject: [PATCH 4/6] pipeline ownership for queues and files

---
 .../azure/storage/file/_shared/base_client.py | 23 ++++++++++++++++++-
 .../storage/file/_shared/base_client_async.py | 22 ++++++++++++++++++
 .../file/aio/directory_client_async.py        | 18 +++++++++++----
 .../file/aio/file_service_client_async.py     | 10 ++++++--
 .../storage/file/aio/share_client_async.py    | 18 +++++++++++----
 .../azure/storage/file/directory_client.py    | 17 +++++++++++---
 .../azure/storage/file/file_service_client.py | 11 ++++++---
 .../azure/storage/file/share_client.py        | 17 +++++++++++---
 .../azure-storage-file/tests/test_share.py    | 13 +++++++++++
 .../tests/test_share_async.py                 | 12 ++++++++++
 .../storage/queue/_shared/base_client.py      | 23 ++++++++++++++++++-
 .../queue/_shared/base_client_async.py        | 23 ++++++++++++++++++-
 .../queue/aio/queue_service_client_async.py   | 11 +++++++--
 .../storage/queue/queue_service_client.py     | 11 +++++++--
 .../azure-storage-queue/tests/test_queue.py   | 12 ++++++++++
 .../tests/test_queue_async.py                 | 13 +++++++++++
 16 files changed, 228 insertions(+), 26 deletions(-)

diff --git a/sdk/storage/azure-storage-file/azure/storage/file/_shared/base_client.py b/sdk/storage/azure-storage-file/azure/storage/file/_shared/base_client.py
index f784ac130cf9..76c4eac11ded 100644
--- a/sdk/storage/azure-storage-file/azure/storage/file/_shared/base_client.py
+++ b/sdk/storage/azure-storage-file/azure/storage/file/_shared/base_client.py
@@ -28,7 +28,7 @@
 from azure.core import Configuration
 from azure.core.exceptions import HttpResponseError
 from azure.core.pipeline import Pipeline
-from azure.core.pipeline.transport import RequestsTransport
+from azure.core.pipeline.transport import RequestsTransport, HttpTransport
 from azure.core.pipeline.policies.distributed_tracing import DistributedTracingPolicy
 from azure.core.pipeline.policies import RedirectPolicy, ContentDecodePolicy, BearerTokenCredentialPolicy, ProxyPolicy
 
@@ -216,6 +216,27 @@ def _batch_send(
             process_storage_error(error)
 
 
+class TransportWrapper(HttpTransport):
+
+    def __init__(self, transport):
+        self._transport = transport
+
+    def send(self, request, **kwargs):
+        return self._transport.send(request, **kwargs)
+
+    def open(self):
+        pass
+
+    def close(self):
+        pass
+
+    def __enter__(self, *args): # pylint: disable=arguments-differ
+        pass
+
+    def __exit__(self, *args): # pylint: disable=arguments-differ
+        pass
+
+
 def format_shared_key_credential(account, credential):
     if isinstance(credential, six.string_types):
         if len(account) < 2:
diff --git a/sdk/storage/azure-storage-file/azure/storage/file/_shared/base_client_async.py b/sdk/storage/azure-storage-file/azure/storage/file/_shared/base_client_async.py
index bdcaba15d86f..a6573d926253 100644
--- a/sdk/storage/azure-storage-file/azure/storage/file/_shared/base_client_async.py
+++ b/sdk/storage/azure-storage-file/azure/storage/file/_shared/base_client_async.py
@@ -18,6 +18,7 @@
     AsyncBearerTokenCredentialPolicy,
     AsyncRedirectPolicy)
 
+from azure.core.pipeline.transport import AsyncHttpTransport
 from .constants import STORAGE_OAUTH_SCOPE, DEFAULT_SOCKET_TIMEOUT
 from .authentication import SharedKeyCredentialPolicy
 from .base_client import create_configuration
@@ -122,3 +123,24 @@ async def _batch_send(
             return response.parts()  # Return an AsyncIterator
         except StorageErrorException as error:
             process_storage_error(error)
+
+
+class AsyncTransportWrapper(AsyncHttpTransport):
+
+    def __init__(self, async_transport):
+        self._transport = async_transport
+
+    async def send(self, request, **kwargs):
+        return await self._transport.send(request, **kwargs)
+
+    async def open(self):
+        pass
+
+    async def close(self):
+        pass
+
+    async def __aenter__(self, *args): # pylint: disable=arguments-differ
+        pass
+
+    async def __aexit__(self, *args): # pylint: disable=arguments-differ
+        pass
diff --git a/sdk/storage/azure-storage-file/azure/storage/file/aio/directory_client_async.py b/sdk/storage/azure-storage-file/azure/storage/file/aio/directory_client_async.py
index 63021bca89bc..0f80d74fd1f4 100644
--- a/sdk/storage/azure-storage-file/azure/storage/file/aio/directory_client_async.py
+++ b/sdk/storage/azure-storage-file/azure/storage/file/aio/directory_client_async.py
@@ -11,7 +11,7 @@
 
 from azure.core.polling import async_poller
 from azure.core.async_paging import AsyncItemPaged
-
+from azure.core.pipeline import AsyncPipeline
 from azure.core.tracing.decorator import distributed_trace
 from azure.core.tracing.decorator_async import distributed_trace_async
 from .._parser import _get_file_permission, _datetime_to_str
@@ -20,7 +20,7 @@
 from .._generated.aio import AzureFileStorage
 from .._generated.version import VERSION
 from .._generated.models import StorageErrorException
-from .._shared.base_client_async import AsyncStorageAccountHostsMixin
+from .._shared.base_client_async import AsyncStorageAccountHostsMixin, AsyncTransportWrapper
 from .._shared.policies_async import ExponentialRetry
 from .._shared.request_handlers import add_metadata_headers
 from .._shared.response_handlers import return_response_headers, process_storage_error
@@ -112,10 +112,15 @@ def get_file_client(self, file_name, **kwargs):
         """
         if self.directory_path:
             file_name = self.directory_path.rstrip('/') + "/" + file_name
+
+        _pipeline = AsyncPipeline(
+            transport=AsyncTransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
+            policies=self._pipeline._impl_policies # pylint: disable = protected-access
+        )
         return FileClient(
             self.url, file_path=file_name, share_name=self.share_name, snapshot=self.snapshot,
             credential=self.credential, _hosts=self._hosts, _configuration=self._config,
-            _pipeline=self._pipeline, _location_mode=self._location_mode, loop=self._loop, **kwargs)
+            _pipeline=_pipeline, _location_mode=self._location_mode, loop=self._loop, **kwargs)
 
     def get_subdirectory_client(self, directory_name, **kwargs):
         # type: (str, Any) -> DirectoryClient
@@ -138,10 +143,15 @@ def get_subdirectory_client(self, directory_name, **kwargs):
                 :caption: Gets the subdirectory client.
         """
         directory_path = self.directory_path.rstrip('/') + "/" + directory_name
+
+        _pipeline = AsyncPipeline(
+            transport=AsyncTransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
+            policies=self._pipeline._impl_policies # pylint: disable = protected-access
+        )
         return DirectoryClient(
             self.url, share_name=self.share_name, directory_path=directory_path, snapshot=self.snapshot,
             credential=self.credential, _hosts=self._hosts, _configuration=self._config,
-            _pipeline=self._pipeline, _location_mode=self._location_mode, loop=self._loop, **kwargs)
+            _pipeline=_pipeline, _location_mode=self._location_mode, loop=self._loop, **kwargs)
 
     @distributed_trace_async
     async def create_directory(self, **kwargs): # type: ignore
diff --git a/sdk/storage/azure-storage-file/azure/storage/file/aio/file_service_client_async.py b/sdk/storage/azure-storage-file/azure/storage/file/aio/file_service_client_async.py
index c043a030f4a3..a2050a86e276 100644
--- a/sdk/storage/azure-storage-file/azure/storage/file/aio/file_service_client_async.py
+++ b/sdk/storage/azure-storage-file/azure/storage/file/aio/file_service_client_async.py
@@ -12,9 +12,10 @@
 
 from azure.core.async_paging import AsyncItemPaged
 from azure.core.tracing.decorator import distributed_trace
+from azure.core.pipeline import AsyncPipeline
 from azure.core.tracing.decorator_async import distributed_trace_async
 
-from .._shared.base_client_async import AsyncStorageAccountHostsMixin
+from .._shared.base_client_async import AsyncStorageAccountHostsMixin, AsyncTransportWrapper
 from .._shared.response_handlers import process_storage_error
 from .._shared.policies_async import ExponentialRetry
 from .._generated.aio import AzureFileStorage
@@ -314,6 +315,11 @@ def get_share_client(self, share, snapshot=None):
             share_name = share.name
         except AttributeError:
             share_name = share
+
+        _pipeline = AsyncPipeline(
+            transport=AsyncTransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
+            policies=self._pipeline._impl_policies # pylint: disable = protected-access
+        )
         return ShareClient(
             self.url, share_name=share_name, snapshot=snapshot, credential=self.credential, _hosts=self._hosts,
-            _configuration=self._config, _pipeline=self._pipeline, _location_mode=self._location_mode, loop=self._loop)
+            _configuration=self._config, _pipeline=_pipeline, _location_mode=self._location_mode, loop=self._loop)
diff --git a/sdk/storage/azure-storage-file/azure/storage/file/aio/share_client_async.py b/sdk/storage/azure-storage-file/azure/storage/file/aio/share_client_async.py
index b5e5ce453024..9ccb64e7281e 100644
--- a/sdk/storage/azure-storage-file/azure/storage/file/aio/share_client_async.py
+++ b/sdk/storage/azure-storage-file/azure/storage/file/aio/share_client_async.py
@@ -10,9 +10,9 @@
 
 from azure.core.tracing.decorator import distributed_trace
 from azure.core.tracing.decorator_async import distributed_trace_async
-
+from azure.core.pipeline import AsyncPipeline
 from .._shared.policies_async import ExponentialRetry
-from .._shared.base_client_async import AsyncStorageAccountHostsMixin
+from .._shared.base_client_async import AsyncStorageAccountHostsMixin, AsyncTransportWrapper
 from .._shared.request_handlers import add_metadata_headers, serialize_iso
 from .._shared.response_handlers import (
     return_response_headers,
@@ -102,9 +102,14 @@ def get_directory_client(self, directory_path=None):
         :returns: A Directory Client.
         :rtype: ~azure.storage.file.aio.directory_client_async.DirectoryClient
         """
+        _pipeline = AsyncPipeline(
+            transport=AsyncTransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
+            policies=self._pipeline._impl_policies # pylint: disable = protected-access
+        )
+
         return DirectoryClient(
             self.url, share_name=self.share_name, directory_path=directory_path or "", snapshot=self.snapshot,
-            credential=self.credential, _hosts=self._hosts, _configuration=self._config, _pipeline=self._pipeline,
+            credential=self.credential, _hosts=self._hosts, _configuration=self._config, _pipeline=_pipeline,
             _location_mode=self._location_mode, loop=self._loop)
 
     def get_file_client(self, file_path):
@@ -117,10 +122,15 @@ def get_file_client(self, file_path):
         :returns: A File Client.
         :rtype: ~azure.storage.file.aio.file_client_async.FileClient
         """
+        _pipeline = AsyncPipeline(
+            transport=AsyncTransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
+            policies=self._pipeline._impl_policies # pylint: disable = protected-access
+        )
+
         return FileClient(
             self.url, share_name=self.share_name, file_path=file_path, snapshot=self.snapshot,
             credential=self.credential, _hosts=self._hosts, _configuration=self._config,
-            _pipeline=self._pipeline, _location_mode=self._location_mode, loop=self._loop)
+            _pipeline=_pipeline, _location_mode=self._location_mode, loop=self._loop)
 
     @distributed_trace_async
     async def create_share(self, **kwargs):  # type: ignore
diff --git a/sdk/storage/azure-storage-file/azure/storage/file/directory_client.py b/sdk/storage/azure-storage-file/azure/storage/file/directory_client.py
index 8b0e73b3c004..7c0119d72571 100644
--- a/sdk/storage/azure-storage-file/azure/storage/file/directory_client.py
+++ b/sdk/storage/azure-storage-file/azure/storage/file/directory_client.py
@@ -18,12 +18,13 @@
 import six
 from azure.core.polling import LROPoller
 from azure.core.paging import ItemPaged
+from azure.core.pipeline import Pipeline
 from azure.core.tracing.decorator import distributed_trace
 
 from ._generated import AzureFileStorage
 from ._generated.version import VERSION
 from ._generated.models import StorageErrorException
-from ._shared.base_client import StorageAccountHostsMixin, parse_connection_str, parse_query
+from ._shared.base_client import StorageAccountHostsMixin, TransportWrapper, parse_connection_str, parse_query
 from ._shared.request_handlers import add_metadata_headers
 from ._shared.response_handlers import return_response_headers, process_storage_error
 from ._shared.parser import _str
@@ -217,10 +218,15 @@ def get_file_client(self, file_name, **kwargs):
         """
         if self.directory_path:
             file_name = self.directory_path.rstrip('/') + "/" + file_name
+
+        _pipeline = Pipeline(
+            transport=TransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
+            policies=self._pipeline._impl_policies # pylint: disable = protected-access
+        )
         return FileClient(
             self.url, file_path=file_name, share_name=self.share_name, napshot=self.snapshot,
             credential=self.credential, _hosts=self._hosts, _configuration=self._config,
-            _pipeline=self._pipeline, _location_mode=self._location_mode, **kwargs)
+            _pipeline=_pipeline, _location_mode=self._location_mode, **kwargs)
 
     def get_subdirectory_client(self, directory_name, **kwargs):
         # type: (str, Any) -> DirectoryClient
@@ -243,9 +249,14 @@ def get_subdirectory_client(self, directory_name, **kwargs):
                 :caption: Gets the subdirectory client.
         """
         directory_path = self.directory_path.rstrip('/') + "/" + directory_name
+
+        _pipeline = Pipeline(
+            transport=TransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
+            policies=self._pipeline._impl_policies # pylint: disable = protected-access
+        )
         return DirectoryClient(
             self.url, share_name=self.share_name, directory_path=directory_path, snapshot=self.snapshot,
-            credential=self.credential, _hosts=self._hosts, _configuration=self._config, _pipeline=self._pipeline,
+            credential=self.credential, _hosts=self._hosts, _configuration=self._config, _pipeline=_pipeline,
             _location_mode=self._location_mode, **kwargs)
 
     @distributed_trace
diff --git a/sdk/storage/azure-storage-file/azure/storage/file/file_service_client.py b/sdk/storage/azure-storage-file/azure/storage/file/file_service_client.py
index e349527fa0f4..812436846081 100644
--- a/sdk/storage/azure-storage-file/azure/storage/file/file_service_client.py
+++ b/sdk/storage/azure-storage-file/azure/storage/file/file_service_client.py
@@ -16,10 +16,10 @@
 
 from azure.core.paging import ItemPaged
 from azure.core.tracing.decorator import distributed_trace
-
+from azure.core.pipeline import Pipeline
 from ._shared.shared_access_signature import SharedAccessSignature
 from ._shared.models import Services
-from ._shared.base_client import StorageAccountHostsMixin, parse_connection_str, parse_query
+from ._shared.base_client import StorageAccountHostsMixin, TransportWrapper, parse_connection_str, parse_query
 from ._shared.response_handlers import process_storage_error
 from ._generated import AzureFileStorage
 from ._generated.models import StorageErrorException, StorageServiceProperties
@@ -431,6 +431,11 @@ def get_share_client(self, share, snapshot=None):
             share_name = share.name
         except AttributeError:
             share_name = share
+
+        _pipeline = Pipeline(
+            transport=TransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
+            policies=self._pipeline._impl_policies # pylint: disable = protected-access
+        )
         return ShareClient(
             self.url, share_name=share_name, snapshot=snapshot, credential=self.credential, _hosts=self._hosts,
-            _configuration=self._config, _pipeline=self._pipeline, _location_mode=self._location_mode)
+            _configuration=self._config, _pipeline=_pipeline, _location_mode=self._location_mode)
diff --git a/sdk/storage/azure-storage-file/azure/storage/file/share_client.py b/sdk/storage/azure-storage-file/azure/storage/file/share_client.py
index 29618ef0cf7a..42168904ceae 100644
--- a/sdk/storage/azure-storage-file/azure/storage/file/share_client.py
+++ b/sdk/storage/azure-storage-file/azure/storage/file/share_client.py
@@ -15,7 +15,8 @@
 
 import six
 from azure.core.tracing.decorator import distributed_trace
-from ._shared.base_client import StorageAccountHostsMixin, parse_connection_str, parse_query
+from azure.core.pipeline import Pipeline
+from ._shared.base_client import StorageAccountHostsMixin, TransportWrapper, parse_connection_str, parse_query
 from ._shared.request_handlers import add_metadata_headers, serialize_iso
 from ._shared.response_handlers import (
     return_response_headers,
@@ -297,9 +298,14 @@ def get_directory_client(self, directory_path=None):
         :returns: A Directory Client.
         :rtype: ~azure.storage.file.DirectoryClient
         """
+        _pipeline = Pipeline(
+            transport=TransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
+            policies=self._pipeline._impl_policies # pylint: disable = protected-access
+        )
+
         return DirectoryClient(
             self.url, share_name=self.share_name, directory_path=directory_path or "", snapshot=self.snapshot,
-            credential=self.credential, _hosts=self._hosts, _configuration=self._config, _pipeline=self._pipeline,
+            credential=self.credential, _hosts=self._hosts, _configuration=self._config, _pipeline=_pipeline,
             _location_mode=self._location_mode)
 
     def get_file_client(self, file_path):
@@ -312,10 +318,15 @@ def get_file_client(self, file_path):
         :returns: A File Client.
         :rtype: ~azure.storage.file.FileClient
         """
+        _pipeline = Pipeline(
+            transport=TransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
+            policies=self._pipeline._impl_policies # pylint: disable = protected-access
+        )
+
         return FileClient(
             self.url, share_name=self.share_name, file_path=file_path, snapshot=self.snapshot,
             credential=self.credential, _hosts=self._hosts, _configuration=self._config,
-            _pipeline=self._pipeline, _location_mode=self._location_mode)
+            _pipeline=_pipeline, _location_mode=self._location_mode)
 
     @distributed_trace
     def create_share(self, **kwargs):  # type: ignore
diff --git a/sdk/storage/azure-storage-file/tests/test_share.py b/sdk/storage/azure-storage-file/tests/test_share.py
index 53f6063e2e13..6a1b5b1d889e 100644
--- a/sdk/storage/azure-storage-file/tests/test_share.py
+++ b/sdk/storage/azure-storage-file/tests/test_share.py
@@ -10,6 +10,7 @@
 
 import pytest
 import requests
+from azure.core.pipeline.transport import RequestsTransport
 from azure.core.exceptions import (
     HttpResponseError,
     ResourceNotFoundError,
@@ -762,6 +763,18 @@ def test_create_permission_for_share(self):
         # server returned permission
         self.assertEquals(permission_key, permission_key2)
 
+    @record
+    def test_transport_closed_only_once(self):
+        transport = RequestsTransport()
+        url = self.get_file_url()
+        credential = self.get_shared_key_credential()
+        share = self._get_share_reference()
+        with FileServiceClient(url, credential=credential, transport=transport) as fsc:
+            assert transport.session is not None
+            with fsc.get_share_client(share.share_name) as fc:
+                assert transport.session is not None
+            assert transport.session is not None  # Right now it's None
+
 # ------------------------------------------------------------------------------
 if __name__ == '__main__':
     unittest.main()
diff --git a/sdk/storage/azure-storage-file/tests/test_share_async.py b/sdk/storage/azure-storage-file/tests/test_share_async.py
index fe4d3687d017..3fcd3eda0c53 100644
--- a/sdk/storage/azure-storage-file/tests/test_share_async.py
+++ b/sdk/storage/azure-storage-file/tests/test_share_async.py
@@ -11,6 +11,7 @@
 import pytest
 import requests
 from azure.core.pipeline.transport import AioHttpTransport
+from azure.core.pipeline.transport import AsyncioRequestsTransport
 from multidict import CIMultiDict, CIMultiDictProxy
 from azure.core.exceptions import (
     HttpResponseError,
@@ -918,6 +919,17 @@ def test_create_permission_for_share_async(self):
         loop = asyncio.get_event_loop()
         loop.run_until_complete(self._test_create_permission_for_share())
 
+    async def test_transport_closed_only_once_async(self):
+        transport = AsyncioRequestsTransport()
+        url = self.get_file_url()
+        credential = self.get_shared_key_credential()
+        share = self._get_share_reference()
+        async with FileServiceClient(url, credential=credential, transport=transport) as fsc:
+            assert transport.session is not None
+            async with fsc.get_share_client(share.share_name) as fc:
+                assert transport.session is not None
+            assert transport.session is not None  # Right now it's None
+
 # ------------------------------------------------------------------------------
 if __name__ == '__main__':
     unittest.main()
diff --git a/sdk/storage/azure-storage-queue/azure/storage/queue/_shared/base_client.py b/sdk/storage/azure-storage-queue/azure/storage/queue/_shared/base_client.py
index f784ac130cf9..76c4eac11ded 100644
--- a/sdk/storage/azure-storage-queue/azure/storage/queue/_shared/base_client.py
+++ b/sdk/storage/azure-storage-queue/azure/storage/queue/_shared/base_client.py
@@ -28,7 +28,7 @@
 from azure.core import Configuration
 from azure.core.exceptions import HttpResponseError
 from azure.core.pipeline import Pipeline
-from azure.core.pipeline.transport import RequestsTransport
+from azure.core.pipeline.transport import RequestsTransport, HttpTransport
 from azure.core.pipeline.policies.distributed_tracing import DistributedTracingPolicy
 from azure.core.pipeline.policies import RedirectPolicy, ContentDecodePolicy, BearerTokenCredentialPolicy, ProxyPolicy
 
@@ -216,6 +216,27 @@ def _batch_send(
             process_storage_error(error)
 
 
+class TransportWrapper(HttpTransport):
+
+    def __init__(self, transport):
+        self._transport = transport
+
+    def send(self, request, **kwargs):
+        return self._transport.send(request, **kwargs)
+
+    def open(self):
+        pass
+
+    def close(self):
+        pass
+
+    def __enter__(self, *args): # pylint: disable=arguments-differ
+        pass
+
+    def __exit__(self, *args): # pylint: disable=arguments-differ
+        pass
+
+
 def format_shared_key_credential(account, credential):
     if isinstance(credential, six.string_types):
         if len(account) < 2:
diff --git a/sdk/storage/azure-storage-queue/azure/storage/queue/_shared/base_client_async.py b/sdk/storage/azure-storage-queue/azure/storage/queue/_shared/base_client_async.py
index bdcaba15d86f..a19df6d89332 100644
--- a/sdk/storage/azure-storage-queue/azure/storage/queue/_shared/base_client_async.py
+++ b/sdk/storage/azure-storage-queue/azure/storage/queue/_shared/base_client_async.py
@@ -17,7 +17,7 @@
     ContentDecodePolicy,
     AsyncBearerTokenCredentialPolicy,
     AsyncRedirectPolicy)
-
+from azure.core.pipeline.transport import AsyncHttpTransport
 from .constants import STORAGE_OAUTH_SCOPE, DEFAULT_SOCKET_TIMEOUT
 from .authentication import SharedKeyCredentialPolicy
 from .base_client import create_configuration
@@ -122,3 +122,24 @@ async def _batch_send(
             return response.parts()  # Return an AsyncIterator
         except StorageErrorException as error:
             process_storage_error(error)
+
+
+class AsyncTransportWrapper(AsyncHttpTransport):
+
+    def __init__(self, async_transport):
+        self._transport = async_transport
+
+    async def send(self, request, **kwargs):
+        return await self._transport.send(request, **kwargs)
+
+    async def open(self):
+        pass
+
+    async def close(self):
+        pass
+
+    async def __aenter__(self, *args): # pylint: disable=arguments-differ
+        pass
+
+    async def __aexit__(self, *args): # pylint: disable=arguments-differ
+        pass
diff --git a/sdk/storage/azure-storage-queue/azure/storage/queue/aio/queue_service_client_async.py b/sdk/storage/azure-storage-queue/azure/storage/queue/aio/queue_service_client_async.py
index ee7835eb5044..f47aca78a97e 100644
--- a/sdk/storage/azure-storage-queue/azure/storage/queue/aio/queue_service_client_async.py
+++ b/sdk/storage/azure-storage-queue/azure/storage/queue/aio/queue_service_client_async.py
@@ -15,12 +15,13 @@
 
 from azure.core.async_paging import AsyncItemPaged
 from azure.core.tracing.decorator import distributed_trace
+from azure.core.pipeline import AsyncPipeline
 from azure.core.tracing.decorator_async import distributed_trace_async
 
 from .._shared.policies_async import ExponentialRetry
 from ..queue_service_client import QueueServiceClient as QueueServiceClientBase
 from .._shared.models import LocationMode
-from .._shared.base_client_async import AsyncStorageAccountHostsMixin
+from .._shared.base_client_async import AsyncStorageAccountHostsMixin, AsyncTransportWrapper
 from .._shared.response_handlers import process_storage_error
 from .._generated.aio import AzureQueueStorage
 from .._generated.models import StorageServiceProperties, StorageErrorException
@@ -372,8 +373,14 @@ def get_queue_client(self, queue, **kwargs):
             queue_name = queue.name
         except AttributeError:
             queue_name = queue
+
+        _pipeline = AsyncPipeline(
+            transport=AsyncTransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
+            policies=self._pipeline._impl_policies # pylint: disable = protected-access
+        )
+
         return QueueClient(
             self.url, queue_name=queue_name, credential=self.credential,
             key_resolver_function=self.key_resolver_function, require_encryption=self.require_encryption,
-            key_encryption_key=self.key_encryption_key, _pipeline=self._pipeline, _configuration=self._config,
+            key_encryption_key=self.key_encryption_key, _pipeline=_pipeline, _configuration=self._config,
             _location_mode=self._location_mode, _hosts=self._hosts, loop=self._loop, **kwargs)
diff --git a/sdk/storage/azure-storage-queue/azure/storage/queue/queue_service_client.py b/sdk/storage/azure-storage-queue/azure/storage/queue/queue_service_client.py
index 86f6ea6c6d67..c2789dcbc4ec 100644
--- a/sdk/storage/azure-storage-queue/azure/storage/queue/queue_service_client.py
+++ b/sdk/storage/azure-storage-queue/azure/storage/queue/queue_service_client.py
@@ -14,10 +14,11 @@
     from urlparse import urlparse # type: ignore
 
 from azure.core.paging import ItemPaged
+from azure.core.pipeline import Pipeline
 from azure.core.tracing.decorator import distributed_trace
 from ._shared.shared_access_signature import SharedAccessSignature
 from ._shared.models import LocationMode, Services
-from ._shared.base_client import StorageAccountHostsMixin, parse_connection_str, parse_query
+from ._shared.base_client import StorageAccountHostsMixin, TransportWrapper, parse_connection_str, parse_query
 from ._shared.response_handlers import process_storage_error
 from ._generated import AzureQueueStorage
 from ._generated.models import StorageServiceProperties, StorageErrorException
@@ -472,8 +473,14 @@ def get_queue_client(self, queue, **kwargs):
             queue_name = queue.name
         except AttributeError:
             queue_name = queue
+
+        _pipeline = Pipeline(
+            transport=TransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
+            policies=self._pipeline._impl_policies # pylint: disable = protected-access
+        )
+
         return QueueClient(
             self.url, queue_name=queue_name, credential=self.credential,
             key_resolver_function=self.key_resolver_function, require_encryption=self.require_encryption,
-            key_encryption_key=self.key_encryption_key, _pipeline=self._pipeline, _configuration=self._config,
+            key_encryption_key=self.key_encryption_key, _pipeline=_pipeline, _configuration=self._config,
             _location_mode=self._location_mode, _hosts=self._hosts, **kwargs)
diff --git a/sdk/storage/azure-storage-queue/tests/test_queue.py b/sdk/storage/azure-storage-queue/tests/test_queue.py
index f7a03392df77..ebb04914e3bf 100644
--- a/sdk/storage/azure-storage-queue/tests/test_queue.py
+++ b/sdk/storage/azure-storage-queue/tests/test_queue.py
@@ -18,6 +18,7 @@
 )
 from devtools_testutils import ResourceGroupPreparer, StorageAccountPreparer
 from azure.mgmt.storage.models import Endpoints
+from azure.core.pipeline.transport import RequestsTransport
 from azure.core.exceptions import (
     HttpResponseError,
     ResourceNotFoundError,
@@ -963,6 +964,17 @@ def test_unicode_update_message_unicode_data(self, resource_group, location, sto
         self.assertIsInstance(message.expires_on, datetime)
         self.assertIsInstance(message.next_visible_on, datetime)
 
+    @ResourceGroupPreparer()
+    @StorageAccountPreparer(name_prefix='pyacrstorage')
+    def test_transport_closed_only_once(self, resource_group, location, storage_account, storage_account_key):
+        transport = RequestsTransport()
+        prefix = TEST_QUEUE_PREFIX
+        queue_name = self.get_resource_name(prefix)
+        with QueueServiceClient(self._account_url(storage_account.name), credential=storage_account_key, transport=transport) as qsc:
+            assert transport.session is not None
+            with qsc.get_queue_client(queue_name) as qc:
+                assert transport.session is not None
+            assert transport.session is not None  # Right now it's None
 
 # ------------------------------------------------------------------------------
 if __name__ == '__main__':
diff --git a/sdk/storage/azure-storage-queue/tests/test_queue_async.py b/sdk/storage/azure-storage-queue/tests/test_queue_async.py
index 69d36debbdd9..ba005f73bdbe 100644
--- a/sdk/storage/azure-storage-queue/tests/test_queue_async.py
+++ b/sdk/storage/azure-storage-queue/tests/test_queue_async.py
@@ -16,6 +16,7 @@
 )
 from devtools_testutils import ResourceGroupPreparer, StorageAccountPreparer
 from multidict import CIMultiDict, CIMultiDictProxy
+from azure.core.pipeline.transport import AsyncioRequestsTransport
 from azure.core.exceptions import (
     HttpResponseError,
     ResourceNotFoundError,
@@ -1025,6 +1026,18 @@ async def test_unicode_update_message_unicode_data(self, resource_group, locatio
         self.assertIsInstance(message.expires_on, datetime)
         self.assertIsInstance(message.next_visible_on, datetime)
 
+    @ResourceGroupPreparer()
+    @StorageAccountPreparer(name_prefix='pyacrstorage')
+    @AsyncQueueTestCase.await_prepared_test
+    async def test_transport_closed_only_once_async(self, resource_group, location, storage_account, storage_account_key):
+        transport = AsyncioRequestsTransport()
+        prefix = TEST_QUEUE_PREFIX
+        queue_name = self.get_resource_name(prefix)
+        async with QueueServiceClient(self._account_url(storage_account.name), credential=storage_account_key, transport=transport) as qsc:
+            assert transport.session is not None
+            async with qsc.get_queue_client(queue_name) as qc:
+                assert transport.session is not None
+            assert transport.session is not None  # Right now it's None
 
 # ------------------------------------------------------------------------------
 if __name__ == '__main__':

From 55e8b7f9e732f3adecfe69bb5c994463d59426a0 Mon Sep 17 00:00:00 2001
From: Krista Pratico <krpratic@microsoft.com>
Date: Fri, 18 Oct 2019 16:29:23 -0700
Subject: [PATCH 5/6] fixes tests, adds docstrings to wrapper classes

---
 .../azure/storage/blob/_shared/base_client.py | 25 +++++++++++++++++-
 .../storage/blob/_shared/base_client_async.py | 25 ++++++++++++++++++
 .../tests/test_common_blob.py                 | 16 ++++++++++++
 .../tests/test_common_blob_async.py           | 21 ++++++++++++++-
 .../azure/storage/file/_shared/base_client.py | 25 ++++++++++++++++++
 .../storage/file/_shared/base_client_async.py | 25 ++++++++++++++++++
 .../azure-storage-file/tests/test_share.py    | 18 +++++++++++++
 .../tests/test_share_async.py                 | 22 ++++++++++++++++
 .../storage/queue/_shared/base_client.py      | 26 ++++++++++++++++++-
 .../queue/_shared/base_client_async.py        | 25 ++++++++++++++++++
 .../azure-storage-queue/tests/test_queue.py   | 16 ++++++++++++
 .../tests/test_queue_async.py                 | 17 ++++++++++++
 12 files changed, 258 insertions(+), 3 deletions(-)

diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client.py b/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client.py
index 32f38cbd9b5b..cb8114aafae0 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client.py
@@ -28,7 +28,7 @@
 from azure.core.configuration import Configuration
 from azure.core.exceptions import HttpResponseError
 from azure.core.pipeline import Pipeline
-from azure.core.pipeline.transport import RequestsTransport
+from azure.core.pipeline.transport import RequestsTransport, HttpTransport
 from azure.core.pipeline.policies import (
     RedirectPolicy,
     ContentDecodePolicy,
@@ -221,6 +221,29 @@ def _batch_send(
         except StorageErrorException as error:
             process_storage_error(error)
 
+class TransportWrapper(HttpTransport):
+    """Wrapper class that ensures that an inner client created
+    by a `get_client` method does not close the outer transport for the parent
+    when used in a context manager.
+    """
+    def __init__(self, transport):
+        self._transport = transport
+
+    def send(self, request, **kwargs):
+        return self._transport.send(request, **kwargs)
+
+    def open(self):
+        pass
+
+    def close(self):
+        pass
+
+    def __enter__(self):
+        pass
+
+    def __exit__(self, *args):  # pylint: disable=arguments-differ
+        pass
+
 
 def format_shared_key_credential(account, credential):
     if isinstance(credential, six.string_types):
diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client_async.py b/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client_async.py
index 0fe9eaa90785..215292a64f3e 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client_async.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client_async.py
@@ -18,6 +18,7 @@
     AsyncRedirectPolicy,
     DistributedTracingPolicy
 )
+from azure.core.pipeline.transport import AsyncHttpTransport
 
 from .constants import STORAGE_OAUTH_SCOPE, DEFAULT_SOCKET_TIMEOUT
 from .authentication import SharedKeyCredentialPolicy
@@ -126,3 +127,27 @@ async def _batch_send(
             return response.parts()  # Return an AsyncIterator
         except StorageErrorException as error:
             process_storage_error(error)
+
+
+class AsyncTransportWrapper(AsyncHttpTransport):
+    """Wrapper class that ensures that an inner client created
+    by a `get_client` method does not close the outer transport for the parent
+    when used in a context manager.
+    """
+    def __init__(self, async_transport):
+        self._transport = async_transport
+
+    async def send(self, request, **kwargs):
+        return await self._transport.send(request, **kwargs)
+
+    async def open(self):
+        pass
+
+    async def close(self):
+        pass
+
+    async def __aenter__(self):
+        pass
+
+    async def __aexit__(self, *args):  # pylint: disable=arguments-differ
+        pass
diff --git a/sdk/storage/azure-storage-blob/tests/test_common_blob.py b/sdk/storage/azure-storage-blob/tests/test_common_blob.py
index 92f66abad986..d9b09a11d593 100644
--- a/sdk/storage/azure-storage-blob/tests/test_common_blob.py
+++ b/sdk/storage/azure-storage-blob/tests/test_common_blob.py
@@ -17,6 +17,7 @@
     ResourceNotFoundError,
     ResourceExistsError,
     ClientAuthenticationError)
+from azure.core.pipeline.transport import RequestsTransport
 from azure.storage.blob import (
     upload_blob_to_url,
     download_blob_from_url,
@@ -1833,6 +1834,21 @@ def test_set_blob_permission(self):
         self.assertEqual(permission.delete, True)
         self.assertEqual(permission.write, True)
         self.assertEqual(permission._str, 'wrdx')
+    
+    def test_transport_closed_only_once(self):
+        if TestMode.need_recording_file(self.test_mode):
+            return
+        transport = RequestsTransport()
+        url = self._get_account_url()
+        credential = self._get_shared_key_credential()
+        blob_name = self._get_blob_reference()
+        with BlobServiceClient(url, credential=credential, transport=transport) as bsc:
+            bsc.get_service_properties()
+            assert transport.session is not None
+            with bsc.get_blob_client(self.container_name, blob_name) as bc:
+                assert transport.session is not None
+            bsc.get_service_properties()
+            assert transport.session is not None
 
 #------------------------------------------------------------------------------
 if __name__ == '__main__':
diff --git a/sdk/storage/azure-storage-blob/tests/test_common_blob_async.py b/sdk/storage/azure-storage-blob/tests/test_common_blob_async.py
index dd732030b4de..ee1118a21cb2 100644
--- a/sdk/storage/azure-storage-blob/tests/test_common_blob_async.py
+++ b/sdk/storage/azure-storage-blob/tests/test_common_blob_async.py
@@ -18,7 +18,7 @@
     ResourceNotFoundError,
     ResourceExistsError,
     ClientAuthenticationError)
-
+from azure.core.pipeline.transport import AsyncioRequestsTransport
 from azure.core.pipeline.transport import AioHttpTransport
 from multidict import CIMultiDict, CIMultiDictProxy
 
@@ -2281,6 +2281,25 @@ def test_upload_to_url_file_with_credential(self):
         loop = asyncio.get_event_loop()
         loop.run_until_complete(self._test_upload_to_url_file_with_credential())
 
+    async def _test_transport_closed_only_once(self):
+        if TestMode.need_recording_file(self.test_mode):
+            return
+        transport = AioHttpTransport()
+        url = self._get_account_url()
+        credential = self._get_shared_key_credential()
+        blob_name = self._get_blob_reference()
+        async with BlobServiceClient(url, credential=credential, transport=transport) as bsc:
+            await bsc.get_service_properties()
+            assert transport.session is not None
+            async with bsc.get_blob_client(self.container_name, blob_name) as bc:
+                assert transport.session is not None
+            await bsc.get_service_properties()
+            assert transport.session is not None
+
+    @record
+    def test_transport_closed_only_once(self):
+        loop = asyncio.get_event_loop()
+        loop.run_until_complete(self._test_transport_closed_only_once())
 # ------------------------------------------------------------------------------
 if __name__ == '__main__':
     unittest.main()
diff --git a/sdk/storage/azure-storage-file/azure/storage/file/_shared/base_client.py b/sdk/storage/azure-storage-file/azure/storage/file/_shared/base_client.py
index 85111a60c90a..6cdf780498d6 100644
--- a/sdk/storage/azure-storage-file/azure/storage/file/_shared/base_client.py
+++ b/sdk/storage/azure-storage-file/azure/storage/file/_shared/base_client.py
@@ -28,6 +28,7 @@
 from azure.core.configuration import Configuration
 from azure.core.exceptions import HttpResponseError
 from azure.core.pipeline import Pipeline
+from azure.core.pipeline.transport import RequestsTransport, HttpTransport
 from azure.core.pipeline.transport import RequestsTransport
 from azure.core.pipeline.policies import (
     RedirectPolicy,
@@ -221,6 +222,30 @@ def _batch_send(
             process_storage_error(error)
 
 
+class TransportWrapper(HttpTransport):
+    """Wrapper class that ensures that an inner client created
+    by a `get_client` method does not close the outer transport for the parent
+    when used in a context manager.
+    """
+    def __init__(self, transport):
+        self._transport = transport
+
+    def send(self, request, **kwargs):
+        return self._transport.send(request, **kwargs)
+
+    def open(self):
+        pass
+
+    def close(self):
+        pass
+
+    def __enter__(self):
+        pass
+
+    def __exit__(self, *args):  # pylint: disable=arguments-differ
+        pass
+
+
 def format_shared_key_credential(account, credential):
     if isinstance(credential, six.string_types):
         if len(account) < 2:
diff --git a/sdk/storage/azure-storage-file/azure/storage/file/_shared/base_client_async.py b/sdk/storage/azure-storage-file/azure/storage/file/_shared/base_client_async.py
index 32a03fcf4f7a..77f6e48ef5ab 100644
--- a/sdk/storage/azure-storage-file/azure/storage/file/_shared/base_client_async.py
+++ b/sdk/storage/azure-storage-file/azure/storage/file/_shared/base_client_async.py
@@ -19,6 +19,7 @@
     DistributedTracingPolicy
 )
 
+from azure.core.pipeline.transport import AsyncHttpTransport
 from .constants import STORAGE_OAUTH_SCOPE, DEFAULT_SOCKET_TIMEOUT
 from .authentication import SharedKeyCredentialPolicy
 from .base_client import create_configuration
@@ -123,3 +124,27 @@ async def _batch_send(
             return response.parts()  # Return an AsyncIterator
         except StorageErrorException as error:
             process_storage_error(error)
+
+
+class AsyncTransportWrapper(AsyncHttpTransport):
+    """Wrapper class that ensures that an inner client created
+    by a `get_client` method does not close the outer transport for the parent
+    when used in a context manager.
+    """
+    def __init__(self, async_transport):
+        self._transport = async_transport
+
+    async def send(self, request, **kwargs):
+        return await self._transport.send(request, **kwargs)
+
+    async def open(self):
+        pass
+
+    async def close(self):
+        pass
+
+    async def __aenter__(self):
+        pass
+
+    async def __aexit__(self, *args):  # pylint: disable=arguments-differ
+        pass
diff --git a/sdk/storage/azure-storage-file/tests/test_share.py b/sdk/storage/azure-storage-file/tests/test_share.py
index 53f6063e2e13..7d1b636ac016 100644
--- a/sdk/storage/azure-storage-file/tests/test_share.py
+++ b/sdk/storage/azure-storage-file/tests/test_share.py
@@ -10,6 +10,7 @@
 
 import pytest
 import requests
+from azure.core.pipeline.transport import RequestsTransport
 from azure.core.exceptions import (
     HttpResponseError,
     ResourceNotFoundError,
@@ -762,6 +763,23 @@ def test_create_permission_for_share(self):
         # server returned permission
         self.assertEquals(permission_key, permission_key2)
 
+    @record
+    def test_transport_closed_only_once(self):
+        if TestMode.need_recording_file(self.test_mode):
+            return
+        transport = RequestsTransport()
+        url = self.get_file_url()
+        credential = self.get_shared_key_credential()
+        prefix = TEST_SHARE_PREFIX
+        share_name = self.get_resource_name(prefix)
+        with FileServiceClient(url, credential=credential, transport=transport) as fsc:
+            fsc.get_service_properties()
+            assert transport.session is not None
+            with fsc.get_share_client(share_name) as fc:
+                assert transport.session is not None
+            fsc.get_service_properties()
+            assert transport.session is not None
+
 # ------------------------------------------------------------------------------
 if __name__ == '__main__':
     unittest.main()
diff --git a/sdk/storage/azure-storage-file/tests/test_share_async.py b/sdk/storage/azure-storage-file/tests/test_share_async.py
index fe4d3687d017..a6974b181de9 100644
--- a/sdk/storage/azure-storage-file/tests/test_share_async.py
+++ b/sdk/storage/azure-storage-file/tests/test_share_async.py
@@ -11,6 +11,7 @@
 import pytest
 import requests
 from azure.core.pipeline.transport import AioHttpTransport
+from azure.core.pipeline.transport import AsyncioRequestsTransport
 from multidict import CIMultiDict, CIMultiDictProxy
 from azure.core.exceptions import (
     HttpResponseError,
@@ -918,6 +919,27 @@ def test_create_permission_for_share_async(self):
         loop = asyncio.get_event_loop()
         loop.run_until_complete(self._test_create_permission_for_share())
 
+    async def _test_transport_closed_only_once_async(self):
+        if TestMode.need_recording_file(self.test_mode):
+            return
+        transport = AioHttpTransport()
+        url = self.get_file_url()
+        credential = self.get_shared_key_credential()
+        prefix = TEST_SHARE_PREFIX
+        share_name = self.get_resource_name(prefix)
+        async with FileServiceClient(url, credential=credential, transport=transport) as fsc:
+            await fsc.get_service_properties()
+            assert transport.session is not None
+            async with fsc.get_share_client(share_name) as fc:
+                assert transport.session is not None
+            await fsc.get_service_properties()
+            assert transport.session is not None
+
+    @record
+    def test_transport_closed_only_once_async(self):
+        loop = asyncio.get_event_loop()
+        loop.run_until_complete(self._test_transport_closed_only_once_async())
+
 # ------------------------------------------------------------------------------
 if __name__ == '__main__':
     unittest.main()
diff --git a/sdk/storage/azure-storage-queue/azure/storage/queue/_shared/base_client.py b/sdk/storage/azure-storage-queue/azure/storage/queue/_shared/base_client.py
index 85111a60c90a..3ef7c427560b 100644
--- a/sdk/storage/azure-storage-queue/azure/storage/queue/_shared/base_client.py
+++ b/sdk/storage/azure-storage-queue/azure/storage/queue/_shared/base_client.py
@@ -28,7 +28,7 @@
 from azure.core.configuration import Configuration
 from azure.core.exceptions import HttpResponseError
 from azure.core.pipeline import Pipeline
-from azure.core.pipeline.transport import RequestsTransport
+from azure.core.pipeline.transport import RequestsTransport, HttpTransport
 from azure.core.pipeline.policies import (
     RedirectPolicy,
     ContentDecodePolicy,
@@ -221,6 +221,30 @@ def _batch_send(
             process_storage_error(error)
 
 
+class TransportWrapper(HttpTransport):
+    """Wrapper class that ensures that an inner client created
+    by a `get_client` method does not close the outer transport for the parent
+    when used in a context manager.
+    """
+    def __init__(self, transport):
+        self._transport = transport
+
+    def send(self, request, **kwargs):
+        return self._transport.send(request, **kwargs)
+
+    def open(self):
+        pass
+
+    def close(self):
+        pass
+
+    def __enter__(self):
+        pass
+
+    def __exit__(self, *args):  # pylint: disable=arguments-differ
+        pass
+
+
 def format_shared_key_credential(account, credential):
     if isinstance(credential, six.string_types):
         if len(account) < 2:
diff --git a/sdk/storage/azure-storage-queue/azure/storage/queue/_shared/base_client_async.py b/sdk/storage/azure-storage-queue/azure/storage/queue/_shared/base_client_async.py
index 32a03fcf4f7a..77f6e48ef5ab 100644
--- a/sdk/storage/azure-storage-queue/azure/storage/queue/_shared/base_client_async.py
+++ b/sdk/storage/azure-storage-queue/azure/storage/queue/_shared/base_client_async.py
@@ -19,6 +19,7 @@
     DistributedTracingPolicy
 )
 
+from azure.core.pipeline.transport import AsyncHttpTransport
 from .constants import STORAGE_OAUTH_SCOPE, DEFAULT_SOCKET_TIMEOUT
 from .authentication import SharedKeyCredentialPolicy
 from .base_client import create_configuration
@@ -123,3 +124,27 @@ async def _batch_send(
             return response.parts()  # Return an AsyncIterator
         except StorageErrorException as error:
             process_storage_error(error)
+
+
+class AsyncTransportWrapper(AsyncHttpTransport):
+    """Wrapper class that ensures that an inner client created
+    by a `get_client` method does not close the outer transport for the parent
+    when used in a context manager.
+    """
+    def __init__(self, async_transport):
+        self._transport = async_transport
+
+    async def send(self, request, **kwargs):
+        return await self._transport.send(request, **kwargs)
+
+    async def open(self):
+        pass
+
+    async def close(self):
+        pass
+
+    async def __aenter__(self):
+        pass
+
+    async def __aexit__(self, *args):  # pylint: disable=arguments-differ
+        pass
diff --git a/sdk/storage/azure-storage-queue/tests/test_queue.py b/sdk/storage/azure-storage-queue/tests/test_queue.py
index 0525c6f644a2..c36feaf7082e 100644
--- a/sdk/storage/azure-storage-queue/tests/test_queue.py
+++ b/sdk/storage/azure-storage-queue/tests/test_queue.py
@@ -18,6 +18,7 @@
 )
 from devtools_testutils import ResourceGroupPreparer, StorageAccountPreparer
 from azure.mgmt.storage.models import Endpoints
+from azure.core.pipeline.transport import RequestsTransport
 from azure.core.exceptions import (
     HttpResponseError,
     ResourceNotFoundError,
@@ -963,6 +964,21 @@ def test_unicode_update_message_unicode_data(self, resource_group, location, sto
         self.assertIsInstance(message.expires_on, datetime)
         self.assertIsInstance(message.next_visible_on, datetime)
 
+    @ResourceGroupPreparer()
+    @StorageAccountPreparer(name_prefix='pyacrstorage')
+    def test_transport_closed_only_once(self, resource_group, location, storage_account, storage_account_key):
+        if not self.is_live:
+            return
+        transport = RequestsTransport()
+        prefix = TEST_QUEUE_PREFIX
+        queue_name = self.get_resource_name(prefix)
+        with QueueServiceClient(self._account_url(storage_account.name), credential=storage_account_key, transport=transport) as qsc:
+            qsc.get_service_properties()
+            assert transport.session is not None
+            with qsc.get_queue_client(queue_name) as qc:
+                assert transport.session is not None
+            qsc.get_service_properties()
+            assert transport.session is not None
 
 # ------------------------------------------------------------------------------
 if __name__ == '__main__':
diff --git a/sdk/storage/azure-storage-queue/tests/test_queue_async.py b/sdk/storage/azure-storage-queue/tests/test_queue_async.py
index 3de50da06d05..8f528532f4a7 100644
--- a/sdk/storage/azure-storage-queue/tests/test_queue_async.py
+++ b/sdk/storage/azure-storage-queue/tests/test_queue_async.py
@@ -16,6 +16,7 @@
 )
 from devtools_testutils import ResourceGroupPreparer, StorageAccountPreparer
 from multidict import CIMultiDict, CIMultiDictProxy
+from azure.core.pipeline.transport import AsyncioRequestsTransport
 from azure.core.exceptions import (
     HttpResponseError,
     ResourceNotFoundError,
@@ -1025,6 +1026,22 @@ async def test_unicode_update_message_unicode_data(self, resource_group, locatio
         self.assertIsInstance(message.expires_on, datetime)
         self.assertIsInstance(message.next_visible_on, datetime)
 
+    @ResourceGroupPreparer()
+    @StorageAccountPreparer(name_prefix='pyacrstorage')
+    @AsyncQueueTestCase.await_prepared_test
+    async def test_transport_closed_only_once_async(self, resource_group, location, storage_account, storage_account_key):
+        if not self.is_live:
+            return
+        transport = AioHttpTransport()
+        prefix = TEST_QUEUE_PREFIX
+        queue_name = self.get_resource_name(prefix)
+        async with QueueServiceClient(self._account_url(storage_account.name), credential=storage_account_key, transport=transport) as qsc:
+            await qsc.get_service_properties()
+            assert transport.session is not None
+            async with qsc.get_queue_client(queue_name) as qc:
+                assert transport.session is not None
+            await qsc.get_service_properties()
+            assert transport.session is not None
 
 # ------------------------------------------------------------------------------
 if __name__ == '__main__':

From 9ee12b16dec45803cc247c9fcd8db87abb24c7f6 Mon Sep 17 00:00:00 2001
From: Krista Pratico <krpratic@microsoft.com>
Date: Fri, 18 Oct 2019 18:45:33 -0700
Subject: [PATCH 6/6] pylint

---
 .../azure/storage/blob/_shared/base_client_async.py              | 1 -
 .../azure-storage-file/azure/storage/file/_shared/base_client.py | 1 -
 2 files changed, 2 deletions(-)

diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client_async.py b/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client_async.py
index 791d01e50c95..215292a64f3e 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client_async.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client_async.py
@@ -130,7 +130,6 @@ async def _batch_send(
 
 
 class AsyncTransportWrapper(AsyncHttpTransport):
-
     """Wrapper class that ensures that an inner client created
     by a `get_client` method does not close the outer transport for the parent
     when used in a context manager.
diff --git a/sdk/storage/azure-storage-file/azure/storage/file/_shared/base_client.py b/sdk/storage/azure-storage-file/azure/storage/file/_shared/base_client.py
index 0bb945d26da2..2b9fade672ea 100644
--- a/sdk/storage/azure-storage-file/azure/storage/file/_shared/base_client.py
+++ b/sdk/storage/azure-storage-file/azure/storage/file/_shared/base_client.py
@@ -29,7 +29,6 @@
 from azure.core.exceptions import HttpResponseError
 from azure.core.pipeline import Pipeline
 from azure.core.pipeline.transport import RequestsTransport, HttpTransport
-from azure.core.pipeline.transport import RequestsTransport
 from azure.core.pipeline.policies import (
     RedirectPolicy,
     ContentDecodePolicy,