gendesign/backend/tests/ops/test_3078_alert_ack.py

"""Кнопка подтверждения инцидента: что реально уходит в Telegram (#3078).

ЗАЧЕМ СЕРВИС ВООБЩЕ. Alertmanager пишет в Telegram сам, но инлайн-клавиатуру его
интеграция не поддерживает. Без кнопки нет обратной связи «человек увидел и взял
в работу» — 27.08 продукты лежали 10 часов, и вопрос «а кто-нибудь это читает»
было не к кому адресовать.

ЧТО СТОРОЖАТ ТЕСТЫ. Сервис маленький, но в нём три места, где ошибка не видна
глазами и проявится только в аварию — то есть тогда, когда проверять уже поздно:

1. Кнопка не должна теряться, но и не должна ронять сообщение. Если прицепить
   клавиатуру не удалось, алерт обязан уйти БЕЗ неё: сообщение важнее кнопки.
2. Повторное нажатие не должно слать второй «принято» — ссылка живёт сутки, по
   ней кликнут дважды, и дубль в теме выглядит как второй человек.
3. Токен должен быть неугадываемым и одноразовым: эндпоинт публичный.

Тесты дёргают настоящие функции модуля, подменяя ровно один шов — вызов Bot API.
Сеть не трогается, а всё, что ушло бы в неё, записывается и проверяется.
"""

from __future__ import annotations

import importlib.util
import logging
import socket
import sys
import threading
from http.server import ThreadingHTTPServer
from pathlib import Path
from types import ModuleType

import pytest

REPO_ROOT = Path(__file__).resolve().parents[3]
APP = REPO_ROOT / "ops" / "metrics" / "alert-ack" / "app.py"

FIRING = {
    "status": "firing",
    "commonLabels": {"alertname": "HostAgentDown", "host": "apps", "severity": "critical"},
    "alerts": [
        {
            "labels": {"alertname": "HostAgentDown"},
            "annotations": {
                "summary": "Агент метрик не отвечает",
                "description": "15 минут тишины",
            },
        }
    ],
}
RESOLVED = {**FIRING, "status": "resolved"}


@pytest.fixture
def app(monkeypatch: pytest.MonkeyPatch) -> ModuleType:
    """Загрузить сервис с предсказуемым окружением и подменённым Bot API."""
    assert APP.is_file(), f"нет {APP} — сервис переехал, тест ослеп"
    monkeypatch.setenv("METRICS_TELEGRAM_BOT_TOKEN", "123:FAKE")
    monkeypatch.setenv("METRICS_TELEGRAM_CHAT_ID", "-1004443088679")
    monkeypatch.setenv("METRICS_TELEGRAM_TOPIC_ID", "158")
    monkeypatch.setenv("METRICS_TELEGRAM_ONCALL", "@leks361")
    monkeypatch.setenv("ALERT_ACK_PUBLIC_URL", "https://metrics.gendsgn.ru")

    spec = importlib.util.spec_from_file_location("alert_ack_under_test", APP)
    assert spec and spec.loader
    mod = importlib.util.module_from_spec(spec)
    sys.modules["alert_ack_under_test"] = mod
    spec.loader.exec_module(mod)

    sent: list[tuple[str, dict]] = []

    def fake_tg(method: str, payload: dict) -> dict:
        sent.append((method, payload))
        return {"ok": True, "result": {"message_id": 1000 + len(sent)}}

    mod._tg = fake_tg  # единственный шов, ради него тест и существует
    mod.sent = sent
    mod._PENDING.clear()
    return mod


def test_gorjaschiy_incident_uhodit_s_knopkoy(app: ModuleType) -> None:
    """У горящего инцидента есть кнопка со ссылкой и живой токен."""
    app._send_alert(FIRING)
    method, payload = app.sent[0]
    assert method == "sendMessage"
    assert "reply_markup" in payload, "кнопка не прицеплена"
    assert "/ack/" in payload["reply_markup"], "в кнопке нет ссылки подтверждения"
    assert len(app._PENDING) == 1, "токен не сохранён — нажатие будет некуда деть"


def test_soobschenie_adresovano_v_temu_i_zovyot_dezhurnogo(app: ModuleType) -> None:
    """Тема форума и упоминание — обе вещи, которые молча теряются."""
    app._send_alert(FIRING)
    _, payload = app.sent[0]
    assert payload.get("message_thread_id") == "158", "уйдёт в общую тему форума"
    assert "@leks361" in payload["text"], "дежурного не позвали"
    assert "КЛИЕНТЫ ЗАТРОНУТЫ" in payload["text"]


def test_vosstanovlenie_bez_knopki(app: ModuleType) -> None:
    """У «восстановлено» подтверждать нечего — кнопки быть не должно."""
    app._send_alert(RESOLVED)
    _, payload = app.sent[0]
    assert "reply_markup" not in payload
    assert not app._PENDING, "токен выдан там, где кнопки нет"


def test_esli_knopka_ne_prikrepilas_soobschenie_vsyo_ravno_uhodit(
    app: ModuleType, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Отказ Bot API на сообщении с клавиатурой → повтор без неё.

    Самый важный из тестов: алерт важнее кнопки. Молчание вместо сообщения —
    ровно тот отказ, ради устранения которого весь стек и заводится.
    """
    calls: list[dict] = []

    def flaky(method: str, payload: dict):
        calls.append(payload)
        # Первая попытка (с клавиатурой) не удалась, вторая — без неё.
        return None if "reply_markup" in payload else {"ok": True, "result": {"message_id": 7}}

    monkeypatch.setattr(app, "_tg", flaky)
    app._send_alert(FIRING)
    assert len(calls) == 2, "не было повтора без кнопки — алерт потерян"
    assert "reply_markup" not in calls[1]
    assert calls[1]["text"] == calls[0]["text"], "во второй попытке потерялся текст"


def test_nazhatie_otvechaet_v_temu_i_snimaet_knopku(app: ModuleType) -> None:
    """Нажатие: ответ в тему + снятие клавиатуры у исходного сообщения."""
    app._send_alert(FIRING)
    token = next(iter(app._PENDING))
    app.sent.clear()

    code, page = app.do_ack(token)
    assert code == 200
    assert "Принято в работу" in page

    methods = [m for m, _ in app.sent]
    assert methods == ["sendMessage", "editMessageReplyMarkup"], methods
    reply = app.sent[0][1]
    assert reply.get("message_thread_id") == "158", "подтверждение уйдёт не в ту тему"
    assert reply.get("reply_to_message_id"), "подтверждение не привязано к сообщению инцидента"
    assert "@leks361" in reply["text"], "не видно, кто именно принял"


def test_povtornoe_nazhatie_ne_shlyot_vtoroy_raz(app: ModuleType) -> None:
    """Идемпотентность: ссылка живёт сутки, по ней кликнут дважды."""
    app._send_alert(FIRING)
    token = next(iter(app._PENDING))
    app.do_ack(token)
    app.sent.clear()

    code, page = app.do_ack(token)
    assert code == 200
    assert "Уже подтверждено" in page
    assert app.sent == [], "второе нажатие отправило дубль в чат"


def test_neizvestnyy_token_nichego_ne_rasskazyvaet(app: ModuleType) -> None:
    """Публичный эндпоинт: перебор не должен получать подсказок и ничего не шлёт."""
    code, page = app.do_ack("нет-такого-токена")
    assert code == 404
    assert app.sent == [], "неизвестный токен что-то отправил в чат"
    assert "недействительна" in page


def test_token_dostatochno_dlinnyy(app: ModuleType) -> None:
    """Ссылка защищена только неугадываемостью — длина токена и есть защита."""
    app._send_alert(FIRING)
    token = next(iter(app._PENDING))
    assert len(token) >= 20, f"слишком короткий токен: {len(token)} символов"


def test_protuhshiy_token_ne_prinimaetsya(app: ModuleType, monkeypatch: pytest.MonkeyPatch) -> None:
    """По истечении срока ссылка мертва — иначе она копится вечно."""
    app._send_alert(FIRING)
    token = next(iter(app._PENDING))
    app._PENDING[token]["created"] -= app.TTL_SEC + 1
    app.sent.clear()

    code, _ = app.do_ack(token)
    assert code == 404
    assert app.sent == []


# ── Секрет из query не попадает в лог (#3576) ────────────────────────────────
# GlitchTip шлёт секрет резервного вебхука только в `?secret=`, а http.server
# печатает строку запроса целиком — и в строке доступа, и в тексте ошибки
# разбора. Значение ниже выдуманное: проверяется, что его нет ни в одной записи.

_LEAK = "leak-probe-3576-VALUE"


@pytest.mark.parametrize(
    ("request_line", "masked", "lines_with_mask"),
    [
        # Строка доступа (log_request), путь резервного вебхука GlitchTip.
        (f"POST /glitchtip?secret={_LEAK} HTTP/1.1", "/glitchtip?secret=***", 1),
        # Имя с префиксом и соседний параметр — остальная строка цела.
        (f"GET /ack/x?a=1&access_token={_LEAK}&b=2 HTTP/1.1", "?a=1&access_token=***&b=2", 1),
        # Ошибка разбора (log_error): stdlib кладёт строку запроса в текст ошибки,
        # затем та же строка идёт в строку доступа с кодом 400 — обе записи.
        (f"POST /glitchtip?secret={_LEAK} junk HTTP/1.1", "/glitchtip?secret=***", 2),
    ],
)
def test_sekret_iz_query_ne_popadaet_v_log(
    app: ModuleType,
    caplog: pytest.LogCaptureFixture,
    request_line: str,
    masked: str,
    lines_with_mask: int,
) -> None:
    caplog.set_level(logging.INFO, logger="alert-ack")
    srv = ThreadingHTTPServer(("127.0.0.1", 0), app.Handler)
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    try:
        with socket.create_connection(srv.server_address, timeout=5) as sock:
            sock.sendall(f"{request_line}\r\nHost: x\r\nContent-Length: 0\r\n\r\n".encode())
            sock.shutdown(socket.SHUT_WR)
            while sock.recv(4096):  # до закрытия: к этому моменту запись лога уже сделана
                pass
    finally:
        srv.shutdown()
        srv.server_close()

    messages = [r.getMessage() for r in caplog.records if r.name == "alert-ack"]
    assert not [m for m in messages if _LEAK in m], f"значение секрета в логе: {messages}"
    assert sum(masked in m for m in messages) == lines_with_mask, messages