diff --git a/core/phases.py b/core/phases.py index 3288ac7..26a3a91 100644 --- a/core/phases.py +++ b/core/phases.py @@ -59,7 +59,7 @@ def create_project_with_phases( conn: sqlite3.Connection, id: str, name: str, - path: str, + path: str | None = None, description: str, selected_roles: list[str], tech_stack: list | None = None, diff --git a/tests/test_telegram.py b/tests/test_telegram.py new file mode 100644 index 0000000..78f9526 --- /dev/null +++ b/tests/test_telegram.py @@ -0,0 +1,300 @@ +""" +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