Issue #9324 create test_compress_compressed_no_header_offline

sdk/core/azure-core/tests/test_streaming.py

@@ -26,7 +26,7 @@
 import pytest
 from azure.core.pipeline.transport import RequestsTransport
 from azure.core import PipelineClient
-from azure.core.exceptions import DecodeError
+from azure.core.exceptions import DecodeError, HttpResponseError
 from azure.core.pipeline.transport import RequestsTransport
 from utils import HTTP_REQUESTS
 
@@ -92,6 +92,19 @@ def test_decompress_compressed_no_header(http_request):
     except UnicodeDecodeError:
         pass
 
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+def test_compress_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))
+    pipeline_response = client._pipeline.run(request, stream=True)
+    response = pipeline_response.http_response
+    data = response.stream_download(client._pipeline, decompress=False)
+    content = b"".join(list(data))
+    with pytest.raises(UnicodeDecodeError):
+        decoded = content.decode('utf-8')
+
+@pytest.mark.live_test_only
 @pytest.mark.parametrize("http_request", HTTP_REQUESTS)
 def test_compress_compressed_no_header(http_request):
     # expect compressed text

sdk/core/azure-core/tests/testserver_tests/coretestserver/coretestserver/test_routes/streams.py

@@ -4,6 +4,8 @@
 # Licensed under the MIT License. See LICENSE.txt in the project root for
 # license information.
 # -------------------------------------------------------------------------
+import os
+import gzip
 from flask import (
     Response,
     Blueprint,
@@ -31,6 +33,15 @@ def streaming_test():
 def stream_compressed_header_error():
     yield b'test'
 
+def stream_compressed_no_header():
+    with gzip.open('test.tar.gz', 'wb') as f:
+        f.write(b"test")
+
+    with open(os.path.join(os.path.abspath('test.tar.gz')), "rb") as fd:
+        yield fd.read()
+
+    os.remove("test.tar.gz")
+
 @streams_api.route('/basic', methods=['GET'])
 def basic():
     return Response(streaming_body(), status=200)
@@ -48,7 +59,11 @@ def string():
     return Response(
         streaming_test(), status=200, mimetype="text/plain"
     )
-
+
+@streams_api.route('/compressed_no_header', methods=['GET'])
+def compressed_no_header():
+    return Response(stream_compressed_no_header(), status=300)
+
 @streams_api.route('/compressed', methods=['GET'])
 def compressed():
     return Response(stream_compressed_header_error(), status=300, headers={"Content-Encoding": "gzip"})