301 lines
11 KiB
Python
301 lines
11 KiB
Python
|
"""
|
|||
|
|
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)."""
|
|||
|
|
monkeypatch.setenv("KIN_TG_BOT_TOKEN", "test-token-abc123")
|
|||
|
|
monkeypatch.setenv("KIN_TG_CHAT_ID", "99887766")
|
|||
|
|
|
|||
|
|
|
|||
|
|
@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
|