[kuhe]

This PR creates a test API for HTTP mocking, which uses the existing capturing transports (smithy-http.test), but adds sugar for declaring what parts of the request/response you're interested in asserting on.

Reviewers

Other than the sample usage below, see new tests in this PR's diff using the mock API by filtering by /functional/.

from aws_sdk_dynamodb.client import DynamoDBClient
from aws_sdk_dynamodb.config import Config
from aws_sdk_dynamodb.models import AttributeValueS, GetItemInput
from smithy_test import mock

client = DynamoDBClient(
    Config.resolve(region="us-west-2", endpoint_uri="https://localhost:8000")
)
controller = mock(client)

controller.expect_request(
    method="POST",
    hostname="localhost",
    port="8000",
    headers={"X-Amz-Target": "DynamoDB_20120810.GetItem"},
    body=lambda body: b'"TableName":"my-table"' in body,
).respond_with(body='{"Item": {"id": {"S": "1"}}}')

output = client.get_item(
    GetItemInput(table_name="my-table", key={"id": AttributeValueS(value="1")})
)
assert output.item == {"id": AttributeValueS(value="1")}

from aws_sdk_dynamodb.aio.client import AsyncDynamoDBClient
from aws_sdk_dynamodb.aio.config import AsyncConfig
from aws_sdk_dynamodb.models import AttributeValueS, GetItemInput
from smithy_test.aio import mock_async

async with AsyncDynamoDBClient(
    await AsyncConfig.resolve(region="us-west-2", endpoint_uri="https://localhost:8000")
) as client:
    controller = await mock_async(client)

    # Without respond_with, the call returns None.
    controller.expect_request(hostname="/^localhost$/", port="8000")

    assert await client.get_item(GetItemInput(table_name="my-table", key={})) is None

[rohangavankar]

This hangs for event-stream calls.

For event streams, the request body stays open while events are sent. This line waits for it to close before responding. But AsyncDuplexEventStream.close() (called by async with) waits for the response before closing it, and so does a test that calls await_output() first.
The mock wants to see the whole request first, then respond. But in these streams the test (or close()) wants the response first, then finishes the request. Each order makes sense alone; together they wait on each other forever.

async with await client.duplex_stream(DuplexStreamInput()) as stream:
    await stream.input_stream.send(event)
# never returns

Suggested fix: for stream bodies, respond right away and read the body in the background:

body = request.body
if isinstance(body, AsyncIterable) or inspect.iscoroutinefunction(
    getattr(body, "read", None)
):
    captured.body = b""

    async def drain() -> None:
        captured.body = await read_streaming_blob_async(body)

    self._drains.append(asyncio.create_task(drain()))
else:
    captured.body = await read_streaming_blob_async(body)