import asyncio

import re

import sys

import time

from unittest import mock

import pytest

import sentry_sdk

from sentry_sdk.profiler.continuous_profiler import get_profiler_id

from sentry_sdk.traces import NoOpStreamedSpan, SpanStatus, StreamedSpan

minimum_python_38 = pytest.mark.skipif(

sys.version_info < (3, 8), reason="Asyncio tests need Python >= 3.8"

)

def test_start_span(sentry_init, capture_items):

sentry_init(

traces_sample_rate=1.0,

_experiments={"trace_lifecycle": "stream"},

)

items = capture_items("span")

with sentry_sdk.traces.start_span(name="segment") as segment:

assert segment._is_segment() is True

with sentry_sdk.traces.start_span(name="child") as child:

assert child._is_segment() is False

assert child._segment == segment

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert len(spans) == 2

child, segment = spans

assert segment["name"] == "segment"

assert segment["attributes"]["sentry.segment.name"] == "segment"

assert child["name"] == "child"

assert child["attributes"]["sentry.segment.name"] == "segment"

assert segment["is_segment"] is True

assert "parent_span_id" not in segment

assert child["is_segment"] is False

assert child["parent_span_id"] == segment["span_id"]

assert child["trace_id"] == segment["trace_id"]

assert segment["start_timestamp"] is not None

assert child["start_timestamp"] is not None

assert segment["end_timestamp"] is not None

assert child["end_timestamp"] is not None

assert child["status"] == "ok"

assert segment["status"] == "ok"

def test_start_span_no_context_manager(sentry_init, capture_items):

sentry_init(

traces_sample_rate=1.0,

_experiments={"trace_lifecycle": "stream"},

)

items = capture_items("span")

segment = sentry_sdk.traces.start_span(name="segment")

child = sentry_sdk.traces.start_span(name="child")

assert child._segment == segment

child.end()

segment.end()

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert len(spans) == 2

child, segment = spans

assert segment["name"] == "segment"

assert segment["attributes"]["sentry.segment.name"] == "segment"

assert child["name"] == "child"

assert child["attributes"]["sentry.segment.name"] == "segment"

assert segment["is_segment"] is True

assert child["is_segment"] is False

assert child["parent_span_id"] == segment["span_id"]

assert child["trace_id"] == segment["trace_id"]

assert segment["start_timestamp"] is not None

assert child["start_timestamp"] is not None

assert segment["end_timestamp"] is not None

assert child["end_timestamp"] is not None

assert child["status"] == "ok"

assert segment["status"] == "ok"

def test_span_sampled_when_created(sentry_init, capture_items):

# Test that if a span is created without the context manager, it is sampled

# at start_span() time

def traces_sampler(sampling_context):

assert "delayed_attribute" not in sampling_context["span_context"]["attributes"]

return 1.0

sentry_init(

traces_sampler=traces_sampler,

_experiments={"trace_lifecycle": "stream"},

)

items = capture_items("span")

segment = sentry_sdk.traces.start_span(name="segment")

segment.set_attribute("delayed_attribute", 12)

segment.end()

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert len(spans) == 1

(segment,) = spans

assert segment["name"] == "segment"

assert segment["attributes"]["delayed_attribute"] == 12

def test_start_span_attributes(sentry_init, capture_items):

sentry_init(

traces_sample_rate=1.0,

_experiments={"trace_lifecycle": "stream"},

)

items = capture_items("span")

with sentry_sdk.traces.start_span(

name="segment", attributes={"my_attribute": "my_value"}

):

...

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert len(spans) == 1

(span,) = spans

assert span["name"] == "segment"

assert span["attributes"]["my_attribute"] == "my_value"

def test_start_span_attributes_in_traces_sampler(sentry_init, capture_items):

def traces_sampler(sampling_context):

assert "attributes" in sampling_context["span_context"]

assert "my_attribute" in sampling_context["span_context"]["attributes"]

assert (

sampling_context["span_context"]["attributes"]["my_attribute"] == "my_value"

)

return 1.0

sentry_init(

traces_sampler=traces_sampler,

_experiments={"trace_lifecycle": "stream"},

)

items = capture_items("span")

with sentry_sdk.traces.start_span(

name="segment", attributes={"my_attribute": "my_value"}

):

...

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert len(spans) == 1

(span,) = spans

assert span["name"] == "segment"

assert span["attributes"]["my_attribute"] == "my_value"

def test_sampling_context(sentry_init, capture_items):

received_trace_id = None

def traces_sampler(sampling_context):

nonlocal received_trace_id

assert "trace_id" in sampling_context["span_context"]

received_trace_id = sampling_context["span_context"]["trace_id"]

assert "parent_span_id" in sampling_context["span_context"]

assert sampling_context["span_context"]["parent_span_id"] is None

assert "parent_sampled" in sampling_context["span_context"]

assert sampling_context["span_context"]["parent_sampled"] is None

assert "attributes" in sampling_context["span_context"]

return 1.0

sentry_init(

traces_sampler=traces_sampler,

_experiments={"trace_lifecycle": "stream"},

)

items = capture_items("span")

with sentry_sdk.traces.start_span(name="span") as span:

trace_id = span._trace_id

assert received_trace_id == trace_id

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert len(spans) == 1

def test_custom_sampling_context(sentry_init):

class MyClass: ...

my_class = MyClass()

def traces_sampler(sampling_context):

assert "class" in sampling_context

assert "string" in sampling_context

assert sampling_context["class"] == my_class

assert sampling_context["string"] == "my string"

return 1.0

sentry_init(

traces_sampler=traces_sampler,

_experiments={"trace_lifecycle": "stream"},

)

sentry_sdk.get_current_scope().set_custom_sampling_context(

{

"class": my_class,

"string": "my string",

}

)

with sentry_sdk.traces.start_span(name="span"):

...

def test_custom_sampling_context_update_to_context_value_persists(sentry_init):

def traces_sampler(sampling_context):

if sampling_context["span_context"]["attributes"]["first"] is True:

assert sampling_context["custom_value"] == 1

else:

assert sampling_context["custom_value"] == 2

return 1.0

sentry_init(

traces_sampler=traces_sampler,

_experiments={"trace_lifecycle": "stream"},

)

sentry_sdk.traces.new_trace()

sentry_sdk.get_current_scope().set_custom_sampling_context({"custom_value": 1})

with sentry_sdk.traces.start_span(name="span", attributes={"first": True}):

...

sentry_sdk.traces.new_trace()

sentry_sdk.get_current_scope().set_custom_sampling_context({"custom_value": 2})

with sentry_sdk.traces.start_span(name="span", attributes={"first": False}):

...

def test_before_send_span_basic(sentry_init, capture_items):

def before_send_span(span, hint):

assert isinstance(span, dict)

span["name"] = "Better span name"

del span["attributes"]["drop"]

span["attributes"]["sanitize"] = "[Removed]"

span["attributes"]["add"] = "new"

return span

sentry_init(

traces_sample_rate=1.0,

_experiments={

"before_send_span": before_send_span,

"trace_lifecycle": "stream",

},

)

items = capture_items("span")

with sentry_sdk.traces.start_span(

name="span",

attributes={

"drop": True,

"sanitize": "myamazingpassword",

},

):

...

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert len(spans) == 1

(span,) = spans

assert span["name"] == "Better span name"

assert "drop" not in span["attributes"]

assert span["attributes"]["sanitize"] == "[Removed]"

assert span["attributes"]["add"] == "new"

@pytest.mark.parametrize(

"return_value",

[None, {}, {"not_a_span": True}],

)

def test_before_send_span_invalid_return_value(

sentry_init, capture_items, return_value

):

def before_send_span(span, hint):

# Spans can't be dropped in before_send_span, so unsupported return

# values will be ignored

return return_value

sentry_init(

traces_sample_rate=1.0,

_experiments={

"before_send_span": before_send_span,

"trace_lifecycle": "stream",

},

)

items = capture_items("span")

with sentry_sdk.traces.start_span(name="span"):

...

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert len(spans) == 1

(span,) = spans

assert span["name"] == "span"

def test_before_send_span_unsupported_edit(sentry_init, capture_items):

def before_send_span(span, hint):

# Anything beyond attribute and name changes will be ignored

span["trace_id"] = "my-trace-id"

return span

sentry_init(

traces_sample_rate=1.0,

_experiments={

"before_send_span": before_send_span,

"trace_lifecycle": "stream",

},

)

items = capture_items("span")

with sentry_sdk.traces.start_span(name="span"):

...

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert len(spans) == 1

(span,) = spans

assert span["name"] == "span"

assert span["trace_id"] != "my-trace-id"

def test_before_send_span_doesnt_receive_ignored_spans(sentry_init, capture_items):

before_send_span_called = False

def before_send_span(span, hint):

nonlocal before_send_span_called

before_send_span_called = True

return span

sentry_init(

traces_sample_rate=1.0,

_experiments={

"before_send_span": before_send_span,

"trace_lifecycle": "stream",

"ignore_spans": [

"ignored",

],

},

)

items = capture_items("span")

with sentry_sdk.traces.start_span(name="ignored"):

...

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert not spans

assert not before_send_span_called

def test_span_attributes(sentry_init, capture_items):

sentry_init(

traces_sample_rate=1.0,

_experiments={"trace_lifecycle": "stream"},

)

items = capture_items("span")

with sentry_sdk.traces.start_span(

name="segment", attributes={"attribute1": "value"}

) as span:

assert span.get_attributes()["attribute1"] == "value"

span.set_attribute("attribute2", 47)

span.remove_attribute("attribute1")

span.set_attributes({"attribute3": 4.5, "attribute4": False})

assert "attribute1" not in span.get_attributes()

attributes = span.get_attributes()

assert attributes["attribute2"] == 47

assert attributes["attribute3"] == 4.5

assert attributes["attribute4"] is False

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert len(spans) == 1

(span,) = spans

assert span["name"] == "segment"

assert "attribute1" not in span["attributes"]

assert span["attributes"]["attribute2"] == 47

assert span["attributes"]["attribute3"] == 4.5

assert span["attributes"]["attribute4"] is False

def test_span_attributes_serialize_early(sentry_init, capture_items):

sentry_init(

traces_sample_rate=1.0,

_experiments={"trace_lifecycle": "stream"},

)

items = capture_items("span")

class Class:

pass

with sentry_sdk.traces.start_span(name="span") as span:

span.set_attributes(

{

# arrays of different types will be serialized

"attribute1": [123, "text"],

# so will custom class instances

"attribute2": Class(),

}

)

attributes = span.get_attributes()

assert isinstance(attributes["attribute1"], str)

assert attributes["attribute1"] == "[123, 'text']"

assert isinstance(attributes["attribute2"], str)

assert "Class" in attributes["attribute2"]

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert len(spans) == 1

(span,) = spans

assert span["attributes"]["attribute1"] == "[123, 'text']"

assert "Class" in span["attributes"]["attribute2"]

def test_traces_sampler_drops_span(sentry_init, capture_items):

def traces_sampler(sampling_context):

assert "attributes" in sampling_context["span_context"]

assert "drop" in sampling_context["span_context"]["attributes"]

if sampling_context["span_context"]["attributes"]["drop"] is True:

return 0.0

return 1.0

sentry_init(

traces_sampler=traces_sampler,

_experiments={"trace_lifecycle": "stream"},

)

items = capture_items("span")

with sentry_sdk.traces.start_span(name="dropped", attributes={"drop": True}):

...

with sentry_sdk.traces.start_span(name="retained", attributes={"drop": False}):

...

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert len(spans) == 1

(span,) = spans

assert span["name"] == "retained"

assert span["attributes"]["drop"] is False

def test_traces_sampler_called_once_per_segment(sentry_init):

traces_sampler_called = 0

span_name_in_traces_sampler = None

def traces_sampler(sampling_context):

nonlocal traces_sampler_called, span_name_in_traces_sampler

traces_sampler_called += 1

span_name_in_traces_sampler = sampling_context["span_context"]["name"]

return 1.0

sentry_init(

traces_sampler=traces_sampler,

_experiments={"trace_lifecycle": "stream"},

)

with sentry_sdk.traces.start_span(name="segment") as segment:

with sentry_sdk.traces.start_span(name="child1"):

...

with sentry_sdk.traces.start_span(name="child2"):

with sentry_sdk.traces.start_span(name="child3"):

...

assert traces_sampler_called == 1

assert span_name_in_traces_sampler == segment.name

def test_start_inactive_span(sentry_init, capture_items):

sentry_init(

traces_sample_rate=1.0,

_experiments={"trace_lifecycle": "stream"},

)

items = capture_items("span")

with sentry_sdk.traces.start_span(name="segment") as segment:

with sentry_sdk.traces.start_span(name="child1", active=False):

with sentry_sdk.traces.start_span(name="child2"):

# Should have segment as parent since child1 is inactive

pass

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert len(spans) == 3

child2, child1, segment = spans

assert segment["is_segment"] is True

assert segment["name"] == "segment"

assert segment["attributes"]["sentry.segment.name"] == "segment"

assert child1["is_segment"] is False

assert child1["name"] == "child1"

assert child1["attributes"]["sentry.segment.name"] == "segment"

assert child1["parent_span_id"] == segment["span_id"]

assert child1["trace_id"] == segment["trace_id"]

assert child2["is_segment"] is False

assert child2["name"] == "child2"

assert child2["attributes"]["sentry.segment.name"] == "segment"

assert child2["parent_span_id"] == segment["span_id"]

assert child2["trace_id"] == segment["trace_id"]

def test_start_span_override_parent(sentry_init, capture_items):

sentry_init(

traces_sample_rate=1.0,

_experiments={"trace_lifecycle": "stream"},

)

items = capture_items("span")

with sentry_sdk.traces.start_span(name="segment") as segment:

with sentry_sdk.traces.start_span(name="child1"):

with sentry_sdk.traces.start_span(name="child2", parent_span=segment):

pass

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert len(spans) == 3

child2, child1, segment = spans

assert segment["name"] == "segment"

assert segment["attributes"]["sentry.segment.name"] == "segment"

assert child1["name"] == "child1"

assert child1["attributes"]["sentry.segment.name"] == "segment"

assert child2["name"] == "child2"

assert child2["attributes"]["sentry.segment.name"] == "segment"

assert segment["is_segment"] is True

assert child1["is_segment"] is False

assert child1["parent_span_id"] == segment["span_id"]

assert child1["trace_id"] == segment["trace_id"]

assert child2["is_segment"] is False

assert child2["parent_span_id"] == segment["span_id"]

assert child2["trace_id"] == segment["trace_id"]

def test_sibling_segments(sentry_init, capture_items):

sentry_init(

traces_sample_rate=1.0,

_experiments={"trace_lifecycle": "stream"},

)

items = capture_items("span")

with sentry_sdk.traces.start_span(name="segment1"):

...

with sentry_sdk.traces.start_span(name="segment2"):

...

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert len(spans) == 2

segment1, segment2 = spans

assert segment1["name"] == "segment1"

assert segment1["attributes"]["sentry.segment.name"] == "segment1"

assert segment1["is_segment"] is True

assert "parent_span_id" not in segment1

assert segment2["name"] == "segment2"

assert segment2["attributes"]["sentry.segment.name"] == "segment2"

assert segment2["is_segment"] is True

assert "parent_span_id" not in segment2

assert segment1["trace_id"] == segment2["trace_id"]

def test_sibling_segments_new_trace(sentry_init, capture_items):

sentry_init(

traces_sample_rate=1.0,

_experiments={"trace_lifecycle": "stream"},

)

items = capture_items("span")

with sentry_sdk.traces.start_span(name="segment1"):

...

sentry_sdk.traces.new_trace()

with sentry_sdk.traces.start_span(name="segment2"):

...

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert len(spans) == 2

segment1, segment2 = spans

assert segment1["name"] == "segment1"

assert segment1["attributes"]["sentry.segment.name"] == "segment1"

assert segment1["is_segment"] is True

assert "parent_span_id" not in segment1

assert segment2["name"] == "segment2"

assert segment2["attributes"]["sentry.segment.name"] == "segment2"

assert segment2["is_segment"] is True

assert "parent_span_id" not in segment2

assert segment1["trace_id"] != segment2["trace_id"]

def test_continue_trace_sampled(sentry_init, capture_items):

sentry_init(

# parent sampling decision takes precedence over traces_sample_rate

traces_sample_rate=0.0,

_experiments={"trace_lifecycle": "stream"},

)

items = capture_items("span")

trace_id = "0af7651916cd43dd8448eb211c80319c"

parent_span_id = "b7ad6b7169203331"

sample_rand = "0.222222"

sampled = "1"

sentry_sdk.traces.continue_trace(

{

"sentry-trace": f"{trace_id}-{parent_span_id}-{sampled}",

"baggage": f"sentry-trace_id={trace_id},sentry-sample_rate=0.5,sentry-sample_rand={sample_rand}",

}

)

with sentry_sdk.traces.start_span(name="segment") as span:

...

assert span.sampled is True

assert span.trace_id == trace_id

assert span._parent_span_id == parent_span_id

assert span._sample_rand == float(sample_rand)

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert len(spans) == 1

(segment,) = spans

assert segment["is_segment"] is True

assert segment["parent_span_id"] == parent_span_id

assert segment["trace_id"] == trace_id

def test_continue_trace_unsampled(sentry_init, capture_items):

sentry_init(

# parent sampling decision takes precedence over traces_sample_rate

traces_sample_rate=1.0,

_experiments={"trace_lifecycle": "stream"},

)

items = capture_items("span")

trace_id = "0af7651916cd43dd8448eb211c80319c"

parent_span_id = "b7ad6b7169203331"

sample_rand = "0.999999"

sampled = "0"

sentry_sdk.traces.continue_trace(

{

"sentry-trace": f"{trace_id}-{parent_span_id}-{sampled}",

"baggage": f"sentry-trace_id={trace_id},sentry-sample_rate=0.5,sentry-sample_rand={sample_rand}",

}

)

with sentry_sdk.traces.start_span(name="segment") as span:

...

assert span.sampled is False

assert span.name == ""

assert span.trace_id == "00000000000000000000000000000000"

assert span.span_id == "0000000000000000"

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert len(spans) == 0

@pytest.mark.parametrize(

("sample_rand", "expected_sampled", "expected_outcome"),

[

("0.100000", True, None),

("0.250000", False, "backpressure"),

("0.300000", False, "backpressure"),

("0.500000", False, "sample_rate"),

("0.700000", False, "sample_rate"),

],

)

def test_backpressure_outcome(

sentry_init,

capture_items,

capture_record_lost_event_calls,

sample_rand,

expected_sampled,

expected_outcome,

):

sentry_init(

traces_sample_rate=0.5,

enable_backpressure_handling=True,

_experiments={"trace_lifecycle": "stream"},

)

items = capture_items("span")

record_lost_event_calls = capture_record_lost_event_calls()

client = sentry_sdk.get_client()

client.monitor._downsample_factor = 1

trace_id = "0af7651916cd43dd8448eb211c80319c"

parent_span_id = "b7ad6b7169203331"

sentry_sdk.traces.continue_trace(

{

"sentry-trace": f"{trace_id}-{parent_span_id}",

"baggage": f"sentry-trace_id={trace_id},sentry-sample_rand={sample_rand}",

}

)

with sentry_sdk.traces.start_span(name="span") as span:

pass

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

# Original traces_sample_rate is 0.5, downsampled sample rate is 0.25, so:

# - sample_rand < 0.25 -> sampled

# - 0.25 < sample_rand < 0.5 -> unsampled because of backpressure (would've been sampled if no backpressure)

# - 0.5 < sample_rand -> unsampled because of sampling rate

assert span.sampled is expected_sampled

if expected_sampled:

assert len(spans) == 1

assert not record_lost_event_calls

else:

assert len(spans) == 0

assert record_lost_event_calls == [

(expected_outcome, "span", None, 1),

]

def test_unsampled_spans_produce_client_report(

sentry_init, capture_items, capture_record_lost_event_calls

):

sentry_init(

traces_sample_rate=0.0,

_experiments={"trace_lifecycle": "stream"},

)

items = capture_items("span")

record_lost_event_calls = capture_record_lost_event_calls()

with sentry_sdk.traces.start_span(name="segment"):

with sentry_sdk.traces.start_span(name="child1"):

pass

with sentry_sdk.traces.start_span(name="child2"):

pass

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert not spans

assert record_lost_event_calls == [

("sample_rate", "span", None, 1),

("sample_rate", "span", None, 1),

("sample_rate", "span", None, 1),

]

def test_no_client_reports_if_tracing_is_off(

sentry_init, capture_items, capture_record_lost_event_calls

):

sentry_init(

traces_sample_rate=None,

_experiments={"trace_lifecycle": "stream"},

)

items = capture_items("span")

record_lost_event_calls = capture_record_lost_event_calls()

with sentry_sdk.traces.start_span(name="segment"):

with sentry_sdk.traces.start_span(name="child1"):

pass

with sentry_sdk.traces.start_span(name="child2"):

pass

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert not spans

assert not record_lost_event_calls

def test_continue_trace_no_sample_rand(sentry_init, capture_items):

sentry_init(

# parent sampling decision takes precedence over traces_sample_rate

traces_sample_rate=0.0,

_experiments={"trace_lifecycle": "stream"},

)

items = capture_items("span")

trace_id = "0af7651916cd43dd8448eb211c80319c"

parent_span_id = "b7ad6b7169203331"

sampled = "1"

sentry_sdk.traces.continue_trace(

{

"sentry-trace": f"{trace_id}-{parent_span_id}-{sampled}",

"baggage": f"sentry-trace_id={trace_id},sentry-sample_rate=0.5",

}

)

with sentry_sdk.traces.start_span(name="segment") as span:

...

assert span.sampled is True

assert span.trace_id == trace_id

assert span._parent_span_id == parent_span_id

assert isinstance(span._sample_rand, float)

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert len(spans) == 1

(segment,) = spans

assert segment["is_segment"] is True

assert segment["parent_span_id"] == parent_span_id

assert segment["trace_id"] == trace_id

def test_outgoing_traceparent_and_baggage(sentry_init, capture_envelopes):

sentry_init(

traces_sample_rate=1.0,

_experiments={"trace_lifecycle": "stream"},

)

sentry_sdk.traces.new_trace()

with sentry_sdk.traces.start_span(name="span") as span:

assert span.sampled is True

trace_id = span.trace_id

span_id = span.span_id

traceparent = sentry_sdk.get_traceparent()

assert traceparent == f"{trace_id}-{span_id}-1"

baggage = sentry_sdk.get_baggage()

baggage_items = dict(tuple(item.split("=")) for item in baggage.split(","))

assert "sentry-trace_id" in baggage_items

assert baggage_items["sentry-trace_id"] == trace_id

assert "sentry-sampled" in baggage_items

assert baggage_items["sentry-sampled"] == "true"

def test_outgoing_traceparent_and_baggage_when_noop_span_is_active(

sentry_init, capture_envelopes

):

sentry_init(

traces_sample_rate=1.0,

_experiments={

"trace_lifecycle": "stream",

"ignore_spans": ["ignored"],

},

)

sentry_sdk.traces.new_trace()

propagation_context = (

sentry_sdk.get_current_scope().get_active_propagation_context()

)

propagation_trace_id = propagation_context.trace_id

propagation_span_id = propagation_context.span_id

with sentry_sdk.traces.start_span(name="ignored") as span:

assert span.sampled is False

noop_trace_id = span.trace_id

noop_span_id = span.span_id

traceparent = sentry_sdk.get_traceparent()

assert traceparent != f"{noop_trace_id}-{noop_span_id}"

assert traceparent == f"{propagation_trace_id}-{propagation_span_id}"

baggage = sentry_sdk.get_baggage()

baggage_items = dict(tuple(item.split("=")) for item in baggage.split(","))

assert "sentry-trace_id" in baggage_items

assert baggage_items["sentry-trace_id"] != noop_trace_id

assert baggage_items["sentry-trace_id"] == propagation_trace_id

def test_trace_decorator(sentry_init, capture_items):

sentry_init(

traces_sample_rate=1.0,

_experiments={"trace_lifecycle": "stream"},

)

items = capture_items("span")

@sentry_sdk.traces.trace

def traced_function(): ...

traced_function()

sentry_sdk.get_client().flush()

spans = [item.payload for item in items]

assert len(spans) == 1

(span,) = spans

assert (

span["name"]

== "test_span_streaming.test_trace_decorator.<locals>.traced_function"

)

assert span["status"] == "ok"

def test_trace_decorator_arguments(sentry_init, capture_items):

sentry_init(