From 0b8c6db9e429e91178d2b01e4f339472b2c03855 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 24 Oct 2026 12:41:18 +0300 Subject: [PATCH] feat: add sentry_lean_error_capture for Litestar --- docs/integrations/litestar.md | 46 +++ docs/introduction/configuration.md | 1 + .../bootstrappers/litestar_bootstrapper.py | 177 +++++++++- lite_bootstrap/import_checker.py | 5 + .../instruments/sentry_instrument.py | 14 +- scripts/floor_smoke.py | 36 ++ tests/test_litestar_sentry_capture.py | 320 ++++++++++++++++++ 7 files changed, 592 insertions(+), 7 deletions(-) create mode 100644 tests/test_litestar_sentry_capture.py diff --git a/docs/integrations/litestar.md b/docs/integrations/litestar.md index dfe33d4..47f9c72 100644 --- a/docs/integrations/litestar.md +++ b/docs/integrations/litestar.md @@ -119,6 +119,52 @@ LitestarConfig( ) ``` +## Sentry + +By default, request errors reach Sentry through sentry-sdk's own `LitestarIntegration`. Even with +`sentry_traces_sample_rate` unset, it builds a transaction and a span per middleware for every request +and then discards them, and that is most of what Sentry costs per request +([getsentry/sentry-python#7400](https://github.com/getsentry/sentry-python/issues/7400)). + +`sentry_lean_error_capture=True` replaces it with a smaller error path: + +```python +LitestarConfig( + service_name="microservice", + sentry_dsn="https://testdsn@localhost/1", + sentry_lean_error_capture=True, +) +``` + +lite-bootstrap then passes `LitestarIntegration` in `disabled_integrations` and adds a middleware and an +`after_exception` hook of its own. The middleware gives each request its own Sentry scope and +release-health session, continues an incoming `sentry-trace` header, and attaches the method, URL, query +string and headers to every event sent during the request. Without `send_default_pii`, it masks the same +headers sentry-sdk masks. The hook reports an exception only when its response is a 5xx, so an exception +that your `exception_handlers` answer with a 404 is not reported. `LitestarIntegration` reports every +exception that is not an `HTTPException` below 500. Sentry's other default integrations stay on. + +You lose Sentry performance monitoring for requests, because no transaction is started, and the user +details `LitestarIntegration` adds when `send_default_pii` is on. + +Measured on [litestar-sqlalchemy-template](https://github.com/modern-python/litestar-sqlalchemy-template) +with one Granian worker on two CPUs and two read endpoints, against the same app without Sentry: + +| Sentry setup | added per request | +|---|---:| +| `LitestarIntegration` (default) | 320-345 µs | +| `sentry_lean_error_capture=True` | about 157 µs | +| the same with `sentry_default_integrations=False` | 32-83 µs | + +Most of what lean capture still costs comes from Sentry's other default integrations, such as the +SQLAlchemy and asyncpg breadcrumbs. `sentry_default_integrations=False` also drops the +`LoggingIntegration` lite-bootstrap would otherwise add, so logged errors stop becoming events. Supply +your own `LoggingIntegration` in `sentry_integrations` if you want them. + +sentry-sdk records installed integrations for the whole process, and `disabled_integrations` cannot +remove one that an earlier `sentry_sdk.init` installed. If that happened, lite-bootstrap warns that +errors may be reported twice. + ## Request body size limit `LitestarBootstrapper` builds its app with `Litestar.from_config()`, which, diff --git a/docs/introduction/configuration.md b/docs/introduction/configuration.md index edd8fba..74b49a5 100644 --- a/docs/introduction/configuration.md +++ b/docs/introduction/configuration.md @@ -22,6 +22,7 @@ Additional parameters can also be supplied through the settings object: - `sentry_additional_params` - additional params, which will be passed to `sentry_sdk.init`, overriding any of the above that map to the same keyword - `sentry_default_integrations` - whether to use sentry's default integrations (default: `True`) - `sentry_before_send` - optional callback chained after the built-in structlog enricher, passed to `sentry_sdk.init(before_send=...)` +- `sentry_lean_error_capture` - Litestar only: replace Sentry's `LitestarIntegration` with lite-bootstrap's own error capture, which starts no transaction per request (default: `False`). See [Litestar](../integrations/litestar.md#sentry). Read more about sentry_sdk params [here](https://docs.sentry.io/platforms/python/configuration/options/). diff --git a/lite_bootstrap/bootstrappers/litestar_bootstrapper.py b/lite_bootstrap/bootstrappers/litestar_bootstrapper.py index 623f031..dcf50e1 100644 --- a/lite_bootstrap/bootstrappers/litestar_bootstrapper.py +++ b/lite_bootstrap/bootstrappers/litestar_bootstrapper.py @@ -26,18 +26,25 @@ from lite_bootstrap.instruments.swagger_instrument import SwaggerConfig, SwaggerInstrument +if typing.TYPE_CHECKING: + from sentry_sdk import _types as sentry_types + if import_checker.is_litestar_installed: import litestar from litestar import status_codes from litestar.config.app import AppConfig from litestar.config.cors import CORSConfig + from litestar.enums import ScopeType + from litestar.exceptions import HTTPException from litestar.logging.config import StructLoggingConfig + from litestar.middleware import ASGIMiddleware from litestar.middleware.logging import LoggingMiddlewareConfig from litestar.openapi import OpenAPIConfig from litestar.openapi.plugins import SwaggerRenderPlugin from litestar.plugins.structlog import StructlogConfig, StructlogPlugin from litestar.static_files import create_static_files_router from litestar.types import Empty + from litestar.types.asgi_types import ASGIApp, Message, Receive, Scope, Send if import_checker.is_litestar_installed and import_checker.is_prometheus_client_installed: # litestar.plugins.prometheus imports prometheus_client, which the `litestar` @@ -47,12 +54,22 @@ from litestar.plugins.prometheus import PrometheusConfig, PrometheusController if import_checker.is_litestar_opentelemetry_installed: - from litestar.middleware import ASGIMiddleware from litestar.plugins.opentelemetry import OpenTelemetryConfig as LitestarOpenTelemetryConfig from litestar.plugins.opentelemetry import OpenTelemetryPlugin - from litestar.types.asgi_types import ASGIApp, Receive, Scope, Send from opentelemetry import trace +if import_checker.is_litestar_installed and import_checker.is_sentry_installed: + import sentry_sdk + from sentry_sdk import sessions + from sentry_sdk.integrations.logging import ignore_logger + from sentry_sdk.utils import event_from_exception, transaction_from_function + + # sentry-sdk 2.1, the declared floor, has only the older name for the same context manager. + track_sentry_session = getattr(sessions, "track_session", None) or sessions.auto_session_tracking_scope + +if import_checker.is_sentry_litestar_integration_installed: + from sentry_sdk.integrations.litestar import LitestarIntegration + if import_checker.is_opentelemetry_installed: from opentelemetry.metrics import get_meter_provider from opentelemetry.trace import get_tracer_provider @@ -110,6 +127,118 @@ async def handle( await next_app(scope, receive, send) +# sentry-sdk's own defaults: the status codes LitestarIntegration reports, and the headers it filters. +_SENTRY_FAILED_REQUEST_STATUS_CODES: typing.Final = range(500, 600) +_SENTRY_SENSITIVE_HEADERS: typing.Final = frozenset( + {"authorization", "cookie", "proxy-authorization", "set-cookie", "x-api-key", "x-forwarded-for", "x-real-ip"} +) +_SENTRY_FILTERED_VALUE: typing.Final = "[Filtered]" +_SENTRY_PENDING_EXCEPTION_KEY: typing.Final = "_lite_bootstrap_sentry_pending_exception" +_SENTRY_TRACE_HEADERS: typing.Final = frozenset({b"sentry-trace", b"baggage"}) + + +@dataclasses.dataclass(slots=True) +class _PendingSentryException: + exception: Exception | None = None + + +def _build_sentry_request_data(scope: "Scope") -> dict[str, typing.Any]: + headers: dict[str, str] = {name.decode("latin-1"): value.decode("latin-1") for name, value in scope["headers"]} + if not sentry_sdk.get_client().should_send_default_pii(): + headers = { + name: _SENTRY_FILTERED_VALUE if name in _SENTRY_SENSITIVE_HEADERS else value + for name, value in headers.items() + } + host: typing.Final = headers.get("host") + request_data: dict[str, typing.Any] = { + "method": scope.get("method"), + "url": f"{scope.get('scheme', 'http')}://{host}{scope['path']}" if host else scope["path"], + "headers": headers, + } + if query_string := scope.get("query_string", b"").decode("latin-1"): + request_data["query_string"] = query_string + return request_data + + +def enrich_sentry_event_from_litestar_scope(event: "sentry_types.Event", scope: "Scope") -> "sentry_types.Event": + event["request"] = {**event.get("request", {}), **_build_sentry_request_data(scope)} + route_handler: typing.Final = scope.get("route_handler") + if route_handler is not None and "transaction" not in event: + handler_function: typing.Final = getattr(route_handler.fn, "value", route_handler.fn) + event["transaction"] = route_handler.name or transaction_from_function(handler_function) or scope["path"] + event["transaction_info"] = {"source": "component"} + return event + + +def _capture_litestar_exception(exception: Exception, scope: "Scope") -> None: + event, hint = event_from_exception( + exception, + client_options=sentry_sdk.get_client().options, + mechanism={"type": "litestar", "handled": False}, + ) + sentry_sdk.capture_event(enrich_sentry_event_from_litestar_scope(event, scope), hint=hint) + + +def capture_litestar_exception_after_handling(exception: Exception, scope: "Scope") -> None: + """Report an exception once its response status is known, or now if no capture middleware will see it. + + Litestar calls ``after_exception`` hooks before the exception handler builds the response, so inside + ``LitestarSentryCaptureMiddleware`` the exception waits for the status the handler chose. Outside it + (routing errors, middleware placed before it) only an ``HTTPException`` carries a status to go by. + """ + pending: typing.Final = typing.cast("dict[str, typing.Any]", scope).get(_SENTRY_PENDING_EXCEPTION_KEY) + if pending is not None: + pending.exception = exception + return + if isinstance(exception, HTTPException) and exception.status_code not in _SENTRY_FAILED_REQUEST_STATUS_CODES: + return + _capture_litestar_exception(exception, scope) + + +if import_checker.is_litestar_installed and import_checker.is_sentry_installed: + + class LitestarSentryCaptureMiddleware(ASGIMiddleware): + scopes = (ScopeType.HTTP,) + + async def handle( + self, + scope: "Scope", + receive: "Receive", + send: "Send", + next_app: "ASGIApp", + ) -> None: + pending: typing.Final = _PendingSentryException() + mutable_scope: typing.Final = typing.cast("dict[str, typing.Any]", scope) + mutable_scope[_SENTRY_PENDING_EXCEPTION_KEY] = pending + + async def send_and_capture(message: "Message") -> None: + if message["type"] == "http.response.start": + exception, pending.exception = pending.exception, None + if exception is not None and message["status"] in _SENTRY_FAILED_REQUEST_STATUS_CODES: + _capture_litestar_exception(exception, scope) + await send(message) + + with ( + sentry_sdk.isolation_scope() as sentry_scope, + track_sentry_session(sentry_scope, session_mode="request"), + ): + sentry_scope.clear_breadcrumbs() + sentry_scope.generate_propagation_context( + { + name.decode("latin-1"): value.decode("latin-1") + for name, value in scope["headers"] + if name in _SENTRY_TRACE_HEADERS + } + ) + sentry_scope.add_event_processor(lambda event, _: enrich_sentry_event_from_litestar_scope(event, scope)) + try: + await next_app(scope, receive, send_and_capture) + finally: + del mutable_scope[_SENTRY_PENDING_EXCEPTION_KEY] + if pending.exception is not None: + _capture_litestar_exception(pending.exception, scope) + + @dataclasses.dataclass(kw_only=True, slots=True, frozen=True) class LitestarConfig( CorsConfig, @@ -128,17 +257,32 @@ class LitestarConfig( prometheus_additional_params: dict[str, typing.Any] = dataclasses.field(default_factory=dict) # Bounds path-label cardinality (Litestar defaults False -> raw URLs leak memory). See litestar#4891. prometheus_group_path: bool = True + sentry_lean_error_capture: bool = False swagger_extra_params: dict[str, typing.Any] = dataclasses.field(default_factory=dict) def __post_init__(self) -> None: # @dataclass(slots=True) replaces the class object, breaking bare super(). super(LitestarConfig, self).__post_init__() + if self.sentry_lean_error_capture: + self._warn_about_lean_sentry_capture() if self.litestar_logging_middleware_config is not None and not self.litestar_logging_middleware_enabled: warn_at_caller( "litestar_logging_middleware_config is ignored while litestar_logging_middleware_enabled is False; " "set litestar_logging_middleware_enabled=True to turn access logging on." ) + def _warn_about_lean_sentry_capture(self) -> None: + if self.sentry_traces_sample_rate is not None: + warn_at_caller( + "sentry_lean_error_capture starts no transaction per request, so sentry_traces_sample_rate " + "only samples transactions your code starts itself." + ) + if any(integration.identifier == "litestar" for integration in self.sentry_integrations): + warn_at_caller( + "sentry_lean_error_capture disables Sentry's LitestarIntegration, so the one in " + "sentry_integrations is dropped." + ) + @dataclasses.dataclass(kw_only=True, slots=True) class LitestarCorsInstrument(CorsInstrument[LitestarConfig]): @@ -277,6 +421,33 @@ class LitestarPrometheusController(PrometheusController): config.application_config.middleware.append(litestar_prometheus_config.middleware) +@dataclasses.dataclass(kw_only=True, slots=True) +class LitestarSentryInstrument(SentryInstrument[LitestarConfig]): + def _build_init_params(self) -> dict[str, typing.Any]: + init_params: typing.Final = super(LitestarSentryInstrument, self)._build_init_params() + if self.bootstrap_config.sentry_lean_error_capture and import_checker.is_sentry_litestar_integration_installed: + init_params["disabled_integrations"] = [*init_params.get("disabled_integrations", []), LitestarIntegration] + return init_params + + def bootstrap(self) -> None: + super(LitestarSentryInstrument, self).bootstrap() + if not self.bootstrap_config.sentry_lean_error_capture: + return + if ( + import_checker.is_sentry_litestar_integration_installed + and sentry_sdk.get_client().get_integration(LitestarIntegration) is not None + ): + warn_at_caller( + "Sentry's LitestarIntegration was installed by an earlier sentry_sdk.init in this process and " + "sentry_lean_error_capture cannot disable it, so request errors may be reported twice." + ) + # Litestar logs the exceptions it handles; LitestarIntegration ignores this logger for the same reason. + ignore_logger("litestar") + application_config: typing.Final = self.bootstrap_config.application_config + application_config.middleware.insert(0, LitestarSentryCaptureMiddleware()) + application_config.after_exception.append(capture_litestar_exception_after_handling) + + @dataclasses.dataclass(kw_only=True) class LitestarSwaggerInstrument(SwaggerInstrument[LitestarConfig]): not_configured_reason = "swagger_path is empty or not valid" @@ -320,7 +491,7 @@ class LitestarBootstrapper(BaseBootstrapper["litestar.Litestar"]): LitestarCorsInstrument, LitestarOpenTelemetryInstrument, PyroscopeInstrument, - SentryInstrument, + LitestarSentryInstrument, LitestarHealthChecksInstrument, LitestarLoggingInstrument, LitestarPrometheusInstrument, diff --git a/lite_bootstrap/import_checker.py b/lite_bootstrap/import_checker.py index fcf299e..5e0bdf8 100644 --- a/lite_bootstrap/import_checker.py +++ b/lite_bootstrap/import_checker.py @@ -36,6 +36,11 @@ def _safe_find_spec(module_name: str) -> bool: and _safe_find_spec("opentelemetry.instrumentation.asgi") and _safe_find_spec("litestar.plugins.opentelemetry") ) +# sentry-sdk added its Litestar integration in 2.13, after `disabled_integrations` (2.11), so this +# also tells whether the option that disables the integration exists. +is_sentry_litestar_integration_installed = ( + is_sentry_installed and is_litestar_installed and _safe_find_spec("sentry_sdk.integrations.litestar") +) is_otlp_grpc_exporter_installed = _safe_find_spec("opentelemetry.exporter.otlp.proto.grpc.trace_exporter") is_otlp_http_exporter_installed = _safe_find_spec("opentelemetry.exporter.otlp.proto.http.trace_exporter") is_pyroscope_installed = find_spec("pyroscope") is not None diff --git a/lite_bootstrap/instruments/sentry_instrument.py b/lite_bootstrap/instruments/sentry_instrument.py index 8f0e19e..90338cd 100644 --- a/lite_bootstrap/instruments/sentry_instrument.py +++ b/lite_bootstrap/instruments/sentry_instrument.py @@ -95,8 +95,11 @@ def run_before_send( return run_before_send +SentryConfigT = typing.TypeVar("SentryConfigT", bound=SentryConfig) + + @dataclasses.dataclass(kw_only=True, slots=True) -class SentryInstrument(BaseInstrument[SentryConfig]): +class SentryInstrument(BaseInstrument[SentryConfigT]): not_configured_reason = "sentry_dsn is empty" missing_dependency_message = "sentry_sdk is not installed" @@ -125,7 +128,7 @@ def _build_integrations(self) -> list["Integration"]: logging_integration_kwargs["sentry_logs_level"] = None return [*config.sentry_integrations, LoggingIntegration(**logging_integration_kwargs)] - def bootstrap(self) -> None: + def _build_init_params(self) -> dict[str, typing.Any]: config = self.bootstrap_config init_params: dict[str, typing.Any] = { "dsn": config.sentry_dsn, @@ -143,8 +146,11 @@ def bootstrap(self) -> None: ), } init_params.update(config.sentry_additional_params) - sentry_sdk.init(**init_params) - tags: dict[str, str] = config.sentry_tags or {} + return init_params + + def bootstrap(self) -> None: + sentry_sdk.init(**self._build_init_params()) + tags: dict[str, str] = self.bootstrap_config.sentry_tags or {} sentry_sdk.set_tags(tags) def teardown(self) -> None: diff --git a/scripts/floor_smoke.py b/scripts/floor_smoke.py index e40a519..d5825bf 100644 --- a/scripts/floor_smoke.py +++ b/scripts/floor_smoke.py @@ -41,6 +41,11 @@ if import_checker.is_opentelemetry_installed: from opentelemetry import trace +if import_checker.is_litestar_installed: + import litestar + from litestar.config.app import AppConfig + from litestar.testing import TestClient + if import_checker.is_sentry_installed: import sentry_sdk @@ -142,7 +147,38 @@ def _fastapi() -> None: asyncio.run(_drive_lifespan(application)) +def _litestar_lean_sentry_capture() -> None: + """Send one failing request through sentry_lean_error_capture, whose scope and session calls have floors too. + + Runs before the default bootstrap: that one installs Sentry's LitestarIntegration process-wide, and the + lean mode can no longer disable it afterwards. + """ + + @litestar.get("/floor-error") + async def floor_error() -> None: + message = "floor smoke" + raise RuntimeError(message) + + bootstrapper = LitestarBootstrapper( + bootstrap_config=LitestarConfig( + service_name="floor-smoke", + service_version="1.0.0", + service_debug=False, + logging_buffer_capacity=0, + sentry_dsn=SENTRY_DSN, + sentry_additional_params=SENTRY_PARAMS, + sentry_lean_error_capture=True, + application_config=AppConfig(route_handlers=[floor_error]), + ) + ) + with TestClient(app=bootstrapper.bootstrap(), raise_server_exceptions=False) as test_client: + expected_status = 500 + assert test_client.get("/floor-error").status_code == expected_status + + def _litestar() -> None: + if import_checker.is_sentry_installed: + _litestar_lean_sentry_capture() bootstrapper = LitestarBootstrapper( bootstrap_config=LitestarConfig( service_name="floor-smoke", diff --git a/tests/test_litestar_sentry_capture.py b/tests/test_litestar_sentry_capture.py new file mode 100644 index 0000000..8b25939 --- /dev/null +++ b/tests/test_litestar_sentry_capture.py @@ -0,0 +1,320 @@ +import logging +import typing +import warnings +from collections.abc import Iterator, Mapping +from unittest.mock import AsyncMock, patch + +import litestar +import pytest +import sentry_sdk +from litestar import status_codes +from litestar.config.app import AppConfig +from litestar.exceptions import InternalServerException, NotFoundException +from litestar.testing import TestClient +from sentry_sdk import integrations as sentry_integrations +from sentry_sdk.integrations.httpx import HttpxIntegration +from sentry_sdk.integrations.litestar import LitestarIntegration +from sentry_sdk.utils import transaction_from_function + +from lite_bootstrap import LitestarBootstrapper, LitestarConfig +from lite_bootstrap.bootstrappers.litestar_bootstrapper import LitestarSentryInstrument +from tests.conftest import SentryTestTransport + + +std_logger = logging.getLogger(__name__) + + +@pytest.fixture(autouse=True) +def _litestar_integration_not_yet_installed() -> Iterator[None]: + """Start each test as a fresh process would, with Sentry's LitestarIntegration not installed. + + sentry-sdk records installed integrations process-wide, and ``disabled_integrations`` cannot remove one + that an earlier ``sentry_sdk.init`` installed. Other test modules bootstrap Litestar with it on. + """ + identifier: typing.Final = LitestarIntegration.identifier + registries: typing.Final = ( + sentry_integrations._processed_integrations, # noqa: SLF001 + sentry_integrations._installed_integrations, # noqa: SLF001 + ) + was_in: typing.Final = [identifier in registry for registry in registries] + registries[1].discard(identifier) + yield + for registry, was_there in zip(registries, was_in, strict=True): + (registry.add if was_there else registry.discard)(identifier) + + +class MappedToNotFoundError(Exception): + pass + + +class MappedToUnavailableError(Exception): + pass + + +class BrokenHandlerError(Exception): + pass + + +@litestar.get("/boom") +async def boom() -> None: + message = "boom" + raise RuntimeError(message) + + +@litestar.get("/mapped-404") +async def mapped_404() -> None: + raise MappedToNotFoundError + + +@litestar.get("/mapped-503") +async def mapped_503() -> None: + raise MappedToUnavailableError + + +@litestar.get("/http-404") +async def http_404() -> None: + raise NotFoundException + + +@litestar.get("/http-500") +async def http_500() -> None: + raise InternalServerException + + +@litestar.get("/broken-handler") +async def broken_handler() -> None: + raise BrokenHandlerError + + +@litestar.get("/logs-then-ok") +async def logs_then_ok() -> str: + std_logger.info("breadcrumb from an earlier request") + return "ok" + + +@litestar.get("/integrations") +async def integrations() -> list[str]: + return sorted(sentry_sdk.get_client().integrations) + + +def _respond_404(_: litestar.Request, __: Exception) -> litestar.Response[None]: + return litestar.Response(None, status_code=status_codes.HTTP_404_NOT_FOUND) + + +def _respond_503(_: litestar.Request, __: Exception) -> litestar.Response[None]: + return litestar.Response(None, status_code=status_codes.HTTP_503_SERVICE_UNAVAILABLE) + + +def _fail_to_respond(_: litestar.Request, __: Exception) -> litestar.Response[None]: + message = "the exception handler failed" + raise RuntimeError(message) + + +def _build_config(transport: SentryTestTransport, **overrides: typing.Any) -> LitestarConfig: # noqa: ANN401 + sentry_additional_params = {"transport": transport, **overrides.pop("sentry_additional_params", {})} + return LitestarConfig( + service_environment="test", + sentry_dsn="https://testdsn@localhost/1", + sentry_additional_params=sentry_additional_params, + sentry_lean_error_capture=overrides.pop("sentry_lean_error_capture", True), + logging_buffer_capacity=0, + application_config=AppConfig( + route_handlers=[ + boom, + mapped_404, + mapped_503, + http_404, + http_500, + broken_handler, + logs_then_ok, + integrations, + ], + exception_handlers={ + MappedToNotFoundError: _respond_404, + MappedToUnavailableError: _respond_503, + BrokenHandlerError: _fail_to_respond, + }, + ), + **overrides, + ) + + +def _error_events(transport: SentryTestTransport) -> list[dict[str, typing.Any]]: + events = ( + typing.cast("dict[str, typing.Any] | None", envelope.get_event()) for envelope in transport.mock_envelopes + ) + return [event for event in events if event is not None and "exception" in event] + + +def _exception_types(transport: SentryTestTransport) -> list[str]: + return [event["exception"]["values"][-1]["type"] for event in _error_events(transport)] + + +def _request(transport: SentryTestTransport, *paths: str, **overrides: typing.Any) -> list[int]: # noqa: ANN401 + application = LitestarBootstrapper(bootstrap_config=_build_config(transport, **overrides)).bootstrap() + with TestClient(app=application, raise_server_exceptions=False) as test_client: + return [test_client.get(path).status_code for path in paths] + + +def test_lean_capture_reports_unhandled_exception_with_request_data(sentry_mock: SentryTestTransport) -> None: + application = LitestarBootstrapper(bootstrap_config=_build_config(sentry_mock)).bootstrap() + with TestClient(app=application, raise_server_exceptions=False) as test_client: + response = test_client.get("/boom?page=2", headers={"x-real-ip": "10.0.0.1", "user-agent": "tests"}) + + assert response.status_code == status_codes.HTTP_500_INTERNAL_SERVER_ERROR + [event] = _error_events(sentry_mock) + exception = event["exception"]["values"][-1] + assert exception["type"] == "RuntimeError" + assert exception["mechanism"] == {"type": "litestar", "handled": False} + assert event["request"]["method"] == "GET" + assert event["request"]["url"] == "http://testserver.local/boom" + assert event["request"]["query_string"] == "page=2" + assert event["request"]["headers"]["x-real-ip"] == "[Filtered]" + assert event["request"]["headers"]["user-agent"] == "tests" + assert event["transaction"] == transaction_from_function(boom.fn) + assert event["transaction_info"] == {"source": "component"} + + +async def test_lean_capture_continues_an_incoming_sentry_trace(sentry_mock: SentryTestTransport) -> None: + # Called without TestClient: sentry-sdk's httpx integration would overwrite the outgoing sentry-trace. + trace_id = "771a43a4192642f0b136d5159a501700" + application = LitestarBootstrapper(bootstrap_config=_build_config(sentry_mock)).bootstrap() + scope = { + "type": "http", + "asgi": {"version": "3.0"}, + "http_version": "1.1", + "method": "GET", + "scheme": "http", + "path": "/boom", + "raw_path": b"/boom", + "root_path": "", + "query_string": b"", + "headers": [(b"host", b"testserver"), (b"sentry-trace", f"{trace_id}-1234567890abcdef-1".encode())], + "client": ("127.0.0.1", 1), + "server": ("testserver", 80), + "state": {}, + } + receive = AsyncMock(return_value={"type": "http.request", "body": b"", "more_body": False}) + + await application(scope, receive, AsyncMock()) # ty: ignore[invalid-argument-type] + + [event] = _error_events(sentry_mock) + assert event["contexts"]["trace"]["trace_id"] == trace_id + + +def test_lean_capture_sends_sensitive_headers_when_pii_is_allowed(sentry_mock: SentryTestTransport) -> None: + config = _build_config(sentry_mock, sentry_additional_params={"send_default_pii": True}) + application = LitestarBootstrapper(bootstrap_config=config).bootstrap() + with TestClient(app=application, raise_server_exceptions=False) as test_client: + test_client.get("/boom", headers={"x-real-ip": "10.0.0.1"}) + + [event] = _error_events(sentry_mock) + assert event["request"]["headers"]["x-real-ip"] == "10.0.0.1" + + +def test_lean_capture_skips_exception_mapped_to_4xx(sentry_mock: SentryTestTransport) -> None: + assert _request(sentry_mock, "/mapped-404") == [status_codes.HTTP_404_NOT_FOUND] + assert _error_events(sentry_mock) == [] + + +def test_lean_capture_reports_exception_mapped_to_5xx(sentry_mock: SentryTestTransport) -> None: + assert _request(sentry_mock, "/mapped-503") == [status_codes.HTTP_503_SERVICE_UNAVAILABLE] + assert _exception_types(sentry_mock) == ["MappedToUnavailableError"] + + +def test_lean_capture_skips_client_http_exceptions(sentry_mock: SentryTestTransport) -> None: + assert _request(sentry_mock, "/http-404", "/no-such-route") == [ + status_codes.HTTP_404_NOT_FOUND, + status_codes.HTTP_404_NOT_FOUND, + ] + assert _error_events(sentry_mock) == [] + + +def test_lean_capture_reports_server_http_exceptions(sentry_mock: SentryTestTransport) -> None: + assert _request(sentry_mock, "/http-500") == [status_codes.HTTP_500_INTERNAL_SERVER_ERROR] + assert _exception_types(sentry_mock) == ["InternalServerException"] + + +def test_lean_capture_reports_both_exceptions_when_the_handler_fails(sentry_mock: SentryTestTransport) -> None: + assert _request(sentry_mock, "/broken-handler") == [status_codes.HTTP_500_INTERNAL_SERVER_ERROR] + assert _exception_types(sentry_mock) == ["BrokenHandlerError", "RuntimeError"] + + +def test_lean_capture_keeps_breadcrumbs_to_their_own_request(sentry_mock: SentryTestTransport) -> None: + _request(sentry_mock, "/logs-then-ok", "/boom") + + [event] = _error_events(sentry_mock) + messages = [crumb.get("message") for crumb in event.get("breadcrumbs", {}).get("values", [])] + assert "breadcrumb from an earlier request" not in messages + + +def test_lean_capture_reports_once_when_litestar_logs_the_exception(sentry_mock: SentryTestTransport) -> None: + assert _request(sentry_mock, "/boom", service_debug=True) == [status_codes.HTTP_500_INTERNAL_SERVER_ERROR] + assert _exception_types(sentry_mock) == ["RuntimeError"] + + +def test_lean_capture_replaces_the_litestar_integration(sentry_mock: SentryTestTransport) -> None: + application = LitestarBootstrapper(bootstrap_config=_build_config(sentry_mock)).bootstrap() + with TestClient(app=application) as test_client: + active = test_client.get("/integrations").json() + + assert LitestarIntegration.identifier not in active + assert "logging" in active + + +def _sentry_init_kwargs(config: LitestarConfig) -> Mapping[str, typing.Any]: + instrument: typing.Final = LitestarSentryInstrument(bootstrap_config=config) + with patch.object(sentry_sdk, "init") as sentry_init: + instrument.bootstrap() + return sentry_init.call_args.kwargs + + +def test_default_capture_keeps_the_litestar_integration(sentry_mock: SentryTestTransport) -> None: + config = _build_config(sentry_mock, sentry_lean_error_capture=False) + + assert "disabled_integrations" not in _sentry_init_kwargs(config) + assert config.application_config.middleware == [] + assert config.application_config.after_exception == [] + + +def test_lean_capture_keeps_user_disabled_integrations(sentry_mock: SentryTestTransport) -> None: + httpx_integration = HttpxIntegration() + config = _build_config(sentry_mock, sentry_additional_params={"disabled_integrations": [httpx_integration]}) + + assert _sentry_init_kwargs(config)["disabled_integrations"] == [httpx_integration, LitestarIntegration] + + +def test_lean_capture_warns_when_an_earlier_init_installed_the_litestar_integration( + sentry_mock: SentryTestTransport, +) -> None: + sentry_integrations._installed_integrations.add(LitestarIntegration.identifier) # noqa: SLF001 + bootstrapper = LitestarBootstrapper(bootstrap_config=_build_config(sentry_mock)) + + with pytest.warns(UserWarning, match="installed by an earlier sentry_sdk.init"): + bootstrapper.bootstrap() + bootstrapper.teardown() + + +def test_lean_capture_warns_that_traces_sample_rate_gets_no_request_transactions( + sentry_mock: SentryTestTransport, +) -> None: + with pytest.warns(UserWarning, match="sentry_traces_sample_rate"): + _build_config(sentry_mock, sentry_traces_sample_rate=1.0) + + +def test_lean_capture_warns_that_a_supplied_litestar_integration_is_dropped( + sentry_mock: SentryTestTransport, +) -> None: + with pytest.warns(UserWarning, match="LitestarIntegration"): + _build_config(sentry_mock, sentry_integrations=[LitestarIntegration()]) + + +def test_default_capture_does_not_warn_about_lean_settings(sentry_mock: SentryTestTransport) -> None: + with warnings.catch_warnings(): + warnings.simplefilter("error") + _build_config( + sentry_mock, + sentry_lean_error_capture=False, + sentry_traces_sample_rate=1.0, + sentry_integrations=[LitestarIntegration()], + )