import asyncio

import gzip

import io

import json

import os

import socket

import threading

import warnings

from collections import namedtuple

from contextlib import contextmanager

from dataclasses import dataclass

from http.server import BaseHTTPRequestHandler, HTTPServer

from threading import Thread

from unittest import mock

from urllib.parse import parse_qs, urlparse

try:

import brotli

except ImportError:

brotli = None

try:

import jsonschema

except ImportError:

jsonschema = None

import pytest

try:

from pytest_localserver.http import WSGIServer

except ImportError:

WSGIServer = None

try:

from werkzeug.wrappers import Request, Response

except ImportError:

Request = None

Response = None

try:

from starlette.testclient import TestClient

# Catch RuntimeError to prevent the following exception in aws_lambda tests.

# RuntimeError: The starlette.testclient module requires the httpx package to be installed.

except (ImportError, RuntimeError):

TestClient = None

try:

import gevent

except ImportError:

gevent = None

try:

import eventlet

except ImportError:

eventlet = None

import sentry_sdk

import sentry_sdk.utils

from sentry_sdk.envelope import Envelope, parse_json

from sentry_sdk.integrations import ( # noqa: F401

_DEFAULT_INTEGRATIONS,

_installed_integrations,

_processed_integrations,

)

from sentry_sdk.profiler import teardown_profiler

from sentry_sdk.profiler.continuous_profiler import teardown_continuous_profiler

from sentry_sdk.transport import Transport

from sentry_sdk.utils import package_version, reraise

try:

import openai

except ImportError:

openai = None

try:

import anthropic

except ImportError:

anthropic = None

try:

import google

except ImportError:

google = None

from typing import TYPE_CHECKING

from tests import _warning_recorder, _warning_recorder_mgr

if TYPE_CHECKING:

from collections.abc import Iterator

from typing import Any, Callable, MutableMapping, Optional

try:

from httpx import (

ASGITransport,

AsyncByteStream,

AsyncClient,

)

from httpx import (

Request as HttpxRequest,

)

from httpx import (

Response as HttpxResponse,

)

except ImportError:

ASGITransport = None

HttpxRequest = None

HttpxResponse = None

AsyncByteStream = None

AsyncClient = None

try:

from anyio import EndOfStream, create_memory_object_stream, create_task_group

from mcp.shared.message import SessionMessage

_MCP_VERSION = package_version("mcp")

_IS_MCP_V2 = _MCP_VERSION is not None and _MCP_VERSION >= (2, 0, 0)

if _IS_MCP_V2:

from mcp_types import (

JSONRPCMessage,

JSONRPCNotification,

JSONRPCRequest,

)

else:

from mcp.types import (

JSONRPCMessage,

JSONRPCNotification,

JSONRPCRequest,

)

except ImportError:

create_memory_object_stream = None

create_task_group = None

EndOfStream = None

JSONRPCMessage = None

JSONRPCNotification = None

JSONRPCRequest = None

SessionMessage = None

_IS_MCP_V2 = False

SENTRY_EVENT_SCHEMA = "./checkouts/data-schemas/relay/event.schema.json"

if not os.path.isfile(SENTRY_EVENT_SCHEMA):

SENTRY_EVENT_SCHEMA = None

else:

with open(SENTRY_EVENT_SCHEMA) as f:

SENTRY_EVENT_SCHEMA = json.load(f)

from sentry_sdk import scope

@pytest.fixture(autouse=True)

def clean_scopes():

"""

Resets the scopes for every test to avoid leaking data between tests.

"""

scope._global_scope = None

scope._isolation_scope.set(None)

scope._current_scope.set(None)

@pytest.fixture(autouse=True)

def internal_exceptions(request):

errors = []

if "tests_internal_exceptions" in request.keywords:

return

def _capture_internal_exception(exc_info):

errors.append(exc_info)

@request.addfinalizer

def _():

# reraise the errors so that this just acts as a pass-through (that

# happens to keep track of the errors which pass through it)

for e in errors:

reraise(*e)

sentry_sdk.utils.capture_internal_exception = _capture_internal_exception

return errors

@pytest.fixture(autouse=True, scope="session")

def _capture_internal_warnings():

yield

_warning_recorder_mgr.__exit__(None, None, None)

recorder = _warning_recorder

for warning in recorder:

try:

if isinstance(warning.message, ResourceWarning):

continue

except NameError:

pass

if "sentry_sdk" not in str(warning.filename) and "sentry-sdk" not in str(

warning.filename

):

continue

# pytest-django

if "getfuncargvalue" in str(warning.message):

continue

# Happens when re-initializing the SDK

if "but it was only enabled on init()" in str(warning.message):

continue

# sanic's usage of aiohttp for test client

if "verify_ssl is deprecated, use ssl=False instead" in str(warning.message):

continue

if "getargspec" in str(warning.message) and warning.filename.endswith(

("pyramid/config/util.py", "pyramid/config/views.py")

):

continue

if "isAlive() is deprecated" in str(

warning.message

) and warning.filename.endswith("celery/utils/timer2.py"):

continue

if "collections.abc" in str(warning.message) and warning.filename.endswith(

("celery/canvas.py", "werkzeug/datastructures.py", "tornado/httputil.py")

):

continue

# Django 1.7 emits a (seemingly) false-positive warning for our test

# app and suggests to use a middleware that does not exist in later

# Django versions.

if "SessionAuthenticationMiddleware" in str(warning.message):

continue

if "Something has already installed a non-asyncio" in str(warning.message):

continue

if "dns.hash" in str(warning.message) or "dns/namedict" in warning.filename:

continue

# On Python < 3.8 we fall back to importing `pkg_resources` to

# enumerate installed modules, which emits a DeprecationWarning.

if "pkg_resources is deprecated as an API" in str(

warning.message

) and warning.filename.endswith("sentry_sdk/utils.py"):

continue

raise AssertionError(warning)

@pytest.fixture

def validate_event_schema(tmpdir):

def inner(event):

assert jsonschema is not None

if SENTRY_EVENT_SCHEMA:

jsonschema.validate(instance=event, schema=SENTRY_EVENT_SCHEMA)

return inner

@pytest.fixture

def reset_integrations():

"""

Use with caution, sometimes we really need to start

with a clean slate to ensure monkeypatching works well,

but this also means some other stuff will be monkeypatched twice.

"""

global _DEFAULT_INTEGRATIONS, _processed_integrations

try:

_DEFAULT_INTEGRATIONS.remove(

"sentry_sdk.integrations.opentelemetry.integration.OpenTelemetryIntegration"

)

except ValueError:

pass

_processed_integrations.clear()

_installed_integrations.clear()

@pytest.fixture

def uninstall_integration():

"""Use to force the next call to sentry_init to re-install/setup an integration."""

def inner(identifier):

_processed_integrations.discard(identifier)

_installed_integrations.discard(identifier)

return inner

def _install_flush_completion_handshake(client: "sentry_sdk.Client") -> None:

"""Make batcher.flush() wait for the flusher thread to drain.

Otherwise, test assertions can run before envelopes are captured.

The span batcher flushes pending items asynchronously with the main thread.

Flushes triggered by segments finishing are asynchronous, and can collect buckets

that would have otherwise been flushed synchronously by `sentry_sdk.flush()`.

"""

batcher = client.span_batcher

if batcher is None:

return

orig_flush_raw = batcher._flush

orig_flush = batcher.flush

lock = threading.Lock()

drained_count = 0

wake = threading.Event()

def _flush(*args: "Any", **kwargs: "Any") -> "Any":

nonlocal drained_count

try:

return orig_flush_raw(*args, **kwargs)

finally:

with lock:

drained_count += 1

wake.set()

def flush() -> None:

nonlocal drained_count

# Re-entrancy guard: if `flush()` is invoked from within a drain (e.g. a

# custom transport), waiting on the flusher thread would deadlock, because

# the flusher is blocked inside our own handler.

if getattr(getattr(batcher, "_active", None), "flag", False):

orig_flush()

return

# If the background flusher thread was never started (no spans have

# been added), there is no thread to drain and the counter will never

# advance. Fall back to the original synchronous flush.

if batcher._flusher is None or not batcher._flusher.is_alive():

orig_flush()

return

with lock:

target = drained_count

batcher._flush_event.set()

while True:

with lock:

if drained_count > target:

break

wake.wait()

wake.clear()

orig_flush()

object.__setattr__(batcher, "_flush", _flush)

object.__setattr__(batcher, "flush", flush)

@pytest.fixture

def sentry_init(request):

def inner(*a, **kw):

kw.setdefault("transport", TestTransport())

client = sentry_sdk.Client(*a, **kw)

_install_flush_completion_handshake(client)

sentry_sdk.get_global_scope().set_client(client)

if request.node.get_closest_marker("forked"):

# Do not run isolation if the test is already running in

# ultimate isolation (seems to be required for celery tests that

# fork)

yield inner

else:

old_client = sentry_sdk.get_global_scope().client

try:

sentry_sdk.get_current_scope().set_client(None)

yield inner

finally:

sentry_sdk.get_global_scope().set_client(old_client)

class TestTransport(Transport):

def __init__(self):

Transport.__init__(self)

def capture_envelope(self, _: Envelope) -> None:

"""No-op capture_envelope for tests"""

pass

class TestTransportWithOptions(Transport):

"""TestTransport above does not pass in the options and for some tests we need them"""

__test__ = False

def __init__(self, options=None):

Transport.__init__(self, options)

def capture_envelope(self, _: Envelope) -> None:

"""No-op capture_envelope for tests"""

pass

@pytest.fixture

def capture_events(monkeypatch):

def inner():

events = []

test_client = sentry_sdk.get_client()

old_capture_envelope = test_client.transport.capture_envelope

def append_event(envelope):

for item in envelope:

if item.headers.get("type") in ("event", "transaction"):

events.append(item.payload.json)

return old_capture_envelope(envelope)

monkeypatch.setattr(test_client.transport, "capture_envelope", append_event)

return events

return inner

@pytest.fixture

def capture_envelopes(monkeypatch):

def inner():

envelopes = []

test_client = sentry_sdk.get_client()

old_capture_envelope = test_client.transport.capture_envelope

def append_envelope(envelope):

envelopes.append(envelope)

return old_capture_envelope(envelope)

monkeypatch.setattr(test_client.transport, "capture_envelope", append_envelope)

return envelopes

return inner

@dataclass

class UnwrappedItem:

type: str

payload: dict

@pytest.fixture

def capture_items(monkeypatch):

"""

Capture envelope payload, unfurling individual items.

Makes it easier to work with both events and attribute-based telemetry in

one test.

"""

def inner(*types):

telemetry = []

test_client = sentry_sdk.get_client()

old_capture_envelope = test_client.transport.capture_envelope

def append_envelope(envelope):

for item in envelope:

if types and item.type not in types:

continue

if item.type in ("trace_metric", "log", "span"):

for i in item.payload.json["items"]:

t = {k: v for k, v in i.items() if k != "attributes"}

t["attributes"] = {

k: v["value"] for k, v in i["attributes"].items()

}

telemetry.append(UnwrappedItem(type=item.type, payload=t))

else:

telemetry.append(

UnwrappedItem(type=item.type, payload=item.payload.json)

)

return old_capture_envelope(envelope)

monkeypatch.setattr(test_client.transport, "capture_envelope", append_envelope)

return telemetry

return inner

@pytest.fixture

def capture_record_lost_event_calls(monkeypatch):

def inner():

calls = []

test_client = sentry_sdk.get_client()

def record_lost_event(reason, data_category=None, item=None, *, quantity=1):

calls.append((reason, data_category, item, quantity))

monkeypatch.setattr(

test_client.transport, "record_lost_event", record_lost_event

)

return calls

return inner

@pytest.fixture

def capture_events_forksafe(monkeypatch, capture_events, request):

def inner():

capture_events()

events_r, events_w = os.pipe()

events_r = os.fdopen(events_r, "rb", 0)

events_w = os.fdopen(events_w, "wb", 0)

test_client = sentry_sdk.get_client()

old_capture_envelope = test_client.transport.capture_envelope

def append(envelope):

event = envelope.get_event() or envelope.get_transaction_event()

if event is not None:

events_w.write(json.dumps(event).encode("utf-8"))

events_w.write(b"\n")

return old_capture_envelope(envelope)

def flush(timeout=None, callback=None):

events_w.write(b"flush\n")

monkeypatch.setattr(test_client.transport, "capture_envelope", append)

monkeypatch.setattr(test_client, "flush", flush)

return EventStreamReader(events_r, events_w)

return inner

@pytest.fixture

def capture_items_forksafe(monkeypatch, capture_items, request):

def inner(*types):

capture_items(*types)

items_r, items_w = os.pipe()

items_r = os.fdopen(items_r, "rb", 0)

items_w = os.fdopen(items_w, "wb", 0)

test_client = sentry_sdk.get_client()

old_capture_envelope = test_client.transport.capture_envelope

telemetry = []

def append(envelope):

for item in envelope:

if types and item.type not in types:

continue

if item.type in ("metric", "log", "span"):

for i in item.payload.json["items"]:

t = {k: v for k, v in i.items() if k != "attributes"}

t["attributes"] = {

k: v["value"] for k, v in i["attributes"].items()

}

telemetry.append({"type": item.type, "payload": t})

else:

telemetry.append({"type": item.type, "payload": item.payload.json})

return old_capture_envelope(envelope)

real_flush = test_client.flush

def flush(timeout=None, callback=None):

real_flush(timeout=timeout, callback=callback)

items_w.write(json.dumps(telemetry).encode("utf-8") + b"\n")

items_w.write(b"flush\n")

monkeypatch.setattr(test_client.transport, "capture_envelope", append)

monkeypatch.setattr(test_client, "flush", flush)

return EventStreamReader(items_r, items_w)

return inner

class EventStreamReader:

def __init__(self, read_file, write_file):

self.read_file = read_file

self.write_file = write_file

def read_event(self):

return json.loads(self.read_file.readline().decode("utf-8"))

def read_flush(self):

assert self.read_file.readline() == b"flush\n"

# scope=session ensures that fixture is run earlier

@pytest.fixture(

scope="session",

params=[None, "eventlet", "gevent"],

ids=("threads", "eventlet", "greenlet"),

)

def maybe_monkeypatched_threading(request):

if request.param == "eventlet":

if eventlet is None:

pytest.skip("no eventlet installed")

try:

eventlet.monkey_patch()

except AttributeError as e:

if "'thread.RLock' object has no attribute" in str(e):

# https://bitbucket.org/pypy/pypy/issues/2962/gevent-cannot-patch-rlock-under-pypy-27-7

pytest.skip("https://github.com/eventlet/eventlet/issues/546")

else:

raise

elif request.param == "gevent":

if gevent is None:

pytest.skip("no gevent installed")

try:

gevent.monkey.patch_all()

except Exception as e:

if "_RLock__owner" in str(e):

pytest.skip("https://github.com/gevent/gevent/issues/1380")

else:

raise

else:

assert request.param is None

return request.param

@pytest.fixture

def render_span_tree():

def inner(spans, root_span=None):

streamed_spans = False

if root_span is None:

streamed_spans = True

by_parent = {}

for span in spans:

if "parent_span_id" not in span:

root_span = span

continue

by_parent.setdefault(span["parent_span_id"], []).append(span)

def render_span(span):

if streamed_spans:

yield "- sentry.op={}: name={}".format(

json.dumps(span["attributes"].get("sentry.op")),

json.dumps(span["name"]),

)

else:

yield "- op={}: description={}".format(

json.dumps(span.get("op")), json.dumps(span.get("description"))

)

for subspan in by_parent.get(span["span_id"]) or ():

for line in render_span(subspan):

yield " {}".format(line)

return "\n".join(render_span(root_span))

return inner

@pytest.fixture(name="StringContaining")

def string_containing_matcher():

"""

An object which matches any string containing the substring passed to the

object at instantiation time.

Useful for assert_called_with, assert_any_call, etc.

Used like this:

>>> f = mock.Mock()

>>> f("dogs are great")

>>> f.assert_any_call("dogs") # will raise AssertionError

Traceback (most recent call last):

...

AssertionError: mock('dogs') call not found

>>> f.assert_any_call(StringContaining("dogs")) # no AssertionError

"""

class StringContaining:

def __init__(self, substring):

self.substring = substring

self.valid_types = (str, bytes)

def __eq__(self, test_string):

if not isinstance(test_string, self.valid_types):

return False

# this is safe even in py2 because as of 2.6, `bytes` exists in py2

# as an alias for `str`

if isinstance(test_string, bytes):

test_string = test_string.decode()

if len(self.substring) > len(test_string):

return False

return self.substring in test_string

def __ne__(self, test_string):

return not self.__eq__(test_string)

return StringContaining

def _safe_is_equal(x, y):

"""

Compares two values, preferring to use the first's __eq__ method if it

exists and is implemented.

Accounts for py2/py3 differences (like ints in py2 not having a __eq__

method), as well as the incomparability of certain types exposed by using

raw __eq__ () rather than ==.

"""

# Prefer using __eq__ directly to ensure that examples like

#

# maisey = Dog()

# maisey.name = "Maisey the Dog"

# maisey == ObjectDescribedBy(attrs={"name": StringContaining("Maisey")})

#

# evaluate to True (in other words, examples where the values in self.attrs

# might also have custom __eq__ methods; this makes sure those methods get

# used if possible)

try:

is_equal = x.__eq__(y)

except AttributeError:

is_equal = NotImplemented

# this can happen on its own, too (i.e. without an AttributeError being

# thrown), which is why this is separate from the except block above

if is_equal == NotImplemented:

# using == smoothes out weird variations exposed by raw __eq__

return x == y

return is_equal

@pytest.fixture(name="DictionaryContaining")

def dictionary_containing_matcher():

"""

An object which matches any dictionary containing all key-value pairs from

the dictionary passed to the object at instantiation time.

Useful for assert_called_with, assert_any_call, etc.

Used like this:

>>> f = mock.Mock()

>>> f({"dogs": "yes", "cats": "maybe"})

>>> f.assert_any_call({"dogs": "yes"}) # will raise AssertionError

Traceback (most recent call last):

...

AssertionError: mock({'dogs': 'yes'}) call not found

>>> f.assert_any_call(DictionaryContaining({"dogs": "yes"})) # no AssertionError

"""

class DictionaryContaining:

def __init__(self, subdict):

self.subdict = subdict

def __eq__(self, test_dict):

if not isinstance(test_dict, dict):

return False

if len(self.subdict) > len(test_dict):

return False

for key, value in self.subdict.items():

try:

test_value = test_dict[key]

except KeyError: # missing key

return False

if not _safe_is_equal(value, test_value):

return False

return True

def __ne__(self, test_dict):

return not self.__eq__(test_dict)

return DictionaryContaining

@pytest.fixture(name="ObjectDescribedBy")

def object_described_by_matcher():

"""

An object which matches any other object with the given properties.

Available properties currently are "type" (a type object) and "attrs" (a

dictionary).

Useful for assert_called_with, assert_any_call, etc.

Used like this:

>>> class Dog:

... pass

...

>>> maisey = Dog()

>>> maisey.name = "Maisey"

>>> maisey.age = 7

>>> f = mock.Mock()

>>> f(maisey)

>>> f.assert_any_call(ObjectDescribedBy(type=Dog)) # no AssertionError

>>> f.assert_any_call(ObjectDescribedBy(attrs={"name": "Maisey"})) # no AssertionError

"""

class ObjectDescribedBy:

def __init__(self, type=None, attrs=None):

self.type = type

self.attrs = attrs

def __eq__(self, test_obj):

if self.type:

if not isinstance(test_obj, self.type):

return False

if self.attrs:

for attr_name, attr_value in self.attrs.items():

try:

test_value = getattr(test_obj, attr_name)

except AttributeError: # missing attribute

return False

if not _safe_is_equal(attr_value, test_value):

return False

return True

def __ne__(self, test_obj):

return not self.__eq__(test_obj)

return ObjectDescribedBy

@pytest.fixture

def teardown_profiling():

# Make sure that a previous test didn't leave the profiler running

teardown_profiler()

teardown_continuous_profiler()

yield

# Make sure that to shut down the profiler after the test

teardown_profiler()

teardown_continuous_profiler()

@pytest.fixture()

def suppress_deprecation_warnings():

"""

Use this fixture to suppress deprecation warnings in a test.

Useful for testing deprecated SDK features.

"""

with warnings.catch_warnings():

warnings.simplefilter("ignore", DeprecationWarning)

yield

def _make_session_message(jsonrpc_msg):

"""Construct a SessionMessage compatible with both MCP v1 and v2."""

if _IS_MCP_V2:

return SessionMessage(message=jsonrpc_msg) # type: ignore

else:

return SessionMessage(message=JSONRPCMessage(root=jsonrpc_msg)) # type: ignore

@pytest.fixture

def get_initialization_payload():

def inner(request_id: str):

return _make_session_message(

JSONRPCRequest( # type: ignore

jsonrpc="2.0",

id=request_id,

method="initialize",

params={

"protocolVersion": "2025-11-25",

"capabilities": {},

"clientInfo": {"name": "test-client", "version": "1.0.0"},

},

)

)

return inner

@pytest.fixture

def get_initialized_notification_payload():

def inner():

return _make_session_message(

JSONRPCNotification( # type: ignore

jsonrpc="2.0",

method="notifications/initialized",

)

)

return inner

@pytest.fixture

def get_mcp_command_payload():

def inner(method: str, params, request_id: str):

return _make_session_message(

JSONRPCRequest( # type: ignore

jsonrpc="2.0",

id=request_id,

method=method,

params=params,

)

)

return inner

@pytest.fixture

def stdio(

get_initialization_payload,

get_initialized_notification_payload,

get_mcp_command_payload,

):

async def inner(server, method: str, params, request_id: str | None = None):

if request_id is None:

request_id = "1"

read_stream_writer, read_stream = create_memory_object_stream(0) # type: ignore

write_stream, write_stream_reader = create_memory_object_stream(0) # type: ignore

result = {}

async def run_server():

await server.run(

read_stream, write_stream, server.create_initialization_options()

)

async def simulate_client(tg, result):

init_request = get_initialization_payload("1")

await read_stream_writer.send(init_request)

await write_stream_reader.receive()

initialized_notification = get_initialized_notification_payload()

await read_stream_writer.send(initialized_notification)

request = get_mcp_command_payload(

method, params=params, request_id=request_id

)

await read_stream_writer.send(request)

result["response"] = await write_stream_reader.receive()

tg.cancel_scope.cancel()

async with create_task_group() as tg: # type: ignore

tg.start_soon(run_server)

tg.start_soon(simulate_client, tg, result)

return result["response"]

return inner

@pytest.fixture()

def json_rpc():

def inner(app, method: str, params, request_id: str):

with TestClient(app) as client: # type: ignore

init_response = client.post(

"/mcp/",

headers={

"Accept": "application/json, text/event-stream",

"Content-Type": "application/json",

},

json={

"jsonrpc": "2.0",

"method": "initialize",

"params": {

"clientInfo": {"name": "test-client", "version": "1.0"},

"protocolVersion": "2025-11-25",

"capabilities": {},

},

"id": request_id,

},

)

session_id = init_response.headers["mcp-session-id"]

# Notification response is mandatory.

# https://modelcontextprotocol.io/specification/2025-11-25/basic/lifecycle

client.post(

"/mcp/",

headers={

"Accept": "application/json, text/event-stream",

"Content-Type": "application/json",

"mcp-session-id": session_id,

},

json={

"jsonrpc": "2.0",

"method": "notifications/initialized",

"params": {},

},