""" 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 . 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 preserved with '# v2.0 feature' marker (static) # --------------------------------------------------------------------------- def test_signal_aggregator_class_preserved_in_telegram(): """SignalAggregator class must still exist in telegram.py (ADR-004, reserved for v2).""" source = (_BACKEND_DIR / "telegram.py").read_text() assert "class SignalAggregator" in source def test_signal_aggregator_has_v2_feature_comment(): """The line immediately before 'class SignalAggregator' must contain '# v2.0 feature'.""" lines = (_BACKEND_DIR / "telegram.py").read_text().splitlines() class_line_idx = next( (i for i, line in enumerate(lines) if "class SignalAggregator" in line), None ) assert class_line_idx is not None, "class SignalAggregator not found in telegram.py" assert class_line_idx > 0, "SignalAggregator is on the first line — no preceding comment line" preceding_line = lines[class_line_idx - 1] assert "# v2.0 feature" in preceding_line, ( f"Expected '# v2.0 feature' on line before class SignalAggregator, got: {preceding_line!r}" )