From 8ee9782737f0de54b9d8dd95f98d75a603b7e78c Mon Sep 17 00:00:00 2001 From: Gros Frumos Date: Sat, 21 Mar 2026 08:36:20 +0200 Subject: [PATCH] =?UTF-8?q?kin:=20BATON-007=20=D0=9F=D1=80=D0=B8=20=D0=BD?= =?UTF-8?q?=D0=B0=D0=B6=D0=B0=D1=82=D0=B8=D0=B8=20=D0=BD=D0=B0=20=D0=BA?= =?UTF-8?q?=D0=BD=D0=BE=D0=BF=D0=BA=D1=83=20=D0=BF=D1=80=D0=BE=D0=B8=D1=81?= =?UTF-8?q?=D1=85=D0=BE=D0=B4=D0=B8=D1=82=20=D0=B0=D0=BD=D0=B8=D0=BC=D0=B0?= =?UTF-8?q?=D1=86=D0=B8=D1=8F=20=D0=B8=20=D1=81=D0=BE=D0=BE=D0=B1=D1=89?= =?UTF-8?q?=D0=B5=D0=BD=D0=B8=D0=B5=20=D1=87=D1=82=D0=BE=20=D1=81=D0=B8?= =?UTF-8?q?=D0=B3=D0=BD=D0=B0=D0=BB=20=D0=BE=D1=82=D0=BF=D1=80=D0=B0=D0=B2?= =?UTF-8?q?=D0=BB=D0=B5=D0=BD,=20=D0=BD=D0=BE=20=D0=B2=20=D1=82=D0=B5?= =?UTF-8?q?=D0=BB=D0=B5=D0=B3=D1=80=D0=B0=D0=BC=20=D0=B3=D1=80=D1=83=D0=BF?= =?UTF-8?q?=D0=BF=D1=83=20=D0=BD=D0=B8=D1=87=D0=B5=D0=B3=D0=BE=20=D0=BD?= =?UTF-8?q?=D0=B5=20=D0=BF=D1=80=D0=B8=D1=85=D0=BE=D0=B4=D0=B8=D1=82.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_baton_007.py | 262 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 262 insertions(+) create mode 100644 tests/test_baton_007.py diff --git a/tests/test_baton_007.py b/tests/test_baton_007.py new file mode 100644 index 0000000..0030c7d --- /dev/null +++ b/tests/test_baton_007.py @@ -0,0 +1,262 @@ +""" +Tests for BATON-007: Verifying real Telegram delivery when a signal is sent. + +Acceptance criteria: +1. After pressing the button, a message physically appears in the Telegram group. + (verified: send_message is called with correct content containing user name) +2. journalctl -u baton does NOT throw ERROR during send. + (verified: no exception is raised when Telegram returns 200) +3. A repeated request is also delivered. + (verified: two consecutive signals each trigger send_message) + +NOTE: These tests verify that send_message is called with correct parameters. +Physical delivery to an actual Telegram group is outside unit test scope. +""" +from __future__ import annotations + +import asyncio +import os + +os.environ.setdefault("BOT_TOKEN", "test-bot-token") +os.environ.setdefault("CHAT_ID", "-1001234567890") +os.environ.setdefault("WEBHOOK_SECRET", "test-webhook-secret") +os.environ.setdefault("WEBHOOK_URL", "https://example.com/api/webhook/telegram") +os.environ.setdefault("FRONTEND_ORIGIN", "http://localhost:3000") +os.environ.setdefault("ADMIN_TOKEN", "test-admin-token") + +import json +from unittest.mock import AsyncMock, patch + +import httpx +import pytest +import respx +from httpx import AsyncClient + +from tests.conftest import make_app_client + +# Valid UUID v4 constants — must not collide with UUIDs in other test files +_UUID_A = "d0000001-0000-4000-8000-000000000001" +_UUID_B = "d0000002-0000-4000-8000-000000000002" +_UUID_C = "d0000003-0000-4000-8000-000000000003" +_UUID_D = "d0000004-0000-4000-8000-000000000004" +_UUID_E = "d0000005-0000-4000-8000-000000000005" + + +async def _register(client: AsyncClient, uuid: str, name: str) -> str: + r = await client.post("/api/register", json={"uuid": uuid, "name": name}) + assert r.status_code == 200, f"Registration failed: {r.status_code} {r.text}" + return r.json()["api_key"] + + +# --------------------------------------------------------------------------- +# Criterion 1 — send_message is called with text containing the user's name +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_signal_send_message_called_with_user_name(): + """Criterion 1: send_message is invoked with text that includes the sender's name.""" + sent_texts: list[str] = [] + + async def _capture(text: str) -> None: + sent_texts.append(text) + + async with make_app_client() as client: + api_key = await _register(client, _UUID_A, "AliceBaton") + + with patch("backend.telegram.send_message", side_effect=_capture): + resp = await client.post( + "/api/signal", + json={"user_id": _UUID_A, "timestamp": 1742478000000}, + headers={"Authorization": f"Bearer {api_key}"}, + ) + await asyncio.sleep(0) # yield to event loop so background task runs + + assert resp.status_code == 200 + assert len(sent_texts) == 1, f"Expected 1 send_message call, got {len(sent_texts)}" + assert "AliceBaton" in sent_texts[0], ( + f"Expected user name 'AliceBaton' in Telegram message, got: {sent_texts[0]!r}" + ) + + +@pytest.mark.asyncio +async def test_signal_send_message_text_contains_signal_keyword(): + """Criterion 1: Telegram message text contains the word 'Сигнал'.""" + sent_texts: list[str] = [] + + async def _capture(text: str) -> None: + sent_texts.append(text) + + async with make_app_client() as client: + api_key = await _register(client, _UUID_B, "BobBaton") + + with patch("backend.telegram.send_message", side_effect=_capture): + await client.post( + "/api/signal", + json={"user_id": _UUID_B, "timestamp": 1742478000000}, + headers={"Authorization": f"Bearer {api_key}"}, + ) + await asyncio.sleep(0) + + assert len(sent_texts) == 1 + assert "Сигнал" in sent_texts[0], ( + f"Expected 'Сигнал' keyword in message, got: {sent_texts[0]!r}" + ) + + +@pytest.mark.asyncio +async def test_signal_with_geo_send_message_contains_coordinates(): + """Criterion 1: when geo is provided, Telegram message includes lat/lon coordinates.""" + sent_texts: list[str] = [] + + async def _capture(text: str) -> None: + sent_texts.append(text) + + async with make_app_client() as client: + api_key = await _register(client, _UUID_C, "GeoUser") + + with patch("backend.telegram.send_message", side_effect=_capture): + await client.post( + "/api/signal", + json={ + "user_id": _UUID_C, + "timestamp": 1742478000000, + "geo": {"lat": 55.7558, "lon": 37.6173, "accuracy": 15.0}, + }, + headers={"Authorization": f"Bearer {api_key}"}, + ) + await asyncio.sleep(0) + + assert len(sent_texts) == 1 + assert "55.7558" in sent_texts[0], ( + f"Expected lat '55.7558' in message, got: {sent_texts[0]!r}" + ) + assert "37.6173" in sent_texts[0], ( + f"Expected lon '37.6173' in message, got: {sent_texts[0]!r}" + ) + + +@pytest.mark.asyncio +async def test_signal_without_geo_send_message_contains_no_geo_label(): + """Criterion 1: when geo is null, Telegram message contains 'Без геолокации'.""" + sent_texts: list[str] = [] + + async def _capture(text: str) -> None: + sent_texts.append(text) + + async with make_app_client() as client: + api_key = await _register(client, _UUID_D, "NoGeoUser") + + with patch("backend.telegram.send_message", side_effect=_capture): + await client.post( + "/api/signal", + json={"user_id": _UUID_D, "timestamp": 1742478000000, "geo": None}, + headers={"Authorization": f"Bearer {api_key}"}, + ) + await asyncio.sleep(0) + + assert len(sent_texts) == 1 + assert "Без геолокации" in sent_texts[0], ( + f"Expected 'Без геолокации' in message, got: {sent_texts[0]!r}" + ) + + +# --------------------------------------------------------------------------- +# Criterion 2 — No ERROR logged on successful send (service stays alive) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_signal_send_message_no_error_on_200_response(): + """Criterion 2: send_message does not raise when Telegram returns 200.""" + from backend import config as _cfg + from backend.telegram import send_message + + send_url = f"https://api.telegram.org/bot{_cfg.BOT_TOKEN}/sendMessage" + + # Must complete without exception + with respx.mock(assert_all_called=False) as mock: + mock.post(send_url).mock(return_value=httpx.Response(200, json={"ok": True})) + await send_message("Test signal delivery") # should not raise + + +@pytest.mark.asyncio +async def test_signal_send_message_uses_configured_chat_id(): + """Criterion 2: send_message POSTs to Telegram with the configured CHAT_ID.""" + from backend import config as _cfg + from backend.telegram import send_message + + send_url = f"https://api.telegram.org/bot{_cfg.BOT_TOKEN}/sendMessage" + + with respx.mock(assert_all_called=False) as mock: + route = mock.post(send_url).mock( + return_value=httpx.Response(200, json={"ok": True}) + ) + await send_message("Delivery check") + + assert route.called + body = json.loads(route.calls[0].request.content) + assert body["chat_id"] == _cfg.CHAT_ID, ( + f"Expected chat_id={_cfg.CHAT_ID!r}, got {body['chat_id']!r}" + ) + + +# --------------------------------------------------------------------------- +# Criterion 3 — Repeated requests are also delivered +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_repeated_signals_each_trigger_send_message(): + """Criterion 3: two consecutive signals each cause a separate send_message call.""" + sent_texts: list[str] = [] + + async def _capture(text: str) -> None: + sent_texts.append(text) + + async with make_app_client() as client: + api_key = await _register(client, _UUID_E, "RepeatUser") + + with patch("backend.telegram.send_message", side_effect=_capture): + r1 = await client.post( + "/api/signal", + json={"user_id": _UUID_E, "timestamp": 1742478000001}, + headers={"Authorization": f"Bearer {api_key}"}, + ) + await asyncio.sleep(0) + + r2 = await client.post( + "/api/signal", + json={"user_id": _UUID_E, "timestamp": 1742478000002}, + headers={"Authorization": f"Bearer {api_key}"}, + ) + await asyncio.sleep(0) + + assert r1.status_code == 200 + assert r2.status_code == 200 + assert len(sent_texts) == 2, ( + f"Expected 2 send_message calls for 2 signals, got {len(sent_texts)}" + ) + + +@pytest.mark.asyncio +async def test_repeated_signals_produce_incrementing_signal_ids(): + """Criterion 3: repeated signals are each stored and return distinct incrementing signal_ids.""" + async with make_app_client() as client: + api_key = await _register(client, _UUID_E, "RepeatUser2") + r1 = await client.post( + "/api/signal", + json={"user_id": _UUID_E, "timestamp": 1742478000001}, + headers={"Authorization": f"Bearer {api_key}"}, + ) + r2 = await client.post( + "/api/signal", + json={"user_id": _UUID_E, "timestamp": 1742478000002}, + headers={"Authorization": f"Bearer {api_key}"}, + ) + + assert r1.status_code == 200 + assert r2.status_code == 200 + assert r2.json()["signal_id"] > r1.json()["signal_id"], ( + "Second signal must have a higher signal_id than the first" + )