"""Тесты для резервного канала GlitchTip → alert-ack (#3471). Продукт (tradein-backend на Selectel) — САМ объект наблюдения GlitchTip. Если он лежит, основной вебхук-получатель лежит вместе с ним, и алерт об этом не доходит именно тогда, когда нужнее всего. `POST /glitchtip` — второй получатель на ДРУГОМ хосте (Beget, рядом с этим сервисом), не зависящий от здоровья продукта. Проверяем на уровне функций, а не полного HTTP-транспорта (тот же приём, что `ops/glitchtip-auth-forwarder/test_forwarder.py`): `BaseHTTPRequestHandler` неудобно поднимать без реального сокета, а бизнес-логика — секрет, рендер, отправка — целиком вынесена в чистые функции модуля. """ from __future__ import annotations import os import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).parent)) # Env — ДО импорта app.py: BOT_TOKEN/CHAT_ID читаются на уровне модуля. os.environ.setdefault("METRICS_TELEGRAM_BOT_TOKEN", "test-bot-token") os.environ.setdefault("METRICS_TELEGRAM_CHAT_ID", "-1001234567890") os.environ.setdefault("METRICS_TELEGRAM_TOPIC_ID", "158") os.environ["ALERT_ACK_GLITCHTIP_SECRET"] = "correct-secret" import app as alert_ack # noqa: E402 def test_verify_glitchtip_secret_accepts_correct_value() -> None: assert alert_ack._verify_glitchtip_secret("correct-secret") is True def test_verify_glitchtip_secret_rejects_wrong_value() -> None: """Неверный секрет — отказ.""" assert alert_ack._verify_glitchtip_secret("wrong-secret") is False def test_verify_glitchtip_secret_rejects_empty_value() -> None: assert alert_ack._verify_glitchtip_secret("") is False def test_verify_glitchtip_secret_fails_closed_when_unconfigured(monkeypatch) -> None: """Пустой ALERT_ACK_GLITCHTIP_SECRET — отказ всем, а не тихий fail-open.""" monkeypatch.setattr(alert_ack, "GLITCHTIP_SECRET", "") assert alert_ack._verify_glitchtip_secret("correct-secret") is False assert alert_ack._verify_glitchtip_secret("") is False def test_render_glitchtip_full_payload_marks_fallback_channel() -> None: payload = { "text": "GlitchTip Alert: Something broke", "attachments": [ { "title": "TypeError: boom", "title_link": "https://errors.gendsgn.ru/issue/1", "text": "подробности ошибки", "color": "#ff0000", "fields": [{"title": "Project", "value": "tradein-backend"}], } ], } text = alert_ack._render_glitchtip(payload) assert "РЕЗЕРВНЫЙ КАНАЛ" in text assert "Проект: tradein-backend" in text assert "TypeError: boom" in text assert "https://errors.gendsgn.ru/issue/1" in text def test_render_glitchtip_survives_payload_without_attachments() -> None: """Тело без attachments не роняет сервис — только текст.""" text = alert_ack._render_glitchtip({"text": "просто текст без вложений"}) assert "РЕЗЕРВНЫЙ КАНАЛ" in text assert "просто текст без вложений" in text def test_render_glitchtip_survives_empty_payload() -> None: """Пустой словарь (нет ни text, ни attachments) — тоже не должен падать.""" text = alert_ack._render_glitchtip({}) assert "РЕЗЕРВНЫЙ КАНАЛ" in text assert "GlitchTip alert" in text def test_render_glitchtip_survives_malformed_attachments() -> None: """attachments/fields неожиданной формы (не список, не словарь, битые типы) — резервный канал не должен падать на кривом теле.""" payload = { "text": "test", "attachments": [ "not-a-dict", {"fields": "not-a-list"}, {"fields": [{"title": "Project"}]}, # value отсутствует None, ], } text = alert_ack._render_glitchtip(payload) assert "РЕЗЕРВНЫЙ КАНАЛ" in text def test_send_glitchtip_alert_forwards_via_tg(monkeypatch) -> None: """Верный секрет уже проверен вызывающей стороной (do_POST) — здесь проверяем, что отрендеренное сообщение реально уходит в тот же chat/topic, что и Alertmanager-алерты этого сервиса, без кнопки подтверждения.""" calls = [] monkeypatch.setattr(alert_ack, "_tg", lambda method, payload: calls.append((method, payload))) alert_ack._send_glitchtip_alert({"text": "boom", "attachments": []}) assert len(calls) == 1 method, sent = calls[0] assert method == "sendMessage" assert sent["chat_id"] == alert_ack.CHAT_ID assert sent["message_thread_id"] == alert_ack.TOPIC_ID assert "reply_markup" not in sent assert "boom" in sent["text"] # ── Keep-alive: отказ не должен ронять СЛЕДУЮЩИЙ запрос ────────────────────── # Эти четыре теста — единственные, что поднимают настоящий сокет. Дефект, # который они стерегут, живёт именно в транспорте и на уровне функций невидим: # ветка отказа отвечала, не вычитав тело запроса, а `protocol_version` здесь # HTTP/1.1, то есть соединение переиспользуется (и Caddy перед сервисом держит # пул к апстриму). Непрочитанное тело оставалось в сокете, и следующий запрос # по тому же соединению начинался с чужих байт. # # Поймано на проде 12.09.2026: зонд без секрета получил 401, а следующий — # уже с верным секретом — вернул 501 Unsupported method ('{"text":"probe"}POST'). # То есть один отказ съедал следующий НАСТОЯЩИЙ алерт. def _serve_in_background(monkeypatch): """Поднимает Handler на эфемерном порту, глушит отправку в Telegram.""" import threading as _threading from http.server import ThreadingHTTPServer sent: list[dict] = [] monkeypatch.setattr(alert_ack, "_send_glitchtip_alert", sent.append) monkeypatch.setattr(alert_ack, "_send_alert", sent.append) srv = ThreadingHTTPServer(("127.0.0.1", 0), alert_ack.Handler) thread = _threading.Thread(target=srv.serve_forever, daemon=True) thread.start() return srv, sent def _raw_post(sock, path: str, body: bytes, headers: str = "") -> str: """Шлёт POST по уже открытому сокету и возвращает статусную строку.""" req = ( f"POST {path} HTTP/1.1\r\n" f"Host: localhost\r\n" f"Content-Type: application/json\r\n" f"Content-Length: {len(body)}\r\n" f"{headers}" f"\r\n" ).encode() + body sock.sendall(req) # Читаем ровно заголовки: тела короткие, Content-Length всегда проставлен. buf = b"" while b"\r\n\r\n" not in buf: chunk = sock.recv(4096) if not chunk: break buf += chunk head, _, rest = buf.partition(b"\r\n\r\n") length = 0 for line in head.split(b"\r\n")[1:]: if line.lower().startswith(b"content-length:"): length = int(line.split(b":")[1]) while len(rest) < length: rest += sock.recv(4096) return head.split(b"\r\n")[0].decode() def _pair_on_one_connection(monkeypatch, first_headers: str, first_path: str = "/glitchtip"): import socket srv, sent = _serve_in_background(monkeypatch) try: sock = socket.create_connection(srv.server_address, timeout=5) try: first = _raw_post(sock, first_path, b'{"text":"probe"}', first_headers) second = _raw_post( sock, "/glitchtip", b'{"text":"real alert"}', "X-GlitchTip-Secret: correct-secret\r\n", ) finally: sock.close() finally: srv.shutdown() srv.server_close() return first, second, sent def test_rejected_request_does_not_break_next_one_on_same_connection(monkeypatch) -> None: """401 без секрета, следом законный алерт по ТОМУ ЖЕ соединению — 200.""" first, second, sent = _pair_on_one_connection(monkeypatch, "") assert "401" in first assert "200" in second, f"второй запрос испорчен первым: {second}" assert sent == [{"text": "real alert"}] def test_unconfigured_secret_does_not_break_next_request(monkeypatch) -> None: """503 при пустом секрете тоже обязан вычитать тело.""" import socket monkeypatch.setattr(alert_ack, "GLITCHTIP_SECRET", "") srv, _sent = _serve_in_background(monkeypatch) try: sock = socket.create_connection(srv.server_address, timeout=5) try: first = _raw_post(sock, "/glitchtip", b'{"text":"probe"}') second = _raw_post(sock, "/glitchtip", b'{"text":"again"}') finally: sock.close() finally: srv.shutdown() srv.server_close() assert "503" in first assert "503" in second, f"второй запрос испорчен первым: {second}" def test_unknown_path_does_not_break_next_request(monkeypatch) -> None: """404 на чужом пути — та же ветка раннего ответа, то же требование.""" first, second, sent = _pair_on_one_connection(monkeypatch, "", first_path="/nope") assert "404" in first assert "200" in second, f"второй запрос испорчен первым: {second}" assert sent == [{"text": "real alert"}] def test_bad_json_on_alertmanager_does_not_break_next_request(monkeypatch) -> None: """400 на неразобранном теле /alertmanager — тело уже вычитано, связь цела.""" import socket srv, sent = _serve_in_background(monkeypatch) try: sock = socket.create_connection(srv.server_address, timeout=5) try: first = _raw_post(sock, "/alertmanager", b"{not json") second = _raw_post( sock, "/glitchtip", b'{"text":"real alert"}', "X-GlitchTip-Secret: correct-secret\r\n", ) finally: sock.close() finally: srv.shutdown() srv.server_close() assert "400" in first assert "200" in second, f"второй запрос испорчен первым: {second}" assert sent == [{"text": "real alert"}]