diff --git a/sdk/core/azure-core/tests/async_tests/test_streaming_async.py b/sdk/core/azure-core/tests/async_tests/test_streaming_async.py
index b7b4f8cc17d..8e005344561 100644
--- a/sdk/core/azure-core/tests/async_tests/test_streaming_async.py
+++ b/sdk/core/azure-core/tests/async_tests/test_streaming_async.py
@@ -24,6 +24,7 @@
 #
 # --------------------------------------------------------------------------
 import os
+import zlib
 import pytest
 from azure.core import AsyncPipelineClient
 from azure.core.exceptions import DecodeError
@@ -121,7 +122,6 @@ async def test_compress_compressed_no_header(http_request):
 @pytest.mark.parametrize("http_request", HTTP_REQUESTS)
 async def test_decompress_plain_header(http_request):
     # expect error
-    import zlib
 
     account_name = "coretests"
     account_url = "https://{}.blob.core.windows.net".format(account_name)
@@ -181,6 +181,20 @@ async def test_decompress_compressed_header(http_request):
         assert decoded == "test"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+async def test_compress_compressed_no_header_offline(port, http_request):
+    # expect compressed text
+    client = AsyncPipelineClient("")
+    async with client:
+        request = http_request(method="GET", url="http://localhost:{}/streams/compressed_no_header".format(port))
+        pipeline_response = await client._pipeline.run(request, stream=True)
+        response = pipeline_response.http_response
+        data = response.stream_download(client._pipeline, decompress=False)
+        with pytest.raises(UnicodeDecodeError):
+            b"".join([d async for d in data]).decode("utf-8")
+
+
 @pytest.mark.live_test_only
 @pytest.mark.asyncio
 @pytest.mark.parametrize("http_request", HTTP_REQUESTS)
@@ -195,11 +209,103 @@ async def test_compress_compressed_header(http_request):
         pipeline_response = await client._pipeline.run(request, stream=True)
         response = pipeline_response.http_response
         data = response.stream_download(client._pipeline, decompress=False)
-        content = b""
-        async for d in data:
-            content += d
-        try:
-            decoded = content.decode("utf-8")
-            assert False
-        except UnicodeDecodeError:
-            pass
+        with pytest.raises(UnicodeDecodeError):
+            b"".join([d async for d in data]).decode("utf-8")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+async def test_decompress_plain_no_header_offline(port, http_request):
+    # expect plain text
+    client = AsyncPipelineClient("")
+    async with client:
+        request = http_request(method="GET", url="http://localhost:{}/streams/string".format(port))
+        pipeline_response = await client._pipeline.run(request, stream=True)
+        response = pipeline_response.http_response
+        data = response.stream_download(client._pipeline, decompress=True)
+        decoded = b"".join([d async for d in data]).decode("utf-8")
+        assert decoded == "test"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+async def test_compress_plain_header_offline(port, http_request):
+    # expect plain text
+    client = AsyncPipelineClient("")
+    async with client:
+        request = http_request(method="GET", url="http://localhost:{}/streams/plain_header".format(port))
+        pipeline_response = await client._pipeline.run(request, stream=True)
+        response = pipeline_response.http_response
+        data = response.stream_download(client._pipeline, decompress=False)
+        decoded = b"".join([d async for d in data]).decode("utf-8")
+        assert decoded == "test"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+async def test_decompress_compressed_no_header_offline(port, http_request):
+    # expect compressed text
+    client = AsyncPipelineClient("")
+    async with client:
+        request = http_request(method="GET", url="http://localhost:{}/streams/compressed_no_header".format(port))
+        pipeline_response = await client._pipeline.run(request, stream=True)
+        response = pipeline_response.http_response
+        data = response.stream_download(client._pipeline, decompress=True)
+
+        with pytest.raises(UnicodeDecodeError):
+            b"".join([d async for d in data]).decode("utf-8")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+async def test_compress_compressed_header_offline(port, http_request):
+    # expect compressed text
+    client = AsyncPipelineClient("")
+    async with client:
+        request = http_request(method="GET", url="http://localhost:{}/streams/compressed_header".format(port))
+        pipeline_response = await client._pipeline.run(request, stream=True)
+        response = pipeline_response.http_response
+        data = response.stream_download(client._pipeline, decompress=False)
+        with pytest.raises(UnicodeDecodeError):
+            b"".join([d async for d in data]).decode("utf-8")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+async def test_decompress_plain_header_offline(port, http_request):
+    # expect error
+    client = AsyncPipelineClient("")
+    async with client:
+        request = http_request(method="GET", url="http://localhost:{}/streams/compressed".format(port))
+        pipeline_response = await client._pipeline.run(request, stream=True)
+        response = pipeline_response.http_response
+        data = response.stream_download(client._pipeline, decompress=True)
+        with pytest.raises((zlib.error, DecodeError)):
+            b"".join([d async for d in data])
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+async def test_compress_plain_no_header_offline(port, http_request):
+    client = AsyncPipelineClient("")
+    async with client:
+        request = http_request(method="GET", url="http://localhost:{}/streams/string".format(port))
+        pipeline_response = await client._pipeline.run(request, stream=True)
+        response = pipeline_response.http_response
+        data = response.stream_download(client._pipeline, decompress=False)
+        decoded = b"".join([d async for d in data]).decode("utf-8")
+        assert decoded == "test"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+async def test_decompress_compressed_header_offline(port, http_request):
+    # expect compressed text
+    client = AsyncPipelineClient("")
+    async with client:
+        request = http_request(method="GET", url="http://localhost:{}/streams/decompress_header".format(port))
+        pipeline_response = await client._pipeline.run(request, stream=True)
+        response = pipeline_response.http_response
+        data = response.stream_download(client._pipeline, decompress=True)
+        decoded = b"".join([d async for d in data]).decode("utf-8")
+        assert decoded == "test"
diff --git a/sdk/core/azure-core/tests/test_streaming.py b/sdk/core/azure-core/tests/test_streaming.py
index 3ce3099fedd..c636fc2637c 100644
--- a/sdk/core/azure-core/tests/test_streaming.py
+++ b/sdk/core/azure-core/tests/test_streaming.py
@@ -221,3 +221,55 @@ def test_compress_compressed_header(http_request):
         assert False
     except UnicodeDecodeError:
         pass
+
+
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+def test_decompress_plain_no_header_offline(port, http_request):
+    # expect plain text
+    request = http_request(method="GET", url="http://localhost:{}/streams/string".format(port))
+    with RequestsTransport() as sender:
+        response = sender.send(request, stream=True)
+        response.raise_for_status()
+        data = response.stream_download(sender, decompress=True)
+        content = b"".join(list(data))
+        decoded = content.decode("utf-8")
+        assert decoded == "test"
+
+
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+def test_compress_plain_header_offline(port, http_request):
+    # expect plain text
+    request = http_request(method="GET", url="http://localhost:{}/streams/plain_header".format(port))
+    with RequestsTransport() as sender:
+        response = sender.send(request, stream=True)
+        response.raise_for_status()
+        data = response.stream_download(sender, decompress=False)
+        content = b"".join(list(data))
+        decoded = content.decode("utf-8")
+        assert decoded == "test"
+
+
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+def test_decompress_compressed_no_header_offline(port, http_request):
+    # expect compressed text
+    client = PipelineClient("")
+    request = http_request(method="GET", url="http://localhost:{}/streams/compressed_no_header".format(port))
+    response = client._pipeline.run(request, stream=True).http_response
+    response.raise_for_status()
+    data = response.stream_download(client._pipeline, decompress=True)
+    content = b"".join(list(data))
+    with pytest.raises(UnicodeDecodeError):
+        content.decode("utf-8")
+
+
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+def test_compress_compressed_header_offline(port, http_request):
+    # expect compressed text
+    client = PipelineClient("")
+    request = http_request(method="GET", url="http://localhost:{}/streams/compressed_header".format(port))
+    response = client._pipeline.run(request, stream=True).http_response
+    response.raise_for_status()
+    data = response.stream_download(client._pipeline, decompress=False)
+    content = b"".join(list(data))
+    with pytest.raises(UnicodeDecodeError):
+        content.decode("utf-8")
diff --git a/sdk/core/azure-core/tests/testserver_tests/coretestserver/coretestserver/test_routes/streams.py b/sdk/core/azure-core/tests/testserver_tests/coretestserver/coretestserver/test_routes/streams.py
index cdf38a895f5..be13b18b15c 100644
--- a/sdk/core/azure-core/tests/testserver_tests/coretestserver/coretestserver/test_routes/streams.py
+++ b/sdk/core/azure-core/tests/testserver_tests/coretestserver/coretestserver/test_routes/streams.py
@@ -59,11 +59,21 @@ def string():
     return Response(streaming_test(), status=200, mimetype="text/plain")
 
 
+@streams_api.route("/plain_header", methods=["GET"])
+def plain_header():
+    return Response(streaming_test(), status=200, mimetype="text/plain", headers={"Content-Encoding": "gzip"})
+
+
 @streams_api.route("/compressed_no_header", methods=["GET"])
 def compressed_no_header():
     return Response(compressed_stream(), status=300)
 
 
+@streams_api.route("/compressed_header", methods=["GET"])
+def compressed_header():
+    return Response(compressed_stream(), status=200, headers={"Content-Encoding": "gzip"})
+
+
 @streams_api.route("/compressed", methods=["GET"])
 def compressed():
     return Response(stream_compressed_header_error(), status=300, headers={"Content-Encoding": "gzip"})