/tests/watch_dog/test_alarms.py

"""Tests for integrations.telegram.credentials and integrations.telegram.alarms."""

from __future__ import annotations

from typing import Any

import pytest

from integrations.telegram.alarms import AlarmDispatcher
from integrations.telegram.credentials import (
    TelegramCredentials,
    load_credentials_from_env,
)
from platform.common.errors import OpenSREError


def _stub_telegram(
    monkeypatch: pytest.MonkeyPatch,
    *,
    ok: bool = True,
    error: str = "",
    captured: list[dict[str, Any]] | None = None,
) -> list[dict[str, Any]]:
    calls: list[dict[str, Any]] = captured if captured is not None else []

    def _fake_post(
        chat_id: str,
        text: str,
        bot_token: str,
        parse_mode: str = "",
    ) -> tuple[bool, str, str]:
        calls.append(
            {
                "chat_id": chat_id,
                "text": text,
                "bot_token": bot_token,
                "parse_mode": parse_mode,
            }
        )
        return ok, error, "1" if ok else ""

    monkeypatch.setattr(
        "integrations.telegram.alarms.post_telegram_message",
        _fake_post,
    )
    return calls


def _patch_clock(monkeypatch: pytest.MonkeyPatch, ticks: list[float]) -> None:
    iterator = iter(ticks)

    def _now() -> float:
        return next(iterator)

    monkeypatch.setattr(AlarmDispatcher, "_now", staticmethod(_now))


@pytest.fixture(autouse=True)
def _isolate_credential_sources(monkeypatch: pytest.MonkeyPatch) -> None:
    """Make credential resolution hermetic.

    By default the integration store is empty and the keyring is disabled, so
    credential tests resolve purely from environment variables unless they opt
    into the store (by patching ``resolve_effective_integrations``) or the
    keyring. Without this, the new store-first resolution would read the real
    ``~/.opensre`` store and make tests depend on local machine state.
    """
    monkeypatch.setattr(
        "integrations.catalog.resolve_effective_integrations",
        lambda: {},
    )
    monkeypatch.setenv("OPENSRE_DISABLE_KEYRING", "1")


def _patch_store(monkeypatch: pytest.MonkeyPatch, config: dict[str, Any]) -> None:
    """Point the store-backed Telegram config at *config*."""
    monkeypatch.setattr(
        "integrations.catalog.resolve_effective_integrations",
        lambda: {"telegram": {"source": "local store", "config": config}},
    )


def test_alarm_credentials_repr_does_not_leak_bot_token() -> None:
    # Auto-generated dataclass __repr__ surfaces in pytest assertion output,
    # tracebacks, and structured log capture. The token must stay out of it.
    creds = TelegramCredentials(bot_token="super-secret-token", chat_id="chat-1")

    rendered = repr(creds)

    assert "super-secret-token" not in rendered
    assert "chat-1" in rendered


def test_load_credentials_reads_env(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "tok-123")
    monkeypatch.setenv("TELEGRAM_DEFAULT_CHAT_ID", "chat-1")

    creds = load_credentials_from_env()

    assert creds == TelegramCredentials(bot_token="tok-123", chat_id="chat-1")


def test_load_credentials_strips_whitespace(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "  tok-123  ")
    monkeypatch.setenv("TELEGRAM_DEFAULT_CHAT_ID", "\tchat-1\n")

    creds = load_credentials_from_env()

    assert creds.bot_token == "tok-123"
    assert creds.chat_id == "chat-1"


def test_load_credentials_chat_id_override_beats_env(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "tok")
    monkeypatch.setenv("TELEGRAM_DEFAULT_CHAT_ID", "from-env")

    creds = load_credentials_from_env(chat_id_override="from-arg")

    assert creds.chat_id == "from-arg"


def test_load_credentials_missing_bot_token(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("TELEGRAM_BOT_TOKEN", raising=False)
    monkeypatch.setenv("TELEGRAM_DEFAULT_CHAT_ID", "chat-1")

    with pytest.raises(OpenSREError) as exc_info:
        load_credentials_from_env()

    assert "TELEGRAM_BOT_TOKEN" in str(exc_info.value)
    assert exc_info.value.suggestion is not None
    assert "TELEGRAM_BOT_TOKEN" in exc_info.value.suggestion


def test_load_credentials_blank_bot_token(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "   ")
    monkeypatch.setenv("TELEGRAM_DEFAULT_CHAT_ID", "chat-1")

    with pytest.raises(OpenSREError):
        load_credentials_from_env()


def test_load_credentials_missing_chat_id(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "tok")
    monkeypatch.delenv("TELEGRAM_DEFAULT_CHAT_ID", raising=False)

    with pytest.raises(OpenSREError) as exc_info:
        load_credentials_from_env()

    assert "chat id" in str(exc_info.value).lower()


def test_load_credentials_missing_chat_id_with_blank_override(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "tok")
    monkeypatch.setenv("TELEGRAM_DEFAULT_CHAT_ID", "chat-from-env")

    # Empty string override falls through to env.
    creds = load_credentials_from_env(chat_id_override="")

    assert creds.chat_id == "chat-from-env"


def test_load_credentials_whitespace_override_falls_back_to_env(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    # A whitespace-only override must fall through to the env var the same
    # way an empty-string override does, not raise a misleading error.
    monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "tok")
    monkeypatch.setenv("TELEGRAM_DEFAULT_CHAT_ID", "chat-from-env")

    creds = load_credentials_from_env(chat_id_override="   ")

    assert creds.chat_id == "chat-from-env"


def test_load_credentials_from_store_only(monkeypatch: pytest.MonkeyPatch) -> None:
    # Guided setup (`opensre integrations setup telegram` / `onboard`) saves the
    # token to the store, not the environment. The watchdog must find it there.
    monkeypatch.delenv("TELEGRAM_BOT_TOKEN", raising=False)
    monkeypatch.delenv("TELEGRAM_DEFAULT_CHAT_ID", raising=False)
    _patch_store(monkeypatch, {"bot_token": "store-tok", "default_chat_id": "store-chat"})

    creds = load_credentials_from_env()

    assert creds == TelegramCredentials(bot_token="store-tok", chat_id="store-chat")


def test_load_credentials_chat_id_from_store(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("TELEGRAM_DEFAULT_CHAT_ID", raising=False)
    _patch_store(monkeypatch, {"bot_token": "store-tok", "default_chat_id": "store-chat"})

    creds = load_credentials_from_env()

    assert creds.chat_id == "store-chat"


def test_load_credentials_store_bot_token_beats_env(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    # Store wins over env, matching the scheduler's precedence.
    monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "env-tok")
    monkeypatch.setenv("TELEGRAM_DEFAULT_CHAT_ID", "chat-1")
    _patch_store(monkeypatch, {"bot_token": "store-tok"})

    creds = load_credentials_from_env()

    assert creds.bot_token == "store-tok"


def test_load_credentials_store_chat_id_beats_env(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "tok")
    monkeypatch.setenv("TELEGRAM_DEFAULT_CHAT_ID", "env-chat")
    _patch_store(monkeypatch, {"bot_token": "tok", "default_chat_id": "store-chat"})

    creds = load_credentials_from_env()

    assert creds.chat_id == "store-chat"


def test_load_credentials_override_beats_store_chat_id(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    _patch_store(monkeypatch, {"bot_token": "tok", "default_chat_id": "store-chat"})

    creds = load_credentials_from_env(chat_id_override="arg-chat")

    assert creds.chat_id == "arg-chat"


def test_load_credentials_from_keyring(monkeypatch: pytest.MonkeyPatch) -> None:
    # No store, no env — the token lives only in the system keyring (as the
    # onboarding wizard writes it via sync_env_secret).
    monkeypatch.delenv("TELEGRAM_BOT_TOKEN", raising=False)
    monkeypatch.setenv("TELEGRAM_DEFAULT_CHAT_ID", "chat-1")
    monkeypatch.setattr(
        "config.llm_credentials.resolve_llm_api_key",
        lambda env_var: "keyring-tok" if env_var == "TELEGRAM_BOT_TOKEN" else "",
    )

    creds = load_credentials_from_env()

    assert creds.bot_token == "keyring-tok"


def test_load_credentials_store_failure_falls_back_to_env(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    # A broken/locked store must not crash the watchdog; resolution falls back
    # to the environment.
    def _boom(*args: Any, **kwargs: Any) -> dict[str, Any]:
        raise RuntimeError("store is locked")

    monkeypatch.setattr("integrations.catalog.resolve_effective_integrations", _boom)
    monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "env-tok")
    monkeypatch.setenv("TELEGRAM_DEFAULT_CHAT_ID", "env-chat")

    creds = load_credentials_from_env()

    assert creds == TelegramCredentials(bot_token="env-tok", chat_id="env-chat")


def test_first_dispatch_calls_telegram(monkeypatch: pytest.MonkeyPatch) -> None:
    calls = _stub_telegram(monkeypatch)
    _patch_clock(monkeypatch, [100.0])

    dispatcher = AlarmDispatcher(
        TelegramCredentials(bot_token="tok", chat_id="chat-1"),
    )

    assert dispatcher.dispatch("max_cpu", "CPU pegged at 95%") is True
    assert len(calls) == 1
    assert calls[0] == {
        "chat_id": "chat-1",
        "text": "CPU pegged at 95%",
        "bot_token": "tok",
        "parse_mode": "",
    }


def test_dispatch_can_use_html_parse_mode(monkeypatch: pytest.MonkeyPatch) -> None:
    calls = _stub_telegram(monkeypatch)
    _patch_clock(monkeypatch, [100.0])

    dispatcher = AlarmDispatcher(
        TelegramCredentials(bot_token="tok", chat_id="chat-1"),
        parse_mode="HTML",
    )

    assert dispatcher.dispatch("max_cpu", "CPU < 95% & rising") is True
    assert calls[0]["parse_mode"] == "HTML"


def test_second_dispatch_within_cooldown_is_suppressed(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    calls = _stub_telegram(monkeypatch)
    # 100s gap < 300s cooldown, second call must be suppressed.
    _patch_clock(monkeypatch, [100.0, 200.0])

    dispatcher = AlarmDispatcher(
        TelegramCredentials(bot_token="tok", chat_id="chat-1"),
        cooldown_seconds=300.0,
    )

    assert dispatcher.dispatch("max_cpu", "first") is True
    assert dispatcher.dispatch("max_cpu", "second") is False
    assert len(calls) == 1
    assert calls[0]["text"] == "first"


def test_second_dispatch_after_cooldown_calls_telegram_again(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    calls = _stub_telegram(monkeypatch)
    # 350s gap > 300s cooldown, second call must go through.
    _patch_clock(monkeypatch, [100.0, 450.0])

    dispatcher = AlarmDispatcher(
        TelegramCredentials(bot_token="tok", chat_id="chat-1"),
        cooldown_seconds=300.0,
    )

    assert dispatcher.dispatch("max_cpu", "first") is True
    assert dispatcher.dispatch("max_cpu", "second") is True
    assert len(calls) == 2
    assert calls[1]["text"] == "second"


def test_cooldown_is_per_threshold_name(monkeypatch: pytest.MonkeyPatch) -> None:
    calls = _stub_telegram(monkeypatch)
    _patch_clock(monkeypatch, [100.0, 110.0])

    dispatcher = AlarmDispatcher(
        TelegramCredentials(bot_token="tok", chat_id="chat-1"),
        cooldown_seconds=300.0,
    )

    assert dispatcher.dispatch("max_cpu", "cpu") is True
    assert dispatcher.dispatch("max_runtime", "runtime") is True
    assert len(calls) == 2


def test_dispatch_returns_false_on_transport_failure(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    # A failed Telegram call still arms cooldown; otherwise a bad token, bad
    # chat id, or network outage can retry on every watchdog sample.
    calls: list[dict[str, Any]] = []
    _stub_telegram(monkeypatch, ok=False, error="network down", captured=calls)
    _patch_clock(monkeypatch, [100.0, 105.0])

    dispatcher = AlarmDispatcher(
        TelegramCredentials(bot_token="tok", chat_id="chat-1"),
        cooldown_seconds=300.0,
    )

    assert dispatcher.dispatch("max_cpu", "first") is False
    assert dispatcher.dispatch("max_cpu", "second") is False
    assert len(calls) == 1
    assert calls[0]["text"] == "first"


def test_failed_dispatch_retries_after_cooldown(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    calls: list[dict[str, Any]] = []
    _stub_telegram(monkeypatch, ok=False, error="network down", captured=calls)
    _patch_clock(monkeypatch, [100.0, 105.0, 450.0])

    dispatcher = AlarmDispatcher(
        TelegramCredentials(bot_token="tok", chat_id="chat-1"),
        cooldown_seconds=300.0,
    )

    assert dispatcher.dispatch("max_cpu", "first") is False
    assert dispatcher.dispatch("max_cpu", "suppressed") is False
    assert dispatcher.dispatch("max_cpu", "retry") is False
    assert [call["text"] for call in calls] == ["first", "retry"]


def test_dispatch_uses_credentials_from_constructor(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    calls = _stub_telegram(monkeypatch)
    _patch_clock(monkeypatch, [100.0])

    dispatcher = AlarmDispatcher(
        TelegramCredentials(bot_token="bot-XYZ", chat_id="my-chat"),
    )
    dispatcher.dispatch("max_runtime", "process exceeded 5m")

    assert calls[0]["bot_token"] == "bot-XYZ"
    assert calls[0]["chat_id"] == "my-chat"


def test_dispatch_truncates_messages_over_telegram_limit(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    # Telegram rejects messages over 4096 chars. Without truncation an
    # oversized alarm would always fail, never arm cooldown, and retry forever.
    calls = _stub_telegram(monkeypatch)
    _patch_clock(monkeypatch, [100.0])

    dispatcher = AlarmDispatcher(
        TelegramCredentials(bot_token="tok", chat_id="chat-1"),
    )
    oversized = "X" * 5000
    assert dispatcher.dispatch("max_cpu", oversized) is True
    assert len(calls[0]["text"]) <= 4096
    assert calls[0]["text"].endswith("…")