gendesign/backend/tests/ops/test_2203_notify_topic.py

"""Оповещения уходят в тему «алерты», а не в общую (#2203/#3078).

Прод-факт, 27.08. Канал доставки включили, `notify()` заработал — и алерты
посыпались в ОБЩУЮ тему форума (в чат переговорки), а не в «алерты». Причина: в
форуме Telegram адрес сообщения — пара «чат + тема». Без `message_thread_id`
сообщение попадает в General, и ошибки при этом нет: `sendMessage` возвращает
200, доставка «успешна», просто не туда.

Отказ ровно того класса, ради которого весь стек и заводится: всё зелёное, а
человек, который должен прочитать алерт, его не видит.

ПОЧЕМУ ТЕСТ ИСПОЛНЯЕТ, А НЕ ЧИТАЕТ. Проверять наличие подстроки
`message_thread_id` в файле бессмысленно: она может стоять в мёртвой ветке, быть
закомментированной или потеряться при подстановке. Тест подсовывает НАСТОЯЩЕЙ
функции подставной `curl`, который записывает свои аргументы, и смотрит, что
реально ушло бы в сеть.

Функция извлекается из файла построчно, а не копируется в тест: копия разошлась
бы с оригиналом на первой же правке. Сорсить файл целиком нельзя — у
`uptime-healthcheck.sh` нет guard'а по `BASH_SOURCE`, и сорсинг запустил бы
настоящие сетевые проверки.
"""

from __future__ import annotations

import functools
import os
import re
import subprocess
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parents[3]
SENDERS = {
    "lib-backup.sh": REPO_ROOT / "ops" / "lib-backup.sh",
    "uptime-healthcheck.sh": REPO_ROOT / "ops" / "uptime-healthcheck.sh",
}

FAKE_CURL = '#!/usr/bin/env bash\nprintf "%s\\n" "$@" >> "$ARGS_DUMP"\nexit 0\n'


@functools.lru_cache(maxsize=1)
def _drive_root() -> str:
    """Куда ЭТОТ bash монтирует диски Windows: `/mnt/c` (WSL) или `/c` (git-bash)."""
    r = subprocess.run(
        ["bash", "-c", "[ -d /mnt/c ] && echo /mnt || echo ''"],
        capture_output=True,
        text=True,
        timeout=30,
    )
    return r.stdout.strip()


def _posix(path: Path) -> str:
    """`C:/Users/x` → `/mnt/c/Users/x` или `/c/Users/x`, смотря какой bash.

    Путь с буквой диска bash не понимает вовсе, и подставной curl тогда просто не
    находится: тест падает с «curl не был вызван», хотя код исправен. Префикс
    зависит от оболочки — WSL и git-bash монтируют диски по-разному, — поэтому
    определяется запуском, а не угадывается. На Linux (CI) пути уже POSIX и
    функция ничего не меняет.
    """
    text = path.as_posix()
    m = re.match(r"^([A-Za-z]):/(.*)$", text)
    return f"{_drive_root()}/{m.group(1).lower()}/{m.group(2)}" if m else text


def _extract_notify(script: Path) -> str:
    """Вырезать тело notify() из настоящего файла — от заголовка до `}` в нулевой колонке."""
    lines = script.read_text(encoding="utf-8").splitlines()
    start = next((i for i, l in enumerate(lines) if l.startswith("notify()")), None)
    assert start is not None, f"{script.name}: не нашёл notify() — отправитель переименован"
    end = next((i for i in range(start + 1, len(lines)) if lines[i] == "}"), None)
    assert end is not None, f"{script.name}: не нашёл конец notify()"
    return "\n".join(lines[start : end + 1])


def _run_notify(tmp_path: Path, script: Path, *, topic: str | None) -> list[str]:
    bin_dir = tmp_path / "bin"
    bin_dir.mkdir()
    fake = bin_dir / "curl"
    fake.write_text(FAKE_CURL, encoding="utf-8", newline="\n")
    fake.chmod(0o755)

    # Заглушки для того, что notify() зовёт помимо curl.
    harness = tmp_path / "harness.sh"
    harness.write_text(
        "log() { :; }\n"
        "notify_fallback_mail() { :; }\n"
        + _extract_notify(script)
        + '\nnotify "тестовое сообщение"\n',
        encoding="utf-8",
        newline="\n",
    )

    dump = tmp_path / "args.txt"
    # Переменные передаются ПРЕФИКСОМ КОМАНДЫ: под Windows git-bash окружение из
    # `env=` у subprocess до скрипта не доходит, и тест был бы зелёным по
    # неверной причине — просто потому, что curl не вызывался вовсе.
    # Значения В КАВЫЧКАХ: в PATH встречается `Program Files (x86)`, и
    # незакавыченное присваивание роняет разбор всей команды на скобке.
    prefix = [
        f'PATH="{_posix(bin_dir)}:$PATH"',
        f'ARGS_DUMP="{_posix(dump)}"',
        'TELEGRAM_BOT_TOKEN="123:FAKE"',
        'TELEGRAM_CHAT_ID="-1004443088679"',
        f"BACKUP_ENV_FILE=\"{_posix(tmp_path / 'missing.env')}\"",
    ]
    if topic is not None:
        prefix.append(f'TELEGRAM_TOPIC_ID="{topic}"')

    subprocess.run(
        ["bash", "-c", " ".join(prefix) + f' bash "{_posix(harness)}"'],
        cwd=str(REPO_ROOT),
        capture_output=True,
        text=True,
        timeout=30,
        env=dict(os.environ),
    )
    return dump.read_text(encoding="utf-8").splitlines() if dump.exists() else []


@pytest.mark.parametrize("name", sorted(SENDERS))
def test_tema_peredayotsya_kogda_zadana(tmp_path: Path, name: str) -> None:
    """При заданном TELEGRAM_TOPIC_ID сообщение адресуется в тему."""
    args = _run_notify(tmp_path, SENDERS[name], topic="158")
    assert args, f"{name}: curl не был вызван — отправка не дошла до сети"
    assert "message_thread_id=158" in args, (
        f"{name}: тема не передана, сообщение уйдёт в общую тему форума:\n{args}"
    )


@pytest.mark.parametrize("name", sorted(SENDERS))
def test_bez_temy_povedenie_prezhnee(tmp_path: Path, name: str) -> None:
    """Без переменной параметр не добавляется вовсе.

    Обратный конец инварианта: пустой `message_thread_id=` Telegram отвергает
    вместе со всем сообщением. Молчащий алерт хуже алерта не в той теме, поэтому
    отсутствие переменной обязано означать отсутствие параметра, а не параметр с
    пустым значением.
    """
    args = _run_notify(tmp_path, SENDERS[name], topic=None)
    assert args, f"{name}: curl не был вызван"
    assert not any("message_thread_id" in a for a in args), (
        f"{name}: параметр темы просочился без переменной:\n{args}"
    )


@pytest.mark.parametrize("name", sorted(SENDERS))
def test_adres_i_tekst_na_meste(tmp_path: Path, name: str) -> None:
    """Тема добавляется, а не вытесняет адресата и текст."""
    args = _run_notify(tmp_path, SENDERS[name], topic="158")
    assert "chat_id=-1004443088679" in args, f"{name}: потерялся chat_id"
    assert any("тестовое сообщение" in a for a in args), f"{name}: потерялся текст"