[Core][Corehttp] Add streaming request tests (#34725)

We already have some tests that test creating HttpRequests with
iterable/generator content, but none actually send the requests through
our transports.

This adds some tests that will do so.

Signed-off-by: Paul Van Eck <[email protected]>

Author: pvaneck

sdk/core/azure-core/tests/async_tests/test_streaming_async.py

@@ -315,3 +315,35 @@ async def test_decompress_compressed_header_offline(port, http_request):
         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_streaming_request_iterable(port, http_request):
+    url = "http://localhost:{}/streams/upload".format(port)
+
+    class Content:
+        async def __aiter__(self):
+            yield b"test 123"
+
+    client = AsyncPipelineClient("")
+    request = http_request(method="POST", url=url, data=Content())
+    response = await client.send_request(request)
+    response.raise_for_status()
+    assert response.text() == "test 123"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+async def test_streaming_request_generator(port, http_request):
+    url = "http://localhost:{}/streams/upload".format(port)
+
+    async def content():
+        yield b"test 123"
+        yield b"test 456"
+
+    client = AsyncPipelineClient("")
+    request = http_request(method="POST", url=url, data=content())
+    response = await client.send_request(request)
+    response.raise_for_status()
+    assert response.text() == "test 123test 456"

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

@@ -277,3 +277,33 @@ def test_compress_compressed_header_offline(port, http_request):
     content = b"".join(list(data))
     with pytest.raises(UnicodeDecodeError):
         content.decode("utf-8")
+
+
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+def test_streaming_request_iterable(port, http_request):
+    url = "http://localhost:{}/streams/upload".format(port)
+
+    class Content:
+        def __iter__(self):
+            yield b"test 123"
+
+    client = PipelineClient("")
+    request = http_request(method="POST", url=url, data=Content())
+    response = client.send_request(request)
+    response.raise_for_status()
+    assert response.text() == "test 123"
+
+
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+def test_streaming_request_generator(port, http_request):
+    url = "http://localhost:{}/streams/upload".format(port)
+
+    def content():
+        yield b"test 123"
+        yield b"test 456"
+
+    client = PipelineClient("")
+    request = http_request(method="POST", url=url, data=content())
+    response = client.send_request(request)
+    response.raise_for_status()
+    assert response.text() == "test 123test 456"

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

@@ -10,6 +10,7 @@
 from flask import (
     Response,
     Blueprint,
+    request,
 )
 
 streams_api = Blueprint("streams_api", __name__)
@@ -91,3 +92,15 @@ def compressed_stream():
 @streams_api.route("/decompress_header", methods=["GET"])
 def decompress_header():
     return Response(compressed_stream(), status=200, headers={"Content-Encoding": "gzip"})
+
+
+@streams_api.route("/upload", methods=["POST"])
+def upload():
+    chunk_size = 1024
+    byte_content = b""
+    while True:
+        chunk = request.stream.read(chunk_size)
+        if len(chunk) == 0:
+            break
+        byte_content += chunk
+    return Response(byte_content, status=200)

sdk/core/corehttp/tests/async_tests/test_streaming_async.py

@@ -6,7 +6,7 @@
 """
 Asynchronous streaming tests.
 
-Test naming convention: test_{1}_{2}
+Test naming convention for streaming response tests: test_{1}_{2}
 
 1:
 compress or decompress. Refers to the stream that is returned from the testserver / streams.py
@@ -294,3 +294,35 @@ async def test_decompress_compressed_header_offline(port, transport):
         data = response.iter_bytes()
         decoded = b"".join([d async for d in data]).decode("utf-8")
         assert decoded == "test"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("transport", ASYNC_TRANSPORTS)
+async def test_streaming_request_iterable(port, transport):
+    url = "http://localhost:{}/streams/upload".format(port)
+
+    class Content:
+        async def __aiter__(self):
+            yield b"test 123"
+
+    client = AsyncPipelineClient(url, transport=transport())
+    request = HttpRequest("POST", url=url, content=Content())
+    response = await client.send_request(request)
+    response.raise_for_status()
+    assert response.text() == "test 123"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("transport", ASYNC_TRANSPORTS)
+async def test_streaming_request_generator(port, transport):
+    url = "http://localhost:{}/streams/upload".format(port)
+
+    async def content():
+        yield b"test 123"
+        yield b"test 456"
+
+    client = AsyncPipelineClient(url, transport=transport())
+    request = HttpRequest("POST", url=url, content=content())
+    response = await client.send_request(request)
+    response.raise_for_status()
+    assert response.text() == "test 123test 456"

sdk/core/corehttp/tests/test_streaming.py

@@ -6,7 +6,7 @@
 """
 Streaming tests.
 
-Test naming convention: test_{1}_{2}
+Test naming convention for streaming response tests: test_{1}_{2}
 
 1:
 compress or decompress. Refers to the stream that is returned from the testserver / streams.py
@@ -259,3 +259,33 @@ def test_compress_compressed_header_offline(port, transport):
     content = b"".join(list(data))
     with pytest.raises(UnicodeDecodeError):
         content.decode("utf-8")
+
+
+@pytest.mark.parametrize("transport", SYNC_TRANSPORTS)
+def test_streaming_request_iterable(port, transport):
+    url = "http://localhost:{}/streams/upload".format(port)
+
+    class Content:
+        def __iter__(self):
+            yield b"test 123"
+
+    client = PipelineClient(url, transport=transport())
+    request = HttpRequest("POST", url=url, content=Content())
+    response = client.send_request(request)
+    response.raise_for_status()
+    assert response.text() == "test 123"
+
+
+@pytest.mark.parametrize("transport", SYNC_TRANSPORTS)
+def test_streaming_request_generator(port, transport):
+    url = "http://localhost:{}/streams/upload".format(port)
+
+    def content():
+        yield b"test 123"
+        yield b"test 456"
+
+    client = PipelineClient(url, transport=transport())
+    request = HttpRequest("POST", url=url, content=content())
+    response = client.send_request(request)
+    response.raise_for_status()
+    assert response.text() == "test 123test 456"

sdk/core/corehttp/tests/testserver_tests/coretestserver/coretestserver/test_routes/streams.py

@@ -9,6 +9,7 @@
 from flask import (
     Response,
     Blueprint,
+    request,
 )
 
 streams_api = Blueprint("streams_api", __name__)
@@ -90,3 +91,15 @@ def compressed_stream():
 @streams_api.route("/decompress_header", methods=["GET"])
 def decompress_header():
     return Response(compressed_stream(), status=200, headers={"Content-Encoding": "gzip"})
+
+
+@streams_api.route("/upload", methods=["POST"])
+def upload():
+    chunk_size = 1024
+    byte_content = b""
+    while True:
+        chunk = request.stream.read(chunk_size)
+        if len(chunk) == 0:
+            break
+        byte_content += chunk
+    return Response(byte_content, status=200)