Add testing utilities to smithy-http (smithy-lang#588)

Author: alexgromero

packages/smithy-http/.changes/next-release/smithy-http-feature-20251024135122.json

@@ -0,0 +1,4 @@
+{
+  "type": "feature",
+  "description": "Added `MockHTTPClient` for testing SDK clients without making real HTTP requests."
+}

packages/smithy-http/README.md

@@ -1,3 +1,44 @@
 # smithy-http
 
 This package provides primitives and interfaces for http functionality in tooling generated by Smithy.
+
+---
+
+## Testing
+
+The `smithy_http.testing` module provides shared utilities for testing HTTP functionality in smithy-python clients.
+
+### MockHTTPClient
+
+The `MockHTTPClient` allows you to test smithy-python clients without making actual network calls. It implements the `HTTPClient` interface and provides configurable responses for functional testing.
+
+#### Basic Usage                                                                                                                                                                                      
+
+```python                                                                                                                                                                                               
+from smithy_http.testing import MockHTTPClient
+
+# Create mock client and configure responses
+mock_client = MockHTTPClient()
+mock_client.add_response(
+   status=200,
+   headers=[("Content-Type", "application/json")],
+   body=b'{"message": "success"}'
+)
+
+# Use with your smithy-python client
+config = Config(transport=mock_client)
+client = TestSmithyServiceClient(config=config)
+
+# Test your client logic
+result = await client.some_operation({"input": "data"})
+
+# Inspect what requests were made
+assert mock_client.call_count == 1
+captured_request = mock_client.captured_requests[0]
+assert result.message == "success"
+```
+
+### Utilities
+
+- `create_test_request()`: Helper for creating test HTTPRequest objects                                                                                                                              
+- `MockHTTPClientError`: Exception raised when no responses are queued                                                                                                                               

packages/smithy-http/src/smithy_http/testing/__init__.py

@@ -0,0 +1,13 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Shared utilities for testing smithy-python clients with an HTTP transport."""
+
+from .mockhttp import MockHTTPClient, MockHTTPClientError
+from .utils import create_test_request
+
+__all__ = (
+    "MockHTTPClient",
+    "MockHTTPClientError",
+    "create_test_request",
+)

packages/smithy-http/src/smithy_http/testing/mockhttp.py

@@ -0,0 +1,100 @@
+#  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#  SPDX-License-Identifier: Apache-2.0
+
+from collections import deque
+from copy import copy
+from typing import Any
+
+from smithy_core.aio.utils import async_list
+
+from smithy_http import tuples_to_fields
+from smithy_http.aio import HTTPResponse
+from smithy_http.aio.interfaces import HTTPClient, HTTPRequest
+from smithy_http.interfaces import HTTPClientConfiguration, HTTPRequestConfiguration
+
+
+class MockHTTPClient(HTTPClient):
+    """Implementation of :py:class:`.interfaces.HTTPClient` solely for testing purposes.
+
+    Simulates HTTP request/response behavior. Responses are queued in FIFO order and
+    requests are captured for inspection.
+    """
+
+    def __init__(
+        self,
+        *,
+        client_config: HTTPClientConfiguration | None = None,
+    ) -> None:
+        """
+        :param client_config: Configuration that applies to all requests made with this
+        client.
+        """
+        self._client_config = client_config
+        self._response_queue: deque[dict[str, Any]] = deque()
+        self._captured_requests: list[HTTPRequest] = []
+
+    def add_response(
+        self,
+        status: int = 200,
+        headers: list[tuple[str, str]] | None = None,
+        body: bytes = b"",
+    ) -> None:
+        """Queue a response for the next request.
+
+        :param status: HTTP status code.
+        :param headers: HTTP response headers as list of (name, value) tuples.
+        :param body: Response body as bytes.
+        """
+        self._response_queue.append(
+            {
+                "status": status,
+                "headers": headers or [],
+                "body": body,
+            }
+        )
+
+    async def send(
+        self,
+        request: HTTPRequest,
+        *,
+        request_config: HTTPRequestConfiguration | None = None,
+    ) -> HTTPResponse:
+        """Send HTTP request and return configured response.
+
+        :param request: The request including destination URI, fields, payload.
+        :param request_config: Configuration specific to this request.
+        :returns: Pre-configured HTTP response from the queue.
+        :raises MockHTTPClientError: If no responses are queued.
+        """
+        self._captured_requests.append(copy(request))
+
+        # Return next queued response or raise error
+        if self._response_queue:
+            response_data = self._response_queue.popleft()
+            return HTTPResponse(
+                status=response_data["status"],
+                fields=tuples_to_fields(response_data["headers"]),
+                body=async_list([response_data["body"]]),
+                reason=None,
+            )
+        else:
+            raise MockHTTPClientError(
+                "No responses queued in MockHTTPClient. Use add_response() to queue responses."
+            )
+
+    @property
+    def call_count(self) -> int:
+        """The number of requests made to this client."""
+        return len(self._captured_requests)
+
+    @property
+    def captured_requests(self) -> list[HTTPRequest]:
+        """The list of all requests captured by this client."""
+        return self._captured_requests.copy()
+
+    def __deepcopy__(self, memo: Any) -> "MockHTTPClient":
+        return self
+
+
+class MockHTTPClientError(Exception):
+    """Exception raised by MockHTTPClient for test setup issues."""

packages/smithy-http/src/smithy_http/testing/utils.py

@@ -0,0 +1,31 @@
+#  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#  SPDX-License-Identifier: Apache-2.0
+
+from smithy_core import URI
+
+from smithy_http import tuples_to_fields
+from smithy_http.aio import HTTPRequest
+
+
+def create_test_request(
+    method: str = "GET",
+    host: str = "test.aws.dev",
+    path: str | None = None,
+    headers: list[tuple[str, str]] | None = None,
+    body: bytes = b"",
+) -> HTTPRequest:
+    """Create test HTTPRequest with defaults.
+
+    :param method: HTTP method (GET, POST, etc.)
+    :param host: Host name (e.g., "test.aws.dev")
+    :param path: Optional path (e.g., "/users")
+    :param headers: Optional headers as list of (name, value) tuples
+    :param body: Request body as bytes
+    :return: Configured HTTPRequest for testing
+    """
+    return HTTPRequest(
+        destination=URI(host=host, path=path),
+        method=method,
+        fields=tuples_to_fields(headers or []),
+        body=body,
+    )

packages/smithy-http/tests/unit/testing/test_mockhttp.py

@@ -0,0 +1,115 @@
+#  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#  SPDX-License-Identifier: Apache-2.0
+
+import pytest
+from smithy_http.testing import MockHTTPClient, MockHTTPClientError, create_test_request
+
+
+async def test_default_response():
+    # Test error when no responses are queued
+    mock_client = MockHTTPClient()
+    request = create_test_request()
+
+    with pytest.raises(MockHTTPClientError, match="No responses queued"):
+        await mock_client.send(request)
+
+
+async def test_queued_responses_fifo():
+    # Test responses are returned in FIFO order
+    mock_client = MockHTTPClient()
+    mock_client.add_response(status=404, body=b"not found")
+    mock_client.add_response(status=500, body=b"server error")
+
+    request = create_test_request()
+
+    response1 = await mock_client.send(request)
+    assert response1.status == 404
+    assert await response1.consume_body_async() == b"not found"
+
+    response2 = await mock_client.send(request)
+    assert response2.status == 500
+    assert await response2.consume_body_async() == b"server error"
+
+    assert mock_client.call_count == 2
+
+
+async def test_captured_requests():
+    # Test all requests are captured for inspection
+    mock_client = MockHTTPClient()
+    mock_client.add_response()
+    mock_client.add_response()
+
+    request1 = create_test_request(
+        method="GET",
+        host="test.aws.dev",
+    )
+    request2 = create_test_request(
+        method="POST",
+        host="test.aws.dev",
+        body=b'{"name": "test"}',
+    )
+
+    await mock_client.send(request1)
+    await mock_client.send(request2)
+
+    captured = mock_client.captured_requests
+    assert len(captured) == 2
+    assert captured[0].method == "GET"
+    assert captured[1].method == "POST"
+    assert captured[1].body == b'{"name": "test"}'
+
+
+async def test_response_headers():
+    # Test response headers are properly set
+    mock_client = MockHTTPClient()
+    mock_client.add_response(
+        status=201,
+        headers=[
+            ("Content-Type", "application/json"),
+            ("X-Amz-Custom", "test"),
+        ],
+        body=b'{"id": 123}',
+    )
+    request = create_test_request()
+    response = await mock_client.send(request)
+
+    assert response.status == 201
+    assert "Content-Type" in response.fields
+    assert response.fields["Content-Type"].as_string() == "application/json"
+    assert "X-Amz-Custom" in response.fields
+    assert response.fields["X-Amz-Custom"].as_string() == "test"
+
+
+async def test_call_count_tracking():
+    # Test call count is tracked correctly
+    mock_client = MockHTTPClient()
+    mock_client.add_response()
+    mock_client.add_response()
+
+    request = create_test_request()
+
+    assert mock_client.call_count == 0
+
+    await mock_client.send(request)
+    assert mock_client.call_count == 1
+
+    await mock_client.send(request)
+    assert mock_client.call_count == 2
+
+
+async def test_captured_requests_copy():
+    # Test that captured_requests returns a copy to prevent modifications
+    mock_client = MockHTTPClient()
+    mock_client.add_response()
+
+    request = create_test_request()
+
+    await mock_client.send(request)
+
+    captured1 = mock_client.captured_requests
+    captured2 = mock_client.captured_requests
+
+    # Should be different list objects
+    assert captured1 is not captured2
+    # But with same content
+    assert len(captured1) == len(captured2) == 1

packages/smithy-http/tests/unit/testing/test_utils.py

@@ -0,0 +1,42 @@
+#  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#  SPDX-License-Identifier: Apache-2.0
+
+from smithy_http.testing import create_test_request
+
+
+def test_create_test_request_defaults():
+    request = create_test_request()
+
+    assert request.method == "GET"
+    assert request.destination.host == "test.aws.dev"
+    assert request.destination.path is None
+    assert request.body == b""
+    assert len(request.fields) == 0
+
+
+def test_create_test_request_custom_values():
+    request = create_test_request(
+        method="POST",
+        host="api.example.com",
+        path="/users",
+        headers=[
+            ("Content-Type", "application/json"),
+            ("Authorization", "AWS4-HMAC-SHA256"),
+        ],
+        body=b'{"name": "test"}',
+    )
+
+    assert request.method == "POST"
+    assert request.destination.host == "api.example.com"
+    assert request.destination.path == "/users"
+    assert request.body == b'{"name": "test"}'
+
+    assert "Content-Type" in request.fields
+    assert request.fields["Content-Type"].as_string() == "application/json"
+    assert "Authorization" in request.fields
+    assert request.fields["Authorization"].as_string() == "AWS4-HMAC-SHA256"
+
+
+def test_create_test_request_empty_headers():
+    request = create_test_request(headers=[])
+    assert len(request.fields) == 0