kin/tests/test_telegram.py

"""
Tests for core/telegram.py — send_telegram_escalation — KIN-BIZ-001.

Covers:
  - Correct Telegram API call parameters (token, chat_id, task_id, agent_role, reason)
  - Graceful failure when Telegram API is unavailable (no exceptions raised)
  - telegram_sent flag written to DB after successful send (mark_telegram_sent)
"""

import json
import urllib.error
from unittest.mock import MagicMock, patch

import pytest

from core import models
from core.db import init_db
from core.telegram import send_telegram_escalation


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------

@pytest.fixture
def db_conn():
    """Fresh in-memory DB for each test."""
    conn = init_db(db_path=":memory:")
    yield conn
    conn.close()


@pytest.fixture
def tg_env(monkeypatch):
    """Inject Telegram credentials via env vars (bypass secrets file).

    Also stubs _load_kin_config so the secrets file doesn't override env vars.
    """
    monkeypatch.setenv("KIN_TG_BOT_TOKEN", "test-token-abc123")
    monkeypatch.setenv("KIN_TG_CHAT_ID", "99887766")
    monkeypatch.setattr("core.telegram._load_kin_config", lambda: {})


@pytest.fixture
def mock_urlopen_ok():
    """Mock urllib.request.urlopen to return HTTP 200."""
    mock_resp = MagicMock()
    mock_resp.status = 200
    mock_resp.__enter__ = lambda s: s
    mock_resp.__exit__ = MagicMock(return_value=False)
    with patch("urllib.request.urlopen", return_value=mock_resp) as m:
        yield m


# ---------------------------------------------------------------------------
# Unit tests: send_telegram_escalation — correct API call parameters
# ---------------------------------------------------------------------------

def test_send_telegram_escalation_url_contains_bot_token(tg_env, mock_urlopen_ok):
    """Запрос уходит на URL с правильным bot token."""
    send_telegram_escalation(
        task_id="KIN-001",
        project_name="Test Project",
        agent_role="backend_dev",
        reason="Cannot access DB",
        pipeline_step="2",
    )
    req = mock_urlopen_ok.call_args[0][0]
    assert "test-token-abc123" in req.full_url
    assert "sendMessage" in req.full_url


def test_send_telegram_escalation_sends_to_correct_chat_id(tg_env, mock_urlopen_ok):
    """В теле POST-запроса содержится правильный chat_id."""
    send_telegram_escalation(
        task_id="KIN-001",
        project_name="Test",
        agent_role="pm",
        reason="Blocked",
        pipeline_step="1",
    )
    req = mock_urlopen_ok.call_args[0][0]
    body = json.loads(req.data.decode())
    assert body["chat_id"] == "99887766"


def test_send_telegram_escalation_includes_task_id_in_message(tg_env, mock_urlopen_ok):
    """task_id присутствует в тексте сообщения."""
    send_telegram_escalation(
        task_id="KIN-TEST-007",
        project_name="My Project",
        agent_role="frontend_dev",
        reason="No API access",
        pipeline_step="3",
    )
    req = mock_urlopen_ok.call_args[0][0]
    body = json.loads(req.data.decode())
    assert "KIN-TEST-007" in body["text"]


def test_send_telegram_escalation_includes_agent_role_in_message(tg_env, mock_urlopen_ok):
    """agent_role присутствует в тексте сообщения."""
    send_telegram_escalation(
        task_id="KIN-001",
        project_name="Test",
        agent_role="sysadmin",
        reason="SSH timeout",
        pipeline_step="1",
    )
    req = mock_urlopen_ok.call_args[0][0]
    body = json.loads(req.data.decode())
    assert "sysadmin" in body["text"]


def test_send_telegram_escalation_includes_reason_in_message(tg_env, mock_urlopen_ok):
    """reason присутствует в тексте сообщения."""
    send_telegram_escalation(
        task_id="KIN-001",
        project_name="Test",
        agent_role="pm",
        reason="Access denied to external API",
        pipeline_step="2",
    )
    req = mock_urlopen_ok.call_args[0][0]
    body = json.loads(req.data.decode())
    assert "Access denied to external API" in body["text"]


def test_send_telegram_escalation_uses_post_method(tg_env, mock_urlopen_ok):
    """Запрос отправляется методом POST."""
    send_telegram_escalation(
        task_id="KIN-001",
        project_name="Test",
        agent_role="pm",
        reason="Reason",
        pipeline_step="1",
    )
    req = mock_urlopen_ok.call_args[0][0]
    assert req.method == "POST"


def test_send_telegram_escalation_returns_true_on_success(tg_env, mock_urlopen_ok):
    """Функция возвращает True при успешном ответе HTTP 200."""
    result = send_telegram_escalation(
        task_id="KIN-001",
        project_name="Test",
        agent_role="pm",
        reason="Reason",
        pipeline_step="1",
    )
    assert result is True


def test_send_telegram_escalation_includes_pipeline_step_in_message(tg_env, mock_urlopen_ok):
    """pipeline_step включён в текст сообщения."""
    send_telegram_escalation(
        task_id="KIN-001",
        project_name="Test",
        agent_role="debugger",
        reason="Reason",
        pipeline_step="5",
    )
    req = mock_urlopen_ok.call_args[0][0]
    body = json.loads(req.data.decode())
    assert "5" in body["text"]


# ---------------------------------------------------------------------------
# Graceful failure tests — Telegram API unavailable
# ---------------------------------------------------------------------------

def test_send_telegram_escalation_returns_false_on_url_error(tg_env):
    """Функция возвращает False (не бросает) при urllib.error.URLError."""
    with patch("urllib.request.urlopen", side_effect=urllib.error.URLError("Connection refused")):
        result = send_telegram_escalation(
            task_id="KIN-001",
            project_name="Test",
            agent_role="pm",
            reason="Reason",
            pipeline_step="1",
        )
    assert result is False


def test_send_telegram_escalation_returns_false_on_unexpected_exception(tg_env):
    """Функция возвращает False (не бросает) при неожиданной ошибке."""
    with patch("urllib.request.urlopen", side_effect=RuntimeError("Unexpected!")):
        result = send_telegram_escalation(
            task_id="KIN-001",
            project_name="Test",
            agent_role="pm",
            reason="Reason",
            pipeline_step="1",
        )
    assert result is False


def test_send_telegram_escalation_never_raises_exception(tg_env):
    """Функция никогда не бросает исключение — пайплайн не должен падать."""
    with patch("urllib.request.urlopen", side_effect=Exception("Anything at all")):
        try:
            result = send_telegram_escalation(
                task_id="KIN-001",
                project_name="Test",
                agent_role="pm",
                reason="Reason",
                pipeline_step="1",
            )
        except Exception as exc:
            pytest.fail(f"send_telegram_escalation raised: {exc!r}")
    assert result is False


def test_send_telegram_escalation_returns_false_on_http_non_200(tg_env):
    """Функция возвращает False при HTTP ответе != 200."""
    mock_resp = MagicMock()
    mock_resp.status = 403
    mock_resp.__enter__ = lambda s: s
    mock_resp.__exit__ = MagicMock(return_value=False)
    with patch("urllib.request.urlopen", return_value=mock_resp):
        result = send_telegram_escalation(
            task_id="KIN-001",
            project_name="Test",
            agent_role="pm",
            reason="Reason",
            pipeline_step="1",
        )
    assert result is False


# ---------------------------------------------------------------------------
# Missing credentials tests
# ---------------------------------------------------------------------------

def test_send_telegram_escalation_returns_false_when_no_bot_token(monkeypatch):
    """Без bot token функция возвращает False, не падает."""
    monkeypatch.delenv("KIN_TG_BOT_TOKEN", raising=False)
    monkeypatch.setenv("KIN_TG_CHAT_ID", "12345")
    with patch("core.telegram._load_kin_config", return_value={}):
        result = send_telegram_escalation(
            task_id="KIN-001",
            project_name="Test",
            agent_role="pm",
            reason="Reason",
            pipeline_step="1",
        )
    assert result is False


def test_send_telegram_escalation_returns_false_when_no_chat_id(monkeypatch):
    """Без KIN_TG_CHAT_ID функция возвращает False, не падает."""
    monkeypatch.setenv("KIN_TG_BOT_TOKEN", "some-token")
    monkeypatch.delenv("KIN_TG_CHAT_ID", raising=False)
    with patch("core.telegram._load_kin_config", return_value={"tg_bot": "some-token"}):
        result = send_telegram_escalation(
            task_id="KIN-001",
            project_name="Test",
            agent_role="pm",
            reason="Reason",
            pipeline_step="1",
        )
    assert result is False


# ---------------------------------------------------------------------------
# DB tests: mark_telegram_sent
# ---------------------------------------------------------------------------

def test_mark_telegram_sent_sets_flag_in_db(db_conn):
    """mark_telegram_sent() устанавливает telegram_sent=1 в БД."""
    models.create_project(db_conn, "proj1", "Project 1", "/proj1")
    models.create_task(db_conn, "PROJ1-001", "proj1", "Task 1")

    task = models.get_task(db_conn, "PROJ1-001")
    assert not bool(task.get("telegram_sent"))

    models.mark_telegram_sent(db_conn, "PROJ1-001")

    task = models.get_task(db_conn, "PROJ1-001")
    assert bool(task["telegram_sent"]) is True


def test_mark_telegram_sent_does_not_affect_other_tasks(db_conn):
    """mark_telegram_sent() обновляет только указанную задачу."""
    models.create_project(db_conn, "proj1", "Project 1", "/proj1")
    models.create_task(db_conn, "PROJ1-001", "proj1", "Task 1")
    models.create_task(db_conn, "PROJ1-002", "proj1", "Task 2")

    models.mark_telegram_sent(db_conn, "PROJ1-001")

    task2 = models.get_task(db_conn, "PROJ1-002")
    assert not bool(task2.get("telegram_sent"))


def test_mark_telegram_sent_idempotent(db_conn):
    """Повторный вызов mark_telegram_sent() не вызывает ошибок."""
    models.create_project(db_conn, "proj1", "Project 1", "/proj1")
    models.create_task(db_conn, "PROJ1-001", "proj1", "Task 1")

    models.mark_telegram_sent(db_conn, "PROJ1-001")
    models.mark_telegram_sent(db_conn, "PROJ1-001")  # second call

    task = models.get_task(db_conn, "PROJ1-001")
    assert bool(task["telegram_sent"]) is True