fix type checks and remove outdated tests

Author: jonathan343

packages/smithy-http/src/smithy_http/aio/crt.py

@@ -14,20 +14,16 @@
     from awscrt import http as crt_http_base
     from awscrt import io as crt_io
     from awscrt.aio.http import (
-        AIOHttp2ClientConnection,
-        AIOHttp2ClientStream,
         AIOHttpClientConnectionUnified,
-        AIOHttpClientStream,
+        AIOHttpClientStreamUnified,
     )
 
 try:
     from awscrt import http as crt_http_base
     from awscrt import io as crt_io
     from awscrt.aio.http import (
-        AIOHttp2ClientConnection,
-        AIOHttp2ClientStream,
         AIOHttpClientConnectionUnified,
-        AIOHttpClientStream,
+        AIOHttpClientStreamUnified,
     )
 
     HAS_CRT = True
@@ -70,7 +66,7 @@ def __init__(
         *,
         status: int,
         fields: Fields,
-        stream: "AIOHttpClientStream | AIOHttp2ClientStream",
+        stream: "AIOHttpClientStreamUnified",
     ) -> None:
         _assert_crt()
         self._status = status
@@ -112,7 +108,7 @@ def __repr__(self) -> str:
 
 
 ConnectionPoolKey = tuple[str, str, int | None]
-ConnectionPoolDict = dict[ConnectionPoolKey, "AIOHttp2ClientConnection"]
+ConnectionPoolDict = dict[ConnectionPoolKey, "AIOHttpClientConnectionUnified"]
 
 
 class AWSCRTHTTPClientConfig(http_interfaces.HTTPClientConfiguration):
@@ -168,7 +164,7 @@ async def send(
         return await self._await_response(crt_stream)
 
     async def _await_response(
-        self, stream: "AIOHttpClientStream | AIOHttp2ClientStream"
+        self, stream: "AIOHttpClientStreamUnified"
     ) -> AWSCRTHTTPResponse:
         status_code = await stream.get_response_status_code()
         headers = await stream.get_response_headers()
@@ -190,15 +186,15 @@ async def _await_response(
 
     async def _create_connection(
         self, url: core_interfaces.URI
-    ) -> "AIOHttp2ClientConnection":
+    ) -> "AIOHttpClientConnectionUnified":
         """Builds and validates connection to ``url``"""
         connection = await self._build_new_connection(url)
         await self._validate_connection(connection)
         return connection
 
     async def _get_connection(
         self, url: core_interfaces.URI
-    ) -> "AIOHttp2ClientConnection":
+    ) -> "AIOHttpClientConnectionUnified":
         # TODO: Use CRT connection pooling instead of this basic kind
         connection_key = (url.scheme, url.host, url.port)
         connection = self._connections.get(connection_key)
@@ -212,7 +208,7 @@ async def _get_connection(
 
     async def _build_new_connection(
         self, url: core_interfaces.URI
-    ) -> "AIOHttp2ClientConnection":
+    ) -> "AIOHttpClientConnectionUnified":
         if url.scheme == "http":
             port = self._HTTP_PORT
             tls_connection_options = None
@@ -229,7 +225,7 @@ async def _build_new_connection(
         if url.port is not None:
             port = url.port
 
-        return await AIOHttp2ClientConnection.new(
+        return await AIOHttpClientConnectionUnified.new(
             bootstrap=self._client_bootstrap,
             host_name=url.host,
             port=port,
@@ -297,11 +293,19 @@ async def _create_body_generator(
         elif isinstance(body, bytearray):
             # Convert bytearray to bytes
             yield bytes(body)
+        elif isinstance(body, AsyncIterable):
+            # Already async iterable, just yield from it.
+            # Check this before AsyncByteStream since AsyncBytesReader implements both.
+            async for chunk in body:
+                if isinstance(chunk, bytearray):
+                    yield bytes(chunk)
+                else:
+                    yield chunk
         elif iscoroutinefunction(getattr(body, "read", None)) and isinstance(
             body,  # type: ignore[reportGeneralTypeIssues]
             core_aio_interfaces.AsyncByteStream,  # type: ignore[reportGeneralTypeIssues]
         ):
-            # AsyncByteStream has async read method
+            # AsyncByteStream has async read method but is not iterable
             while True:
                 chunk = await body.read(65536)  # Read in 64KB chunks
                 if not chunk:
@@ -310,15 +314,8 @@ async def _create_body_generator(
                     yield bytes(chunk)
                 else:
                     yield chunk
-        elif isinstance(body, AsyncIterable):
-            # Already async iterable, just yield from it
-            async for chunk in body:
-                if isinstance(chunk, bytearray):
-                    yield bytes(chunk)
-                else:
-                    yield chunk
         else:
-            # Assume it's a BytesReader, wrap it in AsyncBytesReader
+            # Assume it's a sync BytesReader, wrap it in AsyncBytesReader
             async_reader = AsyncBytesReader(body)
             async for chunk in async_reader:
                 if isinstance(chunk, bytearray):

packages/smithy-http/tests/unit/aio/test_crt.py

@@ -1,25 +1,22 @@
 #  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 #  SPDX-License-Identifier: Apache-2.0
-import asyncio
-from concurrent.futures import Future as ConcurrentFuture
 from copy import deepcopy
 from io import BytesIO
-from unittest.mock import Mock
 
-import pytest
-from awscrt.http import HttpClientStream  # type: ignore
 from smithy_core import URI
 from smithy_http import Fields
 from smithy_http.aio import HTTPRequest
-from smithy_http.aio.crt import AWSCRTHTTPClient, BufferableByteStream, CRTResponseBody
+from smithy_http.aio.crt import AWSCRTHTTPClient
 
 
 def test_deepcopy_client() -> None:
+    """Test that AWSCRTHTTPClient can be deep copied."""
     client = AWSCRTHTTPClient()
     deepcopy(client)
 
 
-async def test_client_marshal_request() -> None:
+def test_client_marshal_request() -> None:
+    """Test that HTTPRequest is correctly marshaled to CRT HttpRequest."""
     client = AWSCRTHTTPClient()
     request = HTTPRequest(
         method="GET",
@@ -29,187 +26,8 @@ async def test_client_marshal_request() -> None:
         body=BytesIO(),
         fields=Fields(),
     )
-    crt_request, _ = await client._marshal_request(request)  # type: ignore
+    crt_request = client._marshal_request(request)  # type: ignore
     assert crt_request.headers.get("host") == "example.com"  # type: ignore
     assert crt_request.headers.get("accept") == "*/*"  # type: ignore
     assert crt_request.method == "GET"  # type: ignore
     assert crt_request.path == "/path?key1=value1&key2=value2"  # type: ignore
-
-
-def test_stream_write() -> None:
-    stream = BufferableByteStream()
-    stream.write(b"foo")
-    assert stream.read() == b"foo"
-
-
-def test_stream_reads_individual_chunks() -> None:
-    stream = BufferableByteStream()
-    stream.write(b"foo")
-    stream.write(b"bar")
-    assert stream.read() == b"foo"
-    assert stream.read() == b"bar"
-
-
-def test_stream_empty_read() -> None:
-    stream = BufferableByteStream()
-    with pytest.raises(BlockingIOError):
-        stream.read()
-
-
-def test_stream_partial_chunk_read() -> None:
-    stream = BufferableByteStream()
-    stream.write(b"foobar")
-    assert stream.read(3) == b"foo"
-    assert stream.read() == b"bar"
-
-
-def test_stream_write_empty_bytes() -> None:
-    stream = BufferableByteStream()
-    stream.write(b"")
-    stream.write(b"foo")
-    stream.write(b"")
-    assert stream.read() == b"foo"
-
-
-def test_stream_write_non_bytes() -> None:
-    stream = BufferableByteStream()
-    with pytest.raises(ValueError):
-        stream.write(memoryview(b"foo"))
-
-
-def test_closed_stream_write() -> None:
-    stream = BufferableByteStream()
-    stream.close()
-    with pytest.raises(IOError):
-        stream.write(b"foo")
-
-
-def test_closed_stream_read() -> None:
-    stream = BufferableByteStream()
-    stream.write(b"foo")
-    stream.close()
-    assert stream.read() == b""
-
-
-def test_done_stream_read() -> None:
-    stream = BufferableByteStream()
-    stream.write(b"foo")
-    stream.end_stream()
-    assert stream.read() == b"foo"
-    assert stream.read() == b""
-
-
-def test_end_empty_stream() -> None:
-    stream = BufferableByteStream()
-    stream.end_stream()
-    assert stream.read() == b""
-
-
-def test_stream_read1() -> None:
-    stream = BufferableByteStream()
-    stream.write(b"foo")
-    stream.write(b"bar")
-    assert stream.read1() == b"foo"
-    assert stream.read1() == b"bar"
-    with pytest.raises(BlockingIOError):
-        stream.read()
-
-
-def test_stream_readinto_memoryview() -> None:
-    buffer = memoryview(bytearray(b"   "))
-    stream = BufferableByteStream()
-    stream.write(b"foobar")
-    stream.readinto(buffer)
-    assert bytes(buffer) == b"foo"
-
-
-def test_stream_readinto_bytearray() -> None:
-    buffer = bytearray(b"   ")
-    stream = BufferableByteStream()
-    stream.write(b"foobar")
-    stream.readinto(buffer)
-    assert bytes(buffer) == b"foo"
-
-
-def test_end_stream() -> None:
-    stream = BufferableByteStream()
-    stream.write(b"foo")
-    stream.end_stream()
-
-    assert not stream.closed
-    assert stream.read() == b"foo"
-    assert stream.closed
-
-
-async def test_response_body_completed_stream() -> None:
-    completion_future = ConcurrentFuture[int]()
-    mock_stream = Mock(spec=HttpClientStream)
-    mock_stream.completion_future = completion_future
-
-    response_body = CRTResponseBody()
-    response_body.set_stream(mock_stream)
-    completion_future.set_result(200)
-
-    assert await response_body.next() == b""
-
-
-async def test_response_body_empty_stream() -> None:
-    completion_future = ConcurrentFuture[int]()
-    mock_stream = Mock(spec=HttpClientStream)
-    mock_stream.completion_future = completion_future
-
-    response_body = CRTResponseBody()
-    response_body.set_stream(mock_stream)
-
-    read_task = asyncio.create_task(response_body.next())
-
-    # Sleep briefly so the read task gets priority. It should
-    # add a chunk future and then await it.
-    await asyncio.sleep(0.01)
-
-    assert len(response_body._chunk_futures) == 1  # type: ignore
-    response_body.on_body(b"foo")
-    assert await read_task == b"foo"
-
-
-async def test_response_body_stream_completion_clears_buffer() -> None:
-    completion_future = ConcurrentFuture[int]()
-    mock_stream = Mock(spec=HttpClientStream)
-    mock_stream.completion_future = completion_future
-
-    response_body = CRTResponseBody()
-    response_body.set_stream(mock_stream)
-
-    read_tasks = (
-        asyncio.create_task(response_body.next()),
-        asyncio.create_task(response_body.next()),
-        asyncio.create_task(response_body.next()),
-        asyncio.create_task(response_body.next()),
-    )
-
-    # Sleep briefly so the read tasks gets priority. It should
-    # add a chunk future and then await it.
-    await asyncio.sleep(0.01)
-
-    assert len(response_body._chunk_futures) == 4  # type: ignore
-    completion_future.set_result(200)
-    await asyncio.sleep(0.01)
-
-    # Tasks should have been drained
-    assert len(response_body._chunk_futures) == 0  # type: ignore
-
-    # Tasks should still be awaited, and should all return empty
-    results = asyncio.gather(*read_tasks)
-    assert results.result() == [b"", b"", b"", b""]
-
-
-async def test_response_body_non_empty_stream() -> None:
-    completion_future = ConcurrentFuture[int]()
-    mock_stream = Mock(spec=HttpClientStream)
-    mock_stream.completion_future = completion_future
-
-    response_body = CRTResponseBody()
-    response_body.set_stream(mock_stream)
-    response_body.on_body(b"foo")
-
-    assert await response_body.next() == b"foo"