import datetime

import importlib

import logging

import os

import sys

import time

from collections import Counter

import pytest

import sentry_sdk

import sentry_sdk.scope

from sentry_sdk import (

Hub,

add_breadcrumb,

capture_event,

capture_exception,

capture_message,

get_client,

isolation_scope,

last_event_id,

new_scope,

push_scope,

start_transaction,

)

from sentry_sdk.integrations import (

_AUTO_ENABLING_INTEGRATIONS,

_DEFAULT_INTEGRATIONS,

DidNotEnable,

Integration,

setup_integrations,

)

from sentry_sdk.integrations.logging import LoggingIntegration

from sentry_sdk.integrations.stdlib import StdlibIntegration

from sentry_sdk.scope import add_global_event_processor

from sentry_sdk.tracing_utils import has_tracing_enabled

from sentry_sdk.utils import datetime_from_isoformat, get_sdk_name, reraise

class NoOpIntegration(Integration):

"""

A simple no-op integration for testing purposes.

"""

identifier = "noop"

@staticmethod

def setup_once() -> None:

pass

def __eq__(self, __value: object) -> bool:

"""

All instances of NoOpIntegration should be considered equal to each other.

"""

return type(__value) == type(self)

def test_processors(sentry_init, capture_events):

sentry_init()

events = capture_events()

def error_processor(event, exc_info):

event["exception"]["values"][0]["value"] += " whatever"

return event

sentry_sdk.get_isolation_scope().add_error_processor(error_processor, ValueError)

try:

raise ValueError("aha!")

except Exception:

capture_exception()

(event,) = events

assert event["exception"]["values"][0]["value"] == "aha! whatever"

class ModuleImportErrorSimulator:

def __init__(self, modules, error_cls=DidNotEnable):

self.modules = modules

self.error_cls = error_cls

for sys_module in list(sys.modules.keys()):

if any(sys_module.startswith(module) for module in modules):

del sys.modules[sys_module]

def find_spec(self, fullname, _path, _target=None):

if fullname in self.modules:

raise self.error_cls("Test import failure for %s" % fullname)

def __enter__(self):

# WARNING: We need to be first to avoid pytest messing with local imports

sys.meta_path.insert(0, self)

def __exit__(self, *_args):

sys.meta_path.remove(self)

def test_auto_enabling_integrations_catches_import_error(sentry_init, caplog):

caplog.set_level(logging.DEBUG)

with ModuleImportErrorSimulator(

[i.rsplit(".", 1)[0] for i in _AUTO_ENABLING_INTEGRATIONS]

):

sentry_init(auto_enabling_integrations=True, debug=True)

for import_string in _AUTO_ENABLING_INTEGRATIONS:

assert any(

record.message.startswith(

"Did not import default integration {}:".format(import_string)

)

for record in caplog.records

), "Problem with checking auto enabling {}".format(import_string)

def test_generic_mechanism(sentry_init, capture_events):

sentry_init()

events = capture_events()

try:

raise ValueError("aha!")

except Exception:

capture_exception()

(event,) = events

assert event["exception"]["values"][0]["mechanism"]["type"] == "generic"

assert event["exception"]["values"][0]["mechanism"]["handled"]

def test_option_before_send(sentry_init, capture_events):

def before_send(event, hint):

event["extra"] = {"before_send_called": True}

return event

def do_this():

try:

raise ValueError("aha!")

except Exception:

capture_exception()

sentry_init(before_send=before_send)

events = capture_events()

do_this()

(event,) = events

assert event["extra"] == {"before_send_called": True}

def test_option_before_send_discard(sentry_init, capture_events):

def before_send_discard(event, hint):

return None

def do_this():

try:

raise ValueError("aha!")

except Exception:

capture_exception()

sentry_init(before_send=before_send_discard)

events = capture_events()

do_this()

assert len(events) == 0

def test_option_before_send_transaction(sentry_init, capture_events):

def before_send_transaction(event, hint):

assert event["type"] == "transaction"

event["extra"] = {"before_send_transaction_called": True}

return event

sentry_init(

before_send_transaction=before_send_transaction,

traces_sample_rate=1.0,

)

events = capture_events()

transaction = start_transaction(name="foo")

transaction.finish()

(event,) = events

assert event["transaction"] == "foo"

assert event["extra"] == {"before_send_transaction_called": True}

def test_option_before_send_transaction_discard(sentry_init, capture_events):

def before_send_transaction_discard(event, hint):

return None

sentry_init(

before_send_transaction=before_send_transaction_discard,

traces_sample_rate=1.0,

)

events = capture_events()

transaction = start_transaction(name="foo")

transaction.finish()

assert len(events) == 0

def test_option_before_breadcrumb(sentry_init, capture_events, monkeypatch):

drop_events = False

drop_breadcrumbs = False

reports = []

def record_lost_event(reason, data_category=None, item=None):

reports.append((reason, data_category))

def before_send(event, hint):

assert isinstance(hint["exc_info"][1], ValueError)

if not drop_events:

event["extra"] = {"foo": "bar"}

return event

def before_breadcrumb(crumb, hint):

assert hint == {"foo": 42}

if not drop_breadcrumbs:

crumb["data"] = {"foo": "bar"}

return crumb

sentry_init(before_send=before_send, before_breadcrumb=before_breadcrumb)

events = capture_events()

monkeypatch.setattr(

sentry_sdk.get_client().transport, "record_lost_event", record_lost_event

)

def do_this():

add_breadcrumb(message="Hello", hint={"foo": 42})

try:

raise ValueError("aha!")

except Exception:

capture_exception()

do_this()

drop_breadcrumbs = True

do_this()

assert not reports

drop_events = True

do_this()

assert reports == [("before_send", "error")]

normal, no_crumbs = events

assert normal["exception"]["values"][0]["type"] == "ValueError"

(crumb,) = normal["breadcrumbs"]["values"]

assert "timestamp" in crumb

assert crumb["message"] == "Hello"

assert crumb["data"] == {"foo": "bar"}

assert crumb["type"] == "default"

@pytest.mark.tests_internal_exceptions

def test_option_before_breadcrumb_exception(sentry_init, capture_events):

"""Exceptions in before_breadcrumb are contained."""

def before_breadcrumb(crumb, hint):

1 / 0

sentry_init(before_breadcrumb=before_breadcrumb)

events = capture_events()

def do_this():

add_breadcrumb(message="Hello", hint={"foo": 42})

try:

raise ValueError("aha!")

except Exception:

capture_exception()

do_this()

(event,) = events

assert event["exception"]["values"][0]["type"] == "ValueError"

(crumb,) = event["breadcrumbs"]["values"]

assert "timestamp" in crumb

assert crumb["message"] == "Hello"

assert crumb["type"] == "default"

@pytest.mark.parametrize(

"enable_tracing, traces_sample_rate, tracing_enabled, updated_traces_sample_rate",

[

(None, None, False, None),

(False, 0.0, False, 0.0),

(False, 1.0, False, 1.0),

(None, 1.0, True, 1.0),

(True, 1.0, True, 1.0),

(None, 0.0, True, 0.0), # We use this as - it's configured but turned off

(True, 0.0, True, 0.0), # We use this as - it's configured but turned off

(True, None, True, 1.0),

],

)

def test_option_enable_tracing(

sentry_init,

enable_tracing,

traces_sample_rate,

tracing_enabled,

updated_traces_sample_rate,

):

sentry_init(enable_tracing=enable_tracing, traces_sample_rate=traces_sample_rate)

options = sentry_sdk.get_client().options

assert has_tracing_enabled(options) is tracing_enabled

assert options["traces_sample_rate"] == updated_traces_sample_rate

def test_breadcrumb_arguments(sentry_init, capture_events):

assert_hint = {"bar": 42}

def before_breadcrumb(crumb, hint):

assert crumb["foo"] == 42

assert hint == assert_hint

sentry_init(before_breadcrumb=before_breadcrumb)

add_breadcrumb(foo=42, hint=dict(bar=42))

add_breadcrumb(dict(foo=42), dict(bar=42))

add_breadcrumb(dict(foo=42), hint=dict(bar=42))

add_breadcrumb(crumb=dict(foo=42), hint=dict(bar=42))

assert_hint.clear()

add_breadcrumb(foo=42)

add_breadcrumb(crumb=dict(foo=42))

def test_push_scope(sentry_init, capture_events, suppress_deprecation_warnings):

sentry_init()

events = capture_events()

with push_scope() as scope:

scope.level = "warning"

try:

1 / 0

except Exception as e:

capture_exception(e)

(event,) = events

assert event["level"] == "warning"

assert "exception" in event

def test_push_scope_null_client(

sentry_init, capture_events, suppress_deprecation_warnings

):

"""

This test can be removed when we remove push_scope and the Hub from the SDK.

"""

sentry_init()

events = capture_events()

Hub.current.bind_client(None)

with push_scope() as scope:

scope.level = "warning"

try:

1 / 0

except Exception as e:

capture_exception(e)

assert len(events) == 0

def test_breadcrumbs(sentry_init, capture_events):

sentry_init(max_breadcrumbs=10)

events = capture_events()

for i in range(20):

add_breadcrumb(

category="auth", message="Authenticated user %s" % i, level="info"

)

capture_exception(ValueError())

(event,) = events

assert len(event["breadcrumbs"]["values"]) == 10

assert "user 10" in event["breadcrumbs"]["values"][0]["message"]

assert "user 19" in event["breadcrumbs"]["values"][-1]["message"]

del events[:]

for i in range(2):

add_breadcrumb(

category="auth", message="Authenticated user %s" % i, level="info"

)

sentry_sdk.get_isolation_scope().clear()

capture_exception(ValueError())

(event,) = events

assert len(event["breadcrumbs"]["values"]) == 0

def test_breadcrumb_ordering(sentry_init, capture_events):

sentry_init()

events = capture_events()

now = datetime.datetime.now(datetime.timezone.utc).replace(microsecond=0)

timestamps = [

now - datetime.timedelta(days=10),

now - datetime.timedelta(days=8),

now - datetime.timedelta(days=12),

]

for timestamp in timestamps:

add_breadcrumb(

message="Authenticated at %s" % timestamp,

category="auth",

level="info",

timestamp=timestamp,

)

capture_exception(ValueError())

(event,) = events

assert len(event["breadcrumbs"]["values"]) == len(timestamps)

timestamps_from_event = [

datetime_from_isoformat(x["timestamp"]) for x in event["breadcrumbs"]["values"]

]

assert timestamps_from_event == sorted(timestamps)

def test_breadcrumb_ordering_different_types(sentry_init, capture_events):

sentry_init()

events = capture_events()

now = datetime.datetime.now(datetime.timezone.utc)

timestamps = [

now - datetime.timedelta(days=10),

now - datetime.timedelta(days=8),

now.replace(microsecond=0) - datetime.timedelta(days=12),

now - datetime.timedelta(days=9),

now - datetime.timedelta(days=13),

now.replace(microsecond=0) - datetime.timedelta(days=11),

]

breadcrumb_timestamps = [

timestamps[0],

timestamps[1].isoformat(),

datetime.datetime.strftime(timestamps[2], "%Y-%m-%dT%H:%M:%S") + "Z",

datetime.datetime.strftime(timestamps[3], "%Y-%m-%dT%H:%M:%S.%f") + "+00:00",

datetime.datetime.strftime(timestamps[4], "%Y-%m-%dT%H:%M:%S.%f") + "+0000",

datetime.datetime.strftime(timestamps[5], "%Y-%m-%dT%H:%M:%S.%f") + "-0000",

]

for i, timestamp in enumerate(timestamps):

add_breadcrumb(

message="Authenticated at %s" % timestamp,

category="auth",

level="info",

timestamp=breadcrumb_timestamps[i],

)

capture_exception(ValueError())

(event,) = events

assert len(event["breadcrumbs"]["values"]) == len(timestamps)

timestamps_from_event = [

datetime_from_isoformat(x["timestamp"]) for x in event["breadcrumbs"]["values"]

]

assert timestamps_from_event == sorted(timestamps)

def test_attachments(sentry_init, capture_envelopes):

sentry_init()

envelopes = capture_envelopes()

this_file = os.path.abspath(__file__.rstrip("c"))

scope = sentry_sdk.get_isolation_scope()

scope.add_attachment(bytes=b"Hello World!", filename="message.txt")

scope.add_attachment(path=this_file)

capture_exception(ValueError())

(envelope,) = envelopes

assert len(envelope.items) == 3

assert envelope.get_event()["exception"] is not None

attachments = [x for x in envelope.items if x.type == "attachment"]

(message, pyfile) = attachments

assert message.headers["filename"] == "message.txt"

assert message.headers["type"] == "attachment"

assert message.headers["content_type"] == "text/plain"

assert message.payload.bytes == message.payload.get_bytes() == b"Hello World!"

assert pyfile.headers["filename"] == os.path.basename(this_file)

assert pyfile.headers["type"] == "attachment"

assert pyfile.headers["content_type"].startswith("text/")

assert pyfile.payload.bytes is None

with open(this_file, "rb") as f:

assert pyfile.payload.get_bytes() == f.read()

@pytest.mark.tests_internal_exceptions

def test_attachments_graceful_failure(

sentry_init, capture_envelopes, internal_exceptions

):

sentry_init()

envelopes = capture_envelopes()

sentry_sdk.get_isolation_scope().add_attachment(path="non_existent")

capture_exception(ValueError())

(envelope,) = envelopes

assert len(envelope.items) == 2

assert envelope.items[1].payload.get_bytes() == b""

def test_attachments_exceptions(sentry_init):

sentry_init()

scope = sentry_sdk.get_isolation_scope()

# bytes and path are None

with pytest.raises(TypeError) as e:

scope.add_attachment()

assert str(e.value) == "path or raw bytes required for attachment"

# filename is None

with pytest.raises(TypeError) as e:

scope.add_attachment(bytes=b"Hello World!")

assert str(e.value) == "filename is required for attachment"

def test_attachments_content_type_is_none(sentry_init, capture_envelopes):

sentry_init()

envelopes = capture_envelopes()

scope = sentry_sdk.get_isolation_scope()

scope.add_attachment(

bytes=b"Hello World!", filename="message.txt", content_type="foo/bar"

)

capture_exception(ValueError())

(envelope,) = envelopes

attachments = [x for x in envelope.items if x.type == "attachment"]

(message,) = attachments

assert message.headers["filename"] == "message.txt"

assert message.headers["content_type"] == "foo/bar"

def test_attachments_repr(sentry_init):

sentry_init()

scope = sentry_sdk.get_isolation_scope()

scope.add_attachment(bytes=b"Hello World!", filename="message.txt")

assert repr(scope._attachments[0]) == "<Attachment 'message.txt'>"

def test_attachments_bytes_callable_payload(sentry_init):

sentry_init()

scope = sentry_sdk.get_isolation_scope()

scope.add_attachment(bytes=bytes, filename="message.txt")

attachment = scope._attachments[0]

item = attachment.to_envelope_item()

assert item.payload.bytes == b""

def test_integration_scoping(sentry_init, capture_events):

logger = logging.getLogger("test_basics")

# This client uses the logging integration

logging_integration = LoggingIntegration(event_level=logging.WARNING)

sentry_init(default_integrations=False, integrations=[logging_integration])

events = capture_events()

logger.warning("This is a warning")

assert len(events) == 1

# This client does not

sentry_init(default_integrations=False)

events = capture_events()

logger.warning("This is not a warning")

assert not events

default_integrations = [

getattr(

importlib.import_module(integration.rsplit(".", 1)[0]),

integration.rsplit(".", 1)[1],

)

for integration in _DEFAULT_INTEGRATIONS

]

@pytest.mark.forked

@pytest.mark.parametrize(

"provided_integrations,default_integrations,disabled_integrations,expected_integrations",

[

([], False, None, set()),

([], False, [], set()),

([LoggingIntegration()], False, None, {LoggingIntegration}),

([], True, None, set(default_integrations)),

(

[],

True,

[LoggingIntegration(), StdlibIntegration],

set(default_integrations) - {LoggingIntegration, StdlibIntegration},

),

],

)

def test_integrations(

sentry_init,

provided_integrations,

default_integrations,

disabled_integrations,

expected_integrations,

reset_integrations,

):

sentry_init(

integrations=provided_integrations,

default_integrations=default_integrations,

disabled_integrations=disabled_integrations,

auto_enabling_integrations=False,

debug=True,

)

assert {

type(integration) for integration in get_client().integrations.values()

} == expected_integrations

def test_scope_event_processor_order(sentry_init, capture_events):

def before_send(event, hint):

event["message"] += "baz"

return event

sentry_init(debug=True, before_send=before_send)

events = capture_events()

with new_scope() as scope:

@scope.add_event_processor

def foo(event, hint):

event["message"] += "foo"

return event

with new_scope() as scope:

@scope.add_event_processor

def bar(event, hint):

event["message"] += "bar"

return event

capture_message("hi")

(event,) = events

assert event["message"] == "hifoobarbaz"

def test_capture_event_with_scope_kwargs(sentry_init, capture_events):

sentry_init()

events = capture_events()

capture_event({}, level="info", extras={"foo": "bar"})

(event,) = events

assert event["level"] == "info"

assert event["extra"]["foo"] == "bar"

def test_dedupe_event_processor_drop_records_client_report(

sentry_init, capture_events, capture_record_lost_event_calls

):

"""

DedupeIntegration internally has an event_processor that filters duplicate exceptions.

We want a duplicate exception to be captured only once and the drop being recorded as

a client report.

"""

sentry_init()

events = capture_events()

record_lost_event_calls = capture_record_lost_event_calls()

try:

raise ValueError("aha!")

except Exception:

try:

capture_exception()

reraise(*sys.exc_info())

except Exception:

capture_exception()

(event,) = events

(lost_event_call,) = record_lost_event_calls

assert event["level"] == "error"

assert "exception" in event

assert lost_event_call == ("event_processor", "error", None, 1)

def test_dedupe_doesnt_take_into_account_dropped_exception(sentry_init, capture_events):

# Two exceptions happen one after another. The first one is dropped in the

# user's before_send. The second one isn't.

# Originally, DedupeIntegration would drop the second exception. This test

# is making sure that that is no longer the case -- i.e., DedupeIntegration

# doesn't consider exceptions dropped in before_send.

count = 0

def before_send(event, hint):

nonlocal count

count += 1

if count == 1:

return None

return event

sentry_init(before_send=before_send)

events = capture_events()

exc = ValueError("aha!")

for _ in range(2):

# The first ValueError will be dropped by before_send. The second

# ValueError will be accepted by before_send, and should be sent to

# Sentry.

try:

raise exc

except Exception:

capture_exception()

assert len(events) == 1

def test_event_processor_drop_records_client_report(

sentry_init, capture_events, capture_record_lost_event_calls

):

sentry_init(traces_sample_rate=1.0)

events = capture_events()

record_lost_event_calls = capture_record_lost_event_calls()

# Ensure full idempotency by restoring the original global event processors list object, not just a copy.

old_processors = sentry_sdk.scope.global_event_processors

try:

sentry_sdk.scope.global_event_processors = (

sentry_sdk.scope.global_event_processors.copy()

)

@add_global_event_processor

def foo(event, hint):

return None

capture_message("dropped")

with start_transaction(name="dropped"):

pass

assert len(events) == 0

# Using Counter because order of record_lost_event calls does not matter

assert Counter(record_lost_event_calls) == Counter(

[

("event_processor", "error", None, 1),

("event_processor", "transaction", None, 1),

("event_processor", "span", None, 1),

]

)

finally:

sentry_sdk.scope.global_event_processors = old_processors

@pytest.mark.tests_internal_exceptions

def test_before_send_exception_records_callback_error(

sentry_init, capture_events, capture_record_lost_event_calls

):

def bad_before_send(event, hint):

raise ValueError("before_send error")

sentry_init(before_send=bad_before_send, default_integrations=False)

events = capture_events()

record_lost_event_calls = capture_record_lost_event_calls()

capture_message("should be dropped")

assert len(events) == 0

assert ("callback_error", "error", None, 1) in record_lost_event_calls

def test_before_send_returning_none_records_before_send(

sentry_init, capture_events, capture_record_lost_event_calls

):

def dropping_before_send(event, hint):

return None

sentry_init(before_send=dropping_before_send)

events = capture_events()

record_lost_event_calls = capture_record_lost_event_calls()

capture_message("should be dropped")

assert len(events) == 0

assert ("before_send", "error", None, 1) in record_lost_event_calls

@pytest.mark.tests_internal_exceptions

def test_before_send_transaction_exception_records_callback_error(

sentry_init, capture_events, capture_record_lost_event_calls

):

def bad_before_send_transaction(event, hint):

raise ValueError("before_send_transaction error")

sentry_init(

traces_sample_rate=1.0,

before_send_transaction=bad_before_send_transaction,

default_integrations=False,

)

events = capture_events()

record_lost_event_calls = capture_record_lost_event_calls()

with start_transaction(name="test"):

pass

assert not any(e.get("type") == "transaction" for e in events)

assert ("callback_error", "transaction", None, 1) in record_lost_event_calls

assert any(

reason == "callback_error" and category == "span"

for reason, category, _, _ in record_lost_event_calls

)

@pytest.mark.parametrize(

"installed_integrations, expected_name",

[

# integrations with own name

(["django"], "sentry.python.django"),

(["flask"], "sentry.python.flask"),

(["fastapi"], "sentry.python.fastapi"),

(["bottle"], "sentry.python.bottle"),

(["falcon"], "sentry.python.falcon"),

(["quart"], "sentry.python.quart"),

(["sanic"], "sentry.python.sanic"),

(["starlette"], "sentry.python.starlette"),

(["starlite"], "sentry.python.starlite"),

(["litestar"], "sentry.python.litestar"),

(["chalice"], "sentry.python.chalice"),

(["serverless"], "sentry.python.serverless"),

(["pyramid"], "sentry.python.pyramid"),

(["tornado"], "sentry.python.tornado"),

(["aiohttp"], "sentry.python.aiohttp"),

(["aws_lambda"], "sentry.python.aws_lambda"),

(["gcp"], "sentry.python.gcp"),

(["beam"], "sentry.python.beam"),

(["asgi"], "sentry.python.asgi"),

(["wsgi"], "sentry.python.wsgi"),

# integrations without name

(["argv"], "sentry.python"),

(["atexit"], "sentry.python"),

(["boto3"], "sentry.python"),

(["celery"], "sentry.python"),

(["dedupe"], "sentry.python"),

(["excepthook"], "sentry.python"),

(["unraisablehook"], "sentry.python"),

(["executing"], "sentry.python"),

(["modules"], "sentry.python"),

(["pure_eval"], "sentry.python"),

(["redis"], "sentry.python"),

(["rq"], "sentry.python"),

(["sqlalchemy"], "sentry.python"),

(["stdlib"], "sentry.python"),

(["threading"], "sentry.python"),

(["trytond"], "sentry.python"),

(["logging"], "sentry.python"),

(["gnu_backtrace"], "sentry.python"),

(["httpx"], "sentry.python"),

# precedence of frameworks

(["flask", "django", "celery"], "sentry.python.django"),

(["fastapi", "flask", "redis"], "sentry.python.flask"),

(["bottle", "fastapi", "httpx"], "sentry.python.fastapi"),

(["falcon", "bottle", "logging"], "sentry.python.bottle"),

(["quart", "falcon", "gnu_backtrace"], "sentry.python.falcon"),

(["sanic", "quart", "sqlalchemy"], "sentry.python.quart"),

(["starlette", "sanic", "rq"], "sentry.python.sanic"),

(["chalice", "starlette", "modules"], "sentry.python.starlette"),

(["chalice", "starlite", "modules"], "sentry.python.starlite"),

(["chalice", "litestar", "modules"], "sentry.python.litestar"),

(["serverless", "chalice", "pure_eval"], "sentry.python.chalice"),

(["pyramid", "serverless", "modules"], "sentry.python.serverless"),

(["tornado", "pyramid", "executing"], "sentry.python.pyramid"),

(["aiohttp", "tornado", "dedupe"], "sentry.python.tornado"),

(["aws_lambda", "aiohttp", "boto3"], "sentry.python.aiohttp"),

(["gcp", "aws_lambda", "atexit"], "sentry.python.aws_lambda"),

(["beam", "gcp", "argv"], "sentry.python.gcp"),

(["asgi", "beam", "stdtlib"], "sentry.python.beam"),

(["wsgi", "asgi", "boto3"], "sentry.python.asgi"),

(["wsgi", "celery", "redis"], "sentry.python.wsgi"),

],

)

def test_get_sdk_name(installed_integrations, expected_name):

assert get_sdk_name(installed_integrations) == expected_name

def _hello_world(word):

return "Hello, {}".format(word)

@pytest.mark.parametrize("span_streaming", [True, False])

def test_functions_to_trace(sentry_init, capture_events, capture_items, span_streaming):

functions_to_trace = [

{"qualified_name": "tests.test_basics._hello_world"},

{"qualified_name": "time.sleep"},

]

global _hello_world

original_hello_world = _hello_world

original_sleep = time.sleep

try:

sentry_init(

traces_sample_rate=1.0,

functions_to_trace=functions_to_trace,

trace_lifecycle="stream" if span_streaming else "static",

)

if span_streaming:

items = capture_items("span")

else:

events = capture_events()

if span_streaming:

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

time.sleep(0)

for word in ["World", "You"]:

_hello_world(word)

sentry_sdk.flush()

spans = [item.payload for item in items]

child_spans = [s for s in spans if not s.get("is_segment")]

child_spans.sort(key=lambda s: s["start_timestamp"])

assert len(child_spans) == 3

assert child_spans[0]["name"] == "time.sleep"

assert child_spans[1]["name"] == "tests.test_basics._hello_world"

assert child_spans[2]["name"] == "tests.test_basics._hello_world"

else:

with start_transaction(name="something"):

time.sleep(0)

for word in ["World", "You"]:

_hello_world(word)

assert len(events) == 1

(event,) = events

assert len(event["spans"]) == 3

assert event["spans"][0]["description"] == "time.sleep"

assert event["spans"][1]["description"] == "tests.test_basics._hello_world"

assert event["spans"][2]["description"] == "tests.test_basics._hello_world"

finally:

_hello_world = original_hello_world

time.sleep = original_sleep

class WorldGreeter:

def __init__(self, word):

self.word = word

def greet(self, new_word=None):

return "Hello, {}".format(new_word if new_word else self.word)

@pytest.mark.parametrize("span_streaming", [True, False])

def test_functions_to_trace_with_class(

sentry_init, capture_events, capture_items, span_streaming

):

functions_to_trace = [

{"qualified_name": "tests.test_basics.WorldGreeter.greet"},

]

original_function = WorldGreeter.greet

try:

sentry_init(

traces_sample_rate=1.0,

functions_to_trace=functions_to_trace,

trace_lifecycle="stream" if span_streaming else "static",

)

if span_streaming:

items = capture_items("span")

else:

events = capture_events()

if span_streaming:

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

wg = WorldGreeter("World")

wg.greet()

wg.greet("You")

sentry_sdk.flush()