kin: BATON-007 При нажатии на кнопку происходит анимация и сообщение что сигнал отправлен, но в телеграм группу ничего не приходит.
This commit is contained in:
Gros Frumos
parent
6142770c0c
commit
cbc15eeedc
1 changed files with 262 additions and 0 deletions
262
tests/test_baton_007.py
Normal file
262
tests/test_baton_007.py
Normal file
|
|
@ -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"
|
||||
)
|
||||
Loading…
Add table
Add a link
Reference in a new issue