47b89ded8d
When signal has geo, show clickable Google Maps link instead of raw coordinates. Without geo, show "Гео нету". Added parse_mode=HTML to send_message for link rendering. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
184 lines
7.2 KiB
Python
184 lines
7.2 KiB
Python
"""
|
|
Tests for BATON-ARCH-002: SignalAggregator disabled in v1 (ADR-004).
|
|
|
|
Acceptance criteria:
|
|
1. No asyncio task for the aggregator is created at lifespan startup.
|
|
2. POST /api/signal calls telegram.send_message directly (no aggregator intermediary).
|
|
3. SignalAggregator class in telegram.py is preserved with '# v2.0 feature' marker.
|
|
|
|
UUID notes: all UUIDs satisfy the UUID v4 pattern.
|
|
BATON-SEC-003: POST /api/signal now requires Authorization: Bearer <api_key>.
|
|
Tests that send signals register first and use the returned api_key.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
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")
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from tests.conftest import make_app_client
|
|
|
|
_BACKEND_DIR = Path(__file__).parent.parent / "backend"
|
|
|
|
# Valid UUID v4 constants
|
|
_UUID_S1 = "a0100001-0000-4000-8000-000000000001"
|
|
_UUID_S2 = "a0100002-0000-4000-8000-000000000002"
|
|
_UUID_S3 = "a0100003-0000-4000-8000-000000000003"
|
|
_UUID_S4 = "a0100004-0000-4000-8000-000000000004"
|
|
_UUID_S5 = "a0100005-0000-4000-8000-000000000005"
|
|
|
|
|
|
async def _register(client, uuid: str, name: str) -> str:
|
|
"""Register user and return api_key."""
|
|
r = await client.post("/api/register", json={"uuid": uuid, "name": name})
|
|
assert r.status_code == 200
|
|
return r.json()["api_key"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Criterion 1 — No asyncio task for aggregator created at startup (static)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_aggregator_task_creation_commented_out_in_main():
|
|
"""aggregator.run() must not appear in an active create_task call in main.py (ADR-004).
|
|
|
|
Note: other create_task calls (e.g. keep-alive) are allowed — only the
|
|
SignalAggregator task is disabled in v1.
|
|
"""
|
|
source = (_BACKEND_DIR / "main.py").read_text()
|
|
active_lines = [
|
|
line
|
|
for line in source.splitlines()
|
|
if "create_task" in line
|
|
and "aggregator" in line
|
|
and not line.strip().startswith("#")
|
|
]
|
|
assert active_lines == [], (
|
|
f"Found active asyncio.create_task(aggregator...) in main.py: {active_lines}"
|
|
)
|
|
|
|
|
|
def test_aggregator_instantiation_commented_out_in_main():
|
|
"""SignalAggregator() must not be instantiated in active code in main.py (ADR-004)."""
|
|
source = (_BACKEND_DIR / "main.py").read_text()
|
|
active_lines = [
|
|
line
|
|
for line in source.splitlines()
|
|
if "SignalAggregator" in line and not line.strip().startswith("#")
|
|
]
|
|
assert active_lines == [], (
|
|
f"Found active SignalAggregator instantiation in main.py: {active_lines}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Criterion 2 — POST /api/signal calls send_message directly (dynamic)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_signal_calls_telegram_send_message_directly():
|
|
"""POST /api/signal must call telegram.send_message directly, not via aggregator (ADR-004)."""
|
|
async with make_app_client() as client:
|
|
api_key = await _register(client, _UUID_S1, "Tester")
|
|
with patch("backend.telegram.send_message", new_callable=AsyncMock) as mock_send:
|
|
resp = await client.post(
|
|
"/api/signal",
|
|
json={"user_id": _UUID_S1, "timestamp": 1742478000000},
|
|
headers={"Authorization": f"Bearer {api_key}"},
|
|
)
|
|
assert resp.status_code == 200
|
|
mock_send.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_signal_message_contains_registered_username():
|
|
"""Message passed to send_message must include the registered user's name."""
|
|
async with make_app_client() as client:
|
|
api_key = await _register(client, _UUID_S2, "Alice")
|
|
with patch("backend.telegram.send_message", new_callable=AsyncMock) as mock_send:
|
|
await client.post(
|
|
"/api/signal",
|
|
json={"user_id": _UUID_S2, "timestamp": 1742478000000},
|
|
headers={"Authorization": f"Bearer {api_key}"},
|
|
)
|
|
text = mock_send.call_args[0][0]
|
|
assert "Alice" in text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_signal_message_without_geo_contains_bez_geolocatsii():
|
|
"""When geo is None, message must contain 'Гео нету'."""
|
|
async with make_app_client() as client:
|
|
api_key = await _register(client, _UUID_S3, "Bob")
|
|
with patch("backend.telegram.send_message", new_callable=AsyncMock) as mock_send:
|
|
await client.post(
|
|
"/api/signal",
|
|
json={"user_id": _UUID_S3, "timestamp": 1742478000000, "geo": None},
|
|
headers={"Authorization": f"Bearer {api_key}"},
|
|
)
|
|
text = mock_send.call_args[0][0]
|
|
assert "Гео нету" in text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_signal_message_with_geo_contains_coordinates():
|
|
"""When geo is provided, message must contain lat and lon values."""
|
|
async with make_app_client() as client:
|
|
api_key = await _register(client, _UUID_S4, "Charlie")
|
|
with patch("backend.telegram.send_message", new_callable=AsyncMock) as mock_send:
|
|
await client.post(
|
|
"/api/signal",
|
|
json={
|
|
"user_id": _UUID_S4,
|
|
"timestamp": 1742478000000,
|
|
"geo": {"lat": 55.7558, "lon": 37.6173, "accuracy": 15.0},
|
|
},
|
|
headers={"Authorization": f"Bearer {api_key}"},
|
|
)
|
|
text = mock_send.call_args[0][0]
|
|
assert "55.7558" in text
|
|
assert "37.6173" in text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_signal_message_contains_utc_marker():
|
|
"""Message passed to send_message must contain 'UTC' timestamp marker."""
|
|
async with make_app_client() as client:
|
|
api_key = await _register(client, _UUID_S5, "Dave")
|
|
with patch("backend.telegram.send_message", new_callable=AsyncMock) as mock_send:
|
|
await client.post(
|
|
"/api/signal",
|
|
json={"user_id": _UUID_S5, "timestamp": 1742478000000},
|
|
headers={"Authorization": f"Bearer {api_key}"},
|
|
)
|
|
text = mock_send.call_args[0][0]
|
|
assert "UTC" in text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Criterion 3 — SignalAggregator removed (BATON-BIZ-004: dead code cleanup)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_signal_aggregator_class_removed_from_telegram():
|
|
"""SignalAggregator must be removed from telegram.py (BATON-BIZ-004)."""
|
|
source = (_BACKEND_DIR / "telegram.py").read_text()
|
|
assert "class SignalAggregator" not in source
|
|
|
|
|
|
def test_signal_aggregator_not_referenced_in_telegram():
|
|
"""telegram.py must not reference SignalAggregator at all (BATON-BIZ-004)."""
|
|
source = (_BACKEND_DIR / "telegram.py").read_text()
|
|
assert "SignalAggregator" not in source
|