From d81085973646a6700778338b003b0a8f2b3f2126 Mon Sep 17 00:00:00 2001 From: Pablo Deputter Date: Thu, 29 Oct 2026 10:52:41 +0100 Subject: [PATCH 1/2] wip: check CI for different botocore versions --- sentry_sdk/consts.py | 6 + sentry_sdk/integrations/boto3/_client.py | 6 + .../integrations/boto3/_services/base.py | 20 ++- .../integrations/boto3/_services/sns.py | 18 ++- .../integrations/boto3/_services/sqs.py | 95 ++++++++++- sentry_sdk/integrations/boto3/_utils.py | 73 +++++++++ tests/integrations/boto3/test_sqs.py | 116 ++++++++++++++ .../boto3/test_trace_propagation.py | 149 ++++++++++++++++++ 8 files changed, 476 insertions(+), 7 deletions(-) diff --git a/sentry_sdk/consts.py b/sentry_sdk/consts.py index a98ec7b01e..c176b0e5ad 100644 --- a/sentry_sdk/consts.py +++ b/sentry_sdk/consts.py @@ -1393,6 +1393,12 @@ class SPANDATA: Example: "client", "server", "producer", "consumer", "internal" """ + SENTRY_LINK_TYPE = "sentry.link.type" + """ + Set on a span link. Describes the relationship between the span and the linked span. + Example: "message_origin" + """ + SENTRY_OP = "sentry.op" """ The operation of a span. diff --git a/sentry_sdk/integrations/boto3/_client.py b/sentry_sdk/integrations/boto3/_client.py index 35ea846e22..54bac75382 100644 --- a/sentry_sdk/integrations/boto3/_client.py +++ b/sentry_sdk/integrations/boto3/_client.py @@ -96,6 +96,10 @@ def sentry_patched_make_api_call( ctx.add_metadata(self) service_ext = _resolve_service(ctx.service_name) span = _start_client_span(ctx, service_ext) + if service_ext is not None: + with capture_internal_exceptions(): + # override the request params for e.g. trace propagation. + api_params = service_ext.get_request_params(ctx, span) or api_params if span is None: return orig_make_api_call(self, operation_name, api_params) @@ -122,6 +126,8 @@ def sentry_patched_make_api_call( attributes.update( service_ext.get_response_attributes(ctx, parsed) ) + with capture_internal_exceptions(): + span.add_links(service_ext.get_response_links(ctx, parsed)) with capture_internal_exceptions(): attributes.update(_get_response_attributes(parsed)) with capture_internal_exceptions(): diff --git a/sentry_sdk/integrations/boto3/_services/base.py b/sentry_sdk/integrations/boto3/_services/base.py index 397499bffb..6a81f8fdb3 100644 --- a/sentry_sdk/integrations/boto3/_services/base.py +++ b/sentry_sdk/integrations/boto3/_services/base.py @@ -1,10 +1,11 @@ from typing import TYPE_CHECKING if TYPE_CHECKING: - from typing import Any, Optional + from typing import Any, Dict, List, Optional - from sentry_sdk._types import Attributes + from sentry_sdk._types import Attributes, SpanLink from sentry_sdk.integrations.boto3._context import AwsCallContext + from sentry_sdk.traces import Span class _ServiceExtension: @@ -36,3 +37,18 @@ def get_response_attributes( ) -> "Attributes": """Return response attributes to add after the AWS request is made.""" return {} + + def get_response_links( + self, ctx: "AwsCallContext", response: "Any" + ) -> "List[SpanLink]": + """Return links to add to the span after the AWS request is made.""" + return [] + + def get_request_params( + self, ctx: "AwsCallContext", span: "Optional[Span]" + ) -> "Optional[Dict[str, Any]]": + """ + Return params to send instead of `ctx.params`, e.g. with trace propagation added, + or `None` to send them unchanged; must not mutate `ctx.params`. + """ + return None diff --git a/sentry_sdk/integrations/boto3/_services/sns.py b/sentry_sdk/integrations/boto3/_services/sns.py index b302d2693a..146a425762 100644 --- a/sentry_sdk/integrations/boto3/_services/sns.py +++ b/sentry_sdk/integrations/boto3/_services/sns.py @@ -3,13 +3,17 @@ import sentry_sdk from sentry_sdk.consts import OP, SPANDATA from sentry_sdk.integrations.boto3._services.base import _ServiceExtension -from sentry_sdk.integrations.boto3._utils import _get_aws_arn_attributes +from sentry_sdk.integrations.boto3._utils import ( + _get_aws_arn_attributes, + _inject_message_attributes, +) if TYPE_CHECKING: - from typing import Any, Optional + from typing import Any, Dict, Optional from sentry_sdk._types import Attributes from sentry_sdk.integrations.boto3._context import AwsCallContext + from sentry_sdk.traces import Span _PUBLISH_OPERATIONS = ("Publish", "PublishBatch") @@ -114,3 +118,13 @@ def get_response_attributes( elif error_codes: attributes[SPANDATA.ERROR_TYPE] = "aws_batch_partial_failure" return attributes + + def get_request_params( + self, ctx: "AwsCallContext", span: "Optional[Span]" + ) -> "Optional[Dict[str, Any]]": + if ctx.operation_name not in _PUBLISH_OPERATIONS: + return None + + return _inject_message_attributes( + ctx, span, "Message", "PublishBatchRequestEntries" + ) diff --git a/sentry_sdk/integrations/boto3/_services/sqs.py b/sentry_sdk/integrations/boto3/_services/sqs.py index 1be4fc1483..0925350c21 100644 --- a/sentry_sdk/integrations/boto3/_services/sqs.py +++ b/sentry_sdk/integrations/boto3/_services/sqs.py @@ -1,16 +1,23 @@ +import json import re from typing import TYPE_CHECKING from urllib.parse import unquote, urlsplit from sentry_sdk.consts import OP, SPANDATA from sentry_sdk.integrations.boto3._services.base import _ServiceExtension -from sentry_sdk.integrations.boto3._utils import _get_aws_arn_attributes +from sentry_sdk.integrations.boto3._utils import ( + _get_aws_arn_attributes, + _inject_message_attributes, +) +from sentry_sdk.traces import BAGGAGE_HEADER_NAME, SENTRY_TRACE_HEADER_NAME, NoOpSpan +from sentry_sdk.tracing_utils import extract_sentrytrace_data if TYPE_CHECKING: - from typing import Any, Optional + from typing import Any, Dict, List, Optional, Tuple - from sentry_sdk._types import Attributes + from sentry_sdk._types import Attributes, SpanLink from sentry_sdk.integrations.boto3._context import AwsCallContext + from sentry_sdk.traces import Span # operation -> (`sentry.op`, `messaging.operation.type`) _MESSAGING_OPERATIONS = { @@ -22,6 +29,44 @@ } +def _message_link(message: "Dict[str, Any]") -> "Optional[SpanLink]": + # message was sent straight to SQS, or through SNS with raw message delivery on. + sentry_trace = ( + message.get("MessageAttributes", {}) + .get(SENTRY_TRACE_HEADER_NAME, {}) + .get("StringValue") + ) + + # SNS wraps the message and its attributes in a JSON envelope when raw message delivery is off. + # https://docs.aws.amazon.com/sns/latest/dg/sns-large-payload-raw-message-delivery.html + body = message.get("Body", "") + if sentry_trace is None and f'"{SENTRY_TRACE_HEADER_NAME}"' in body: + try: + envelope = json.loads(body) + except ValueError: + envelope = None + if isinstance(envelope, dict) and envelope.get("Type") == "Notification": + sentry_trace = ( + envelope.get("MessageAttributes", {}) + .get(SENTRY_TRACE_HEADER_NAME, {}) + .get("Value") + ) + + data = extract_sentrytrace_data(sentry_trace) + if data is None or not data["trace_id"] or not data["parent_span_id"]: + return None + + link: "SpanLink" = { + "trace_id": str(data["trace_id"]), + "span_id": str(data["parent_span_id"]), + # Sentry drops other link attributes for now; add `messaging.message.id` in the future. + "attributes": {SPANDATA.SENTRY_LINK_TYPE: "message_origin"}, + } + if data["parent_sampled"] is not None: + link["sampled"] = bool(data["parent_sampled"]) + return link + + class _SQSExtension(_ServiceExtension): def get_span_op(self, ctx: "AwsCallContext") -> "Optional[str]": if ctx.operation_name in _MESSAGING_OPERATIONS: @@ -111,3 +156,47 @@ def get_response_attributes( else "aws_batch_partial_failure" ) return attributes + + def get_response_links( + self, ctx: "AwsCallContext", response: "Any" + ) -> "List[SpanLink]": + if ctx.operation_name != "ReceiveMessage": + return [] + + # one link per producer context; batch entries of one send share it. + links: "Dict[Tuple[str, str], SpanLink]" = {} + for message in response.get("Messages", ()): + # returns a link to the producer context a received message carries, if valid. + link = _message_link(message) + if link is not None: + links.setdefault((link["trace_id"], link["span_id"]), link) + return list(links.values()) + + def get_request_params( + self, ctx: "AwsCallContext", span: "Optional[Span]" + ) -> "Optional[Dict[str, Any]]": + if ctx.operation_name in ("SendMessage", "SendMessageBatch"): + return _inject_message_attributes(ctx, span, "MessageBody", "Entries") + + # only request the carrier when the receive span can record links. + if ( + ctx.operation_name != "ReceiveMessage" + or span is None + or isinstance(span, NoOpSpan) + ): + return None + + # SQS only returns the message attributes named in `MessageAttributeNames`. + # https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ReceiveMessage.html + names = ctx.params.get("MessageAttributeNames", []) + missing = [ + name + for name in (SENTRY_TRACE_HEADER_NAME, BAGGAGE_HEADER_NAME) + if name not in names + ] + # `All` and `.*` already return every attribute, including ours. + if not missing or "All" in names or ".*" in names: + return None + + # append ours to the caller's names, in a copy of their params. + return {**ctx.params, "MessageAttributeNames": [*names, *missing]} diff --git a/sentry_sdk/integrations/boto3/_utils.py b/sentry_sdk/integrations/boto3/_utils.py index 9f124639a6..8271391a92 100644 --- a/sentry_sdk/integrations/boto3/_utils.py +++ b/sentry_sdk/integrations/boto3/_utils.py @@ -1,15 +1,28 @@ from typing import TYPE_CHECKING +import sentry_sdk from sentry_sdk.consts import SPANDATA +from sentry_sdk.tracing_utils import should_propagate_trace if TYPE_CHECKING: from typing import Any, Dict, Iterable, Optional, Sequence, Tuple from sentry_sdk._types import Attributes + from sentry_sdk.integrations.boto3._context import AwsCallContext + from sentry_sdk.traces import Span # tuple of (request param name, span attribute name). _AttributeSpec = Tuple[str, str] +# SQS and SNS reject a message with more than 10 message attributes. +# https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-message-metadata.html +_MAX_MESSAGE_ATTRIBUTES = 10 + +# body and attributes must fit the queue's or topic's configurable `MaximumMessageSize`; +# assume the 256 KiB default of SNS topics and older SQS queues; batch uses the same limit. +# https://docs.aws.amazon.com/sns/latest/api/API_PublishBatch.html +_MAX_MESSAGE_SIZE = 256 * 1024 + def _extract_attributes( source: "Dict[str, Any]", specs: "Sequence[_AttributeSpec]" @@ -38,3 +51,63 @@ def _get_aws_arn_attributes( return attributes return {} + + +def _inject_message_attributes( + ctx: "AwsCallContext", span: "Optional[Span]", body_key: str, entries_key: str +) -> "Optional[Dict[str, Any]]": + """ + Return a copy of `ctx.params` with `sentry-trace` and `baggage` added in + string message attributes, or to each batch entry with room for them. + """ + if ctx.endpoint_url is None or not should_propagate_trace( + sentry_sdk.get_client(), ctx.endpoint_url + ): + return None + + carrier = { + name: {"DataType": "String", "StringValue": value} + for name, value in sentry_sdk.get_current_scope().iter_trace_propagation_headers( + span=span + ) + } + if not carrier: + return None + + def _get_message_size(body: str, attributes: "Dict[str, Any]") -> int: + size = len(body.encode()) + + # names, types and values of message attributes count towards the message size. + # https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-message-metadata.html + for name, value in attributes.items(): + size += len(name.encode()) + len(value["DataType"].encode()) + if "StringValue" in value: + size += len(value["StringValue"].encode()) + if "BinaryValue" in value: + binary = value["BinaryValue"] + size += len(binary.encode() if isinstance(binary, str) else binary) + return size + + is_batch = entries_key in ctx.params + messages = [] + for message in ctx.params[entries_key] if is_batch else (ctx.params,): + attributes = message.get("MessageAttributes", {}) + injected = {**attributes, **carrier} + if ( + # never replace the caller's carrier. + not carrier.keys() & attributes.keys() + # don't make AWS reject the message. + and len(injected) <= _MAX_MESSAGE_ATTRIBUTES + and _get_message_size(message[body_key], injected) <= _MAX_MESSAGE_SIZE + ): + message = {**message, "MessageAttributes": injected} + messages.append(message) + + total_size = sum( + _get_message_size(message[body_key], message.get("MessageAttributes", {})) + for message in messages + ) + if total_size > _MAX_MESSAGE_SIZE: + return None + + return {**ctx.params, entries_key: messages} if is_batch else messages[0] diff --git a/tests/integrations/boto3/test_sqs.py b/tests/integrations/boto3/test_sqs.py index f28f8cb74b..f3455a2258 100644 --- a/tests/integrations/boto3/test_sqs.py +++ b/tests/integrations/boto3/test_sqs.py @@ -1,3 +1,6 @@ +import copy +import json + import pytest from botocore.exceptions import ClientError from botocore.stub import Stubber @@ -306,3 +309,116 @@ def test_delete_message( assert attributes[SPANDATA.MESSAGING_OPERATION_NAME] == "settle" assert attributes.get(SPANDATA.MESSAGING_BATCH_MESSAGE_COUNT) == message_count assert "receipt-handle" not in str(span) + + +@pytest.mark.parametrize( + "names,sent_names", + [ + pytest.param(None, ["sentry-trace", "baggage"], id="default"), + pytest.param( + ["orderId", "baggage"], + ["orderId", "baggage", "sentry-trace"], + id="caller-names", + ), + pytest.param( + ["baggage", "sentry-trace"], + ["baggage", "sentry-trace"], + id="already-requested", + ), + pytest.param(["All"], ["All"], id="all"), + pytest.param([".*"], [".*"], id="all-pattern"), + ], +) +def test_receive_message_requests_carrier( + client_factory, capture_items, names, sent_names +): + client = client_factory("sqs") + queue_url = "https://sqs.eu-north-1.amazonaws.com/123456789012/orders" + params = {} if names is None else {"MessageAttributeNames": names} + caller_params = copy.deepcopy(params) + + with Stubber(client) as stubber: + stubber.add_response( + "receive_message", + {}, + {"QueueUrl": queue_url, "MessageAttributeNames": sent_names}, + ) + capture_spans_by_op( + lambda: client.receive_message(QueueUrl=queue_url, **params), + capture_items, + ) + + assert params == caller_params + + +def test_receive_message_links(client_factory, capture_items): + client = client_factory("sqs") + sentry_trace = { + "sentry-trace": { + "DataType": "String", + "StringValue": "771a43a4192642f0b136d5159a501700-a3e9e5ba6f99f4d6-1", + } + } + sns_envelope = { + "Type": "Notification", + "MessageId": "sns-message-3", + "TopicArn": "arn:aws:sns:eu-north-1:123456789012:alerts", + "Message": "hi", + "MessageAttributes": { + "sentry-trace": { + "Type": "String", + "Value": "b2c1b1d6cb5f4a0d8a6a2f1c3e9d7b40-9c4f3a2b1e0d8f7a", + } + }, + } + response = { + "Messages": [ + # batch entries of one send share a context. + {"MessageId": "1", "Body": "hi", "MessageAttributes": sentry_trace}, + {"MessageId": "2", "Body": "hi", "MessageAttributes": sentry_trace}, + # SNS-to-SQS delivery without raw message delivery. + {"MessageId": "3", "Body": json.dumps(sns_envelope)}, + # trace id with malformed sentry-trace. + { + "MessageId": "4", + "Body": "hi", + "MessageAttributes": { + "sentry-trace": { + "DataType": "String", + "StringValue": "771a43a4192642f0b136d5159a501700", + } + }, + }, + # JSON body that is not an SNS envelope. + {"MessageId": "5", "Body": '{"sentry-trace": "not an SNS envelope"}'}, + {"MessageId": "6", "Body": "hi"}, + ] + } + + with Stubber(client) as stubber: + stubber.add_response("receive_message", response) + spans = capture_spans_by_op( + lambda: client.receive_message( + QueueUrl="https://sqs.eu-north-1.amazonaws.com/123456789012/orders" + ), + capture_items, + ) + + (span,) = spans[OP.QUEUE_RECEIVE] + link_type = { + SPANDATA.SENTRY_LINK_TYPE: {"value": "message_origin", "type": "string"} + } + assert span["links"] == [ + # messages 1 and 2 share a context, so one link. + { + "trace_id": "771a43a4192642f0b136d5159a501700", + "span_id": "a3e9e5ba6f99f4d6", + "sampled": True, + "attributes": link_type, + }, + { + "trace_id": "b2c1b1d6cb5f4a0d8a6a2f1c3e9d7b40", + "span_id": "9c4f3a2b1e0d8f7a", + "attributes": link_type, + }, + ] diff --git a/tests/integrations/boto3/test_trace_propagation.py b/tests/integrations/boto3/test_trace_propagation.py index 41e3226215..6df105007b 100644 --- a/tests/integrations/boto3/test_trace_propagation.py +++ b/tests/integrations/boto3/test_trace_propagation.py @@ -1,18 +1,26 @@ +import copy from http.server import BaseHTTPRequestHandler, HTTPServer from threading import Thread from urllib.parse import parse_qs, urlparse import boto3 +import pytest from botocore.auth import SIGNED_HEADERS_BLACKLIST from botocore.config import Config +from botocore.stub import Stubber import sentry_sdk +from sentry_sdk.consts import OP from sentry_sdk.integrations.boto3 import Boto3Integration from sentry_sdk.integrations.stdlib import StdlibIntegration from sentry_sdk.utils import ( _get_aws_sigv4_signed_headers_from_authorization_header, _get_aws_sigv4_signed_headers_from_url_query_string, ) +from tests.integrations.boto3.helpers import ( + capture_spans_by_op, + require_botocore_model_fields, +) # botocore 1.43.115+ never signs `baggage`. BOTOCORE_SIGNS_BAGGAGE = "baggage" not in SIGNED_HEADERS_BLACKLIST @@ -207,3 +215,144 @@ def test_presigned_urls_do_not_require_sentry_headers(sentry_init): # presigned URL query must not contain `sentry-trace` or `baggage`. assert "sentry-trace" not in url assert "baggage" not in url + + +def _send(sentry_init, capture_items, service, messages, **options): + """ + Send `messages`, i.e. `(body, message attributes)`, with the service's + (batch) API; return the publish span and the sent messages. + """ + sentry_init(traces_sample_rate=1.0, integrations=[Boto3Integration()], **options) + + client = boto3.client( # type: ignore[attr-defined] + service, + region_name="eu-north-1", + aws_access_key_id="-", + aws_secret_access_key="-", + ) + + # service -> (single method, batch method, destination, body key, entries key) + single, batch, destination, body_key, entries_key = { + "sqs": ( + "send_message", + "send_message_batch", + {"QueueUrl": "https://sqs.eu-north-1.amazonaws.com/123456789012/orders"}, + "MessageBody", + "Entries", + ), + "sns": ( + "publish", + "publish_batch", + {"TopicArn": "arn:aws:sns:eu-north-1:123456789012:alerts"}, + "Message", + "PublishBatchRequestEntries", + ), + }[service] + + entries = [ + {"Id": str(i), body_key: body, "MessageAttributes": attributes} + for i, (body, attributes) in enumerate(messages) + ] + if len(entries) == 1: + method, response = single, {"MessageId": "1"} + (params,) = entries + # single-method API uses `MessageId` instead of `Id`. + del params["Id"] + params.update(destination) + else: + method, response = batch, {"Successful": [], "Failed": []} + params = {**destination, entries_key: entries} + + require_botocore_model_fields(client, method) + + sent_params = [] + client.meta.events.register( + "before-parameter-build", lambda params, **kwargs: sent_params.append(params) + ) + caller_params = copy.deepcopy(params) + with Stubber(client) as stubber: + stubber.add_response(method, response) + spans = capture_spans_by_op( + lambda: getattr(client, method)(**params), capture_items + ) + + # the caller's params are left as-is. + assert params == caller_params + (sent,) = sent_params + (span,) = spans[OP.QUEUE_PUBLISH] + return span, sent.get(entries_key, [sent]) + + +@pytest.mark.parametrize("message_count", [1, 2], ids=["single", "batch"]) +@pytest.mark.parametrize("service", ["sqs", "sns"]) +def test_message_attributes_propagation( + sentry_init, capture_items, service, message_count +): + span, sent = _send( + sentry_init, + capture_items, + service, + [("hi", {"orderId": {"DataType": "String", "StringValue": "42"}})] + * message_count, + ) + + # every batch entry carries the context of the publish span. + assert len(sent) == message_count + for message in sent: + attributes = message["MessageAttributes"] + assert attributes.keys() == {"orderId", "sentry-trace", "baggage"} + assert attributes["sentry-trace"] == { + "DataType": "String", + "StringValue": f"{span['trace_id']}-{span['span_id']}-1", + } + assert attributes["baggage"]["DataType"] == "String" + assert ( + f"sentry-trace_id={span['trace_id']}" + in attributes["baggage"]["StringValue"] + ) + + +@pytest.mark.parametrize( + "messages,options", + [ + pytest.param( + [("hi", {"baggage": {"DataType": "String", "StringValue": "vendor=1"}})], + {}, + id="caller-carrier", + ), + pytest.param( + [ + ( + "hi", + { + f"attribute{i}": {"DataType": "String", "StringValue": "1"} + for i in range(9) + }, + ) + ], + {}, + id="attribute-limit", + ), + pytest.param([("x" * 1024 * 256, {})], {}, id="message-size-limit"), + # 2 * (131_000 + ~385) > 262_144 which exceeds the message size limit. + pytest.param( + [("x" * 131_000, {}), ("x" * 131_000, {})], + {}, + id="batch-size-limit", + ), + pytest.param( + [("hi", {})], + {"trace_propagation_targets": ["example.com"]}, + id="trace-propagation-targets", + ), + ], +) +@pytest.mark.parametrize("service", ["sqs", "sns"]) +def test_message_attributes_propagation_skipped( + sentry_init, capture_items, service, messages, options +): + _, sent = _send(sentry_init, capture_items, service, messages, **options) + + assert [message["MessageAttributes"] for message in sent] == [ + attributes for _, attributes in messages + ] From bb65d326f7a38b72fa12b41e97cb1e0cd55be0f1 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Mon, 9 Nov 2026 08:01:48 +0000 Subject: [PATCH 2/2] test(boto3): Round-trip SQS/SNS trace propagation Send a message, receive what was injected, and check that the receive span links to the send span. Covers SQS and an SNS notification delivered to SQS without raw message delivery. Co-Authored-By: johdax <114897+dingsdax@users.noreply.github.com> --- .../boto3/test_trace_propagation.py | 47 +++++++++++++++++++ 1 file changed, 47 insertions(+) diff --git a/tests/integrations/boto3/test_trace_propagation.py b/tests/integrations/boto3/test_trace_propagation.py index 6df105007b..2c73ff7f2c 100644 --- a/tests/integrations/boto3/test_trace_propagation.py +++ b/tests/integrations/boto3/test_trace_propagation.py @@ -1,4 +1,5 @@ import copy +import json from http.server import BaseHTTPRequestHandler, HTTPServer from threading import Thread from urllib.parse import parse_qs, urlparse @@ -356,3 +357,49 @@ def test_message_attributes_propagation_skipped( assert [message["MessageAttributes"] for message in sent] == [ attributes for _, attributes in messages ] + + +@pytest.mark.parametrize("service", ["sqs", "sns"]) +def test_message_attributes_round_trip(sentry_init, capture_items, service): + """A receive span links to the send span whose context the message carries.""" + publish_span, (sent,) = _send(sentry_init, capture_items, service, [("hi", {})]) + + if service == "sqs": + message = {"MessageAttributes": sent["MessageAttributes"], "Body": "hi"} + else: + # an SNS subscription without raw message delivery wraps the message and its + # attributes in a JSON notification. + notification = { + "Type": "Notification", + "MessageId": "sns-1", + "TopicArn": "arn:aws:sns:eu-north-1:123456789012:alerts", + "Message": "hi", + "MessageAttributes": { + name: {"Type": value["DataType"], "Value": value["StringValue"]} + for name, value in sent["MessageAttributes"].items() + }, + } + message = {"Body": json.dumps(notification)} + + client = boto3.client( # type: ignore[attr-defined] + "sqs", + region_name="eu-north-1", + aws_access_key_id="-", + aws_secret_access_key="-", + ) + with Stubber(client) as stubber: + stubber.add_response( + "receive_message", {"Messages": [{"MessageId": "1", **message}]} + ) + spans = capture_spans_by_op( + lambda: client.receive_message( + QueueUrl="https://sqs.eu-north-1.amazonaws.com/123456789012/orders" + ), + capture_items, + ) + + (receive_span,) = spans[OP.QUEUE_RECEIVE] + (link,) = receive_span["links"] + assert link["trace_id"] == publish_span["trace_id"] + assert link["span_id"] == publish_span["span_id"] + assert link["sampled"] is True