gendesign/backend/tests/test_sentry_init.py

"""Unit-тесты логики инициализации GlitchTip SDK.

Проверяем что init-блок в main.py / celery_app.py вызывает sentry_sdk.init()
только при непустом GLITCHTIP_DSN, что release-fallback работает корректно,
что scrub_sensitive_query redact-ит api keys из URL spans, что scrub_pii_event
redact-ит consumer-PII (client_name/client_phone/client_email/phone/email/name/
company/message) из request.data/extra/contexts, и что composed-хендлер
scrub_event реально повешен на ОБА канала (before_send И
before_send_transaction) в main.py/celery_app.py (#2457-review).
"""

import os
import pathlib
from unittest.mock import patch

import sentry_sdk

_BACKEND_ROOT = pathlib.Path(__file__).resolve().parents[1]


def test_sdk_imports_without_error() -> None:
    """Все интеграции импортируются без ModuleNotFoundError."""
    from sentry_sdk.integrations.celery import CeleryIntegration
    from sentry_sdk.integrations.fastapi import FastApiIntegration
    from sentry_sdk.integrations.httpx import HttpxIntegration
    from sentry_sdk.integrations.logging import LoggingIntegration
    from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration
    from sentry_sdk.integrations.starlette import StarletteIntegration

    assert StarletteIntegration()
    assert FastApiIntegration()
    assert CeleryIntegration()
    assert SqlalchemyIntegration()
    assert HttpxIntegration()
    assert LoggingIntegration()


def test_no_sdk_init_when_dsn_empty() -> None:
    """Если GLITCHTIP_DSN пустой, sentry_sdk.init() не должен вызываться."""
    with patch("sentry_sdk.init") as mock_init:
        glitchtip_dsn = None
        if glitchtip_dsn:
            sentry_sdk.init(dsn=glitchtip_dsn)
        mock_init.assert_not_called()


def test_sdk_init_called_when_dsn_set() -> None:
    """Если GLITCHTIP_DSN задан, sentry_sdk.init() вызывается с правильными параметрами."""
    dsn = "https://key@errors.gendsgn.ru/1"
    with patch("sentry_sdk.init") as mock_init:
        glitchtip_dsn = dsn
        if glitchtip_dsn:
            sentry_sdk.init(
                dsn=glitchtip_dsn,
                environment="test",
                release="unknown",
                traces_sample_rate=0.05,
                profiles_sample_rate=0.0,
                send_default_pii=False,
                integrations=[],
            )
        mock_init.assert_called_once()
        call_kwargs = mock_init.call_args.kwargs
        assert call_kwargs["dsn"] == dsn
        assert call_kwargs["send_default_pii"] is False
        assert call_kwargs["profiles_sample_rate"] == 0.0


# ── release fallback ──────────────────────────────────────────────────────────


def test_release_uses_git_sha_when_set() -> None:
    """GIT_SHA имеет приоритет над SENTRY_RELEASE."""
    env = {"GIT_SHA": "abc1234", "SENTRY_RELEASE": "v1.0.0"}
    with patch.dict(os.environ, env, clear=False):
        release = os.getenv("GIT_SHA") or os.getenv("SENTRY_RELEASE") or "unknown"
    assert release == "abc1234"


def test_release_falls_back_to_sentry_release() -> None:
    """Если GIT_SHA не задан, используется SENTRY_RELEASE."""
    env = {"SENTRY_RELEASE": "v1.2.3"}
    with patch.dict(os.environ, env, clear=False):
        # Убираем GIT_SHA если он есть
        os.environ.pop("GIT_SHA", None)
        release = os.getenv("GIT_SHA") or os.getenv("SENTRY_RELEASE") or "unknown"
    assert release == "v1.2.3"


def test_release_falls_back_to_unknown() -> None:
    """Если ни одна переменная не задана, release='unknown'."""
    with patch.dict(os.environ, {}, clear=False):
        os.environ.pop("GIT_SHA", None)
        os.environ.pop("SENTRY_RELEASE", None)
        release = os.getenv("GIT_SHA") or os.getenv("SENTRY_RELEASE") or "unknown"
    assert release == "unknown"


# ── sentry_scrub ──────────────────────────────────────────────────────────────


def test_scrub_redacts_apikey_in_span_url() -> None:
    """scrub_sensitive_query заменяет apiKey= в span data['url']."""
    from app.observability.sentry_scrub import scrub_sensitive_query

    event: dict = {
        "spans": [
            {
                "data": {
                    "url": "https://api.objctv.ru/v2/Report?apiKey=supersecret&group=EKB",
                    "http.url": "https://api.objctv.ru/v2/Report?api_key=topsecret",
                }
            }
        ]
    }
    result = scrub_sensitive_query(event, {})
    span_data = result["spans"][0]["data"]
    assert "[REDACTED]" in span_data["url"]
    assert "supersecret" not in span_data["url"]
    assert "[REDACTED]" in span_data["http.url"]
    assert "topsecret" not in span_data["http.url"]


def test_scrub_redacts_token_in_description() -> None:
    """scrub_sensitive_query заменяет token= в span description."""
    from app.observability.sentry_scrub import scrub_sensitive_query

    event: dict = {
        "spans": [{"description": "GET https://example.com?token=mysecrettoken&foo=bar"}]
    }
    result = scrub_sensitive_query(event, {})
    assert "[REDACTED]" in result["spans"][0]["description"]
    assert "mysecrettoken" not in result["spans"][0]["description"]


def test_scrub_redacts_request_url() -> None:
    """scrub_sensitive_query заменяет token в event['request']['url']."""
    from app.observability.sentry_scrub import scrub_sensitive_query

    event: dict = {"request": {"url": "https://example.com?access_token=abc123&other=val"}}
    result = scrub_sensitive_query(event, {})
    assert "[REDACTED]" in result["request"]["url"]
    assert "abc123" not in result["request"]["url"]


def test_scrub_passes_through_clean_event() -> None:
    """scrub_sensitive_query не трогает URL без чувствительных параметров."""
    from app.observability.sentry_scrub import scrub_sensitive_query

    event: dict = {
        "spans": [{"data": {"url": "https://example.com?foo=bar&page=1"}}],
        "request": {"url": "https://example.com/api/health"},
    }
    result = scrub_sensitive_query(event, {})
    assert result["spans"][0]["data"]["url"] == "https://example.com?foo=bar&page=1"
    assert result["request"]["url"] == "https://example.com/api/health"


def test_scrub_handles_missing_spans() -> None:
    """scrub_sensitive_query не падает если 'spans' отсутствует."""
    from app.observability.sentry_scrub import scrub_sensitive_query

    event: dict = {"request": {"url": "https://example.com"}}
    result = scrub_sensitive_query(event, {})
    assert result["request"]["url"] == "https://example.com"


# ── scrub_pii_event (портировано из tradein-mvp, #396 / аудит-фикс) ────────────
#
# send_default_pii=False в sentry_sdk.init НЕ закрывает consumer-PII, попадающий
# в request.data / extra / contexts (лиды pilot.py, chat.py) — это user-data,
# а не PII-заголовки/cookies, которые режет сам флаг (проверено на sentry-sdk
# 2.58). scrub_pii_event закрывает этот вектор ключ-based редактором.


def test_scrub_pii_redacts_request_data() -> None:
    """scrub_pii_event заменяет client_name/phone/email в request.data."""
    from app.observability.sentry_scrub import scrub_pii_event

    event: dict = {
        "request": {
            "data": {
                "client_name": "Иван Иванов",
                "phone": "+79991234567",
                "email": "ivan@example.com",
                "address": "Екатеринбург, ул. Ленина 1",
            }
        }
    }
    result = scrub_pii_event(event, {})
    data = result["request"]["data"]
    assert data["client_name"] == "[REDACTED]"
    assert data["phone"] == "[REDACTED]"
    assert data["email"] == "[REDACTED]"
    # non-PII поле не трогаем
    assert data["address"] == "Екатеринбург, ул. Ленина 1"


def test_scrub_pii_redacts_pilot_request_company_and_message() -> None:
    """scrub_pii_event заменяет company/message — свободный текст
    PilotRequestInput (app/api/v1/pilot.py), куда чаще всего прилетают
    телефоны/имена/адреса, а не только фиксированные name/phone/email
    (#2457-review)."""
    from app.observability.sentry_scrub import scrub_pii_event

    event: dict = {
        "request": {
            "data": {
                "company": "ООО Ромашка",
                "message": "Меня зовут Иван, звоните на +79991234567",
                "source": "landing",
            }
        }
    }
    result = scrub_pii_event(event, {})
    data = result["request"]["data"]
    assert data["company"] == "[REDACTED]"
    assert data["message"] == "[REDACTED]"
    # non-PII поле не трогаем
    assert data["source"] == "landing"


def test_scrub_pii_redacts_client_prefixed_keys() -> None:
    """Полный набор ключей МЕРЫ (client_name/client_phone/client_email, #396) —
    Птица их сегодня не использует, но одинаковый механизм на оба продукта
    проще сопровождать (#2457-review)."""
    from app.observability.sentry_scrub import scrub_pii_event

    event: dict = {
        "extra": {
            "client_name": "Иван",
            "client_phone": "+79991234567",
            "client_email": "ivan@example.com",
        }
    }
    result = scrub_pii_event(event, {})
    extra = result["extra"]
    assert extra["client_name"] == "[REDACTED]"
    assert extra["client_phone"] == "[REDACTED]"
    assert extra["client_email"] == "[REDACTED]"


def test_scrub_pii_redacts_extra() -> None:
    """scrub_pii_event заменяет PII-ключи в extra, не трогая остальное."""
    from app.observability.sentry_scrub import scrub_pii_event

    event: dict = {
        "extra": {
            "phone": "+79990000000",
            "email": "x@y.ru",
            "name": "Пётр",
            "lead_id": 42,
        }
    }
    result = scrub_pii_event(event, {})
    extra = result["extra"]
    assert extra["phone"] == "[REDACTED]"
    assert extra["email"] == "[REDACTED]"
    assert extra["name"] == "[REDACTED]"
    assert extra["lead_id"] == 42


def test_scrub_pii_is_case_insensitive() -> None:
    """Сверка PII-ключей case-insensitive."""
    from app.observability.sentry_scrub import scrub_pii_event

    event: dict = {"extra": {"Client_Name": "Анна", "PHONE": "+7900"}}
    result = scrub_pii_event(event, {})
    assert result["extra"]["Client_Name"] == "[REDACTED]"
    assert result["extra"]["PHONE"] == "[REDACTED]"


def test_scrub_pii_redacts_nested_contexts() -> None:
    """scrub_pii_event проходит вложенные dict в contexts, не трогая non-PII."""
    from app.observability.sentry_scrub import scrub_pii_event

    event: dict = {"contexts": {"trace": {"op": "http"}, "lead": {"email": "z@z.ru"}}}
    result = scrub_pii_event(event, {})
    assert result["contexts"]["lead"]["email"] == "[REDACTED]"
    assert result["contexts"]["trace"]["op"] == "http"


def test_scrub_pii_leaves_non_pii_untouched() -> None:
    """scrub_pii_event не трогает event без PII-ключей."""
    from app.observability.sentry_scrub import scrub_pii_event

    event: dict = {
        "request": {"data": {"region": "66", "area_sqm": 50}},
        "extra": {"job": "geocode"},
        "level": "error",
    }
    result = scrub_pii_event(event, {})
    assert result["request"]["data"] == {"region": "66", "area_sqm": 50}
    assert result["extra"] == {"job": "geocode"}
    assert result["level"] == "error"


def test_scrub_pii_handles_missing_sections() -> None:
    """scrub_pii_event не падает на пустом event."""
    from app.observability.sentry_scrub import scrub_pii_event

    result = scrub_pii_event({}, {})
    assert result == {}


def test_scrub_pii_handles_none_and_non_dict_sections() -> None:
    """scrub_pii_event не бросает исключений на None/не-dict секциях."""
    from app.observability.sentry_scrub import scrub_pii_event

    event: dict = {"request": None, "extra": None, "contexts": "not-a-dict"}
    result = scrub_pii_event(event, {})
    assert result is event


def test_scrub_pii_returns_event_not_none() -> None:
    """before_send должен вернуть event (не None) — иначе SDK дропнет отчёт."""
    from app.observability.sentry_scrub import scrub_pii_event

    event: dict = {"request": {"data": {"client_name": "X"}}}
    result = scrub_pii_event(event, {})
    assert result is not None
    assert result is event


# ── scrub_event (composed before_send / before_send_transaction handler) ───────
#
# scrub_event — ЕДИНЫЙ хендлер, которым в main.py/celery_app.py вешаются ОБА
# канала (before_send И before_send_transaction). До #2457-review composed-хук
# висел только на before_send, а before_send_transaction оставался на голом
# scrub_sensitive_query (только URL) — Starlette-интеграция кладёт request.data
# на transaction-scope так же, как на error-scope, поэтому PII продолжало течь
# через transaction-канал при glitchtip_traces_sample_rate > 0.


def test_scrub_event_composes_pii_and_url_secret_scrub() -> None:
    """scrub_event применяет PII-scrub (ключ-based) И URL query-string secret
    redact (regex) оба разом, не заменяя друг друга — разные классы данных."""
    from app.observability.sentry_scrub import scrub_event

    event: dict = {
        "request": {
            "data": {"client_name": "Иван"},
            "url": "https://example.com?api_key=supersecret",
        }
    }
    result = scrub_event(event, {})
    assert result is not None
    assert result["request"]["data"]["client_name"] == "[REDACTED]"
    assert "[REDACTED]" in result["request"]["url"]
    assert "supersecret" not in result["request"]["url"]


def test_scrub_event_returns_event_not_none() -> None:
    """scrub_event всегда возвращает event (не None) — иначе SDK дропнет отчёт."""
    from app.observability.sentry_scrub import scrub_event

    event: dict = {"request": {"data": {"name": "X"}}}
    result = scrub_event(event, {})
    assert result is not None
    assert result is event


def test_scrub_event_survives_scrub_pii_event_exception() -> None:
    """try/except в scrub_event — предохранитель: sentry_sdk оборачивает
    before_send в capture_internal_exceptions, который при исключении ТОЛЬКО
    логирует и ДРОПАЕТ event целиком (SDK никогда не узнает, что редактор упал).
    Если scrub_pii_event падает — scrub_event обязан вернуть event, а не
    пробросить исключение дальше (#2457-review)."""
    from app.observability.sentry_scrub import scrub_event

    event: dict = {"request": {"data": {"client_name": "X"}}}
    with patch(
        "app.observability.sentry_scrub.scrub_pii_event",
        side_effect=RuntimeError("boom"),
    ):
        result = scrub_event(event, {})
    assert result is not None
    assert result is event


def test_scrub_event_survives_scrub_sensitive_query_exception() -> None:
    """То же самое для второго шага композиции (URL-secret redact)."""
    from app.observability.sentry_scrub import scrub_event

    event: dict = {"request": {"data": {"name": "X"}}}
    with patch(
        "app.observability.sentry_scrub.scrub_sensitive_query",
        side_effect=RuntimeError("boom"),
    ):
        result = scrub_event(event, {})
    assert result is not None
    assert result is event


# ── wiring: before_send/before_send_transaction реально используют scrub_event ──
#
# Source-grep вместо мока sentry_sdk.init: main.py/celery_app.py вызывают
# sentry_sdk.init() на module-level import, поэтому мок пришлось бы ставить ДО
# импорта app.main — фрагильно и не переиспользуемо между тестами (модуль уже
# закэширован в sys.modules к моменту первого теста). Прямая проверка исходника
# — детерминированный, дешёвый и точный регрессионный гейт на саму строку,
# которую правил review (#2457).


def test_main_wires_scrub_event_to_both_channels() -> None:
    """app/main.py: before_send И before_send_transaction ОБА на scrub_event."""
    text = (_BACKEND_ROOT / "app" / "main.py").read_text(encoding="utf-8")
    assert "before_send=scrub_event" in text
    assert "before_send_transaction=scrub_event" in text


def test_celery_app_wires_scrub_event_to_both_channels() -> None:
    """app/workers/celery_app.py: before_send И before_send_transaction ОБА на
    scrub_event (раньше before_send не было вообще)."""
    text = (_BACKEND_ROOT / "app" / "workers" / "celery_app.py").read_text(encoding="utf-8")
    assert "before_send=scrub_event" in text
    assert "before_send_transaction=scrub_event" in text