diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2ee86abad..80cb8a5d4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -82,7 +82,7 @@ jobs: - name: Run protocol tests run: | - make generate-protocol-tests test-protocols + make generate-protocol-tests test-protocols test-integration - name: Type-check protocol test stubs if: ${{ matrix.python-version == '3.12' }} diff --git a/Makefile b/Makefile index 5b44d9a89..d0e5baefc 100644 --- a/Makefile +++ b/Makefile @@ -1,6 +1,6 @@ .DEFAULT_GOAL:=help -.PHONY: help install format-java check-java build-java test-protocols typecheck-protocols generate-protocol-tests generate-codegen-output-py api-snapshot ci lint-py check-py test-py benchmark build-py clean +.PHONY: help install format-java check-java build-java test-protocols typecheck-protocols generate-protocol-tests generate-codegen-output-py api-snapshot ci lint-py check-py test-py test-integration benchmark build-py clean help: ## Show this help. @awk 'BEGIN {FS = ":.*##"; printf "\nUsage:\n make \033[36m\033[0m\n\nTargets:\n"} /^[a-zA-Z_-]+:.*?##/ { printf " \033[36m%-25s\033[0m %s\n", $$1, $$2 }' $(MAKEFILE_LIST) @@ -98,7 +98,7 @@ ci: ## Runs the full CI sequence locally (mirrors the CI workflow steps). make build-py make build-java make generate-codegen-output-py - make generate-protocol-tests test-protocols + make generate-protocol-tests test-protocols test-integration make typecheck-protocols @@ -106,6 +106,11 @@ test-py: ## Runs tests for the python packages. uv run pytest packages +test-integration: ## Runs integration tests that drive generated clients over a real transport. Requires `make test-protocols` (or generate-protocol-tests) to have installed the generated clients first. + uv pip install h2 trustme "packages/smithy-http[awscrt]" + uv run pytest packages/*/tests/integration --no-cov + + benchmark: ## Runs ser/deser throughput benchmarks (coverage disabled; not part of CI or the default test run). uv run pytest packages/*/tests/benchmark --no-cov -s -q diff --git a/codegen-output/aws-json-1-0/src/awsjson10/config.py b/codegen-output/aws-json-1-0/src/awsjson10/config.py index dd232cff4..00a483f1b 100644 --- a/codegen-output/aws-json-1-0/src/awsjson10/config.py +++ b/codegen-output/aws-json-1-0/src/awsjson10/config.py @@ -34,6 +34,7 @@ from smithy_core.shapes import ShapeID from smithy_http.aio.aiohttp import AIOHTTPClient +from ._private.schemas import JSON_RPC10 from .auth import HTTPAuthSchemeResolver @@ -124,7 +125,9 @@ _ServiceInterceptor = Any _PROTOCOL_SETTINGS = ProtocolSettings( - namespace="aws.protocoltests.json10", service_target="JsonRpc10" + namespace="aws.protocoltests.json10", + service_target="JsonRpc10", + service_schema=JSON_RPC10, ) diff --git a/codegen-output/aws-json-1-1/src/awsjson11/config.py b/codegen-output/aws-json-1-1/src/awsjson11/config.py index 7a8e0b041..28a03ab97 100644 --- a/codegen-output/aws-json-1-1/src/awsjson11/config.py +++ b/codegen-output/aws-json-1-1/src/awsjson11/config.py @@ -34,6 +34,7 @@ from smithy_core.shapes import ShapeID from smithy_http.aio.aiohttp import AIOHTTPClient +from ._private.schemas import JSON_PROTOCOL from .auth import HTTPAuthSchemeResolver @@ -118,7 +119,9 @@ _ServiceInterceptor = Any _PROTOCOL_SETTINGS = ProtocolSettings( - namespace="aws.protocoltests.json", service_target="JsonProtocol" + namespace="aws.protocoltests.json", + service_target="JsonProtocol", + service_schema=JSON_PROTOCOL, ) diff --git a/codegen-output/aws-query/src/awsquery/config.py b/codegen-output/aws-query/src/awsquery/config.py index 3a0526d78..813939d49 100644 --- a/codegen-output/aws-query/src/awsquery/config.py +++ b/codegen-output/aws-query/src/awsquery/config.py @@ -34,6 +34,7 @@ from smithy_core.shapes import ShapeID from smithy_http.aio.aiohttp import AIOHTTPClient +from ._private.schemas import AWS_QUERY from .auth import HTTPAuthSchemeResolver @@ -172,7 +173,10 @@ _ServiceInterceptor = Any _PROTOCOL_SETTINGS = ProtocolSettings( - namespace="aws.protocoltests.query", service_target="AwsQuery", version="2020-01-08" + namespace="aws.protocoltests.query", + service_target="AwsQuery", + version="2020-01-08", + service_schema=AWS_QUERY, ) diff --git a/codegen-output/ec2-query/src/ec2query/config.py b/codegen-output/ec2-query/src/ec2query/config.py index 39fe67fe6..032738972 100644 --- a/codegen-output/ec2-query/src/ec2query/config.py +++ b/codegen-output/ec2-query/src/ec2query/config.py @@ -34,6 +34,7 @@ from smithy_core.shapes import ShapeID from smithy_http.aio.aiohttp import AIOHTTPClient +from ._private.schemas import AWS_EC2 from .auth import HTTPAuthSchemeResolver @@ -141,7 +142,10 @@ _ServiceInterceptor = Any _PROTOCOL_SETTINGS = ProtocolSettings( - namespace="aws.protocoltests.ec2", service_target="AwsEc2", version="2020-01-08" + namespace="aws.protocoltests.ec2", + service_target="AwsEc2", + version="2020-01-08", + service_schema=AWS_EC2, ) diff --git a/codegen-output/my-handwritten-model/README.md b/codegen-output/my-handwritten-model/README.md new file mode 100644 index 000000000..cc10b198b --- /dev/null +++ b/codegen-output/my-handwritten-model/README.md @@ -0,0 +1,5 @@ + + +## My Handwritten Model Client + +my_handwritten_model client diff --git a/codegen-output/my-handwritten-model/pyproject.toml b/codegen-output/my-handwritten-model/pyproject.toml new file mode 100644 index 000000000..844350a25 --- /dev/null +++ b/codegen-output/my-handwritten-model/pyproject.toml @@ -0,0 +1,71 @@ +# Code generated by smithy-python-codegen DO NOT EDIT. + +[project] +name = "my_handwritten_model" +version = "0.0.1" +description = "my_handwritten_model client" +readme = "README.md" +requires-python = ">=3.12" +keywords = ["smithy", "my_handwritten_model"] +license = {text = "Apache-2.0"} +classifiers = [ + "Development Status :: 2 - Pre-Alpha", + "Intended Audience :: Developers", + "Intended Audience :: System Administrators", + "Natural Language :: English", + "License :: OSI Approved :: Apache Software License", + "Programming Language :: Python", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3 :: Only", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", + "Programming Language :: Python :: 3.15" +] + +dependencies = [ + "smithy_aws_core[eventstream]~=0.12.0", + "smithy_core~=0.9.0", + "smithy_http[cbor, aiohttp]~=0.6.0" +] + +[project.optional-dependencies] +awscrt = [ + "smithy_http[awscrt]~=0.6.0" +] + +[dependency-groups] +test = [ + "pytest>=9.0.1,<10.0.0", + "pytest-asyncio>=1.3.0,<1.4.0", + "smithy_test~=0.1.0" +] + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.pyright] +typeCheckingMode = "strict" +reportPrivateUsage = false +reportUnusedFunction = false +reportUnusedVariable = false +reportUnnecessaryComparison = false +reportUnusedClass = false +enableExperimentalFeatures = true +# models.py is the dynamic runtime; models.pyi carries the typed +# surface for consumers, so don't type-check the implementation body. +exclude = ["**/models.py"] + +[tool.ruff] +target-version = "py312" + +[tool.ruff.lint] +ignore = ["F841"] + +[tool.ruff.format] +skip-magic-trailing-comma = true + +[tool.pytest.ini_options] +python_classes = ["!Test"] +asyncio_mode = "auto" diff --git a/codegen-output/my-handwritten-model/src/my_handwritten_model/__init__.py b/codegen-output/my-handwritten-model/src/my_handwritten_model/__init__.py new file mode 100644 index 000000000..30f6d44e9 --- /dev/null +++ b/codegen-output/my-handwritten-model/src/my_handwritten_model/__init__.py @@ -0,0 +1,3 @@ +# Code generated by smithy-python-codegen DO NOT EDIT. + +__version__: str = "0.0.1" diff --git a/codegen-output/my-handwritten-model/src/my_handwritten_model/_private/__init__.py b/codegen-output/my-handwritten-model/src/my_handwritten_model/_private/__init__.py new file mode 100644 index 000000000..247be3e3d --- /dev/null +++ b/codegen-output/my-handwritten-model/src/my_handwritten_model/_private/__init__.py @@ -0,0 +1 @@ +# Code generated by smithy-python-codegen DO NOT EDIT. diff --git a/codegen-output/my-handwritten-model/src/my_handwritten_model/_private/schemas.py b/codegen-output/my-handwritten-model/src/my_handwritten_model/_private/schemas.py new file mode 100644 index 000000000..f9ef0c3bb --- /dev/null +++ b/codegen-output/my-handwritten-model/src/my_handwritten_model/_private/schemas.py @@ -0,0 +1,269 @@ +# Code generated by smithy-python-codegen DO NOT EDIT. + +from types import MappingProxyType + +from smithy_core._schema_compact import StaticOperationSchema, StaticSchema + + +n0 = "smithy.python.tests.handwritten" +_A = "Alpha" +_i = "id" +_t = "timestamp" +_DSN = "DifferentShapeName" +_n = "name" +_nu = "number" +_IL = "IntegerList" +_m = "member" +_GP = "GammaPayload" +_me = "message" +_v = "values" +_G = "Gamma" +_sN = "sequenceNumber" +_sH = "smithy.api#eventHeader" +_p = "payload" +_sP = "smithy.api#eventPayload" +_HE = "HeartbeatEvent" +_LE = "LogEvent" +_l = "level" +_ME = "MetricEvent" +_va = "value" +_PES = "PublishEventStream" +_s = "smithy.api#streaming" +_lo = "log" +_met = "metric" +_PEI = "PublishEventsInput" +_sm = "smithy.api#input" +_c = "channel" +_e = "events" +_PEO = "PublishEventsOutput" +_smi = "smithy.api#output" +_eC = "eventCount" +_PE = "PublishEvents" +_STEI = "SubscribeToEventsInput" +_mE = "maxEvents" +_NE = "NotificationEvent" +_to = "topic" +_SES = "SubscribeEventStream" +_no = "notification" +_h = "heartbeat" +_STEO = "SubscribeToEventsOutput" +_sI = "subscriptionId" +_STE = "SubscribeToEvents" +_SSE = "StreamServiceError" +_smit = "smithy.api#error" +_se = "server" +_TE = "ThrottlingEvent" +_cl = "client" +_TEr = "TradeEvents" +_a = "alpha" +_b = "beta" +_g = "gamma" +_d = "delta" +_th = "throttle" +_TESI = "TradeEventStreamInput" +_sIe = "sessionId" +_TESO = "TradeEventStreamOutput" +_gr = "greeting" +_TES = "TradeEventStream" +_MHM = "MyHandwrittenModel" +_sC = "smithy.protocols#rpcv2Cbor" +_sB = "smithy.rules#endpointBdd" + +ALPHA: StaticSchema = (18, n0, _A, 0, (_i, _t), ((3, 0), (4, 0))) + +DIFFERENT_SHAPE_NAME: StaticSchema = (18, n0, _DSN, 0, (_n, _nu), ((3, 0), (7, 0))) + +INTEGER_LIST: StaticSchema = (16, n0, _IL, 0, (_m,), ((7, 0),)) + +GAMMA_PAYLOAD: StaticSchema = (18, n0, _GP, 0, (_me, _v), ((3, 0), (INTEGER_LIST, 0))) + +GAMMA: StaticSchema = ( + 18, + n0, + _G, + 0, + (_sN, _p), + ((7, {_sH: None}), (GAMMA_PAYLOAD, {_sP: None})), +) + +HEARTBEAT_EVENT: StaticSchema = (18, n0, _HE, 0, (_t,), ((4, 0),)) + +LOG_EVENT: StaticSchema = (18, n0, _LE, 0, (_l, _me), ((3, 0), (3, 0))) + +METRIC_EVENT: StaticSchema = (18, n0, _ME, 0, (_n, _va), ((3, 0), (10, 0))) + +PUBLISH_EVENT_STREAM: StaticSchema = ( + 19, + n0, + _PES, + {_s: None}, + (_lo, _met), + ((LOG_EVENT, 0), (METRIC_EVENT, 0)), +) + +PUBLISH_EVENTS_INPUT: StaticSchema = ( + 18, + n0, + _PEI, + {_sm: None}, + (_c, _e), + ((3, 0), (PUBLISH_EVENT_STREAM, 0)), +) + +PUBLISH_EVENTS_OUTPUT: StaticSchema = ( + 18, + n0, + _PEO, + {_smi: None}, + (_eC, _me), + ((7, 0), (3, 0)), +) + +PUBLISH_EVENTS: StaticOperationSchema = ( + 23, + n0, + _PE, + 0, + PUBLISH_EVENTS_INPUT, + PUBLISH_EVENTS_OUTPUT, + (), +) + +SUBSCRIBE_TO_EVENTS_INPUT: StaticSchema = ( + 18, + n0, + _STEI, + {_sm: None}, + (_c, _mE), + ((3, 0), (7, 0)), +) + +NOTIFICATION_EVENT: StaticSchema = (18, n0, _NE, 0, (_to, _p), ((3, 0), (3, 0))) + +SUBSCRIBE_EVENT_STREAM: StaticSchema = ( + 19, + n0, + _SES, + {_s: None}, + (_no, _h), + ((NOTIFICATION_EVENT, 0), (HEARTBEAT_EVENT, 0)), +) + +SUBSCRIBE_TO_EVENTS_OUTPUT: StaticSchema = ( + 18, + n0, + _STEO, + {_smi: None}, + (_sI, _e), + ((3, 0), (SUBSCRIBE_EVENT_STREAM, 0)), +) + +SUBSCRIBE_TO_EVENTS: StaticOperationSchema = ( + 23, + n0, + _STE, + 0, + SUBSCRIBE_TO_EVENTS_INPUT, + SUBSCRIBE_TO_EVENTS_OUTPUT, + (), +) + +STREAM_SERVICE_ERROR: StaticSchema = (18, n0, _SSE, {_smit: _se}, (_me,), ((3, 0),)) + +THROTTLING_EVENT: StaticSchema = (18, n0, _TE, {_smit: _cl}, (_me,), ((3, 0),)) + +TRADE_EVENTS: StaticSchema = ( + 19, + n0, + _TEr, + {_s: None}, + (_a, _b, _g, _d, _th), + ((ALPHA, 0), (-1, 0), (GAMMA, 0), (DIFFERENT_SHAPE_NAME, 0), (THROTTLING_EVENT, 0)), +) + +TRADE_EVENT_STREAM_INPUT: StaticSchema = ( + 18, + n0, + _TESI, + {_sm: None}, + (_sIe, _e), + ((3, 0), (TRADE_EVENTS, 0)), +) + +TRADE_EVENT_STREAM_OUTPUT: StaticSchema = ( + 18, + n0, + _TESO, + {_smi: None}, + (_gr, _e), + ((3, 0), (TRADE_EVENTS, 0)), +) + +TRADE_EVENT_STREAM: StaticOperationSchema = ( + 23, + n0, + _TES, + 0, + TRADE_EVENT_STREAM_INPUT, + TRADE_EVENT_STREAM_OUTPUT, + (STREAM_SERVICE_ERROR,), +) + +MY_HANDWRITTEN_MODEL: StaticSchema = ( + 21, + n0, + _MHM, + { + _sC: MappingProxyType({"http": ("h2", "http/1.1"), "eventStreamHttp": ("h2",)}), + _sB: MappingProxyType( + { + "version": "1.1", + "parameters": MappingProxyType( + { + "endpoint": MappingProxyType( + { + "builtIn": "SDK::Endpoint", + "required": False, + "documentation": "Endpoint used for making requests. Should be formatted as a URI.", + "type": "string", + } + ) + } + ), + "conditions": ( + MappingProxyType( + { + "fn": "isSet", + "argv": (MappingProxyType({"ref": "endpoint"}),), + } + ), + ), + "results": ( + MappingProxyType( + { + "conditions": (), + "endpoint": MappingProxyType( + { + "url": MappingProxyType({"ref": "endpoint"}), + "properties": MappingProxyType[str, object]({}), + "headers": MappingProxyType[str, object]({}), + } + ), + "type": "endpoint", + } + ), + MappingProxyType( + { + "conditions": (), + "error": "(default endpointRuleSet) endpoint is not set - you must configure an endpoint.", + "type": "error", + } + ), + ), + "root": 2, + "nodeCount": 2, + "nodes": "/////wAAAAH/////AAAAAAX14QEF9eEC", + } + ), + }, +) diff --git a/codegen-output/my-handwritten-model/src/my_handwritten_model/client.py b/codegen-output/my-handwritten-model/src/my_handwritten_model/client.py new file mode 100644 index 000000000..789367846 --- /dev/null +++ b/codegen-output/my-handwritten-model/src/my_handwritten_model/client.py @@ -0,0 +1,205 @@ +# Code generated by smithy-python-codegen DO NOT EDIT. + +from copy import deepcopy +import logging +from typing import Any + +from smithy_core.aio.client import AsyncClient, ClientCall, RequestPipeline +from smithy_core.aio.eventstream import ( + DuplexEventStream, + InputEventStream, + OutputEventStream, +) +from smithy_core.deserializers import DeserializeableShape +from smithy_core.exceptions import ExpectationNotMetError +from smithy_core.schemas import APIOperation +from smithy_core.serializers import SerializeableShape +from smithy_http.plugins import user_agent_plugin + +from .config import Config, Plugin +from .models import ( + PUBLISH_EVENTS, + PublishEventStream, + PublishEventsInput, + PublishEventsOutput, + SUBSCRIBE_TO_EVENTS, + SubscribeEventStream, + SubscribeToEventsInput, + SubscribeToEventsOutput, + TRADE_EVENT_STREAM, + TradeEventStreamInput, + TradeEventStreamOutput, + TradeEvents, + _SubscribeEventStreamDeserializer, + _TradeEventsDeserializer, +) +from .user_agent import aws_user_agent_plugin + + +logger = logging.getLogger(__name__) + + +class MyHandwrittenModel(AsyncClient): + """Client for MyHandwrittenModel""" + + def __init__( + self, config: Config | None = None, plugins: list[Plugin] | None = None + ): + """ + Constructor for `MyHandwrittenModel`. + + Args: + config: + Optional configuration for the client. Here you can set things like + the endpoint for HTTP services or auth credentials. + plugins: + A list of callables applied once to the client's base configuration. + Their changes are inherited by every operation invocation. + """ + super().__init__() + self._config = config + self._plugins = plugins + self._client_plugins: list[Plugin] = [aws_user_agent_plugin, user_agent_plugin] + + async def _ensure_setup(self) -> Config: + if not self._setup_done: + async with self._derive_lock: + if self._closed: + raise RuntimeError( + "Cannot invoke an operation on a client that has been closed." + ) + if not self._setup_done: + if self._config is None: + config = Config() + else: + # Copy so plugins don't mutate the caller's config. + config = deepcopy(self._config) + for plugin in self._client_plugins: + plugin(config) + if self._plugins: + for plugin in self._plugins: + plugin(config) + # Publish state only after setup fully succeeds, so a + # failed _post_setup leaves the caller's config untouched + # (no re-applied plugins) and the transport closeable. + await self._post_setup(config) + self._transport = config.transport + self._config = config + self._setup_done = True + assert self._config is not None + return self._config + + async def _post_setup(self, config: Config) -> None: + pass + + async def _prepare_call[I: SerializeableShape, O: DeserializeableShape]( + self, + input: I, + operation: APIOperation[I, O], + default_plugins: list[Plugin], + plugins: list[Plugin] | None, + ) -> tuple[RequestPipeline[Any, Any], ClientCall[I, O]]: + if self._closed: + raise RuntimeError( + "Cannot invoke an operation on a client that has been closed." + ) + config = await self._ensure_setup() + if default_plugins or plugins: + # Keep operation-plugin mutations scoped to this call. + config = deepcopy(config) + for plugin in default_plugins: + plugin(config) + if plugins: + for plugin in plugins: + plugin(config) + if ( + config.protocol is None + or config.transport is None + or config.endpoint_resolver is None + or config.auth_scheme_resolver is None + or config.auth_schemes is None + ): + raise ExpectationNotMetError( + "protocol, transport, endpoint_resolver, auth_scheme_resolver," + " and auth_schemes MUST be set on the config to make calls." + ) + retry_strategy = await self._retry_strategy_resolver.resolve_retry_strategy( + retry_strategy=config.retry_strategy + ) + return self._build_call( + input, operation, config=config, retry_strategy=retry_strategy + ) + + async def publish_events( + self, input: PublishEventsInput, plugins: list[Plugin] | None = None + ) -> InputEventStream[PublishEventStream, PublishEventsOutput]: + """ + Input-only stream: the client sends events, the server returns a + summary. + + Args: + input: + An instance of `PublishEventsInput`. + plugins: + A list of callables that modify the configuration dynamically. + Changes made by these plugins only apply for the duration of the + operation execution and will not affect any other operation + invocations. + + Returns: + An `InputEventStream` for client-to-server streaming. + """ + pipeline, call = await self._prepare_call(input, PUBLISH_EVENTS, [], plugins) + return await pipeline.input_stream(call, PublishEventStream) + + async def subscribe_to_events( + self, input: SubscribeToEventsInput, plugins: list[Plugin] | None = None + ) -> OutputEventStream[SubscribeEventStream, SubscribeToEventsOutput]: + """ + Output-only stream: the client subscribes, the server streams events. + + Args: + input: + An instance of `SubscribeToEventsInput`. + plugins: + A list of callables that modify the configuration dynamically. + Changes made by these plugins only apply for the duration of the + operation execution and will not affect any other operation + invocations. + + Returns: + An `OutputEventStream` for server-to-client streaming. + """ + pipeline, call = await self._prepare_call( + input, SUBSCRIBE_TO_EVENTS, [], plugins + ) + return await pipeline.output_stream( + call, SubscribeEventStream, _SubscribeEventStreamDeserializer().deserialize + ) + + async def trade_event_stream( + self, input: TradeEventStreamInput, plugins: list[Plugin] | None = None + ) -> DuplexEventStream[TradeEvents, TradeEvents, TradeEventStreamOutput]: + """ + Bidirectional (duplex) stream. The non-stream members (sessionId, + greeting) are carried as the initial-request and initial-response + messages. + + Args: + input: + An instance of `TradeEventStreamInput`. + plugins: + A list of callables that modify the configuration dynamically. + Changes made by these plugins only apply for the duration of the + operation execution and will not affect any other operation + invocations. + + Returns: + A `DuplexEventStream` for bidirectional streaming. + """ + pipeline, call = await self._prepare_call( + input, TRADE_EVENT_STREAM, [], plugins + ) + return await pipeline.duplex_stream( + call, TradeEvents, TradeEvents, _TradeEventsDeserializer().deserialize + ) diff --git a/codegen-output/my-handwritten-model/src/my_handwritten_model/config.py b/codegen-output/my-handwritten-model/src/my_handwritten_model/config.py new file mode 100644 index 000000000..bcc398f3e --- /dev/null +++ b/codegen-output/my-handwritten-model/src/my_handwritten_model/config.py @@ -0,0 +1,166 @@ +# Code generated by smithy-python-codegen DO NOT EDIT. + +from dataclasses import dataclass +from typing import Any, Callable, TYPE_CHECKING, TypeAlias, Union, cast + +from smithy_aws_core.endpoints.standard_regional import ( + StandardRegionalEndpointsResolver, +) +from smithy_core.aio.interfaces import ( + ClientProtocol, + ClientTransport, + EndpointResolver, + ProtocolConstructor, + ProtocolSettings, +) +from smithy_core.aio.interfaces.auth import AuthScheme +from smithy_core.aio.interfaces.retries import RetryStrategy +from smithy_core.auth import DefaultAuthResolver +from smithy_core.interfaces import URI +from smithy_core.interfaces.auth import AuthSchemeResolver +from smithy_core.retries import RetryStrategyOptions +from smithy_core.shapes import ShapeID +from smithy_http.aio.aiohttp import AIOHTTPClient +from smithy_http.aio.protocols import RpcV2CborClientProtocol +from smithy_http.interfaces import HTTPRequestConfiguration + +from ._private.schemas import MY_HANDWRITTEN_MODEL + + +if TYPE_CHECKING: + from smithy_core.interceptors import Interceptor + from .models import ( + PublishEventsInput, + PublishEventsOutput, + SubscribeToEventsInput, + SubscribeToEventsOutput, + TradeEventStreamInput, + TradeEventStreamOutput, + ) + + _ServiceInterceptor = Union[ + Interceptor[PublishEventsInput, PublishEventsOutput, Any, Any], + Interceptor[SubscribeToEventsInput, SubscribeToEventsOutput, Any, Any], + Interceptor[TradeEventStreamInput, TradeEventStreamOutput, Any, Any], + ] +else: + _ServiceInterceptor = Any + +_PROTOCOL_SETTINGS = ProtocolSettings( + namespace="smithy.python.tests.handwritten", + service_target="MyHandwrittenModel", + service_schema=MY_HANDWRITTEN_MODEL, +) + + +@dataclass(init=False) +class Config: + """Configuration for MyHandwrittenModel.""" + + auth_scheme_resolver: AuthSchemeResolver + """ + An auth scheme resolver that determines the auth scheme for each + operation. + """ + + auth_schemes: dict[ShapeID, AuthScheme[Any, Any, Any, Any]] + """A map of auth scheme ids to auth schemes.""" + + endpoint_resolver: EndpointResolver + """ + The endpoint resolver used to resolve the final endpoint per-operation + based on the configuration. + """ + + endpoint_uri: str | URI | None + """A static URI to route requests to.""" + + http_request_config: HTTPRequestConfiguration | None + """Configuration for individual HTTP requests.""" + + interceptors: list[_ServiceInterceptor] + """ + The list of interceptors, which are hooks that are called during the + execution of a request. + """ + + protocol: ClientProtocol[Any, Any] | None + """ + Pass a protocol class reference from smithy_aws_core.aio.protocols to + select the protocol, e.g. protocol=AwsJson10ClientProtocol. For custom + protocols a protocol instance may also be passed. + """ + + region: str | None + """ + The AWS region to connect to. The configured region is used to determine + the service endpoint. + """ + + retry_strategy: RetryStrategy | RetryStrategyOptions | None + """ + The retry strategy or options for configuring retry behavior. Can be + either a configured RetryStrategy or RetryStrategyOptions to create one. + """ + + sdk_ua_app_id: str | None + """ + A unique and opaque application ID that is appended to the User-Agent + header. + """ + + transport: ClientTransport[Any, Any] | None + """ + The transport to use to send requests (e.g. an HTTP client). Operations + with bidirectional event streams require a DuplexClientTransport, such + as AWSCRTHTTPClient. Transports are assumed not to support duplex + streaming unless they explicitly set SUPPORTS_DUPLEX_STREAMING to True. + """ + + user_agent_extra: str | None + """Additional suffix to be added to the User-Agent header.""" + + def __init__( + self, + *, + auth_scheme_resolver: AuthSchemeResolver | None = None, + auth_schemes: dict[ShapeID, AuthScheme[Any, Any, Any, Any]] | None = None, + endpoint_resolver: EndpointResolver | None = None, + endpoint_uri: str | URI | None = None, + http_request_config: HTTPRequestConfiguration | None = None, + interceptors: list[_ServiceInterceptor] | None = None, + protocol: ClientProtocol[Any, Any] + | ProtocolConstructor[ClientProtocol[Any, Any]] + | None = None, + region: str | None = None, + retry_strategy: RetryStrategy | RetryStrategyOptions | None = None, + sdk_ua_app_id: str | None = None, + transport: ClientTransport[Any, Any] | None = None, + user_agent_extra: str | None = None, + ): + self.auth_scheme_resolver = auth_scheme_resolver or DefaultAuthResolver() + self.auth_schemes = auth_schemes or {} + self.endpoint_resolver = endpoint_resolver or StandardRegionalEndpointsResolver( + endpoint_prefix="MyHandwrittenModel" + ) + self.endpoint_uri = endpoint_uri + self.http_request_config = http_request_config + self.interceptors = interceptors or [] + if isinstance(protocol, type): + protocol = protocol(_PROTOCOL_SETTINGS) + self.protocol = cast( + "ClientProtocol[Any, Any]", protocol + ) or RpcV2CborClientProtocol(_PROTOCOL_SETTINGS) + self.region = region + self.retry_strategy = retry_strategy + self.sdk_ua_app_id = sdk_ua_app_id + self.transport = transport or AIOHTTPClient() + self.user_agent_extra = user_agent_extra + + +Plugin: TypeAlias = Callable[[Config], None] +""" +A callable that customizes a client configuration. Service-level plugins +are applied once to the base configuration inherited by every operation. +Operation-level plugins apply only to a single operation invocation. +""" diff --git a/codegen-output/my-handwritten-model/src/my_handwritten_model/models.py b/codegen-output/my-handwritten-model/src/my_handwritten_model/models.py new file mode 100644 index 000000000..45a366c86 --- /dev/null +++ b/codegen-output/my-handwritten-model/src/my_handwritten_model/models.py @@ -0,0 +1,111 @@ +# Code generated by smithy-python-codegen DO NOT EDIT. + +from ._private import schemas as _schemas +from smithy_core._dynamic_models import ModelRegistry +from smithy_core.exceptions import ModeledError +from smithy_core.schemas import APIOperation +from smithy_core.shapes import ShapeID + +from ._private.schemas import ( + PUBLISH_EVENTS as _SCHEMA_PUBLISH_EVENTS, + SUBSCRIBE_TO_EVENTS as _SCHEMA_SUBSCRIBE_TO_EVENTS, + TRADE_EVENT_STREAM as _SCHEMA_TRADE_EVENT_STREAM, +) + + +class ServiceError(ModeledError): + """ + Base error for all errors in the service. + + Some exceptions do not extend from this class, including + synthetic, implicit, and shared exception types. + """ + + +_registry = ModelRegistry( + _schemas, + renames={}, + index={ + "Alpha": "ALPHA", + "DifferentShapeName": "DIFFERENT_SHAPE_NAME", + "Gamma": "GAMMA", + "GammaPayload": "GAMMA_PAYLOAD", + "HeartbeatEvent": "HEARTBEAT_EVENT", + "LogEvent": "LOG_EVENT", + "MetricEvent": "METRIC_EVENT", + "NotificationEvent": "NOTIFICATION_EVENT", + "PublishEventsInput": "PUBLISH_EVENTS_INPUT", + "PublishEventsOutput": "PUBLISH_EVENTS_OUTPUT", + "PublishEventStream": "PUBLISH_EVENT_STREAM", + "PublishEventStreamLog": "PUBLISH_EVENT_STREAM", + "PublishEventStreamMetric": "PUBLISH_EVENT_STREAM", + "PublishEventStreamUnknown": "PUBLISH_EVENT_STREAM", + "StreamServiceError": "STREAM_SERVICE_ERROR", + "SubscribeEventStream": "SUBSCRIBE_EVENT_STREAM", + "SubscribeEventStreamNotification": "SUBSCRIBE_EVENT_STREAM", + "SubscribeEventStreamHeartbeat": "SUBSCRIBE_EVENT_STREAM", + "SubscribeEventStreamUnknown": "SUBSCRIBE_EVENT_STREAM", + "SubscribeToEventsInput": "SUBSCRIBE_TO_EVENTS_INPUT", + "SubscribeToEventsOutput": "SUBSCRIBE_TO_EVENTS_OUTPUT", + "ThrottlingEvent": "THROTTLING_EVENT", + "TradeEvents": "TRADE_EVENTS", + "TradeEventsAlpha": "TRADE_EVENTS", + "TradeEventsBeta": "TRADE_EVENTS", + "TradeEventsGamma": "TRADE_EVENTS", + "TradeEventsDelta": "TRADE_EVENTS", + "TradeEventsThrottle": "TRADE_EVENTS", + "TradeEventsUnknown": "TRADE_EVENTS", + "TradeEventStreamInput": "TRADE_EVENT_STREAM_INPUT", + "TradeEventStreamOutput": "TRADE_EVENT_STREAM_OUTPUT", + }, + shape_names=frozenset( + { + "Alpha", + "DifferentShapeName", + "Gamma", + "GammaPayload", + "HeartbeatEvent", + "LogEvent", + "MetricEvent", + "NotificationEvent", + "PublishEventStream", + "PublishEventsInput", + "PublishEventsOutput", + "StreamServiceError", + "SubscribeEventStream", + "SubscribeToEventsInput", + "SubscribeToEventsOutput", + "ThrottlingEvent", + "TradeEventStreamInput", + "TradeEventStreamOutput", + "TradeEvents", + } + ), + service_error="ServiceError", +) + + +def __getattr__(name: str): + return _registry.build(name) + + +PUBLISH_EVENTS = APIOperation( + static_schema=_SCHEMA_PUBLISH_EVENTS, + model_registry=_registry, + error_ids=[], + effective_auth_schemes=[ShapeID("smithy.api#noAuth")], +) + +SUBSCRIBE_TO_EVENTS = APIOperation( + static_schema=_SCHEMA_SUBSCRIBE_TO_EVENTS, + model_registry=_registry, + error_ids=[], + effective_auth_schemes=[ShapeID("smithy.api#noAuth")], +) + +TRADE_EVENT_STREAM = APIOperation( + static_schema=_SCHEMA_TRADE_EVENT_STREAM, + model_registry=_registry, + error_ids=[ShapeID("smithy.python.tests.handwritten#StreamServiceError")], + effective_auth_schemes=[ShapeID("smithy.api#noAuth")], +) diff --git a/codegen-output/my-handwritten-model/src/my_handwritten_model/models.pyi b/codegen-output/my-handwritten-model/src/my_handwritten_model/models.pyi new file mode 100644 index 000000000..48e7e79d9 --- /dev/null +++ b/codegen-output/my-handwritten-model/src/my_handwritten_model/models.pyi @@ -0,0 +1,283 @@ +# Code generated by smithy-python-codegen DO NOT EDIT. + +from dataclasses import dataclass +from datetime import datetime +from typing import Self, Union + +from smithy_core.deserializers import ShapeDeserializer +from smithy_core.exceptions import ModeledError +from smithy_core.response import ResponseMetadata +from smithy_core.schemas import APIOperation +from smithy_core.serializers import ShapeSerializer + +class _SerdeShape: + def serialize(self, serializer: ShapeSerializer) -> None: ... + def serialize_members(self, serializer: ShapeSerializer) -> None: ... + @classmethod + def deserialize(cls, deserializer: ShapeDeserializer) -> Self: ... + +class ServiceError(ModeledError): + """ + Base error for all errors in the service. + + Some exceptions do not extend from this class, including + synthetic, implicit, and shared exception types. + """ + +@dataclass(kw_only=True) +class Unit(_SerdeShape): + """Dataclass for Unit structure.""" + +@dataclass(kw_only=True) +class Alpha(_SerdeShape): + """Dataclass for Alpha structure.""" + + id: str | None = ... + timestamp: datetime | None = ... + +@dataclass(kw_only=True) +class DifferentShapeName(_SerdeShape): + """Dataclass for DifferentShapeName structure.""" + + name: str | None = ... + number: int | None = ... + +@dataclass(kw_only=True) +class Gamma(_SerdeShape): + """Dataclass for Gamma structure.""" + + sequence_number: int | None = ... + payload: GammaPayload | None = ... + +@dataclass(kw_only=True) +class GammaPayload(_SerdeShape): + """Dataclass for GammaPayload structure.""" + + message: str | None = ... + values: list[int] | None = ... + +@dataclass(kw_only=True) +class HeartbeatEvent(_SerdeShape): + """Dataclass for HeartbeatEvent structure.""" + + timestamp: datetime | None = ... + +@dataclass(kw_only=True) +class LogEvent(_SerdeShape): + """Dataclass for LogEvent structure.""" + + level: str | None = ... + message: str | None = ... + +@dataclass(kw_only=True) +class MetricEvent(_SerdeShape): + """Dataclass for MetricEvent structure.""" + + name: str | None = ... + value: float | None = ... + +@dataclass(kw_only=True) +class NotificationEvent(_SerdeShape): + """Dataclass for NotificationEvent structure.""" + + topic: str | None = ... + payload: str | None = ... + +@dataclass(kw_only=True) +class PublishEventsInput(_SerdeShape): + """Dataclass for PublishEventsInput structure.""" + + channel: str | None = ... + events: PublishEventStream | None = ... + +@dataclass(kw_only=True) +class PublishEventsOutput(_SerdeShape): + """Dataclass for PublishEventsOutput structure.""" + + event_count: int | None = ... + message: str | None = ... + response_metadata: ResponseMetadata = ... + """ + Metadata about the response that produced this output. + + Use this to recover the request identifiers a service's support team + needs in order to investigate a call. Members of the metadata are + individually optional. + """ + +@dataclass +class PublishEventStreamLog(_SerdeShape): + value: LogEvent + +@dataclass +class PublishEventStreamMetric(_SerdeShape): + value: MetricEvent + +@dataclass +class PublishEventStreamUnknown(_SerdeShape): + """ + Represents an unknown variant. + + If you receive this value, you will need to update your library to + receive the parsed value. + + This value may not be deliberately sent. + """ + + tag: str + +PublishEventStream = Union[ + PublishEventStreamLog | PublishEventStreamMetric | PublishEventStreamUnknown +] + +class _PublishEventStreamDeserializer: + def deserialize(self, deserializer: ShapeDeserializer) -> PublishEventStream: ... + +@dataclass(kw_only=True) +class StreamServiceError(ServiceError, _SerdeShape): + """Dataclass for StreamServiceError structure.""" + + response_metadata: ResponseMetadata = ... + """ + Metadata about the response that produced this output. + + Use this to recover the request identifiers a service's support team + needs in order to investigate a call. Members of the metadata are + individually optional. + """ + +@dataclass +class SubscribeEventStreamNotification(_SerdeShape): + value: NotificationEvent + +@dataclass +class SubscribeEventStreamHeartbeat(_SerdeShape): + value: HeartbeatEvent + +@dataclass +class SubscribeEventStreamUnknown(_SerdeShape): + """ + Represents an unknown variant. + + If you receive this value, you will need to update your library to + receive the parsed value. + + This value may not be deliberately sent. + """ + + tag: str + +SubscribeEventStream = Union[ + SubscribeEventStreamNotification + | SubscribeEventStreamHeartbeat + | SubscribeEventStreamUnknown +] + +class _SubscribeEventStreamDeserializer: + def deserialize(self, deserializer: ShapeDeserializer) -> SubscribeEventStream: ... + +@dataclass(kw_only=True) +class SubscribeToEventsInput(_SerdeShape): + """Dataclass for SubscribeToEventsInput structure.""" + + channel: str | None = ... + max_events: int | None = ... + +@dataclass(kw_only=True) +class SubscribeToEventsOutput(_SerdeShape): + """Dataclass for SubscribeToEventsOutput structure.""" + + subscription_id: str | None = ... + events: SubscribeEventStream | None = ... + response_metadata: ResponseMetadata = ... + """ + Metadata about the response that produced this output. + + Use this to recover the request identifiers a service's support team + needs in order to investigate a call. Members of the metadata are + individually optional. + """ + +@dataclass(kw_only=True) +class ThrottlingEvent(ServiceError, _SerdeShape): + """Dataclass for ThrottlingEvent structure.""" + + response_metadata: ResponseMetadata = ... + """ + Metadata about the response that produced this output. + + Use this to recover the request identifiers a service's support team + needs in order to investigate a call. Members of the metadata are + individually optional. + """ + +@dataclass +class TradeEventsAlpha(_SerdeShape): + value: Alpha + +@dataclass +class TradeEventsBeta(_SerdeShape): + value: Unit + +@dataclass +class TradeEventsGamma(_SerdeShape): + value: Gamma + +@dataclass +class TradeEventsDelta(_SerdeShape): + value: DifferentShapeName + +@dataclass +class TradeEventsThrottle(_SerdeShape): + value: ThrottlingEvent + +@dataclass +class TradeEventsUnknown(_SerdeShape): + """ + Represents an unknown variant. + + If you receive this value, you will need to update your library to + receive the parsed value. + + This value may not be deliberately sent. + """ + + tag: str + +TradeEvents = Union[ + TradeEventsAlpha + | TradeEventsBeta + | TradeEventsGamma + | TradeEventsDelta + | TradeEventsThrottle + | TradeEventsUnknown +] + +class _TradeEventsDeserializer: + def deserialize(self, deserializer: ShapeDeserializer) -> TradeEvents: ... + +@dataclass(kw_only=True) +class TradeEventStreamInput(_SerdeShape): + """Dataclass for TradeEventStreamInput structure.""" + + session_id: str | None = ... + events: TradeEvents | None = ... + +@dataclass(kw_only=True) +class TradeEventStreamOutput(_SerdeShape): + """Dataclass for TradeEventStreamOutput structure.""" + + greeting: str | None = ... + events: TradeEvents | None = ... + response_metadata: ResponseMetadata = ... + """ + Metadata about the response that produced this output. + + Use this to recover the request identifiers a service's support team + needs in order to investigate a call. Members of the metadata are + individually optional. + """ + +PUBLISH_EVENTS: APIOperation[PublishEventsInput, PublishEventsOutput] +SUBSCRIBE_TO_EVENTS: APIOperation[SubscribeToEventsInput, SubscribeToEventsOutput] +TRADE_EVENT_STREAM: APIOperation[TradeEventStreamInput, TradeEventStreamOutput] diff --git a/codegen-output/my-handwritten-model/src/my_handwritten_model/user_agent.py b/codegen-output/my-handwritten-model/src/my_handwritten_model/user_agent.py new file mode 100644 index 000000000..f7386afdb --- /dev/null +++ b/codegen-output/my-handwritten-model/src/my_handwritten_model/user_agent.py @@ -0,0 +1,17 @@ +# Code generated by smithy-python-codegen DO NOT EDIT. + +from smithy_aws_core.interceptors.user_agent import UserAgentInterceptor + +from . import __version__ +from .config import Config + + +def aws_user_agent_plugin(config: Config): + config.interceptors.append( + UserAgentInterceptor( + ua_suffix=config.user_agent_extra, + ua_app_id=config.sdk_ua_app_id, + sdk_version=__version__, + service_id="MyHandwrittenModel", + ) + ) diff --git a/codegen-output/my-handwritten-model/tests/__init__.py b/codegen-output/my-handwritten-model/tests/__init__.py new file mode 100644 index 000000000..247be3e3d --- /dev/null +++ b/codegen-output/my-handwritten-model/tests/__init__.py @@ -0,0 +1 @@ +# Code generated by smithy-python-codegen DO NOT EDIT. diff --git a/codegen-output/my-handwritten-model/tests/_typecheck.py b/codegen-output/my-handwritten-model/tests/_typecheck.py new file mode 100644 index 000000000..e477355a6 --- /dev/null +++ b/codegen-output/my-handwritten-model/tests/_typecheck.py @@ -0,0 +1,43 @@ +# Code generated by smithy-python-codegen DO NOT EDIT. + +from typing import Any + +from my_handwritten_model.models import ( + Alpha, + DifferentShapeName, + Gamma, + GammaPayload, + HeartbeatEvent, + LogEvent, + MetricEvent, + NotificationEvent, + PublishEventsInput, + PublishEventsOutput, + SubscribeToEventsInput, + SubscribeToEventsOutput, + TradeEventStreamInput, + TradeEventStreamOutput, + Unit, +) + + +# Static type fixture: pyright over this validates models.pyi. +# Not executed as a test; it only needs to type-check. + +_: Any = ... + +Unit() +Alpha() +DifferentShapeName() +Gamma() +GammaPayload() +HeartbeatEvent() +LogEvent() +MetricEvent() +NotificationEvent() +PublishEventsInput() +PublishEventsOutput() +SubscribeToEventsInput() +SubscribeToEventsOutput() +TradeEventStreamInput() +TradeEventStreamOutput() diff --git a/codegen-output/my-handwritten-model/tests/test_model_conformance.py b/codegen-output/my-handwritten-model/tests/test_model_conformance.py new file mode 100644 index 000000000..5aff6a05d --- /dev/null +++ b/codegen-output/my-handwritten-model/tests/test_model_conformance.py @@ -0,0 +1,13 @@ +# Code generated by smithy-python-codegen DO NOT EDIT. + +from pathlib import Path + +from smithy_test import verify_models + +from my_handwritten_model import models as _models +from my_handwritten_model._private import schemas as _schemas + + +def test_model_conformance() -> None: + pyi = Path(_models.__file__).with_suffix(".pyi") + verify_models(_models, _schemas, pyi_path=str(pyi) if pyi.exists() else None) diff --git a/codegen-output/my-handwritten-model/tests/test_rpcv2cbor_protocol.py b/codegen-output/my-handwritten-model/tests/test_rpcv2cbor_protocol.py new file mode 100644 index 000000000..8c06bd27c --- /dev/null +++ b/codegen-output/my-handwritten-model/tests/test_rpcv2cbor_protocol.py @@ -0,0 +1,69 @@ +# Code generated by smithy-python-codegen DO NOT EDIT. + +from smithy_core.aio.utils import async_list +from smithy_http import tuples_to_fields +from smithy_http.aio import HTTPResponse as _smithy_http_aio_HTTPResponse +from smithy_http.aio.interfaces import ( + HTTPRequest, + HTTPResponse as _smithy_http_aio_interfaces_HTTPResponse, +) +from smithy_http.interfaces import HTTPClientConfiguration, HTTPRequestConfiguration + +from my_handwritten_model.models import ServiceError + + +class TestHttpServiceError(ServiceError): + """A test error that subclasses the service-error for protocol tests.""" + + def __init__(self, request: HTTPRequest): + self.request = request + + +class RequestTestHTTPClient: + """An asynchronous HTTP client solely for testing purposes.""" + + TIMEOUT_EXCEPTIONS = () + SUPPORTS_DUPLEX_STREAMING: bool = True + + def __init__(self, *, client_config: HTTPClientConfiguration | None = None): + self._client_config = client_config + + async def send( + self, + request: HTTPRequest, + *, + request_config: HTTPRequestConfiguration | None = None, + ) -> _smithy_http_aio_interfaces_HTTPResponse: + # Raise the exception with the request object to bypass actual request handling + raise TestHttpServiceError(request) + + +class ResponseTestHTTPClient: + """An asynchronous HTTP client solely for testing purposes.""" + + TIMEOUT_EXCEPTIONS = () + SUPPORTS_DUPLEX_STREAMING: bool = True + + def __init__( + self, + *, + client_config: HTTPClientConfiguration | None = None, + status: int = 200, + headers: list[tuple[str, str]] | None = None, + body: bytes = b"", + ): + self._client_config = client_config + self.status = status + self.fields = tuples_to_fields(headers or []) + self.body = body + + async def send( + self, + request: HTTPRequest, + *, + request_config: HTTPRequestConfiguration | None = None, + ) -> _smithy_http_aio_HTTPResponse: + # Pre-construct the response from the request and return it + return _smithy_http_aio_HTTPResponse( + status=self.status, fields=self.fields, body=async_list([self.body]) + ) diff --git a/codegen-output/rest-json-1/src/restjson/config.py b/codegen-output/rest-json-1/src/restjson/config.py index b73cec319..d5508b341 100644 --- a/codegen-output/rest-json-1/src/restjson/config.py +++ b/codegen-output/rest-json-1/src/restjson/config.py @@ -34,6 +34,7 @@ from smithy_core.shapes import ShapeID from smithy_http.aio.aiohttp import AIOHTTPClient +from ._private.schemas import REST_JSON from .auth import HTTPAuthSchemeResolver @@ -603,7 +604,9 @@ _ServiceInterceptor = Any _PROTOCOL_SETTINGS = ProtocolSettings( - namespace="aws.protocoltests.restjson", service_target="RestJson" + namespace="aws.protocoltests.restjson", + service_target="RestJson", + service_schema=REST_JSON, ) diff --git a/codegen-output/rest-xml-with-namespace/src/restxmlwithnamespace/config.py b/codegen-output/rest-xml-with-namespace/src/restxmlwithnamespace/config.py index bac66920a..50b8ed4a2 100644 --- a/codegen-output/rest-xml-with-namespace/src/restxmlwithnamespace/config.py +++ b/codegen-output/rest-xml-with-namespace/src/restxmlwithnamespace/config.py @@ -34,6 +34,7 @@ from smithy_core.shapes import ShapeID from smithy_http.aio.aiohttp import AIOHTTPClient +from ._private.schemas import REST_XML_WITH_NAMESPACE from .auth import HTTPAuthSchemeResolver @@ -53,6 +54,7 @@ namespace="aws.protocoltests.restxml.xmlns", service_target="RestXmlWithNamespace", xml_namespace="https://example.com", + service_schema=REST_XML_WITH_NAMESPACE, ) diff --git a/codegen-output/rest-xml/src/restxml/config.py b/codegen-output/rest-xml/src/restxml/config.py index 60ccec2d5..da84db427 100644 --- a/codegen-output/rest-xml/src/restxml/config.py +++ b/codegen-output/rest-xml/src/restxml/config.py @@ -34,6 +34,7 @@ from smithy_core.shapes import ShapeID from smithy_http.aio.aiohttp import AIOHTTPClient +from ._private.schemas import REST_XML from .auth import HTTPAuthSchemeResolver @@ -340,7 +341,9 @@ _ServiceInterceptor = Any _PROTOCOL_SETTINGS = ProtocolSettings( - namespace="aws.protocoltests.restxml", service_target="RestXml" + namespace="aws.protocoltests.restxml", + service_target="RestXml", + service_schema=REST_XML, ) diff --git a/codegen-output/rpcv2-cbor/src/rpcv2cbor/config.py b/codegen-output/rpcv2-cbor/src/rpcv2cbor/config.py index 038301797..7347dab38 100644 --- a/codegen-output/rpcv2-cbor/src/rpcv2cbor/config.py +++ b/codegen-output/rpcv2-cbor/src/rpcv2cbor/config.py @@ -24,6 +24,8 @@ from smithy_http.aio.protocols import RpcV2CborClientProtocol from smithy_http.interfaces import HTTPRequestConfiguration +from ._private.schemas import RPC_V2_PROTOCOL + if TYPE_CHECKING: from smithy_core.interceptors import Interceptor @@ -80,7 +82,9 @@ _ServiceInterceptor = Any _PROTOCOL_SETTINGS = ProtocolSettings( - namespace="smithy.protocoltests.rpcv2Cbor", service_target="RpcV2Protocol" + namespace="smithy.protocoltests.rpcv2Cbor", + service_target="RpcV2Protocol", + service_schema=RPC_V2_PROTOCOL, ) diff --git a/codegen/aws/core/src/main/java/software/amazon/smithy/python/aws/codegen/AwsJson10ProtocolGenerator.java b/codegen/aws/core/src/main/java/software/amazon/smithy/python/aws/codegen/AwsJson10ProtocolGenerator.java index f9953dd46..2e601574d 100644 --- a/codegen/aws/core/src/main/java/software/amazon/smithy/python/aws/codegen/AwsJson10ProtocolGenerator.java +++ b/codegen/aws/core/src/main/java/software/amazon/smithy/python/aws/codegen/AwsJson10ProtocolGenerator.java @@ -13,6 +13,7 @@ import software.amazon.smithy.python.codegen.GenerationContext; import software.amazon.smithy.python.codegen.HttpProtocolTestGenerator; import software.amazon.smithy.python.codegen.generators.ProtocolGenerator; +import software.amazon.smithy.python.codegen.generators.ProtocolSettingsField; import software.amazon.smithy.python.codegen.writer.PythonWriter; import software.amazon.smithy.utils.SmithyInternalApi; @@ -56,6 +57,13 @@ public void initializeProtocol(GenerationContext context, PythonWriter writer) { writer.write("AwsJson10ClientProtocol(_PROTOCOL_SETTINGS)"); } + @Override + public Set requiredProtocolSettings(GenerationContext context) { + return Set.of( + ProtocolSettingsField.NAMESPACE, + ProtocolSettingsField.SERVICE_TARGET); + } + @Override public void generateProtocolTests(GenerationContext context) { context.writerDelegator() diff --git a/codegen/aws/core/src/main/java/software/amazon/smithy/python/aws/codegen/AwsJson11ProtocolGenerator.java b/codegen/aws/core/src/main/java/software/amazon/smithy/python/aws/codegen/AwsJson11ProtocolGenerator.java index c3ad4946f..f7793aacc 100644 --- a/codegen/aws/core/src/main/java/software/amazon/smithy/python/aws/codegen/AwsJson11ProtocolGenerator.java +++ b/codegen/aws/core/src/main/java/software/amazon/smithy/python/aws/codegen/AwsJson11ProtocolGenerator.java @@ -13,6 +13,7 @@ import software.amazon.smithy.python.codegen.GenerationContext; import software.amazon.smithy.python.codegen.HttpProtocolTestGenerator; import software.amazon.smithy.python.codegen.generators.ProtocolGenerator; +import software.amazon.smithy.python.codegen.generators.ProtocolSettingsField; import software.amazon.smithy.python.codegen.writer.PythonWriter; import software.amazon.smithy.utils.SmithyInternalApi; @@ -50,6 +51,13 @@ public void initializeProtocol(GenerationContext context, PythonWriter writer) { writer.write("AwsJson11ClientProtocol(_PROTOCOL_SETTINGS)"); } + @Override + public Set requiredProtocolSettings(GenerationContext context) { + return Set.of( + ProtocolSettingsField.NAMESPACE, + ProtocolSettingsField.SERVICE_TARGET); + } + @Override public void generateProtocolTests(GenerationContext context) { context.writerDelegator() diff --git a/codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/ConfigGenerator.java b/codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/ConfigGenerator.java index 8f1e16064..e6a2b51d6 100644 --- a/codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/ConfigGenerator.java +++ b/codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/ConfigGenerator.java @@ -346,6 +346,14 @@ private void stageProtocolSettings(GenerationContext context, PythonWriter write params.append(", xml_namespace=$S"); args.add(xmlNamespace.get().getUri()); } + // The protocol reads its own eventStreamHttp off the service schema's protocol + // trait, so pass the schema rather than a flattened per-service value. + var serviceSchema = context.symbolProvider() + .toSymbol(service) + .expectProperty(SymbolProperties.SCHEMA) + .getSymbol(); + params.append(", service_schema=$T"); + args.add(serviceSchema); args.add(0, RuntimeTypes.PROTOCOL_SETTINGS); writer.write("_PROTOCOL_SETTINGS = $T(" + params + ")", args.toArray()); diff --git a/codegen/core/src/main/java/software/amazon/smithy/python/codegen/integrations/Rpcv2CborProtocolGenerator.java b/codegen/core/src/main/java/software/amazon/smithy/python/codegen/integrations/Rpcv2CborProtocolGenerator.java index 550a7c913..4019e4fad 100644 --- a/codegen/core/src/main/java/software/amazon/smithy/python/codegen/integrations/Rpcv2CborProtocolGenerator.java +++ b/codegen/core/src/main/java/software/amazon/smithy/python/codegen/integrations/Rpcv2CborProtocolGenerator.java @@ -67,7 +67,9 @@ public void initializeProtocol(GenerationContext context, PythonWriter writer) { public Set requiredProtocolSettings(GenerationContext context) { // The RPC URI is /service/{service_target}/operation/{name}, and the codec // resolves document discriminators against the service namespace. - return Set.of(ProtocolSettingsField.NAMESPACE, ProtocolSettingsField.SERVICE_TARGET); + return Set.of( + ProtocolSettingsField.NAMESPACE, + ProtocolSettingsField.SERVICE_TARGET); } @Override diff --git a/codegen/protocol-test/model/my-handwritten-model.smithy b/codegen/protocol-test/model/my-handwritten-model.smithy new file mode 100644 index 000000000..36955195e --- /dev/null +++ b/codegen/protocol-test/model/my-handwritten-model.smithy @@ -0,0 +1,142 @@ +$version: "2.0" + +namespace smithy.python.tests.handwritten + +use smithy.protocols#rpcv2Cbor + +@rpcv2Cbor( + http: ["h2", "http/1.1"] + eventStreamHttp: ["h2"] +) +@title("My Handwritten Model") +service MyHandwrittenModel { + version: "2020-07-14" + operations: [ + TradeEventStream + PublishEvents + SubscribeToEvents + ] +} + +/// Bidirectional (duplex) stream. The non-stream members (sessionId, greeting) are +/// carried as the initial-request and initial-response messages. +operation TradeEventStream { + input := { + sessionId: String + events: TradeEvents + } + output := { + greeting: String + events: TradeEvents + } + errors: [ + StreamServiceError + ] +} + +@streaming +union TradeEvents { + alpha: Alpha + + beta: Unit + + gamma: Gamma + + // The :event-type header must use the member name `delta`, not the shape name. + delta: DifferentShapeName + + throttle: ThrottlingEvent +} + +structure Alpha { + id: String + timestamp: Timestamp +} + +structure Gamma { + @eventHeader + sequenceNumber: Integer + + @eventPayload + payload: GammaPayload +} + +structure GammaPayload { + message: String + values: IntegerList +} + +structure DifferentShapeName { + name: String + number: Integer +} + +@error("client") +@retryable(throttling: true) +structure ThrottlingEvent { + message: String +} + +/// Input-only stream: the client sends events, the server returns a summary. +operation PublishEvents { + input := { + channel: String + events: PublishEventStream + } + output := { + eventCount: Integer + message: String + } +} + +@streaming +union PublishEventStream { + log: LogEvent + metric: MetricEvent +} + +structure LogEvent { + level: String + message: String +} + +structure MetricEvent { + name: String + value: Double +} + +/// Output-only stream: the client subscribes, the server streams events. +operation SubscribeToEvents { + input := { + channel: String + maxEvents: Integer + } + output := { + subscriptionId: String + events: SubscribeEventStream + } +} + +@streaming +union SubscribeEventStream { + notification: NotificationEvent + heartbeat: HeartbeatEvent +} + +structure NotificationEvent { + topic: String + payload: String +} + +structure HeartbeatEvent { + timestamp: Timestamp +} + +list IntegerList { + member: Integer +} + +@error("server") +structure StreamServiceError { + message: String +} diff --git a/codegen/protocol-test/smithy-build.json b/codegen/protocol-test/smithy-build.json index f1acd24af..7f06db784 100644 --- a/codegen/protocol-test/smithy-build.json +++ b/codegen/protocol-test/smithy-build.json @@ -176,6 +176,28 @@ "moduleVersion": "0.0.1" } } + }, + "my-handwritten-model": { + "transforms": [ + { + "name": "includeServices", + "args": { + "services": [ + "smithy.python.tests.handwritten#MyHandwrittenModel" + ] + } + }, + { + "name": "removeUnusedShapes" + } + ], + "plugins": { + "python-client-codegen": { + "service": "smithy.python.tests.handwritten#MyHandwrittenModel", + "module": "my_handwritten_model", + "moduleVersion": "0.0.1" + } + } } } } diff --git a/packages/smithy-aws-core/.changes/next-release/smithy-aws-core-feature-6f6f0c8027434ec293dc683d88648e30.json b/packages/smithy-aws-core/.changes/next-release/smithy-aws-core-feature-6f6f0c8027434ec293dc683d88648e30.json new file mode 100644 index 000000000..65e11c7b7 --- /dev/null +++ b/packages/smithy-aws-core/.changes/next-release/smithy-aws-core-feature-6f6f0c8027434ec293dc683d88648e30.json @@ -0,0 +1,4 @@ +{ + "type": "feature", + "description": "Added client-side event stream support for the awsJson1.0 and awsJson1.1 protocols." +} \ No newline at end of file diff --git a/packages/smithy-aws-core/src/smithy_aws_core/aio/protocols.py b/packages/smithy-aws-core/src/smithy_aws_core/aio/protocols.py index 3a15d8b3e..2a6c340fd 100644 --- a/packages/smithy-aws-core/src/smithy_aws_core/aio/protocols.py +++ b/packages/smithy-aws-core/src/smithy_aws_core/aio/protocols.py @@ -1,19 +1,14 @@ # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 -from collections.abc import Callable from dataclasses import dataclass, replace -from inspect import iscoroutinefunction from io import BytesIO from typing import TYPE_CHECKING, Any, ClassVar, Final from xml.etree.ElementTree import Element from smithy_core import URI as _URI from smithy_core.aio.interfaces import ( - AsyncWriter, ProtocolSettings, ) -from smithy_core.aio.interfaces.auth import AuthScheme -from smithy_core.aio.interfaces.eventstream import EventPublisher, EventReceiver from smithy_core.aio.types import AsyncBytesReader from smithy_core.codecs import Codec from smithy_core.deserializers import DeserializeableShape, ShapeDeserializer @@ -24,7 +19,6 @@ ExpectationNotMetError, MissingDependencyError, ModeledError, - UnsupportedStreamError, ) from smithy_core.interfaces import BytesReader, BytesWriter, TypedProperties, URI from smithy_core.interfaces import StreamingBlob as SyncStreamingBlob @@ -40,7 +34,7 @@ from smithy_http.aio.interfaces import HTTPErrorIdentifier, HTTPRequest, HTTPResponse from smithy_http.aio.protocols import ( HttpBindingClientProtocol, - HttpClientProtocol, + RpcClientProtocol, ) from smithy_http.deserializers import HTTPResponseDeserializer @@ -79,25 +73,14 @@ _HAS_XML = False # type: ignore try: - from smithy_aws_event_stream.aio import ( - AWSEventPublisher, - AWSEventReceiver, - SigningConfig, - ) - - _HAS_EVENT_STREAM = True + from smithy_aws_event_stream.exceptions import MissingInitialResponse except ImportError: - _HAS_EVENT_STREAM = False # type: ignore + pass if TYPE_CHECKING: - from smithy_aws_event_stream.aio import ( - AWSEventPublisher, - AWSEventReceiver, - SigningConfig, - ) + from smithy_aws_event_stream.exceptions import MissingInitialResponse from smithy_json import JSONCodec, JSONDocument from smithy_xml import XMLCodec - from typing_extensions import TypeForm __all__ = [ @@ -126,14 +109,6 @@ def _assert_xml() -> None: ) -def _assert_event_stream() -> None: - if not _HAS_EVENT_STREAM: - raise MissingDependencyError( - "Attempted to use event streams, but smithy-aws-event-stream " - "is not installed." - ) - - class AWSErrorIdentifier(HTTPErrorIdentifier): _HEADER_KEY: Final = "x-amzn-errortype" @@ -195,74 +170,11 @@ class _AWSHttpBindingClientProtocol( ): """Base for the AWS protocols that use HTTP bindings: restJson1 and restXml. - Adds AWS response metadata and event streams. Events are framed as - ``application/vnd.amazon.eventstream``, and structured event payloads use the - protocol's payload codec. + Adds AWS response metadata. Event streams are inherited from + ``HttpClientProtocol``: events are framed as ``application/vnd.amazon.eventstream`` + and structured event payloads use the protocol's payload codec. """ - def create_event_publisher[ - OperationInput: SerializeableShape, - OperationOutput: DeserializeableShape, - Event: SerializeableShape, - ]( - self, - *, - operation: "APIOperation[OperationInput, OperationOutput]", - request: HTTPRequest, - event_type: "TypeForm[Event]", - context: TypedProperties, - auth_scheme: AuthScheme[Any, Any, Any, Any] | None = None, - ) -> EventPublisher[Event]: - _assert_event_stream() - signing_config: SigningConfig | None = None - if auth_scheme is not None: - event_signer = auth_scheme.event_signer(request=request) - if event_signer is not None: - signing_config = SigningConfig( - signer=event_signer, - signing_properties=auth_scheme.signer_properties(context=context), - identity_resolver=auth_scheme.identity_resolver(context=context), - identity_properties=auth_scheme.identity_properties( - context=context - ), - ) - - # The HTTP body must be an async writeable. The HTTP serializers are responsible - # for ensuring this. - body = request.body - if not isinstance(body, AsyncWriter) or not iscoroutinefunction(body.write): - raise UnsupportedStreamError( - "Input streams require an async write function, but none was present " - "on the serialized HTTP request." - ) - - return AWSEventPublisher[Event]( - payload_codec=self.payload_codec, - async_writer=body, - signing_config=signing_config, - ) - - def create_event_receiver[ - OperationInput: SerializeableShape, - OperationOutput: DeserializeableShape, - Event: DeserializeableShape, - ]( - self, - *, - operation: "APIOperation[OperationInput, OperationOutput]", - request: HTTPRequest, - response: HTTPResponse, - event_type: "TypeForm[Event]", - event_deserializer: Callable[[ShapeDeserializer], Event], - context: TypedProperties, - ) -> EventReceiver[Event]: - _assert_event_stream() - return AWSEventReceiver( - payload_codec=self.payload_codec, - source=AsyncBytesReader(response.body), - deserializer=event_deserializer, - ) - class RestJsonClientProtocol(_AWSHttpBindingClientProtocol): """An implementation of the aws.protocols#restJson1 protocol.""" @@ -431,7 +343,7 @@ def create_deserializer(self, source: bytes | BytesReader) -> ShapeDeserializer: ) -class _AWSJSONClientProtocol(_AWSResponseMetadataMixin, HttpClientProtocol): +class _AWSJSONClientProtocol(_AWSResponseMetadataMixin, RpcClientProtocol): _error_identifier: Final = AWSErrorIdentifier() _id: ClassVar[ShapeID] @@ -440,6 +352,9 @@ class _AWSJSONClientProtocol(_AWSResponseMetadataMixin, HttpClientProtocol): def __init__(self, settings: ProtocolSettings) -> None: _assert_json() self._service_name: Final = settings.service_target + self._event_stream_http = self._resolve_event_stream_http( + settings.service_schema, self._id + ) self._codec: Final = JSONCodec( document_class=AWSJSONDocument, default_namespace=settings.namespace, @@ -474,6 +389,29 @@ def serialize_request[ endpoint: URI, context: TypedProperties, ) -> HTTPRequest: + # An event-stream operation must negotiate HTTP/2 (see _requires_http2); the + # transport does so per operation, leaving non-streaming calls on HTTP/1.1. + http2_required = self._requires_http2(operation) + + if self._is_event_stream_member(operation.input_stream_member): + # RPC event streams carry the input as an initial-request event at the head + # of the stream body; the envelope keeps x-amz-target as a normal request. + return _HTTPRequest( + method="POST", + destination=_URI(host="", path="/"), + fields=tuples_to_fields( + [ + ("content-type", "application/vnd.amazon.eventstream"), + ( + "x-amz-target", + f"{self._service_name}.{operation.schema.id.name}", + ), + ] + ), + body=self._build_initial_request_body(input, context), + http2_required=http2_required, + ) + payload = self.payload_codec.serialize(shape=input) return _HTTPRequest( method="POST", @@ -489,6 +427,7 @@ def serialize_request[ ] ), body=AsyncBytesReader(payload), + http2_required=http2_required, ) async def deserialize_response[ @@ -503,6 +442,14 @@ async def deserialize_response[ error_registry: TypeRegistry, context: TypedProperties, ) -> OperationOutput: + if ( + self._is_event_stream_member(operation.output_stream_member) + and response.status == 200 + ): + return await self._deserialize_event_stream_response( + operation=operation, response=response, context=context + ) + body = await response.consume_body_async() if not self._is_success(operation, context, response): @@ -518,6 +465,18 @@ async def deserialize_response[ body = b"{}" return self.payload_codec.deserialize(source=body, shape=operation.output) + def _default_output[ + OperationInput: SerializeableShape, + OperationOutput: DeserializeableShape, + ]( + self, operation: APIOperation[OperationInput, OperationOutput] + ) -> OperationOutput: + # initial-response is required only if an empty document can't satisfy the + # output; clients fill defaults, so only a required-without-default member does. + if self._requires_initial_response(operation.output_schema): + raise MissingInitialResponse() + return self.payload_codec.deserialize(source=b"{}", shape=operation.output) + def _is_success( self, operation: APIOperation[Any, Any], @@ -655,7 +614,7 @@ class _RecordedQueryRequestId: """ -class AwsQueryClientProtocol(_AWSResponseMetadataMixin, HttpClientProtocol): +class AwsQueryClientProtocol(_AWSResponseMetadataMixin, RpcClientProtocol): """An implementation of the aws.protocols#awsQuery protocol.""" _id: ClassVar[ShapeID] = AwsQueryTrait.id diff --git a/packages/smithy-aws-core/tests/integration/_h2_server.py b/packages/smithy-aws-core/tests/integration/_h2_server.py new file mode 100644 index 000000000..0f5ecba7e --- /dev/null +++ b/packages/smithy-aws-core/tests/integration/_h2_server.py @@ -0,0 +1,186 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +"""A localhost HTTP/2 server harness for driving generated clients over a real +bidirectional transport. Test-runtime concerns (which client, which shapes, what +to assert) live in the test modules that use this harness.""" + +import asyncio +import ssl +from collections.abc import Awaitable, Callable +from io import BytesIO + +import h2.config +import h2.connection +import h2.events +import trustme +from smithy_aws_event_stream.events import Event, EventMessage + +EVENT_STREAM_CONTENT_TYPE = "application/vnd.amazon.eventstream" + + +def encode_event(event_type: str, payload: bytes) -> bytes: + return EventMessage( + headers={ + ":message-type": "event", + ":event-type": event_type, + ":content-type": "application/cbor", + }, + payload=payload, + ).encode() + + +# A referee drives the turn-based exchange. It is called once with ``None`` to open +# the game (after the client's initial-request lands), then once per subsequent client +# event frame. It sends server frames through the ServerStream and returns True when +# the game is over. +Referee = Callable[["Event | None", "ServerStream"], Awaitable[bool]] + + +class ServerStream: + """The server's writable side of one event stream, handed to the referee.""" + + def __init__(self, conn: h2.connection.H2Connection, stream_id: int) -> None: + self._conn = conn + self._stream_id = stream_id + + def send(self, event_type: str, payload: bytes) -> None: + self._conn.send_data( + self._stream_id, encode_event(event_type, payload), end_stream=False + ) + + def end(self) -> None: + self._conn.send_data(self._stream_id, b"", end_stream=True) + + +class H2BidiServer: + """A localhost HTTP/2 server that runs a referee over one duplex event stream. + + After the client's ``initial-request`` arrives, the referee is opened (``None``) + and thereafter handed each client event frame, deciding what to send back. The + ``initial-request`` itself is recorded in ``received_event_types`` but not passed + to the referee -- it is request setup, not a game move. ``received_event_types`` + records every client frame's ``:event-type`` for post-hoc assertions. + """ + + def __init__(self, referee: Referee) -> None: + self._referee = referee + self._ca = trustme.CA() + self._server_cert = self._ca.issue_cert("localhost", "127.0.0.1") + self._ca_cert_ctx = self._ca.cert_pem.tempfile() + self.ca_cert_path = self._ca_cert_ctx.__enter__() + self.host = "127.0.0.1" + self.port = 0 + self._server: asyncio.base_events.Server | None = None + self._tasks: set[asyncio.Task[None]] = set() + self.received_event_types: list[str] = [] + + async def __aenter__(self) -> "H2BidiServer": + ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + ssl_context.set_alpn_protocols(["h2"]) + self._server_cert.configure_cert(ssl_context) + self._server = await asyncio.start_server( + self._track, self.host, 0, ssl=ssl_context + ) + self.port = self._server.sockets[0].getsockname()[1] + return self + + async def __aexit__(self, *exc: object) -> None: + for task in self._tasks: + task.cancel() + if self._server is not None: + self._server.close() + await self._server.wait_closed() + self._ca_cert_ctx.__exit__(None, None, None) + + def _track( + self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + task = asyncio.ensure_future(self._handle(reader, writer)) + self._tasks.add(task) + task.add_done_callback(self._tasks.discard) + + async def _handle( + self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + conn = h2.connection.H2Connection( + config=h2.config.H2Configuration(client_side=False) + ) + conn.initiate_connection() + writer.write(conn.data_to_send()) + await writer.drain() + + decoder = _FrameDecoder() + stream: ServerStream | None = None + done = False + + try: + while not writer.is_closing() and not done: + data = await reader.read(65536) + if not data: + break + for h2_event in conn.receive_data(data): + if isinstance(h2_event, h2.events.RequestReceived): + stream = ServerStream(conn, h2_event.stream_id) + self._respond_headers(conn, stream) + elif isinstance(h2_event, h2.events.DataReceived): + conn.acknowledge_received_data( + len(h2_event.data), h2_event.stream_id + ) + if stream is None: + continue + for frame in decoder.feed(h2_event.data): + event_type = str(frame.message.headers.get(":event-type")) + self.received_event_types.append(event_type) + if event_type == "initial-request": + # Request setup, not a move: open the game off it. + done = await self._referee(None, stream) + else: + done = await self._referee(frame, stream) + if done: + break + + to_send = conn.data_to_send() + if to_send: + writer.write(to_send) + await writer.drain() + except (asyncio.CancelledError, ConnectionError): + pass + finally: + writer.close() + + def _respond_headers( + self, conn: h2.connection.H2Connection, stream: ServerStream + ) -> None: + conn.send_headers( + stream._stream_id, + [(":status", "200"), ("content-type", EVENT_STREAM_CONTENT_TYPE)], + ) + + +class _FrameDecoder: + """Decodes whole event-stream frames out of a growing byte buffer. + + ``Event.decode`` needs a complete frame; feeding it a half-arrived frame raises. + This buffers bytes across reads and yields only frames whose prelude-declared + length has fully landed, leaving any trailing partial frame for the next read. + """ + + def __init__(self) -> None: + self._buffer = bytearray() + + def feed(self, data: bytes) -> list[Event]: + self._buffer.extend(data) + frames: list[Event] = [] + while (frame := self._take_frame()) is not None: + frames.append(frame) + return frames + + def _take_frame(self) -> Event | None: + if len(self._buffer) < 12: + return None + total_length = int.from_bytes(self._buffer[:4], "big") + if len(self._buffer) < total_length: + return None + frame = Event.decode(BytesIO(bytes(self._buffer[:total_length]))) + del self._buffer[:total_length] + return frame diff --git a/packages/smithy-aws-core/tests/integration/test_rpcv2cbor_event_stream.py b/packages/smithy-aws-core/tests/integration/test_rpcv2cbor_event_stream.py new file mode 100644 index 000000000..a58f2262a --- /dev/null +++ b/packages/smithy-aws-core/tests/integration/test_rpcv2cbor_event_stream.py @@ -0,0 +1,168 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +"""End-to-end duplex event streaming against the generated rpcv2Cbor client. + +Drives the generated local client (``codegen-output/my-handwritten-model``) over the +real CRT HTTP/2 duplex transport against a local HTTP/2 server +(``_h2_server.H2BidiServer``) that referees a game of Simon Says. + +The game makes the bidirectional exchange prove its own thoroughness. The server +(Simon) streams a scripted sequence of command events; each command's payload either +begins with ``simon says `` or does not. The client (the test's player loop) must send +a *move* event echoing the action when and only when Simon said so, and a *pass* event +otherwise. Because every turn produces exactly one client frame, the exchange is +deterministic: the referee advances one command per client frame and asserts the +client obeyed. A client that never put its events on the wire, or obeyed the wrong +commands, fails -- which a single ``initial-request``-only assertion could not catch. +""" + +from collections.abc import AsyncIterator + +import pytest + +pytest.importorskip("smithy_cbor") +pytest.importorskip("my_handwritten_model") +pytest.importorskip("awscrt") +pytest.importorskip("h2") +pytest.importorskip("trustme") + +from my_handwritten_model.client import MyHandwrittenModel +from my_handwritten_model.config import Config +from my_handwritten_model.models import ( + Alpha, + TradeEventsAlpha, + TradeEventsBeta, + TradeEventStreamInput, + TradeEventStreamOutput, + Unit, +) +from smithy_aws_event_stream.events import Event +from smithy_cbor import CBORCodec +from smithy_core import URI +from smithy_http.aio.crt import ( + AWSCRTHTTPClient, + AWSCRTHTTPClientConfig, +) + +from ._h2_server import H2BidiServer, ServerStream + +_CODEC = CBORCodec() + +# The scripted game. Each entry is (command text, whether Simon ordered it). A +# well-behaved client MOVES (echoes the action) on the ordered ones and PASSES on +# the rest. The action is the command text with any "simon says " prefix stripped. +_SIMON_SAYS = "simon says " +_SCRIPT: list[str] = [ + "simon says touch your nose", + "jump", + "simon says raise your left hand", + "simon says spin around", + "clap twice", + "sit down", + "simon says take one step forward", +] + + +def _action(command: str) -> str: + return command.removeprefix(_SIMON_SAYS) + + +def _ordered(command: str) -> bool: + return command.startswith(_SIMON_SAYS) + + +def _simon_referee() -> "tuple[object, list[str]]": + """Build the server-side referee and the list it records the client's moves in. + + The referee sends the initial-response, then one command per turn, validating the + client's reply to the previous command before issuing the next. ``moves`` ends up + holding one entry per turn: the moved action, or ``""``. + """ + moves: list[str] = [] + state = {"turn": -1} + + async def referee(frame: Event | None, stream: ServerStream) -> bool: + turn = state["turn"] + + if frame is None: + # Open the game: initial-response, then the first command. + stream.send( + "initial-response", + _CODEC.serialize(shape=TradeEventStreamOutput(greeting="lets play")), + ) + state["turn"] = 0 + stream.send("alpha", _CODEC.serialize(shape=Alpha(id=_SCRIPT[0]))) + return False + + # A client frame is the reply to command `turn`. alpha = a move carrying the + # action id; beta = a pass. Record it, then issue the next command or finish. + event_type = str(frame.message.headers.get(":event-type")) + if event_type == "alpha": + deser = _CODEC.create_deserializer(frame.message.payload) + moves.append(Alpha.deserialize(deser).id or "") + elif event_type == "beta": + moves.append("") + else: + moves.append(f"") + + turn += 1 + state["turn"] = turn + if turn >= len(_SCRIPT): + stream.end() + return True + stream.send("alpha", _CODEC.serialize(shape=Alpha(id=_SCRIPT[turn]))) + return False + + return referee, moves + + +async def _client(server: H2BidiServer) -> tuple[MyHandwrittenModel, AWSCRTHTTPClient]: + transport = AWSCRTHTTPClient( + client_config=AWSCRTHTTPClientConfig( + force_http_2=True, trust_store_path=str(server.ca_cert_path) + ) + ) + client = MyHandwrittenModel( + Config( + endpoint_uri=URI(host=server.host, port=server.port, scheme="https"), + transport=transport, + ) + ) + return client, transport + + +async def test_simon_says_duplex() -> None: + referee, moves = _simon_referee() + async with H2BidiServer(referee) as server: + client, transport = await _client(server) + + stream = await client.trade_event_stream( + TradeEventStreamInput(session_id="player-1") + ) + output, output_stream = await stream.await_output() + assert output.greeting == "lets play" + + # Player loop: for each command Simon streams, move iff ordered, else pass. + async for command in _commands(output_stream): + if _ordered(command): + await stream.input_stream.send( + TradeEventsAlpha(value=Alpha(id=_action(command))) + ) + else: + await stream.input_stream.send(TradeEventsBeta(value=Unit())) + await stream.input_stream.close() + + await transport.close() + + expected = [_action(cmd) if _ordered(cmd) else "" for cmd in _SCRIPT] + assert moves == expected + # Every ordered command was obeyed and no pass leaked a move onto the wire. + assert [m for m in moves if m != ""] == [ + _action(cmd) for cmd in _SCRIPT if _ordered(cmd) + ] + + +async def _commands(output_stream: AsyncIterator[object]) -> AsyncIterator[str]: + async for event in output_stream: + assert isinstance(event, TradeEventsAlpha) + yield event.value.id or "" diff --git a/packages/smithy-aws-core/tests/unit/aio/test_event_streams.py b/packages/smithy-aws-core/tests/unit/aio/test_event_streams.py new file mode 100644 index 000000000..138e8cf5d --- /dev/null +++ b/packages/smithy-aws-core/tests/unit/aio/test_event_streams.py @@ -0,0 +1,230 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Event-stream wiring for awsJson1.1 via the shared RpcClientProtocol base. + +Mirrors the rpcv2Cbor coverage in smithy-http: an ``initial-request`` frame on the +request body and an ``initial-response`` peel feeding the stream's initial output +before the receiver continues. +""" + +from typing import Any + +from smithy_aws_core.aio.protocols import AwsJson11ClientProtocol +from smithy_aws_event_stream.aio import AWSEventReceiver +from smithy_aws_event_stream.events import Event, EventMessage +from smithy_core import URI +from smithy_core.aio.interfaces import AsyncWriter, ProtocolSettings +from smithy_core.aio.types import AsyncBytesProvider, AsyncBytesReader +from smithy_core.deserializers import ShapeDeserializer +from smithy_core.documents import TypeRegistry +from smithy_core.prelude import STRING +from smithy_core.schemas import APIOperation, Schema +from smithy_core.serializers import ShapeSerializer +from smithy_core.shapes import ShapeID, ShapeType +from smithy_core.traits import StreamingTrait +from smithy_core.types import TypedProperties +from smithy_http import Fields +from smithy_http.aio import HTTPRequest, HTTPResponse +from smithy_json import JSONCodec + +_CODEC = JSONCodec() + +_MESSAGE_EVENT = Schema.collection( + id=ShapeID("smithy.example#MessageEvent"), + members={"body": {"target": STRING}}, +) +_EVENT_STREAM = Schema.collection( + id=ShapeID("smithy.example#EventStream"), + shape_type=ShapeType.UNION, + traits=[StreamingTrait()], + members={"message": {"target": _MESSAGE_EVENT}}, +) +_STREAM_IO_SCHEMA = Schema.collection( + id=ShapeID("smithy.example#StreamOpIO"), + members={ + "message": {"target": STRING}, + "stream": {"target": _EVENT_STREAM}, + }, +) + + +class InitialMessage: + def __init__(self, message: str | None = None) -> None: + self.message = message + + def serialize(self, serializer: ShapeSerializer) -> None: + with serializer.begin_struct(_STREAM_IO_SCHEMA) as s: + if self.message is not None: + s.write_string(_STREAM_IO_SCHEMA.members["message"], self.message) + + @classmethod + def deserialize(cls, deserializer: ShapeDeserializer) -> "InitialMessage": + instance = cls() + + def _consume(schema: Schema, de: ShapeDeserializer) -> None: + instance.message = de.read_string(_STREAM_IO_SCHEMA.members["message"]) + + deserializer.read_struct(_STREAM_IO_SCHEMA, _consume) + return instance + + +class _MessageEvent: + def __init__(self, body: str | None = None) -> None: + self.body = body + + def serialize(self, serializer: ShapeSerializer) -> None: + with serializer.begin_struct(_MESSAGE_EVENT) as s: + if self.body is not None: + s.write_string(_MESSAGE_EVENT.members["body"], self.body) + + @classmethod + def deserialize(cls, deserializer: ShapeDeserializer) -> "_MessageEvent": + instance = cls() + + def _consume(schema: Schema, de: ShapeDeserializer) -> None: + instance.body = de.read_string(_MESSAGE_EVENT.members["body"]) + + deserializer.read_struct(_MESSAGE_EVENT, _consume) + return instance + + +class _StubRegistry: + def build(self, name: str) -> type: + return InitialMessage + + +def _message_event_deserializer(deserializer: ShapeDeserializer) -> _MessageEvent: + result: _MessageEvent | None = None + + def _consume(schema: Schema, de: ShapeDeserializer) -> None: + nonlocal result + if schema.expect_member_index() == 0: + result = _MessageEvent.deserialize(de) + + deserializer.read_struct(_EVENT_STREAM, _consume) + if result is None: + raise AssertionError("expected a message event") + return result + + +def _operation() -> APIOperation[Any, Any]: + return APIOperation( + static_schema=( + ShapeType.OPERATION.value, + "smithy.example", + "StreamOp", + {}, + _STREAM_IO_SCHEMA, + _STREAM_IO_SCHEMA, + (), + ), + model_registry=_StubRegistry(), + effective_auth_schemes=[], + ) + + +def _protocol() -> AwsJson11ClientProtocol: + return AwsJson11ClientProtocol( + ProtocolSettings( + namespace="smithy.example", + service_target="StreamService", + service_schema=Schema.collection( + id=ShapeID("smithy.example#StreamService"), + shape_type=ShapeType.SERVICE, + ), + ) + ) + + +def _request() -> HTTPRequest: + return HTTPRequest( + destination=URI(host="example.com"), method="POST", fields=Fields() + ) + + +async def test_serialize_request_defers_initial_request_to_publisher() -> None: + context = TypedProperties() + request = _protocol().serialize_request( + operation=_operation(), + input=InitialMessage(message="hello"), + endpoint=URI(host="example.com"), + context=context, + ) + + assert ( + request.fields["content-type"].as_string() + == "application/vnd.amazon.eventstream" + ) + # The x-amz-target envelope is preserved for streaming operations. + assert request.fields["x-amz-target"].as_string() == "StreamService.StreamOp" + body = request.body + assert isinstance(body, AsyncBytesProvider) + assert isinstance(body, AsyncWriter) + + from io import BytesIO + + publisher = await _protocol().create_event_publisher( + operation=_operation(), + request=request, + event_type=Any, + context=context, + ) + assert publisher is not None + + frame_bytes = await anext(aiter(body)) + event = Event.decode(BytesIO(frame_bytes)) + assert event is not None + assert event.message.headers[":event-type"] == "initial-request" + assert ( + _CODEC.deserialize(source=event.message.payload, shape=InitialMessage).message + == "hello" + ) + + +async def test_deserialize_response_peels_initial_response_then_streams() -> None: + operation = _operation() + initial = EventMessage( + headers={ + ":message-type": "event", + ":event-type": "initial-response", + ":content-type": "application/json", + }, + payload=_CODEC.serialize(shape=InitialMessage(message="the initial response")), + ) + event = EventMessage( + headers={ + ":message-type": "event", + ":event-type": "message", + ":content-type": "application/json", + }, + payload=_CODEC.serialize(shape=_MessageEvent("first event")), + ) + response = HTTPResponse( + status=200, + fields=Fields(), + body=AsyncBytesReader(initial.encode() + event.encode()), + ) + + context = TypedProperties() + output = await _protocol().deserialize_response( + operation=operation, + request=_request(), + response=response, + error_registry=TypeRegistry({}), + context=context, + ) + assert isinstance(output, InitialMessage) + assert output.message == "the initial response" + + receiver = _protocol().create_event_receiver( + operation=operation, + request=_request(), + response=response, + event_type=Any, + event_deserializer=_message_event_deserializer, + context=context, + ) + assert isinstance(receiver, AWSEventReceiver) + received = await receiver.receive() + assert received is not None + assert received.body == "first event" diff --git a/packages/smithy-aws-core/tests/unit/aio/test_protocols.py b/packages/smithy-aws-core/tests/unit/aio/test_protocols.py index b9408e6ef..2ea04b61a 100644 --- a/packages/smithy-aws-core/tests/unit/aio/test_protocols.py +++ b/packages/smithy-aws-core/tests/unit/aio/test_protocols.py @@ -32,6 +32,11 @@ from smithy_http.aio import HTTPRequest, HTTPResponse from smithy_json import JSONSettings +_SERVICE_SCHEMA = Schema.collection( + id=ShapeID("com.test#Service"), + shape_type=ShapeType.SERVICE, +) + class _StubRegistry: def __init__(self, classes: dict[str, type]) -> None: @@ -208,12 +213,18 @@ def _mock_operation( operation = Mock(spec=APIOperation) operation.schema = schema operation.error_schemas = error_schemas or [] + operation.input_stream_member = None + operation.output_stream_member = None return cast("APIOperation[Any, Any]", operation) def _aws_json11_protocol() -> AwsJson11ClientProtocol: return AwsJson11ClientProtocol( - ProtocolSettings(namespace="com.test", service_target="JsonService") + ProtocolSettings( + namespace="com.test", + service_target="JsonService", + service_schema=_SERVICE_SCHEMA, + ) ) @@ -493,7 +504,10 @@ async def test_aws_json11_raises_parse_error_for_invalid_error_body() -> None: async def test_aws_query_serializes_base_request_shape() -> None: protocol = AwsQueryClientProtocol( ProtocolSettings( - namespace="com.test", service_target="QueryService", version="2020-01-08" + namespace="com.test", + service_target="QueryService", + version="2020-01-08", + service_schema=_SERVICE_SCHEMA, ) ) request = protocol.serialize_request( @@ -517,7 +531,10 @@ async def test_aws_query_serializes_base_request_shape() -> None: async def test_aws_query_resolves_modeled_error_from_query_error_trait() -> None: protocol = AwsQueryClientProtocol( ProtocolSettings( - namespace="com.test", service_target="QueryService", version="2020-01-08" + namespace="com.test", + service_target="QueryService", + version="2020-01-08", + service_schema=_SERVICE_SCHEMA, ) ) with patch( @@ -552,7 +569,10 @@ async def test_aws_query_resolves_modeled_error_from_query_error_trait() -> None async def test_aws_query_parses_success_body_once() -> None: protocol = AwsQueryClientProtocol( ProtocolSettings( - namespace="com.test", service_target="QueryService", version="2020-01-08" + namespace="com.test", + service_target="QueryService", + version="2020-01-08", + service_schema=_SERVICE_SCHEMA, ) ) operation = _mock_operation(_operation_schema("EmptyOperation")) @@ -584,7 +604,10 @@ async def test_aws_query_resolves_modeled_error_from_default_namespace_fallback( ): protocol = AwsQueryClientProtocol( ProtocolSettings( - namespace="com.test", service_target="QueryService", version="2020-01-08" + namespace="com.test", + service_target="QueryService", + version="2020-01-08", + service_schema=_SERVICE_SCHEMA, ) ) with pytest.raises(_ModeledQueryError) as exc_info: @@ -611,7 +634,10 @@ async def test_aws_query_resolves_modeled_error_from_default_namespace_fallback( async def test_aws_query_returns_generic_error_for_unknown_code() -> None: protocol = AwsQueryClientProtocol( ProtocolSettings( - namespace="com.test", service_target="QueryService", version="2020-01-08" + namespace="com.test", + service_target="QueryService", + version="2020-01-08", + service_schema=_SERVICE_SCHEMA, ) ) with pytest.raises(CallError) as exc_info: @@ -640,7 +666,10 @@ async def test_aws_query_returns_generic_error_for_unknown_code() -> None: async def test_aws_query_parses_invalid_error_body_once() -> None: protocol = AwsQueryClientProtocol( ProtocolSettings( - namespace="com.test", service_target="QueryService", version="2020-01-08" + namespace="com.test", + service_target="QueryService", + version="2020-01-08", + service_schema=_SERVICE_SCHEMA, ) ) @@ -667,7 +696,10 @@ async def test_aws_query_parses_invalid_error_body_once() -> None: async def test_aws_query_reports_request_id_from_the_response_body() -> None: protocol = AwsQueryClientProtocol( ProtocolSettings( - namespace="com.test", service_target="QueryService", version="2020-01-08" + namespace="com.test", + service_target="QueryService", + version="2020-01-08", + service_schema=_SERVICE_SCHEMA, ) ) context = TypedProperties() @@ -702,7 +734,10 @@ async def test_aws_query_reports_request_id_from_the_response_body() -> None: async def test_aws_query_prefers_a_request_id_header_when_one_is_sent() -> None: protocol = AwsQueryClientProtocol( ProtocolSettings( - namespace="com.test", service_target="QueryService", version="2020-01-08" + namespace="com.test", + service_target="QueryService", + version="2020-01-08", + service_schema=_SERVICE_SCHEMA, ) ) context = TypedProperties() @@ -724,7 +759,10 @@ async def test_aws_query_prefers_a_request_id_header_when_one_is_sent() -> None: async def test_aws_query_reports_no_request_id_when_the_body_has_none() -> None: protocol = AwsQueryClientProtocol( ProtocolSettings( - namespace="com.test", service_target="QueryService", version="2020-01-08" + namespace="com.test", + service_target="QueryService", + version="2020-01-08", + service_schema=_SERVICE_SCHEMA, ) ) response = HTTPResponse( @@ -740,7 +778,10 @@ async def test_aws_query_reports_no_request_id_when_the_body_has_none() -> None: async def test_aws_query_does_not_report_a_previous_attempts_request_id() -> None: protocol = AwsQueryClientProtocol( ProtocolSettings( - namespace="com.test", service_target="QueryService", version="2020-01-08" + namespace="com.test", + service_target="QueryService", + version="2020-01-08", + service_schema=_SERVICE_SCHEMA, ) ) context = TypedProperties() @@ -779,7 +820,10 @@ async def test_aws_query_does_not_report_a_previous_attempts_request_id_when_des ): protocol = AwsQueryClientProtocol( ProtocolSettings( - namespace="com.test", service_target="QueryService", version="2020-01-08" + namespace="com.test", + service_target="QueryService", + version="2020-01-08", + service_schema=_SERVICE_SCHEMA, ) ) context = TypedProperties() @@ -854,6 +898,7 @@ def _rest_xml_protocol() -> RestXmlClientProtocol: namespace="com.test", service_target="XmlService", xml_namespace="https://xml.example.com", + service_schema=_SERVICE_SCHEMA, ) ) diff --git a/packages/smithy-aws-event-stream/src/smithy_aws_event_stream/aio/__init__.py b/packages/smithy-aws-event-stream/src/smithy_aws_event_stream/aio/__init__.py index df28c6eb2..b86024aa8 100644 --- a/packages/smithy-aws-event-stream/src/smithy_aws_event_stream/aio/__init__.py +++ b/packages/smithy-aws-event-stream/src/smithy_aws_event_stream/aio/__init__.py @@ -1,6 +1,7 @@ # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 import asyncio +import contextlib import logging from collections.abc import Callable from dataclasses import dataclass @@ -45,46 +46,42 @@ def __init__( async_writer: AsyncWriter, signing_config: SigningConfig | None = None, is_client_mode: bool = True, + initial_message: EventMessage | None = None, ): self._writer = async_writer self._signing_config = signing_config self._serializer = _EventSerializer( payload_codec=payload_codec, is_client_mode=is_client_mode ) + self._initial_message = initial_message self._closed = False + async def prepare(self) -> None: + """Flush the initial-request event, if any, before the first user event. + + Duplex peers may gate their initial-response on it, so the caller flushes it + at stream creation rather than leaving it until the first ``send``. + """ + await self._write_initial() + async def send(self, event: E) -> None: if self._closed: raise OSError("Attempted to write to closed stream.") - logger.debug("Preparing to publish event: %s", event) + await self._write_initial() event.serialize(self._serializer) result = self._serializer.get_result() if result is None: raise ExpectationNotMetError( "Expected an event message to be serialized, but was None." ) - - if self._signing_config is not None: - identity = await self._signing_config.identity_resolver.get_identity( - properties=self._signing_config.identity_properties - ) - result = await self._signing_config.signer.sign( - event=result, - identity=identity, - properties=self._signing_config.signing_properties, - ) - - encoded_result = result.encode() - try: - logger.debug("Publishing serialized event: %s", result) - await self._writer.write(encoded_result) - except Exception as e: - await self.close() - raise OSError("Failed to write to stream.") from e + await self._write(result) async def close(self) -> None: if self._closed: return + # The initial-request is part of the stream even if no user event follows, so + # it must go out before the terminating frame. + await self._write_initial() self._closed = True try: @@ -112,6 +109,41 @@ async def close(self) -> None: def closed(self) -> bool: return self._closed + async def _write_initial(self) -> None: + # The initial-request seeds the event signature chain off the request + # signature, so it must be the first frame signed and written -- not poured + # into the body unsigned ahead of the signer. + if (initial := self._initial_message) is not None: + # Clear only after a successful write: signing or the socket write can + # fail, and the frame must stay pending rather than be silently skipped. + await self._write(initial) + self._initial_message = None + + async def _write(self, message: EventMessage) -> None: + if self._signing_config is not None: + identity = await self._signing_config.identity_resolver.get_identity( + properties=self._signing_config.identity_properties + ) + message = await self._signing_config.signer.sign( + event=message, + identity=identity, + properties=self._signing_config.signing_properties, + ) + try: + logger.debug("Publishing serialized event: %s", message) + await self._writer.write(message.encode()) + except Exception as e: + # The stream is broken. Abort the underlying writer directly rather than + # routing through close(), which re-flushes the pending initial frame onto + # the dead writer and tries to send a signed terminator that would also fail. + self._closed = True + self._initial_message = None + if (close := getattr(self._writer, "close", None)) is not None: + with contextlib.suppress(Exception): + if asyncio.iscoroutine(result := close()): + await result + raise OSError("Failed to write to stream.") from e + class AWSEventReceiver[E: DeserializeableShape](EventReceiver[E]): def __init__( diff --git a/packages/smithy-aws-event-stream/tests/unit/aio/test_publisher.py b/packages/smithy-aws-event-stream/tests/unit/aio/test_publisher.py new file mode 100644 index 000000000..90bf1cf58 --- /dev/null +++ b/packages/smithy-aws-event-stream/tests/unit/aio/test_publisher.py @@ -0,0 +1,145 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Publisher behavior for the initial-request event. + +The initial-request must be the first frame on the wire and, when signing is +configured, the first event signed -- it seeds the event signature chain off the +request signature, so a user event must never precede or outrank it. +""" + +from io import BytesIO +from typing import Any + +import pytest +from smithy_aws_event_stream._private.serializers import EventSerializer +from smithy_aws_event_stream.aio import AWSEventPublisher, SigningConfig +from smithy_aws_event_stream.events import Event, EventMessage +from smithy_json import JSONCodec + +from .._private import ( + EventStreamMessageEvent, + EventStreamOperationInputOutput, + MessageEvent, +) + + +class _CollectingWriter: + def __init__(self) -> None: + self.frames: list[bytes] = [] + + async def write(self, data: bytes) -> None: + self.frames.append(data) + + +def _event_types(frames: list[bytes]) -> list[str]: + out: list[str] = [] + for frame in frames: + event = Event.decode(BytesIO(frame)) + assert event is not None + out.append(str(event.message.headers.get(":event-type"))) + return out + + +def _initial() -> EventMessage: + serializer = EventSerializer(payload_codec=JSONCodec(), is_client_mode=True) + EventStreamOperationInputOutput(message="init").serialize(serializer) + result = serializer.get_result() + assert result is not None + return result + + +def _user_event() -> EventStreamMessageEvent: + return EventStreamMessageEvent(value=MessageEvent(body_member="one")) + + +def _publisher(writer: _CollectingWriter, **kwargs: Any) -> AWSEventPublisher[Any]: + return AWSEventPublisher[Any]( + payload_codec=JSONCodec(), + async_writer=writer, + initial_message=_initial(), + **kwargs, + ) + + +async def test_prepare_writes_initial_request_first() -> None: + writer = _CollectingWriter() + publisher = _publisher(writer) + + await publisher.prepare() + await publisher.send(_user_event()) + + assert _event_types(writer.frames) == ["initial-request", "message"] + + +async def test_initial_request_flushed_before_first_send_without_prepare() -> None: + # Even if the caller skips prepare(), the first send() must not jump the queue. + writer = _CollectingWriter() + publisher = _publisher(writer) + + await publisher.send(_user_event()) + + assert _event_types(writer.frames) == ["initial-request", "message"] + + +async def test_initial_request_is_signed_first() -> None: + signed_order: list[str] = [] + + class _RecordingSigner: + async def sign( + self, *, event: EventMessage, identity: Any, properties: Any + ) -> EventMessage: + signed_order.append(str(event.headers.get(":event-type"))) + return event + + async def sign_empty( + self, *, event: EventMessage, identity: Any, properties: Any + ) -> EventMessage: + signed_order.append("") + return event + + class _StubResolver: + async def get_identity(self, *, properties: Any) -> Any: + return object() + + signing_config = SigningConfig( + signer=_RecordingSigner(), # type: ignore[arg-type] + signing_properties={}, # type: ignore[arg-type] + identity_resolver=_StubResolver(), # type: ignore[arg-type] + identity_properties={}, # type: ignore[arg-type] + ) + + writer = _CollectingWriter() + publisher = _publisher(writer, signing_config=signing_config) + + await publisher.send(_user_event()) + await publisher.close() + + # Chain order: initial-request signed first, then the user event, then the empty + # terminating frame. + assert signed_order == ["initial-request", "message", ""] + + +async def test_write_failure_aborts_without_reflushing() -> None: + # A failed write must abort: close the underlying writer and mark the publisher + # closed, NOT route through close() (which would re-flush the pending initial frame + # onto the dead writer and attempt a terminator). + class _FailingWriter: + def __init__(self) -> None: + self.frames: list[bytes] = [] + self.closed = False + + async def write(self, data: bytes) -> None: + raise OSError("socket is down") + + async def close(self) -> None: + self.closed = True + + writer = _FailingWriter() + publisher = _publisher(writer) # type: ignore[arg-type] + + with pytest.raises(OSError, match=r"Failed to write to stream\."): + await publisher.send(_user_event()) + + assert writer.frames == [] # nothing reached the wire + assert writer.closed is True # underlying writer torn down + assert publisher.closed is True # subsequent send/close no-op diff --git a/packages/smithy-cbor/tests/benchmark/model.py b/packages/smithy-cbor/tests/benchmark/model.py index cb47dea50..d1b98b1df 100644 --- a/packages/smithy-cbor/tests/benchmark/model.py +++ b/packages/smithy-cbor/tests/benchmark/model.py @@ -95,7 +95,14 @@ shape_type=ShapeType.SERVICE, ) -SETTINGS = ProtocolSettings(namespace=_NS, service_target="BenchmarkService") +SETTINGS = ProtocolSettings( + namespace=_NS, + service_target="BenchmarkService", + service_schema=Schema.collection( + id=ShapeID(f"{_NS}#BenchmarkService"), + shape_type=ShapeType.SERVICE, + ), +) # --- Shape dataclasses (mirroring models.py) ----------------------------------- diff --git a/packages/smithy-core/.changes/next-release/smithy-core-breaking-8393a56560a540f2b2bc0314a3856535.json b/packages/smithy-core/.changes/next-release/smithy-core-breaking-8393a56560a540f2b2bc0314a3856535.json new file mode 100644 index 000000000..be9d4ce88 --- /dev/null +++ b/packages/smithy-core/.changes/next-release/smithy-core-breaking-8393a56560a540f2b2bc0314a3856535.json @@ -0,0 +1,4 @@ +{ + "type": "breaking", + "description": "ProtocolSettings now requires a service_schema field; the event_stream_http field was removed and RPC protocols read eventStreamHttp from the service schema's protocol trait." +} \ No newline at end of file diff --git a/packages/smithy-core/src/smithy_core/aio/client.py b/packages/smithy-core/src/smithy_core/aio/client.py index f2e1eb7a2..719fb9341 100644 --- a/packages/smithy-core/src/smithy_core/aio/client.py +++ b/packages/smithy-core/src/smithy_core/aio/client.py @@ -1,6 +1,7 @@ # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 import asyncio +import contextlib import logging from asyncio import Future, sleep from collections.abc import Awaitable, Callable, Sequence @@ -159,13 +160,21 @@ async def input_stream[ self._await_output(self._execute_request(call, request_future)) ) request_context = await request_future - input_stream = self.protocol.create_event_publisher( - operation=call.operation, - request=request_context.transport_request, - event_type=event_type, - context=request_context.properties, - auth_scheme=request_context.properties.get(AUTH_SCHEME), - ) + try: + input_stream = await self.protocol.create_event_publisher( + operation=call.operation, + request=request_context.transport_request, + event_type=event_type, + context=request_context.properties, + auth_scheme=request_context.properties.get(AUTH_SCHEME), + ) + except BaseException: + # Publisher creation failed; the request task is still running, so cancel + # and drain it rather than leaking a pending task. + output_future.cancel() + with contextlib.suppress(BaseException): + await output_future + raise return InputEventStream(input_stream=input_stream, output_future=output_future) async def _await_output[I: SerializeableShape, O: DeserializeableShape]( @@ -238,13 +247,21 @@ async def duplex_stream[ request_future = Future[RequestContext[I, TRequest]]() execute_task = asyncio.create_task(self._execute_request(call, request_future)) request_context = await request_future - input_stream = self.protocol.create_event_publisher( - operation=call.operation, - request=request_context.transport_request, - event_type=input_event_type, - context=request_context.properties, - auth_scheme=request_context.properties.get(AUTH_SCHEME), - ) + try: + input_stream = await self.protocol.create_event_publisher( + operation=call.operation, + request=request_context.transport_request, + event_type=input_event_type, + context=request_context.properties, + auth_scheme=request_context.properties.get(AUTH_SCHEME), + ) + except BaseException: + # Publisher creation failed; the request task is still running, so cancel + # and drain it rather than leaking a pending task. + execute_task.cancel() + with contextlib.suppress(BaseException): + await execute_task + raise output_future = asyncio.create_task( self._await_output_stream( call=call, diff --git a/packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py b/packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py index 720945b49..70b10b61a 100644 --- a/packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py +++ b/packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py @@ -14,9 +14,10 @@ if TYPE_CHECKING: from typing_extensions import TypeForm + from ..._schema_compact import StaticSchema from ...deserializers import DeserializeableShape, ShapeDeserializer from ...response import ResponseMetadata - from ...schemas import APIOperation + from ...schemas import APIOperation, Schema from ...serializers import SerializeableShape from ...shapes import ShapeID from .auth import AuthScheme @@ -132,6 +133,16 @@ class ProtocolSettings: """The URI of the service's ``@xmlNamespace``, applied by restXml to request payloads that don't declare their own namespace.""" + service_schema: "StaticSchema | Schema" + """The service shape's schema. + + Carries the service's traits, including the protocol trait whose ``eventStreamHttp`` + the RPC protocols read to negotiate HTTP/2 for event streams. The protocol reads its + own trait off this schema, so the value is inherently per-protocol rather than a + single flattened field. Codegen passes the compact schema form; it is hydrated on + first read. + """ + type ProtocolConstructor[T] = Callable[[ProtocolSettings], T] """A callable that builds a protocol instance from ``ProtocolSettings``. @@ -225,7 +236,7 @@ def extract_response_metadata( """ ... - def create_event_publisher[ + async def create_event_publisher[ OperationInput: "SerializeableShape", OperationOutput: "DeserializeableShape", Event: "SerializeableShape", diff --git a/packages/smithy-core/tests/unit/aio/_pipeline_harness.py b/packages/smithy-core/tests/unit/aio/_pipeline_harness.py index 12ff9a717..30d6c8ed0 100644 --- a/packages/smithy-core/tests/unit/aio/_pipeline_harness.py +++ b/packages/smithy-core/tests/unit/aio/_pipeline_harness.py @@ -221,7 +221,7 @@ def extract_response_metadata(self, **kwargs: Any) -> ResponseMetadata: raise self.extract_response_metadata_error return self.stub_response_metadata - def create_event_publisher(self, **kwargs: Any) -> StubEventPublisher: + async def create_event_publisher(self, **kwargs: Any) -> StubEventPublisher: self.create_event_publisher_calls += 1 return StubEventPublisher() diff --git a/packages/smithy-http/.changes/next-release/smithy-http-enhancement-6707a5d4fd7545c28358196547693139.json b/packages/smithy-http/.changes/next-release/smithy-http-enhancement-6707a5d4fd7545c28358196547693139.json new file mode 100644 index 000000000..787545771 --- /dev/null +++ b/packages/smithy-http/.changes/next-release/smithy-http-enhancement-6707a5d4fd7545c28358196547693139.json @@ -0,0 +1,4 @@ +{ + "type": "enhancement", + "description": "AWSCRTHTTPClient now negotiates HTTP/2 per request for operations that require it (event streams), advertising only HTTP/2 in ALPN and pooling those connections separately from HTTP/1.1 ones, so a single client serves both HTTP/1.1 operations and HTTP/2 event streams without configuration. The client-wide force_http_2 config remains as an override." +} diff --git a/packages/smithy-http/.changes/next-release/smithy-http-enhancement-f30efea6cdbd4fd1a2a7c30264629aac.json b/packages/smithy-http/.changes/next-release/smithy-http-enhancement-f30efea6cdbd4fd1a2a7c30264629aac.json new file mode 100644 index 000000000..da9329dc3 --- /dev/null +++ b/packages/smithy-http/.changes/next-release/smithy-http-enhancement-f30efea6cdbd4fd1a2a7c30264629aac.json @@ -0,0 +1,4 @@ +{ + "type": "enhancement", + "description": "Added a trust_store_path option to AWSCRTHTTPClientConfig for overriding the TLS trust store with a custom CA bundle." +} \ No newline at end of file diff --git a/packages/smithy-http/.changes/next-release/smithy-http-feature-c3a19ac086e9446c95fb57068cc8b355.json b/packages/smithy-http/.changes/next-release/smithy-http-feature-c3a19ac086e9446c95fb57068cc8b355.json new file mode 100644 index 000000000..530fd799f --- /dev/null +++ b/packages/smithy-http/.changes/next-release/smithy-http-feature-c3a19ac086e9446c95fb57068cc8b355.json @@ -0,0 +1,4 @@ +{ + "type": "feature", + "description": "Added a RpcClientProtocol base class and client-side event stream support for the rpcv2Cbor protocol." +} \ No newline at end of file diff --git a/packages/smithy-http/src/smithy_http/aio/__init__.py b/packages/smithy-http/src/smithy_http/aio/__init__.py index d853f0799..6daa2e852 100644 --- a/packages/smithy-http/src/smithy_http/aio/__init__.py +++ b/packages/smithy-http/src/smithy_http/aio/__init__.py @@ -20,6 +20,10 @@ class HTTPRequest(http_aio_interfaces.HTTPRequest): body: core_aio_interfaces.StreamingBlob = field(repr=False, default=b"") method: str fields: http_interfaces.Fields + http2_required: bool = False + """Transport hint: negotiate HTTP/2 for this request and don't reuse a non-h2 + connection for it. Set for event-stream operations whose protocol permits h2; + not sent on the wire.""" # HTTPResponse implements http_interfaces.HTTPResponse but cannot be explicitly diff --git a/packages/smithy-http/src/smithy_http/aio/crt.py b/packages/smithy-http/src/smithy_http/aio/crt.py index 50e924b85..7fa30fbec 100644 --- a/packages/smithy-http/src/smithy_http/aio/crt.py +++ b/packages/smithy-http/src/smithy_http/aio/crt.py @@ -123,7 +123,7 @@ def __repr__(self) -> str: ) -ConnectionPoolKey = tuple[str, str, int | None] +ConnectionPoolKey = tuple[str, str, int | None, bool] ConnectionPoolDict = dict[ConnectionPoolKey, "AIOHttpClientConnectionUnified"] @@ -133,9 +133,13 @@ class AWSCRTHTTPClientConfig(http_interfaces.HTTPClientConfiguration): :param read_buffer_size: The buffer size in bytes to use when reading from streams. Defaults to 8192 (8 KB). + :param trust_store_path: Path to a PEM CA bundle that overrides the default system + trust store when verifying TLS peers. Enables connecting to a server whose + certificate chains to a private CA. """ read_buffer_size: int = DEFAULT_READ_BUFFER_SIZE + trust_store_path: str | None = None def __post_init__(self) -> None: _assert_crt() @@ -171,7 +175,12 @@ def __init__( eventloop = _AWSCRTEventLoop() self._eventloop = eventloop self._client_bootstrap = self._eventloop.bootstrap - self._tls_ctx = crt_io.ClientTlsContext(crt_io.TlsContextOptions()) + tls_ctx_options = crt_io.TlsContextOptions() + if self._config.trust_store_path is not None: + tls_ctx_options.override_default_trust_store_from_path( + None, self._config.trust_store_path + ) + self._tls_ctx = crt_io.ClientTlsContext(tls_ctx_options) self._socket_options = crt_io.SocketOptions() self._connections: ConnectionPoolDict = {} self._closed = False @@ -194,7 +203,12 @@ async def send( try: crt_request = self._marshal_request(request) - connection = await self._get_connection(request.destination) + require_http2 = ( + getattr(request, "http2_required", False) or self._config.force_http_2 + ) + connection = await self._get_connection( + request.destination, require_http2=require_http2 + ) # request_body_generator is HTTP/2-only in CRT; HTTP/1.1 must use body_stream if connection.version is crt_http.HttpVersion.Http2: @@ -257,29 +271,31 @@ async def _await_response( ) async def _create_connection( - self, url: core_interfaces.URI + self, url: core_interfaces.URI, *, require_http2: bool ) -> "AIOHttpClientConnectionUnified": """Builds and validates connection to ``url``""" - connection = await self._build_new_connection(url) - await self._validate_connection(connection) + connection = await self._build_new_connection(url, require_http2=require_http2) + await self._validate_connection(connection, require_http2=require_http2) return connection async def _get_connection( - self, url: core_interfaces.URI + self, url: core_interfaces.URI, *, require_http2: bool = False ) -> "AIOHttpClientConnectionUnified": # TODO: Use CRT connection pooling instead of this basic kind - connection_key = (url.scheme, url.host, url.port) + # Key by require_http2 so an h2-required event stream never reuses a pooled + # HTTP/1.1 connection to the same host (and vice versa). + connection_key = (url.scheme, url.host, url.port, require_http2) connection = self._connections.get(connection_key) if connection and connection.is_open(): return connection - connection = await self._create_connection(url) + connection = await self._create_connection(url, require_http2=require_http2) self._connections[connection_key] = connection return connection async def _build_new_connection( - self, url: core_interfaces.URI + self, url: core_interfaces.URI, *, require_http2: bool = False ) -> "AIOHttpClientConnectionUnified": if url.scheme == "http": port = self._HTTP_PORT @@ -288,8 +304,13 @@ async def _build_new_connection( port = self._HTTPS_PORT tls_connection_options = self._tls_ctx.new_connection_options() tls_connection_options.set_server_name(url.host) - # TODO: Support TLS configuration, including alpn - tls_connection_options.set_alpn_list(["h2", "http/1.1"]) + # Advertise only h2 when it is required so endpoints that would otherwise + # negotiate down to http/1.1 cannot, which would then fail validation. + # TODO: Support broader TLS configuration. + if require_http2: + tls_connection_options.set_alpn_list(["h2"]) + else: + tls_connection_options.set_alpn_list(["h2", "http/1.1"]) else: raise SmithyHTTPError( f"AWSCRTHTTPClient does not support URL scheme {url.scheme}" @@ -306,15 +327,14 @@ async def _build_new_connection( ) async def _validate_connection( - self, connection: "AIOHttpClientConnectionUnified" + self, connection: "AIOHttpClientConnectionUnified", *, require_http2: bool ) -> None: - """Validates an existing connection against the client config. + """Validates an existing connection against the request's requirement. Checks performed: - * If ``force_http_2`` is enabled: Is the connection HTTP/2? + * If HTTP/2 is required: Is the connection HTTP/2? """ - force_http_2 = self._config.force_http_2 - if force_http_2 and connection.version is not crt_http.HttpVersion.Http2: + if require_http2 and connection.version is not crt_http.HttpVersion.Http2: await connection.close() negotiated = crt_http.HttpVersion(connection.version).name raise SmithyHTTPError(f"HTTP/2 could not be negotiated: {negotiated}") diff --git a/packages/smithy-http/src/smithy_http/aio/protocols.py b/packages/smithy-http/src/smithy_http/aio/protocols.py index c16286c81..7356eb311 100644 --- a/packages/smithy-http/src/smithy_http/aio/protocols.py +++ b/packages/smithy-http/src/smithy_http/aio/protocols.py @@ -1,22 +1,31 @@ # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 +import asyncio import os -from collections.abc import AsyncIterable +from collections.abc import AsyncIterable, Callable, Mapping, Sequence from inspect import iscoroutinefunction from typing import TYPE_CHECKING, Any, cast from smithy_core import URI as _URI -from smithy_core.aio.interfaces import AsyncByteStream, ClientProtocol, ProtocolSettings +from smithy_core.aio.interfaces import ( + AsyncByteStream, + AsyncWriter, + ClientProtocol, + ProtocolSettings, +) from smithy_core.aio.interfaces import StreamingBlob as AsyncStreamingBlob +from smithy_core.aio.interfaces.auth import AuthScheme +from smithy_core.aio.interfaces.eventstream import EventPublisher, EventReceiver from smithy_core.aio.types import AsyncBytesReader from smithy_core.codecs import Codec -from smithy_core.deserializers import DeserializeableShape +from smithy_core.deserializers import DeserializeableShape, ShapeDeserializer from smithy_core.documents import TypeRegistry from smithy_core.exceptions import ( CallError, ExpectationNotMetError, MissingDependencyError, ModeledError, + UnsupportedStreamError, ) from smithy_core.interfaces import ( Endpoint, @@ -29,11 +38,14 @@ from smithy_core.response import ResponseMetadata from smithy_core.schemas import APIOperation, Schema from smithy_core.serializers import SerializeableShape, SerializeableStruct -from smithy_core.shapes import ShapeID +from smithy_core.shapes import ShapeID, ShapeType from smithy_core.traits import ( ORIGINAL_SHAPE_ID, + DefaultTrait, + RequiredTrait, RpcV2CborTrait, ) +from smithy_core.types import PropertyKey from .. import tuples_to_fields from ..deserializers import HTTPResponseDeserializer @@ -48,13 +60,98 @@ __all__ = [ "HttpBindingClientProtocol", "HttpClientProtocol", + "RpcClientProtocol", "RpcV2CborClientProtocol", ] +try: + from smithy_aws_event_stream._private.deserializers import ( + EventDeserializer as _EventDeserializer, + ) + from smithy_aws_event_stream._private.serializers import ( + EventSerializer as _EventSerializer, + ) + from smithy_aws_event_stream.aio import ( + AWSEventPublisher, + AWSEventReceiver, + SigningConfig, + ) + from smithy_aws_event_stream.events import Event as _Event + from smithy_aws_event_stream.events import EventMessage as _EventMessage + from smithy_aws_event_stream.exceptions import MissingInitialResponse + from smithy_core.aio.types import AsyncBytesProvider + + _HAS_EVENT_STREAM = True +except ImportError: + _HAS_EVENT_STREAM = False # type: ignore + +if TYPE_CHECKING: + from smithy_aws_event_stream._private.deserializers import ( + EventDeserializer as _EventDeserializer, + ) + from smithy_aws_event_stream._private.serializers import ( + EventSerializer as _EventSerializer, + ) + from smithy_aws_event_stream.aio import ( + AWSEventPublisher, + AWSEventReceiver, + SigningConfig, + ) + from smithy_aws_event_stream.events import Event as _Event + from smithy_aws_event_stream.events import EventMessage as _EventMessage + from smithy_aws_event_stream.exceptions import MissingInitialResponse + from smithy_core._schema_compact import StaticSchema + from smithy_core.aio.types import AsyncBytesProvider + from typing_extensions import TypeForm + + +_EVENT_STREAM_CONTENT_TYPE = "application/vnd.amazon.eventstream" + +_EVENT_STREAM_SOURCE = PropertyKey( + key="smithy_http_event_stream_source", value_type=AsyncByteStream +) +"""Carries the response-body reader already positioned past the initial-response event. + +RPC output streams read their initial response off the first event frame in +``deserialize_response``; the receiver then continues from this same reader so the +frames stay in order. +""" + +_PREFETCHED_EVENT = PropertyKey( + key="smithy_http_event_stream_prefetched", value_type=object +) +"""Carries a first stream event already decoded off the body in ``deserialize_response``. + +When the output has no ``initial-response`` the first frame is a real event, not the +output; it is stashed here so the receiver replays it before reading the next frame. +""" + +_INITIAL_REQUEST_EVENT = PropertyKey( + key="smithy_http_event_stream_initial_request", value_type=object +) +"""Carries the serialized ``initial-request`` event awaiting the publisher. + +The frame is NOT poured into the request body: it must be signed as the first event +in the signature chain (off the request signature), so the publisher writes it. +""" + + +def _assert_event_stream() -> None: + if not _HAS_EVENT_STREAM: + raise MissingDependencyError( + "Attempted to use event streams, but smithy-aws-event-stream " + "is not installed." + ) + class HttpClientProtocol(ClientProtocol[HTTPRequest, HTTPResponse]): """An HTTP-based protocol.""" + @property + def payload_codec(self) -> Codec: + """The codec used for the serde of event payloads and HTTP bodies.""" + raise NotImplementedError() + def set_service_endpoint( self, *, @@ -101,6 +198,77 @@ def extract_response_metadata( """ return ResponseMetadata(http_status_code=response.status) + async def create_event_publisher[ + OperationInput: SerializeableShape, + OperationOutput: DeserializeableShape, + Event: SerializeableShape, + ]( + self, + *, + operation: "APIOperation[OperationInput, OperationOutput]", + request: HTTPRequest, + event_type: "TypeForm[Event]", + context: TypedProperties, + auth_scheme: AuthScheme[Any, Any, Any, Any] | None = None, + ) -> EventPublisher[Event]: + _assert_event_stream() + signing_config: SigningConfig | None = None + if auth_scheme is not None: + event_signer = auth_scheme.event_signer(request=request) + if event_signer is not None: + signing_config = SigningConfig( + signer=event_signer, + signing_properties=auth_scheme.signer_properties(context=context), + identity_resolver=auth_scheme.identity_resolver(context=context), + identity_properties=auth_scheme.identity_properties( + context=context + ), + ) + + # The HTTP body must be an async writeable. Protocols that serialize an event + # stream request are responsible for setting one. + body = request.body + if not isinstance(body, AsyncWriter) or not iscoroutinefunction(body.write): + raise UnsupportedStreamError( + "Input streams require an async write function, but none was present " + "on the serialized HTTP request." + ) + + initial_message = cast( + "_EventMessage | None", context.get(_INITIAL_REQUEST_EVENT) + ) + publisher = AWSEventPublisher[Event]( + payload_codec=self.payload_codec, + async_writer=body, + signing_config=signing_config, + initial_message=initial_message, + ) + # Flush initial-request now so a duplex peer that gates its initial-response on + # it isn't deadlocked waiting for a frame the publisher would defer. + await publisher.prepare() + return publisher + + def create_event_receiver[ + OperationInput: SerializeableShape, + OperationOutput: DeserializeableShape, + Event: DeserializeableShape, + ]( + self, + *, + operation: "APIOperation[OperationInput, OperationOutput]", + request: HTTPRequest, + response: HTTPResponse, + event_type: "TypeForm[Event]", + event_deserializer: Callable[[ShapeDeserializer], Event], + context: TypedProperties, + ) -> EventReceiver[Event]: + _assert_event_stream() + return AWSEventReceiver( + payload_codec=self.payload_codec, + source=AsyncBytesReader(response.body), + deserializer=event_deserializer, + ) + class HttpBindingClientProtocol(HttpClientProtocol): """An HTTP-based protocol that uses HTTP binding traits.""" @@ -355,7 +523,181 @@ def _is_unit(schema: Schema | None) -> bool: return original is not None and original.document_value == str(UNIT.id) -class _RpcV2ClientProtocol(HttpClientProtocol): +class RpcClientProtocol(HttpClientProtocol): + """Base for the RPC-based protocols (awsJson, awsQuery, rpcv2). + + RPC protocols serialize the whole operation input through a codec rather than via + HTTP bindings. This base carries the event-stream framing they share: per the + Amazon event stream specification, an RPC event stream carries the operation input + and output as ``initial-request`` and ``initial-response`` events at the head of + the stream rather than as the HTTP request and response bodies. + + Subclasses build their own HTTP request envelope (path, headers) and response + handling; they delegate the event-stream body to the helpers here, gated on the + operation's ``input_stream_member`` / ``output_stream_member``. A protocol that + never models an event stream (awsQuery) simply never reaches these helpers. + """ + + # The protocol trait's eventStreamHttp ALPN list (empty if the trait left it unset). + _event_stream_http: tuple[str, ...] = () + + @staticmethod + def _resolve_event_stream_http( + service_schema: "StaticSchema | Schema", protocol_id: ShapeID + ) -> tuple[str, ...]: + """Read this protocol's ``eventStreamHttp`` ALPN list off the service schema. + + Keyed by the protocol's own id, so a service carrying several protocol traits + yields each protocol its own value rather than a flattened one. The compact + schema codegen emits is hydrated on first read (cached thereafter). + """ + from smithy_core._schema_compact import hydrate + + trait = hydrate(service_schema).get_trait(protocol_id) + if trait is None: + return () + value = trait.document_value + if not isinstance(value, Mapping): + return () + versions = value.get("eventStreamHttp") + if not isinstance(versions, Sequence) or isinstance(versions, str | bytes): + return () + return tuple(str(v) for v in versions) + + @staticmethod + def _is_event_stream_member(member: Schema | None) -> bool: + """Whether a ``@streaming`` member is an EVENT stream (union), not a data stream. + + ``input_stream_member``/``output_stream_member`` match any ``@streaming`` + member; a blob target is a data stream, only a union target is an event stream. + """ + return member is not None and member.expect_member_target().shape_type is ( + ShapeType.UNION + ) + + def _requires_http2(self, operation: APIOperation[Any, Any]) -> bool: + # Only EVENT streams (a @streaming UNION member) force HTTP/2 -- a @streaming + # blob is a data stream and runs fine over HTTP/1.1. input_stream_member / + # output_stream_member match either, so check the target is a union here. + if not ( + self._is_event_stream_member(operation.input_stream_member) + or self._is_event_stream_member(operation.output_stream_member) + ): + return False + # Require HTTP/2 unless the model EXPLICITLY allows only HTTP/1.1: no list, or + # any list containing h2, requires h2; only an h1-and-nothing-else list opts out. + return "h2" in self._event_stream_http or not self._event_stream_http + + @staticmethod + def _requires_initial_response(output: Schema) -> bool: + """Whether the output needs an ``initial-response`` to be populated. + + An empty document satisfies any member that is optional or has a default + (clients fill defaults), so only a ``@required`` member without a ``@default`` + forces an initial-response. + """ + for member in output.members.values(): + if member.get_trait(RequiredTrait) is not None and ( + member.get_trait(DefaultTrait) is None + ): + return True + return False + + def _build_initial_request_body( + self, input: SerializeableShape, context: TypedProperties + ) -> "AsyncBytesProvider": + """Serialize the ``initial-request`` event and stash it for the publisher. + + The frame is not written into the returned body: the publisher must write it + as its first (signed) frame. The body is an empty async writer the publisher + appends to. + """ + _assert_event_stream() + serializer = _EventSerializer( + payload_codec=self.payload_codec, is_client_mode=True + ) + input.serialize(serializer) + initial = serializer.get_result() + if initial is None: + raise ExpectationNotMetError( + "Expected an initial-request event to be serialized, but was None." + ) + context[_INITIAL_REQUEST_EVENT] = initial + return AsyncBytesProvider() + + async def _deserialize_event_stream_response[ + OperationInput: SerializeableShape, + OperationOutput: DeserializeableShape, + ]( + self, + *, + operation: APIOperation[OperationInput, OperationOutput], + response: HTTPResponse, + context: TypedProperties, + ) -> OperationOutput: + # initial-response is optional: the spec omits it when every non-stream output + # member is optional. Peek the first frame's :event-type -- consume it as the + # output only when it IS the initial-response; otherwise synthesize the default + # output and stash the frame so the receiver replays it as the stream's first + # event rather than dropping it. + _assert_event_stream() + source = AsyncBytesReader(response.body) + event = await _Event.decode_async(source) + context[_EVENT_STREAM_SOURCE] = source + if event is not None and event.message.headers.get(":event-type") == ( + "initial-response" + ): + deserializer = _EventDeserializer( + event=event, payload_codec=self.payload_codec, is_client_mode=True + ) + return operation.output.deserialize(deserializer) + if event is not None: + context[_PREFETCHED_EVENT] = event + return self._default_output(operation) + + def _default_output[ + OperationInput: SerializeableShape, + OperationOutput: DeserializeableShape, + ]( + self, operation: APIOperation[OperationInput, OperationOutput] + ) -> OperationOutput: + """Build the output when no ``initial-response`` precedes the stream. + + :raises MissingInitialResponse: when the output has required members, so an + absent initial-response cannot be filled by defaults. + """ + raise NotImplementedError + + def create_event_receiver[ + OperationInput: SerializeableShape, + OperationOutput: DeserializeableShape, + Event: DeserializeableShape, + ]( + self, + *, + operation: "APIOperation[OperationInput, OperationOutput]", + request: HTTPRequest, + response: HTTPResponse, + event_type: "TypeForm[Event]", + event_deserializer: Callable[[ShapeDeserializer], Event], + context: TypedProperties, + ) -> EventReceiver[Event]: + _assert_event_stream() + # Continue from the reader that already consumed the initial-response frame in + # deserialize_response, so subsequent events stay in order. + source = context.get(_EVENT_STREAM_SOURCE) or AsyncBytesReader(response.body) + prefetched = context.get(_PREFETCHED_EVENT) + if prefetched is not None: + # No initial-response: replay the first frame ahead of the live stream. + source = _PrependedEventStream(cast("_Event", prefetched), source) + return AWSEventReceiver( + payload_codec=self.payload_codec, + source=source, + deserializer=event_deserializer, + ) + + +class _RpcV2ClientProtocol(RpcClientProtocol): """Shared framing for the Smithy RPC v2 protocol family (rpcv2Cbor, rpcv2Json). Subclasses supply the codec-specific pieces: the protocol id, content type, @@ -404,13 +746,41 @@ def serialize_request[ operation_name = operation.schema.id.name path = f"/service/{self._service_name}/operation/{operation_name}" - # The protocol omits the body AND the Content-Type header for operations whose - # input is the unit type (no modeled input). Every other operation serializes - # the input shape, even when it has no members. + # An event-stream operation must negotiate HTTP/2 (see _requires_http2); mark + # the request so the transport does so for this operation alone, leaving + # non-streaming operations on HTTP/1.1. + http2_required = self._requires_http2(operation) + + # Accept describes the RESPONSE and Content-Type the REQUEST body, so they turn + # on output_stream_member and input_stream_member independently: a one-way + # stream advertises the event-stream media type on only the streaming side. + # Only an EVENT stream (union member) counts; a @streaming blob is a data stream + # and takes the normal media type. + accept = ( + _EVENT_STREAM_CONTENT_TYPE + if self._is_event_stream_member(operation.output_stream_member) + else self._content_type + ) fields: list[tuple[str, str]] = [ ("Smithy-Protocol", self._smithy_protocol), - ("Accept", self._content_type), + ("Accept", accept), ] + + if self._is_event_stream_member(operation.input_stream_member): + # RPC event streams carry the input as an initial-request event at the head + # of the stream body; the HTTP envelope is otherwise a normal RPC request. + fields.append(("Content-Type", _EVENT_STREAM_CONTENT_TYPE)) + return _HTTPRequest( + method="POST", + destination=_URI(host="", path=path), + fields=tuples_to_fields(fields), + body=self._build_initial_request_body(input, context), + http2_required=http2_required, + ) + + # The protocol omits the body AND the Content-Type header for operations whose + # input is the unit type (no modeled input). Every other operation serializes + # the input shape, even when it has no members. if _is_unit(operation.input_schema): payload = b"" else: @@ -423,6 +793,7 @@ def serialize_request[ destination=_URI(host="", path=path), fields=tuples_to_fields(fields), body=AsyncBytesReader(payload), + http2_required=http2_required, ) async def deserialize_response[ @@ -437,6 +808,14 @@ async def deserialize_response[ error_registry: TypeRegistry, context: TypedProperties, ) -> OperationOutput: + if ( + self._is_event_stream_member(operation.output_stream_member) + and response.status == 200 + ): + return await self._deserialize_event_stream_response( + operation=operation, response=response, context=context + ) + body = await response.consume_body_async() if response.status != 200: raise self._create_error( @@ -452,6 +831,22 @@ async def deserialize_response[ body = self._empty_document return self._codec.deserialize(source=body, shape=operation.output) + def _default_output[ + OperationInput: SerializeableShape, + OperationOutput: DeserializeableShape, + ]( + self, operation: APIOperation[OperationInput, OperationOutput] + ) -> OperationOutput: + # initial-response is required only if the output cannot be satisfied by an + # empty document. Clients populate missing members from their defaults, so a + # required member WITH a default is fine; only a required member with no + # default genuinely needs the initial-response. + if self._requires_initial_response(operation.output_schema): + raise MissingInitialResponse() + return self._codec.deserialize( + source=self._empty_document, shape=operation.output + ) + def _create_error( self, *, @@ -518,7 +913,40 @@ class RpcV2CborClientProtocol(_RpcV2ClientProtocol): def __init__(self, settings: ProtocolSettings) -> None: _assert_cbor() self._service_name = settings.service_target + self._event_stream_http = self._resolve_event_stream_http( + settings.service_schema, self._id + ) self._codec = CBORCodec(default_namespace=settings.namespace) def _decode_document(self, body: bytes) -> object: return _cbor_loads(body) # type: ignore[possibly-unbound] + + +class _PrependedEventStream: + """Serves one already-decoded event's bytes ahead of a live byte stream. + + Lets the receiver read a first event that ``deserialize_response`` pre-fetched + (an output with no initial-response) as though it were still on the wire, then + continues from the real source. + """ + + def __init__(self, event: "_Event", source: AsyncByteStream) -> None: + self._buffer = event.message.encode() + self._source = source + + async def read(self, size: int = -1) -> bytes: + if self._buffer: + if size < 0: + head, self._buffer = self._buffer, b"" + return head + await self._source.read(size) + head = self._buffer[:size] + self._buffer = self._buffer[size:] + return head + return await self._source.read(size) + + async def close(self) -> None: + # Forward close to the wrapped source so the receiver can tear down the + # underlying response body; without this the connection leaks. + if (close := getattr(self._source, "close", None)) is not None: + if asyncio.iscoroutine(result := close()): + await result diff --git a/packages/smithy-http/tests/unit/aio/test_crt.py b/packages/smithy-http/tests/unit/aio/test_crt.py index 65750b9d3..a972a039a 100644 --- a/packages/smithy-http/tests/unit/aio/test_crt.py +++ b/packages/smithy-http/tests/unit/aio/test_crt.py @@ -15,7 +15,6 @@ from smithy_http.aio import HTTPRequest from smithy_http.aio.crt import ( AWSCRTHTTPClient, - AWSCRTHTTPClientConfig, AWSCRTHTTPResponse, ) from smithy_http.exceptions import SmithyHTTPError @@ -31,8 +30,8 @@ async def test_close_closes_and_clears_pooled_connections() -> None: client = AWSCRTHTTPClient() connections = [AsyncMock(), AsyncMock()] client._connections = { - ("https", "one.example.com", None): connections[0], - ("https", "two.example.com", None): connections[1], + ("https", "one.example.com", None, False): connections[0], + ("https", "two.example.com", None, True): connections[1], } await client.close() @@ -392,32 +391,85 @@ async def test_build_connection_unsupported_scheme() -> None: async def test_validate_connection_http2_required() -> None: - """Test connection validation when force_http_2 is enabled.""" - config = AWSCRTHTTPClientConfig(force_http_2=True) - client = AWSCRTHTTPClient(client_config=config) - - # Mock HTTP/1.1 connection + """A non-HTTP/2 connection is rejected when HTTP/2 is required.""" + client = AWSCRTHTTPClient() mock_connection = AsyncMock() mock_connection.version = crt_http.HttpVersion.Http1_1 mock_connection.close = AsyncMock() with pytest.raises(SmithyHTTPError, match="HTTP/2 could not be negotiated"): - await client._validate_connection(mock_connection) + await client._validate_connection(mock_connection, require_http2=True) mock_connection.close.assert_called_once() async def test_validate_connection_http2_success() -> None: - """Test connection validation succeeds with HTTP/2.""" - config = AWSCRTHTTPClientConfig(force_http_2=True) - client = AWSCRTHTTPClient(client_config=config) - - # Mock HTTP/2 connection + """Validation passes for an HTTP/2 connection when HTTP/2 is required.""" + client = AWSCRTHTTPClient() mock_connection = AsyncMock() mock_connection.version = crt_http.HttpVersion.Http2 # Should not raise - await client._validate_connection(mock_connection) + await client._validate_connection(mock_connection, require_http2=True) + + +async def test_validate_connection_http1_allowed_when_not_required() -> None: + """An HTTP/1.1 connection is fine when HTTP/2 is not required.""" + client = AWSCRTHTTPClient() + mock_connection = AsyncMock() + mock_connection.version = crt_http.HttpVersion.Http1_1 + + # Should not raise + await client._validate_connection(mock_connection, require_http2=False) + + +async def _captured_alpn(require_http2: bool) -> list[str]: + client = AWSCRTHTTPClient() + tls_options = Mock() + client._tls_ctx = Mock() + client._tls_ctx.new_connection_options.return_value = tls_options + with patch("smithy_http.aio.crt.AIOHttpClientConnectionUnified.new", AsyncMock()): + await client._build_new_connection( + URI(scheme="https", host="example.com"), require_http2=require_http2 + ) + (alpn_list,), _ = tls_options.set_alpn_list.call_args + return alpn_list + + +async def test_alpn_h2_only_when_http2_required() -> None: + # When h2 is required, advertise only h2 so an endpoint that prefers http/1.1 + # cannot negotiate down below the required HTTP/2. + assert await _captured_alpn(require_http2=True) == ["h2"] + + +async def test_alpn_offers_both_when_not_required() -> None: + assert await _captured_alpn(require_http2=False) == ["h2", "http/1.1"] + + +async def test_pool_separates_http2_required_connections() -> None: + # An h2-required request must not reuse a pooled HTTP/1.1 connection to the same + # host, nor vice versa. + client = AWSCRTHTTPClient() + url = URI(scheme="https", host="example.com") + h1 = AsyncMock() + h1.version = crt_http.HttpVersion.Http1_1 + h1.is_open = Mock(return_value=True) + h2 = AsyncMock() + h2.version = crt_http.HttpVersion.Http2 + h2.is_open = Mock(return_value=True) + + with patch.object( + client, "_create_connection", AsyncMock(side_effect=[h1, h2]) + ) as created: + first = await client._get_connection(url, require_http2=False) + second = await client._get_connection(url, require_http2=True) + # Same host, different h2 requirement -> two distinct pooled connections. + assert first is h1 + assert second is h2 + assert created.call_count == 2 + # The h1 request reuses its own pooled connection. + assert await client._get_connection(url, require_http2=False) is h1 + assert created.call_count == 2 async def test_connection_pooling() -> None: diff --git a/packages/smithy-http/tests/unit/aio/test_event_streams.py b/packages/smithy-http/tests/unit/aio/test_event_streams.py new file mode 100644 index 000000000..f368664d5 --- /dev/null +++ b/packages/smithy-http/tests/unit/aio/test_event_streams.py @@ -0,0 +1,442 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +"""End-to-end event-stream wiring for the RPC protocols (rpcv2Cbor). + +Exercises the two halves the RPC protocols add over the shared factory: the +``initial-request`` frame written into a writable request body, and the +``initial-response`` peel that feeds the stream's initial output before the event +receiver continues from the same reader. +""" + +from io import BytesIO +from typing import Any + +import pytest +from smithy_core import URI +from smithy_core.aio.interfaces import AsyncWriter, ProtocolSettings +from smithy_core.aio.types import AsyncBytesProvider, AsyncBytesReader +from smithy_core.deserializers import ShapeDeserializer +from smithy_core.documents import TypeRegistry +from smithy_core.prelude import STRING +from smithy_core.schemas import APIOperation, Schema +from smithy_core.serializers import ShapeSerializer +from smithy_core.shapes import ShapeID, ShapeType +from smithy_core.traits import RequiredTrait, RpcV2CborTrait, StreamingTrait +from smithy_core.types import TypedProperties +from smithy_http import Fields +from smithy_http.aio import HTTPRequest, HTTPResponse +from smithy_http.aio.protocols import RpcV2CborClientProtocol + +pytest.importorskip("smithy_cbor") + +from smithy_aws_event_stream.aio import AWSEventReceiver +from smithy_aws_event_stream.events import Event, EventMessage +from smithy_aws_event_stream.exceptions import MissingInitialResponse +from smithy_cbor import CBORCodec + +_CODEC = CBORCodec() + +_MESSAGE_EVENT = Schema.collection( + id=ShapeID("smithy.example#MessageEvent"), + members={"body": {"target": STRING}}, +) + +_EVENT_STREAM = Schema.collection( + id=ShapeID("smithy.example#EventStream"), + shape_type=ShapeType.UNION, + traits=[StreamingTrait()], + members={"message": {"target": _MESSAGE_EVENT}}, +) + + +class InitialMessage: + """The operation input/output; ``message`` is the initial payload.""" + + def __init__(self, message: str | None = None) -> None: + self.message = message + + def serialize(self, serializer: ShapeSerializer) -> None: + with serializer.begin_struct(_STREAM_IO_SCHEMA) as s: + if self.message is not None: + s.write_string(_STREAM_IO_SCHEMA.members["message"], self.message) + + @classmethod + def deserialize(cls, deserializer: ShapeDeserializer) -> "InitialMessage": + instance = cls() + + def _consume(schema: Schema, de: ShapeDeserializer) -> None: + instance.message = de.read_string(_STREAM_IO_SCHEMA.members["message"]) + + deserializer.read_struct(_STREAM_IO_SCHEMA, _consume) + return instance + + +class _MessageEvent: + def __init__(self, body: str | None = None) -> None: + self.body = body + + def serialize(self, serializer: ShapeSerializer) -> None: + with serializer.begin_struct(_MESSAGE_EVENT) as s: + if self.body is not None: + s.write_string(_MESSAGE_EVENT.members["body"], self.body) + + @classmethod + def deserialize(cls, deserializer: ShapeDeserializer) -> "_MessageEvent": + instance = cls() + + def _consume(schema: Schema, de: ShapeDeserializer) -> None: + instance.body = de.read_string(_MESSAGE_EVENT.members["body"]) + + deserializer.read_struct(_MESSAGE_EVENT, _consume) + return instance + + +# The event-stream member is NOT part of the operation IO schema in codegen, but the +# protocol detects streaming via input_stream_member/output_stream_member, which scans +# the schema members for a @streaming target. So the IO schema used by the operation +# carries the stream member here. +_STREAM_IO_SCHEMA = Schema.collection( + id=ShapeID("smithy.example#StreamOpIO"), + members={ + "message": {"target": STRING}, + "stream": {"target": _EVENT_STREAM}, + }, +) + +# A non-streaming IO schema, for the quiet side of a one-way stream operation. +_PLAIN_IO_SCHEMA = Schema.collection( + id=ShapeID("smithy.example#PlainIO"), + members={"message": {"target": STRING}}, +) + +# A DATA stream (a @streaming blob), NOT an event stream. It must not force HTTP/2. +_DATA_STREAM_BLOB = Schema.collection( + id=ShapeID("smithy.example#DataStream"), + shape_type=ShapeType.BLOB, + traits=[StreamingTrait()], +) +_DATA_STREAM_IO_SCHEMA = Schema.collection( + id=ShapeID("smithy.example#DataStreamOpIO"), + members={"body": {"target": _DATA_STREAM_BLOB}}, +) + +# An output event stream whose non-stream member is @required with no default, so an +# absent initial-response cannot be filled by defaults and must raise. +_REQUIRED_OUTPUT_IO_SCHEMA = Schema.collection( + id=ShapeID("smithy.example#RequiredOutputIO"), + members={ + "message": {"target": STRING, "traits": [RequiredTrait()]}, + "stream": {"target": _EVENT_STREAM}, + }, +) + + +class _StubRegistry: + def build(self, name: str) -> type: + return InitialMessage + + +def _operation( + input_schema: Schema = _STREAM_IO_SCHEMA, + output_schema: Schema = _STREAM_IO_SCHEMA, +) -> APIOperation[Any, Any]: + return APIOperation( + static_schema=( + ShapeType.OPERATION.value, + "smithy.example", + "StreamOp", + {}, + input_schema, + output_schema, + (), + ), + model_registry=_StubRegistry(), + effective_auth_schemes=[], + ) + + +def _message_event_deserializer(deserializer: ShapeDeserializer) -> _MessageEvent: + """Union deserializer: dispatch the event-stream union to its member event.""" + result: _MessageEvent | None = None + + def _consume(schema: Schema, de: ShapeDeserializer) -> None: + nonlocal result + if schema.expect_member_index() == 0: + result = _MessageEvent.deserialize(de) + + deserializer.read_struct(_EVENT_STREAM, _consume) + if result is None: + raise AssertionError("expected a message event") + return result + + +def _service_schema(trait: RpcV2CborTrait | None = None) -> Schema: + return Schema.collection( + id=ShapeID("smithy.example#StreamService"), + shape_type=ShapeType.SERVICE, + traits=[trait] if trait is not None else None, + ) + + +def _protocol(service_schema: Schema | None = None) -> RpcV2CborClientProtocol: + return RpcV2CborClientProtocol( + ProtocolSettings( + namespace="smithy.example", + service_target="StreamService", + service_schema=service_schema or _service_schema(), + ) + ) + + +def _request() -> HTTPRequest: + return HTTPRequest( + destination=URI(host="example.com"), method="POST", fields=Fields() + ) + + +async def test_serialize_request_defers_initial_request_to_publisher() -> None: + # The frame is NOT poured into the body: it is stashed for the publisher, which + # writes (and would sign) it as the first event. The serialized body starts empty. + context = TypedProperties() + request = _protocol().serialize_request( + operation=_operation(), + input=InitialMessage(message="hello"), + endpoint=URI(host="example.com"), + context=context, + ) + assert ( + request.fields["Content-Type"].as_string() + == "application/vnd.amazon.eventstream" + ) + body = request.body + assert isinstance(body, AsyncBytesProvider) + assert isinstance(body, AsyncWriter) + + publisher = await _protocol().create_event_publisher( + operation=_operation(), + request=request, + event_type=Any, + context=context, + ) + assert publisher is not None + + # prepare() ran inside create_event_publisher, so the initial-request frame is now + # the first (and only) thing on the body. + frame_bytes = await anext(aiter(body)) + event = Event.decode(BytesIO(frame_bytes)) + assert event is not None + assert event.message.headers[":message-type"] == "event" + assert event.message.headers[":event-type"] == "initial-request" + decoded = _CODEC.deserialize(source=event.message.payload, shape=InitialMessage) + assert decoded.message == "hello" + + +async def test_deserialize_response_peels_initial_response_then_streams() -> None: + operation = _operation() + initial = EventMessage( + headers={ + ":message-type": "event", + ":event-type": "initial-response", + ":content-type": "application/cbor", + }, + payload=_CODEC.serialize(shape=InitialMessage(message="the initial response")), + ) + event = EventMessage( + headers={ + ":message-type": "event", + ":event-type": "message", + ":content-type": "application/cbor", + }, + payload=_CODEC.serialize(shape=_MessageEvent("first event")), + ) + response = HTTPResponse( + status=200, + fields=Fields(), + body=AsyncBytesReader(initial.encode() + event.encode()), + ) + + context = TypedProperties() + output = await _protocol().deserialize_response( + operation=operation, + request=_request(), + response=response, + error_registry=TypeRegistry({}), + context=context, + ) + + assert isinstance(output, InitialMessage) + assert output.message == "the initial response" + + receiver = _protocol().create_event_receiver( + operation=operation, + request=_request(), + response=response, + event_type=Any, + event_deserializer=_message_event_deserializer, + context=context, + ) + assert isinstance(receiver, AWSEventReceiver) + received = await receiver.receive() + assert received is not None + assert received.body == "first event" + + +def _serialize(operation: APIOperation[Any, Any]): + return _protocol().serialize_request( + operation=operation, + input=InitialMessage(message="hello"), + endpoint=URI(host="example.com"), + context=TypedProperties(), + ) + + +async def test_duplex_advertises_event_stream_both_ways() -> None: + # Both input and output stream: Accept and Content-Type are the event-stream type. + request = _serialize(_operation()) + assert request.fields["Accept"].as_string() == "application/vnd.amazon.eventstream" + assert ( + request.fields["Content-Type"].as_string() + == "application/vnd.amazon.eventstream" + ) + + +async def test_input_only_stream_keeps_normal_accept() -> None: + # PublishEvents shape: streaming request, non-streaming response. Content-Type is + # the event-stream type (request body is a stream); Accept is the codec type. + request = _serialize( + _operation(input_schema=_STREAM_IO_SCHEMA, output_schema=_PLAIN_IO_SCHEMA) + ) + assert request.fields["Accept"].as_string() == "application/cbor" + assert ( + request.fields["Content-Type"].as_string() + == "application/vnd.amazon.eventstream" + ) + + +async def test_output_only_stream_advertises_event_stream_accept() -> None: + # SubscribeToEvents shape: non-streaming request, streaming response. Accept is the + # event-stream type (response is a stream); Content-Type is the codec type. + request = _serialize( + _operation(input_schema=_PLAIN_IO_SCHEMA, output_schema=_STREAM_IO_SCHEMA) + ) + assert request.fields["Accept"].as_string() == "application/vnd.amazon.eventstream" + assert request.fields["Content-Type"].as_string() == "application/cbor" + + +async def test_event_stream_requires_http2() -> None: + # A @streaming UNION member is an event stream: it must negotiate HTTP/2. + request = _serialize(_operation()) + assert getattr(request, "http2_required") is True + + +async def test_data_stream_does_not_require_http2() -> None: + # A @streaming BLOB member is a DATA stream, not an event stream: it must NOT be + # forced onto HTTP/2. input_stream_member/output_stream_member match both, so this + # guards against treating a data stream as an event stream. + request = _serialize( + _operation(input_schema=_PLAIN_IO_SCHEMA, output_schema=_DATA_STREAM_IO_SCHEMA) + ) + assert getattr(request, "http2_required") is False + + +async def test_output_data_stream_keeps_normal_accept() -> None: + # A @streaming BLOB output is a data stream: Accept must stay the codec type, not + # the event-stream media type. output_stream_member matches a blob too, so this + # guards the Accept header against treating a data stream as an event stream. + request = _serialize( + _operation(input_schema=_PLAIN_IO_SCHEMA, output_schema=_DATA_STREAM_IO_SCHEMA) + ) + assert request.fields["Accept"].as_string() == "application/cbor" + + +async def test_non_streaming_does_not_require_http2() -> None: + request = _serialize( + _operation(input_schema=_PLAIN_IO_SCHEMA, output_schema=_PLAIN_IO_SCHEMA) + ) + assert getattr(request, "http2_required") is False + + +async def test_event_stream_http_opt_out_read_from_service_schema() -> None: + # eventStreamHttp is read off the service schema's protocol trait, keyed by the + # protocol id. An h1-only list opts the event stream out of forced HTTP/2. + protocol = _protocol( + service_schema=_service_schema( + RpcV2CborTrait({"eventStreamHttp": ["http/1.1"]}) + ) + ) + request = protocol.serialize_request( + operation=_operation(), + input=InitialMessage(message="hello"), + endpoint=URI(host="example.com"), + context=TypedProperties(), + ) + assert getattr(request, "http2_required") is False + # The output members are all optional, so the stream may omit initial-response and + # start with a real event. The output defaults, and that first event is not lost -- + # the receiver replays it. + operation = _operation() + event = EventMessage( + headers={ + ":message-type": "event", + ":event-type": "message", + ":content-type": "application/cbor", + }, + payload=_CODEC.serialize(shape=_MessageEvent("first event")), + ) + response = HTTPResponse( + status=200, fields=Fields(), body=AsyncBytesReader(event.encode()) + ) + + context = TypedProperties() + output = await _protocol().deserialize_response( + operation=operation, + request=_request(), + response=response, + error_registry=TypeRegistry({}), + context=context, + ) + assert isinstance(output, InitialMessage) + assert output.message is None + + receiver = _protocol().create_event_receiver( + operation=operation, + request=_request(), + response=response, + event_type=Any, + event_deserializer=_message_event_deserializer, + context=context, + ) + received = await receiver.receive() + assert received is not None + assert received.body == "first event" + + +async def test_deserialize_response_empty_stream_defaults_output() -> None: + # An empty stream with no initial-response yields the default output rather than + # raising, because every output member is optional. + response = HTTPResponse(status=200, fields=Fields(), body=AsyncBytesReader(b"")) + output = await _protocol().deserialize_response( + operation=_operation(), + request=_request(), + response=response, + error_registry=TypeRegistry({}), + context=TypedProperties(), + ) + assert isinstance(output, InitialMessage) + assert output.message is None + + +async def test_missing_initial_response_raises_when_required_member_has_no_default() -> ( + None +): + # The output has a @required member with no default, so an empty document cannot + # satisfy it: an absent initial-response must raise rather than fabricate a value. + response = HTTPResponse(status=200, fields=Fields(), body=AsyncBytesReader(b"")) + with pytest.raises(MissingInitialResponse): + await _protocol().deserialize_response( + operation=_operation(output_schema=_REQUIRED_OUTPUT_IO_SCHEMA), + request=_request(), + response=response, + error_registry=TypeRegistry({}), + context=TypedProperties(), + ) diff --git a/pyproject.toml b/pyproject.toml index 17b89a56f..8c2cbe851 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -46,6 +46,12 @@ smithy_python = { workspace = true } [tool.pyright] typeCheckingMode = "strict" enableExperimentalFeatures = true +# Drives a stubless h2 server and the dynamically-typed generated client; strict +# typing there is noise, not signal. +exclude = [ + "**/__pycache__", + "packages/smithy-aws-core/tests/integration", +] [tool.pytest.ini_options] asyncio_mode = "auto" # makes pytest run async tests without having to be marked with the @pytest.mark.asyncio decorator diff --git a/scripts/api-snapshot/api.json b/scripts/api-snapshot/api.json index d82b8bfe0..8c13bf091 100644 --- a/scripts/api-snapshot/api.json +++ b/scripts/api-snapshot/api.json @@ -74,12 +74,8 @@ "smithy_aws_core.aio.protocols": { "APIOperation": "rex-class", "AWSErrorIdentifier": "class", - "AWSEventPublisher": "rex-class[async]", - "AWSEventReceiver": "rex-class[async]", "AWSJSONDocument": "rex-class", "AsyncBytesReader": "rex-class[async]", - "AsyncWriter": "rex-protocol[async]", - "AuthScheme": "rex-protocol", "AwsJson10ClientProtocol": "class[async]", "AwsJson11ClientProtocol": "class[async]", "AwsJson1_0Trait": "rex-class", @@ -96,8 +92,6 @@ "Ec2QueryClientProtocol": "class[async]", "Ec2QueryShapeSerializer": "rex-class", "Ec2QueryTrait": "rex-class", - "EventPublisher": "rex-protocol[async]", - "EventReceiver": "rex-protocol[async]", "ExpectationNotMetError": "rex-exception", "HTTPErrorIdentifier": "rex-class", "HTTPRequest": "rex-protocol[async]", @@ -105,10 +99,10 @@ "HTTPResponseDeserializer": "rex-class", "HTTPTrait": "rex-class", "HttpBindingClientProtocol": "rex-class[async]", - "HttpClientProtocol": "rex-class[async]", "JSONCodec": "rex-class", "JSONDocument": "rex-class", "MissingDependencyError": "rex-exception", + "MissingInitialResponse": "rex-exception", "ModeledError": "rex-exception", "PropertyKey": "rex-class", "ProtocolSettings": "rex-class", @@ -118,18 +112,17 @@ "RestJsonClientProtocol": "class[async]", "RestXmlClientProtocol": "class[async]", "RestXmlTrait": "rex-class", + "RpcClientProtocol": "rex-class[async]", "SerializeableShape": "rex-protocol", "ShapeDeserializer": "rex-protocol", "ShapeID": "rex-class", "ShapeSerializer": "rex-protocol", "ShapeType": "rex-class", - "SigningConfig": "rex-class", "SyncStreamingBlob": "rex-type_alias", "TimestampFormat": "rex-class", "TypeRegistry": "rex-class", "TypedProperties": "rex-protocol", "URI": "rex-protocol", - "UnsupportedStreamError": "rex-exception", "XMLCodec": "rex-class", "create_aws_query_error": "rex-function", "parse_aws_query_request_id": "rex-function", @@ -1203,16 +1196,24 @@ }, "smithy_http.aio.protocols": { "APIOperation": "rex-class", + "AWSEventPublisher": "rex-class[async]", + "AWSEventReceiver": "rex-class[async]", "AsyncByteStream": "rex-protocol[async]", + "AsyncBytesProvider": "rex-class[async]", "AsyncBytesReader": "rex-class[async]", "AsyncStreamingBlob": "rex-type_alias", + "AsyncWriter": "rex-protocol[async]", + "AuthScheme": "rex-protocol", "CBORCodec": "rex-class", "CallError": "rex-exception", "ClientProtocol": "rex-protocol[async]", "Codec": "rex-protocol", "DOCUMENT": "rex-Schema", + "DefaultTrait": "rex-class", "DeserializeableShape": "rex-protocol", "Endpoint": "rex-protocol", + "EventPublisher": "rex-protocol[async]", + "EventReceiver": "rex-protocol[async]", "ExpectationNotMetError": "rex-exception", "HTTPBindingSerializer": "rex-class", "HTTPErrorIdentifier": "rex-class", @@ -1225,22 +1226,30 @@ "HttpBindingClientProtocol": "class[async]", "HttpClientProtocol": "class[async]", "MissingDependencyError": "rex-exception", + "MissingInitialResponse": "rex-exception", "ModeledError": "rex-exception", "ORIGINAL_SHAPE_ID": "rex-ShapeID", + "PropertyKey": "rex-class", "ProtocolSettings": "rex-class", + "RequiredTrait": "rex-class", "ResponseMetadata": "rex-class", + "RpcClientProtocol": "class[async]", "RpcV2CborClientProtocol": "class[async]", "RpcV2CborTrait": "rex-class", "Schema": "rex-class", "SeekableBytesReader": "rex-protocol", "SerializeableShape": "rex-protocol", "SerializeableStruct": "rex-protocol", + "ShapeDeserializer": "rex-protocol", "ShapeID": "rex-class", + "ShapeType": "rex-class", + "SigningConfig": "rex-class", "SyncStreamingBlob": "rex-type_alias", "TypeRegistry": "rex-class", "TypedProperties": "rex-protocol", "UNIT": "rex-Schema", "URI": "rex-protocol", + "UnsupportedStreamError": "rex-exception", "tuples_to_fields": "rex-function" }, "smithy_http.aio.restjson": {