304 lines
11 KiB
Python
304 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).
|
||
|
||
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
|