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/phone/email/name) из
request.data/extra/contexts перед отправкой в GlitchTip.
"""

import os
from unittest.mock import patch

import sentry_sdk


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


def test_scrub_pii_composes_with_url_secret_scrub() -> None:
    """Композиция, реально используемая в app.main._before_send /
    app.workers.celery_app._before_send: PII-scrub (ключ-based) и URL
    query-string secret redact (regex) применяются оба, не заменяя друг друга —
    разные классы данных."""
    from app.observability.sentry_scrub import scrub_pii_event, scrub_sensitive_query

    event: dict = {
        "request": {
            "data": {"client_name": "Иван"},
            "url": "https://example.com?api_key=supersecret",
        }
    }

    def composed_before_send(evt: dict, hint: dict) -> dict | None:
        scrubbed = scrub_pii_event(evt, hint)
        if scrubbed is None:
            return None
        return scrub_sensitive_query(scrubbed, hint)

    result = composed_before_send(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"]