pelmen/baton
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

kin: BATON-BIZ-002 Убрать hardcoded VAPID key из meta-тега, читать с /api/push/public-key

Browse source
This commit is contained in:
Gros Frumos 2026-03-21 12:43:35 +02:00
parent ea06309a6e
commit 6444b30d17
7 changed files with 488 additions and 159 deletions
Download patch file Download diff file Expand all files Collapse all files

157
tests/test_telegram.py
View file

@ -1,5 +1,5 @@
"""
Tests for backend/telegram.py: send_message, set_webhook, SignalAggregator.
Tests for backend/telegram.py: send_message, set_webhook, validate_bot_token.
NOTE: respx routes must be registered INSIDE the 'with mock:' block to be
intercepted properly. Registering them before entering the context does not
@ -25,8 +25,6 @@ def _safe_aiosqlite_await(self):
aiosqlite.core.Connection.__await__ = _safe_aiosqlite_await # type: ignore[method-assign]
import json
import os as _os
import tempfile
from unittest.mock import AsyncMock, patch
import httpx
@ -34,7 +32,7 @@ import pytest
import respx
from backend import config
from backend.telegram import SignalAggregator, send_message, set_webhook, validate_bot_token
from backend.telegram import send_message, set_webhook, validate_bot_token
SEND_URL = f"https://api.telegram.org/bot{config.BOT_TOKEN}/sendMessage"
@ -186,127 +184,6 @@ async def test_set_webhook_raises_on_non_200():
await set_webhook(url="https://example.com/webhook", secret="s")
# ---------------------------------------------------------------------------
# SignalAggregator helpers
# ---------------------------------------------------------------------------
async def _init_db_with_tmp() -> str:
"""Init a temp-file DB and return its path."""
from backend import config as _cfg, db as _db
path = tempfile.mktemp(suffix=".db")
_cfg.DB_PATH = path
await _db.init_db()
return path
def _cleanup(path: str) -> None:
for ext in ("", "-wal", "-shm"):
try:
_os.unlink(path + ext)
except FileNotFoundError:
pass
# ---------------------------------------------------------------------------
# SignalAggregator tests
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_aggregator_single_signal_calls_send_message():
"""Flushing an aggregator with one signal calls send_message once."""
path = await _init_db_with_tmp()
try:
agg = SignalAggregator(interval=9999)
await agg.add_signal(
user_uuid="a9900001-0000-4000-8000-000000000001",
user_name="Alice",
timestamp=1742478000000,
geo={"lat": 55.0, "lon": 37.0, "accuracy": 10.0},
signal_id=1,
)
with respx.mock(assert_all_called=False) as mock:
send_route = mock.post(SEND_URL).mock(
return_value=httpx.Response(200, json={"ok": True})
)
with patch("backend.telegram.db.save_telegram_batch", new_callable=AsyncMock):
with patch("backend.telegram.asyncio.sleep", new_callable=AsyncMock):
await agg.flush()
assert send_route.call_count == 1
finally:
_cleanup(path)
@pytest.mark.asyncio
async def test_aggregator_multiple_signals_one_message():
"""5 signals flushed at once produce exactly one send_message call."""
path = await _init_db_with_tmp()
try:
agg = SignalAggregator(interval=9999)
for i in range(5):
await agg.add_signal(
user_uuid=f"a990000{i}-0000-4000-8000-00000000000{i}",
user_name=f"User{i}",
timestamp=1742478000000 + i * 1000,
geo=None,
signal_id=i + 1,
)
with respx.mock(assert_all_called=False) as mock:
send_route = mock.post(SEND_URL).mock(
return_value=httpx.Response(200, json={"ok": True})
)
with patch("backend.telegram.db.save_telegram_batch", new_callable=AsyncMock):
with patch("backend.telegram.asyncio.sleep", new_callable=AsyncMock):
await agg.flush()
assert send_route.call_count == 1
finally:
_cleanup(path)
@pytest.mark.asyncio
async def test_aggregator_empty_buffer_no_send():
"""Flushing an empty aggregator must NOT call send_message."""
agg = SignalAggregator(interval=9999)
# No routes registered — if a POST is made it will raise AllMockedAssertionError
with respx.mock(assert_all_called=False) as mock:
send_route = mock.post(SEND_URL).mock(
return_value=httpx.Response(200, json={"ok": True})
)
await agg.flush()
assert send_route.call_count == 0
@pytest.mark.asyncio
async def test_aggregator_buffer_cleared_after_flush():
"""After flush, the aggregator buffer is empty."""
path = await _init_db_with_tmp()
try:
agg = SignalAggregator(interval=9999)
await agg.add_signal(
user_uuid="a9900099-0000-4000-8000-000000000099",
user_name="Test",
timestamp=1742478000000,
geo=None,
signal_id=99,
)
assert len(agg._buffer) == 1
with respx.mock(assert_all_called=False) as mock:
mock.post(SEND_URL).mock(return_value=httpx.Response(200, json={"ok": True}))
with patch("backend.telegram.db.save_telegram_batch", new_callable=AsyncMock):
with patch("backend.telegram.asyncio.sleep", new_callable=AsyncMock):
await agg.flush()
assert len(agg._buffer) == 0
finally:
_cleanup(path)
# ---------------------------------------------------------------------------
# BATON-007: 400 "chat not found" handling
# ---------------------------------------------------------------------------
@ -371,33 +248,3 @@ async def test_send_message_all_5xx_retries_exhausted_does_not_raise():
# Must not raise — message is dropped, service stays alive
await send_message("test all retries exhausted")
@pytest.mark.asyncio
async def test_aggregator_unknown_user_shows_uuid_prefix():
"""If user_name is None, the message shows first 8 chars of uuid."""
path = await _init_db_with_tmp()
try:
agg = SignalAggregator(interval=9999)
test_uuid = "abcdef1234567890"
await agg.add_signal(
user_uuid=test_uuid,
user_name=None,
timestamp=1742478000000,
geo=None,
signal_id=1,
)
sent_texts: list[str] = []
async def _fake_send(text: str) -> None:
sent_texts.append(text)
with patch("backend.telegram.send_message", side_effect=_fake_send):
with patch("backend.telegram.db.save_telegram_batch", new_callable=AsyncMock):
with patch("backend.telegram.asyncio.sleep", new_callable=AsyncMock):
await agg.flush()
assert len(sent_texts) == 1
assert test_uuid[:8] in sent_texts[0]
finally:
_cleanup(path)