gendesign/backend/tests/_sentry_wiring_probe.py

"""Проба проводки GlitchTip: запускается ОТДЕЛЬНЫМ процессом из test_sentry_init.py.

Зачем подпроцесс. `app/main.py` и `app/workers/celery_app.py` зовут
`sentry_sdk.init()` на импорте модуля и только при непустом `GLITCHTIP_DSN`. В
процессе pytest этот путь недостижим (модуль уже в `sys.modules`, DSN пуст), а
если бы и был достижим — глобальный клиент SDK остался бы живым для всех
последующих тестов. Отдельный процесс даёт настоящую инициализацию и умирает
вместе с ней.

Наружу не уходит ничего: `capture_envelope` подменяется ДО первого события, а
DSN в тесте указывает на несуществующий хост. Значения-маркеры генерируются
случайно на каждый запуск — кадр стека несёт не только переменные, но и строки
исходника, поэтому литерал в коде пробы сделал бы проверку вечно красной.

stdout — одна строка JSON: counts / scrub_handler_entries / markers / payloads
(тело каждого канала отдельно — см. `main`).
"""

from __future__ import annotations

import importlib
import io
import itertools
import json
import sys
import uuid
from typing import Any

import sentry_sdk

_FAILURE_CAP = 3


def _fresh(prefix: str) -> str:
    return f"{prefix}-{uuid.uuid4().hex}"


def _leaking_event(markers: dict[str, str]) -> dict[str, Any]:
    """Событие с ПДн в трёх местах, которые закрывает scrub_event."""
    return {
        "message": "sentry-wiring-probe",
        "level": "error",
        "request": {
            "data": {"phone": markers["phone"], "message": markers["free_text"]},
            "url": f"https://example.invalid/probe?api_key={markers['url_secret']}",
        },
    }


def main(module: str) -> int:
    importlib.import_module(module)  # ← здесь отрабатывает sentry_sdk.init()

    client = sentry_sdk.get_client()
    sent: list[str] = []

    def _record(envelope: Any) -> None:
        buf = io.BytesIO()
        envelope.serialize_into(buf)
        sent.append(buf.getvalue().decode("utf-8", "replace"))

    client.transport.capture_envelope = _record  # type: ignore[union-attr,method-assign]

    markers = {
        "phone": _fresh("probe-phone"),
        "free_text": _fresh("probe-free-text"),
        "url_secret": _fresh("probe-url-secret"),
        "local_var": _fresh("probe-local-var"),
    }

    # 1. Канал ошибок (before_send).
    sentry_sdk.capture_event(_leaking_event(markers))
    after_error = len(sent)

    # 2. Канал транзакций (before_send_transaction) — Starlette кладёт
    #    request.data на transaction-scope так же, как на error-scope.
    transaction = _leaking_event(markers)
    transaction["type"] = "transaction"
    transaction["transaction"] = "sentry-wiring-probe-tx"
    transaction["contexts"] = {"trace": {"trace_id": "0" * 32, "span_id": "0" * 16}}
    transaction["start_timestamp"] = "2026-01-01T00:00:00.000000Z"
    transaction["timestamp"] = "2026-01-01T00:00:01.000000Z"
    transaction["spans"] = []
    sentry_sdk.capture_event(transaction)
    after_transaction = len(sent)

    # 3. Локальные переменные кадра стека (include_local_variables). Имя
    #    переменной произвольное — ключевой скраб такое не ловит по построению.
    def _raise_with_local() -> None:
        applicant_note = markers["local_var"]  # noqa: F841 — ради кадра стека
        raise RuntimeError("sentry-wiring-probe boom")

    try:
        _raise_with_local()
    except RuntimeError:
        sentry_sdk.capture_exception()
    after_exception = len(sent)

    # 4. Сбой самого скраба не должен порождать ВТОРОЕ событие: иначе строка
    #    журнала об отказе уходит в мониторинг через LoggingIntegration, снова
    #    попадает в скраб, снова падает — рекурсия (#2753; на коде до фикса
    #    проверено: не завершается, 1000+ вложенных трассировок за минуту).
    #    Считаем ВХОДЫ в обработчик; после _FAILURE_CAP перестаём падать, иначе
    #    проба на сломанном коде висела бы вместо того, чтобы честно покраснеть.
    from app.observability import sentry_scrub

    original = sentry_scrub.scrub_pii_event
    entries: list[int] = []

    def _boom(*_a: Any, **_kw: Any) -> Any:
        entries.append(1)
        if len(entries) > _FAILURE_CAP:
            return None
        raise RuntimeError("sentry-wiring-probe scrubber failure")

    sentry_scrub.scrub_pii_event = _boom  # type: ignore[assignment]
    try:
        # Без маркеров: это событие по замыслу уходит НЕОЧИЩЕННЫМ ("as-is").
        sentry_sdk.capture_event({"message": "sentry-wiring-probe-failure", "level": "error"})
    finally:
        sentry_scrub.scrub_pii_event = original  # type: ignore[assignment]
    after_scrub_failure = len(sent)

    # Тело каждого канала — отдельно: иначе утечка из одного (напр. локальные
    # переменные шага 3 несут те же маркеры, что тело запроса шага 1) красит
    # чужую проверку и мешает понять, что именно сломано.
    bounds = [0, after_error, after_transaction, after_exception, after_scrub_failure]
    names = ["error", "transaction", "exception", "scrub_failure"]
    spans = dict(zip(names, itertools.pairwise(bounds), strict=True))

    print(
        json.dumps(
            {
                "counts": {name: end - start for name, (start, end) in spans.items()},
                "scrub_handler_entries": len(entries),
                "markers": markers,
                "payloads": {
                    name: "\n".join(sent[start:end]) for name, (start, end) in spans.items()
                },
            }
        )
    )
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1]))