- release= now: GIT_SHA or SENTRY_RELEASE or "unknown" (covers existing deploy.yml which writes SENTRY_RELEASE=$IMAGE_TAG, not GIT_SHA) - new module app/observability/sentry_scrub.py: before_send_transaction hook that redacts apiKey/token/access_token/secret query params from HttpxIntegration spans before they reach GlitchTip - config.py: model_validator promotes legacy SENTRY_DSN -> glitchtip_dsn with DeprecationWarning for backward compat on existing VPS env - .env.example: added deprecated SENTRY_DSN entry with comment - tests: 8 new cases covering release fallback + scrub module (11 total)
158 lines
6.5 KiB
Python
158 lines
6.5 KiB
Python
"""Unit-тесты логики инициализации GlitchTip SDK.
|
||
|
||
Проверяем что init-блок в main.py / celery_app.py вызывает sentry_sdk.init()
|
||
только при непустом GLITCHTIP_DSN, что release-fallback работает корректно,
|
||
и что scrub_sensitive_query redact-ит api keys из URL spans.
|
||
"""
|
||
|
||
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"
|