"""

scope._global_scope = None

scope._isolation_scope.set(None)

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

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

def inner(event):

assert jsonschema is not None

if SENTRY_EVENT_SCHEMA:

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

"""

global _DEFAULT_INTEGRATIONS, _processed_integrations

try:

_DEFAULT_INTEGRATIONS.remove(

"sentry_sdk.integrations.opentelemetry.integration.OpenTelemetryIntegration"

)

except ValueError:

pass

_processed_integrations.clear()

"""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)

def inner(*a, **kw):

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

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

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:

def __init__(self):

Transport.__init__(self)

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

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

"""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"""

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

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

"""

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

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

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)

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)

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):

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

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))

"""

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)

"""

# 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

"""

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)

"""

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)

# 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()

"""

with warnings.catch_warnings():

warnings.simplefilter("ignore", DeprecationWarning)

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

if _IS_MCP_V2:

return SessionMessage(message=jsonrpc_msg) # type: ignore

else:

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"},

},

)

)

def inner():

return _make_session_message(

JSONRPCNotification( # type: ignore

jsonrpc="2.0",

method="notifications/initialized",

)

)

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,

)

)

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"]

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": {},

},

)

response = client.post(

"/mcp/",

headers={

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

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

"mcp-session-id": session_id,

},

json={

"jsonrpc": "2.0",

"method": method,

"params": params,

"id": request_id,

},

)

return session_id, response

def inner(events):

return [

event

for event in events

if event["type"] == "transaction"

and event["contexts"]["trace"]["op"] == "mcp.server"

]

def inner(events, method_name):

return [

transaction

for transaction in events

if transaction["type"] == "transaction"

and any(

span["data"].get("mcp.method.name") == method_name

for span in transaction.get("spans", [])

)

]

class StreamingASGITransport(ASGITransport):

"""

def __init__(

self,

app: "Callable",

keep_sse_alive: "asyncio.Event",

) -> None:

self.keep_sse_alive = keep_sse_alive

super().__init__(app)

async def handle_async_request(

self, request: "HttpxRequest"