""" 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