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

582 lines
23 KiB
Python
Raw Normal View History

kin: BATON-008-backend_dev
2026-03-21 09:19:50 +02:00
"""
Tests for BATON-008: Registration flow with Telegram admin approval.
Acceptance criteria:
1. POST /api/auth/register returns 201 with status='pending' on valid input
2. POST /api/auth/register returns 409 on email or login conflict
3. POST /api/auth/register returns 422 on invalid email/login/password
4. Telegram notification is fire-and-forget — 201 is returned even if Telegram fails
5. Webhook callback_query approve → db status='approved', push task fired if subscription present
6. Webhook callback_query reject → db status='rejected'
7. Webhook callback_query with unknown reg_id → returns {"ok": True} gracefully
"""
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")
from unittest.mock import AsyncMock, patch
import pytest
from tests.conftest import make_app_client
_WEBHOOK_SECRET = "test-webhook-secret"
_WEBHOOK_HEADERS = {"X-Telegram-Bot-Api-Secret-Token": _WEBHOOK_SECRET}
_VALID_PAYLOAD = {
"email": "user@example.com",
"login": "testuser",
"password": "strongpassword123",
}
# ---------------------------------------------------------------------------
# 1. Happy path — 201
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_auth_register_returns_201_pending():
"""Valid registration request returns 201 with status='pending'."""
async with make_app_client() as client:
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
resp = await client.post("/api/auth/register", json=_VALID_PAYLOAD)
assert resp.status_code == 201, f"Expected 201, got {resp.status_code}: {resp.text}"
body = resp.json()
assert body["status"] == "pending"
assert "message" in body
@pytest.mark.asyncio
async def test_auth_register_fire_and_forget_telegram_error_still_returns_201():
"""Telegram failure must not break 201 — fire-and-forget pattern."""
async with make_app_client() as client:
with patch(
"backend.telegram.send_registration_notification",
new_callable=AsyncMock,
side_effect=Exception("Telegram down"),
):
resp = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "email": "other@example.com", "login": "otheruser"},
)
await asyncio.sleep(0)
assert resp.status_code == 201, f"Telegram error must not break 201, got {resp.status_code}"
# ---------------------------------------------------------------------------
# 2. Conflict — 409
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_auth_register_409_on_duplicate_email():
"""Duplicate email returns 409."""
async with make_app_client() as client:
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
r1 = await client.post("/api/auth/register", json=_VALID_PAYLOAD)
assert r1.status_code == 201, f"First registration failed: {r1.text}"
r2 = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "login": "differentlogin"},
)
assert r2.status_code == 409, f"Expected 409 on duplicate email, got {r2.status_code}"
@pytest.mark.asyncio
async def test_auth_register_409_on_duplicate_login():
"""Duplicate login returns 409."""
async with make_app_client() as client:
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
r1 = await client.post("/api/auth/register", json=_VALID_PAYLOAD)
assert r1.status_code == 201, f"First registration failed: {r1.text}"
r2 = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "email": "different@example.com"},
)
assert r2.status_code == 409, f"Expected 409 on duplicate login, got {r2.status_code}"
# ---------------------------------------------------------------------------
# 3. Validation — 422
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_auth_register_422_invalid_email():
"""Invalid email format returns 422."""
async with make_app_client() as client:
resp = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "email": "not-an-email"},
)
assert resp.status_code == 422, f"Expected 422 on invalid email, got {resp.status_code}"
@pytest.mark.asyncio
async def test_auth_register_422_short_login():
"""Login shorter than 3 chars returns 422."""
async with make_app_client() as client:
resp = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "login": "ab"},
)
assert resp.status_code == 422, f"Expected 422 on short login, got {resp.status_code}"
@pytest.mark.asyncio
async def test_auth_register_422_login_invalid_chars():
"""Login with spaces/special chars returns 422."""
async with make_app_client() as client:
resp = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "login": "invalid login!"},
)
assert resp.status_code == 422, f"Expected 422 on login with spaces, got {resp.status_code}"
@pytest.mark.asyncio
async def test_auth_register_422_short_password():
"""Password shorter than 8 chars returns 422."""
async with make_app_client() as client:
resp = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "password": "short"},
)
assert resp.status_code == 422, f"Expected 422 on short password, got {resp.status_code}"
# ---------------------------------------------------------------------------
# 4. Telegram notification is sent to ADMIN_CHAT_ID
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_auth_register_sends_notification_to_admin():
"""Registration triggers send_registration_notification with correct data."""
calls: list[dict] = []
async def _capture(reg_id, login, email, created_at):
calls.append({"reg_id": reg_id, "login": login, "email": email})
async with make_app_client() as client:
with patch("backend.telegram.send_registration_notification", side_effect=_capture):
await client.post("/api/auth/register", json=_VALID_PAYLOAD)
await asyncio.sleep(0)
assert len(calls) == 1, f"Expected 1 notification call, got {len(calls)}"
assert calls[0]["login"] == _VALID_PAYLOAD["login"]
assert calls[0]["email"] == _VALID_PAYLOAD["email"]
# ---------------------------------------------------------------------------
# 5. Webhook callback_query — approve
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_webhook_callback_approve_updates_db_status():
"""approve callback updates registration status to 'approved' in DB."""
from backend import db as _db
async with make_app_client() as client:
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
reg_resp = await client.post("/api/auth/register", json=_VALID_PAYLOAD)
assert reg_resp.status_code == 201
# We need reg_id — get it from DB directly
reg_id = None
from tests.conftest import temp_db as _temp_db # noqa: F401 — already active
from backend import config as _cfg
import aiosqlite
async with aiosqlite.connect(_cfg.DB_PATH) as conn:
conn.row_factory = aiosqlite.Row
async with conn.execute("SELECT id FROM registrations LIMIT 1") as cur:
row = await cur.fetchone()
reg_id = row["id"] if row else None
assert reg_id is not None, "Registration not found in DB"
cb_payload = {
"callback_query": {
"id": "cq_001",
"data": f"approve:{reg_id}",
"message": {
"message_id": 42,
"chat": {"id": 5694335584},
},
}
}
resp = await client.post(
"/api/webhook/telegram",
json=cb_payload,
headers=_WEBHOOK_HEADERS,
)
await asyncio.sleep(0)
assert resp.status_code == 200
assert resp.json() == {"ok": True}
# Verify DB status updated
reg = await _db.get_registration(reg_id)
assert reg is not None
assert reg["status"] == "approved", f"Expected status='approved', got {reg['status']!r}"
@pytest.mark.asyncio
async def test_webhook_callback_approve_fires_push_when_subscription_present():
"""approve callback triggers send_push when push_subscription is set."""
push_sub = {
"endpoint": "https://fcm.googleapis.com/fcm/send/test",
"keys": {"p256dh": "BQABC", "auth": "xyz"},
}
async with make_app_client() as client:
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
reg_resp = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "push_subscription": push_sub},
)
assert reg_resp.status_code == 201
from backend import config as _cfg
import aiosqlite
async with aiosqlite.connect(_cfg.DB_PATH) as conn:
conn.row_factory = aiosqlite.Row
async with conn.execute("SELECT id FROM registrations LIMIT 1") as cur:
row = await cur.fetchone()
reg_id = row["id"] if row else None
assert reg_id is not None
cb_payload = {
"callback_query": {
"id": "cq_002",
"data": f"approve:{reg_id}",
"message": {"message_id": 43, "chat": {"id": 5694335584}},
}
}
push_calls: list = []
async def _capture_push(sub_json, title, body):
push_calls.append(sub_json)
with patch("backend.push.send_push", side_effect=_capture_push):
await client.post("/api/webhook/telegram", json=cb_payload, headers=_WEBHOOK_HEADERS)
await asyncio.sleep(0)
assert len(push_calls) == 1, f"Expected 1 push call, got {len(push_calls)}"
# ---------------------------------------------------------------------------
# 6. Webhook callback_query — reject
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_webhook_callback_reject_updates_db_status():
"""reject callback updates registration status to 'rejected' in DB."""
from backend import db as _db
async with make_app_client() as client:
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
reg_resp = await client.post("/api/auth/register", json=_VALID_PAYLOAD)
assert reg_resp.status_code == 201
from backend import config as _cfg
import aiosqlite
async with aiosqlite.connect(_cfg.DB_PATH) as conn:
conn.row_factory = aiosqlite.Row
async with conn.execute("SELECT id FROM registrations LIMIT 1") as cur:
row = await cur.fetchone()
reg_id = row["id"] if row else None
assert reg_id is not None
cb_payload = {
"callback_query": {
"id": "cq_003",
"data": f"reject:{reg_id}",
"message": {"message_id": 44, "chat": {"id": 5694335584}},
}
}
resp = await client.post(
"/api/webhook/telegram", json=cb_payload, headers=_WEBHOOK_HEADERS
)
await asyncio.sleep(0)
assert resp.status_code == 200
reg = await _db.get_registration(reg_id)
assert reg is not None
assert reg["status"] == "rejected", f"Expected status='rejected', got {reg['status']!r}"
# ---------------------------------------------------------------------------
# 7. Unknown reg_id — graceful handling
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_webhook_callback_unknown_reg_id_returns_ok():
"""callback_query with unknown reg_id returns ok without error."""
async with make_app_client() as client:
cb_payload = {
"callback_query": {
"id": "cq_999",
"data": "approve:99999",
"message": {"message_id": 1, "chat": {"id": 5694335584}},
}
}
resp = await client.post(
"/api/webhook/telegram", json=cb_payload, headers=_WEBHOOK_HEADERS
)
await asyncio.sleep(0)
assert resp.status_code == 200
assert resp.json() == {"ok": True}
kin: BATON-FIX-008 [TECH DEBT] Серверный код (backend/main.py, middleware.py) расходится с worktree — у сервера нет rate_limit_signal в middleware, серверный main.py пропатчен вручную через sed
2026-03-21 09:25:08 +02:00
# ---------------------------------------------------------------------------
# 8. Registration without push_subscription
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_register_without_push_subscription():
"""Registration with push_subscription=null returns 201."""
async with make_app_client() as client:
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
resp = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "email": "nopush@example.com", "login": "nopushuser"},
)
assert resp.status_code == 201
assert resp.json()["status"] == "pending"
# ---------------------------------------------------------------------------
# 9. reject does NOT trigger Web Push
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_webhook_callback_reject_does_not_send_push():
"""reject callback does NOT call send_push."""
async with make_app_client() as client:
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
reg_resp = await client.post("/api/auth/register", json=_VALID_PAYLOAD)
assert reg_resp.status_code == 201
from backend import config as _cfg
import aiosqlite
async with aiosqlite.connect(_cfg.DB_PATH) as conn:
conn.row_factory = aiosqlite.Row
async with conn.execute("SELECT id FROM registrations LIMIT 1") as cur:
row = await cur.fetchone()
reg_id = row["id"] if row else None
assert reg_id is not None
cb_payload = {
"callback_query": {
"id": "cq_r001",
"data": f"reject:{reg_id}",
"message": {"message_id": 50, "chat": {"id": 5694335584}},
}
}
push_calls: list = []
async def _capture_push(sub_json, title, body):
push_calls.append(sub_json)
with patch("backend.push.send_push", side_effect=_capture_push):
await client.post("/api/webhook/telegram", json=cb_payload, headers=_WEBHOOK_HEADERS)
await asyncio.sleep(0)
assert len(push_calls) == 0, f"Expected 0 push calls on reject, got {len(push_calls)}"
# ---------------------------------------------------------------------------
# 10. approve calls editMessageText with ✅ text
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_webhook_callback_approve_edits_message():
"""approve callback calls editMessageText with '✅ Пользователь ... одобрен'."""
async with make_app_client() as client:
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
reg_resp = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "email": "edit@example.com", "login": "edituser"},
)
assert reg_resp.status_code == 201
from backend import config as _cfg
import aiosqlite
async with aiosqlite.connect(_cfg.DB_PATH) as conn:
conn.row_factory = aiosqlite.Row
async with conn.execute("SELECT id FROM registrations LIMIT 1") as cur:
row = await cur.fetchone()
reg_id = row["id"] if row else None
assert reg_id is not None
cb_payload = {
"callback_query": {
"id": "cq_e001",
"data": f"approve:{reg_id}",
"message": {"message_id": 51, "chat": {"id": 5694335584}},
}
}
edit_calls: list[str] = []
async def _capture_edit(chat_id, message_id, text):
edit_calls.append(text)
with patch("backend.telegram.edit_message_text", side_effect=_capture_edit):
await client.post("/api/webhook/telegram", json=cb_payload, headers=_WEBHOOK_HEADERS)
assert len(edit_calls) == 1, f"Expected 1 editMessageText call, got {len(edit_calls)}"
assert "✅" in edit_calls[0], f"Expected ✅ in edit text, got: {edit_calls[0]!r}"
assert "edituser" in edit_calls[0], f"Expected login in edit text, got: {edit_calls[0]!r}"
# ---------------------------------------------------------------------------
# 11. answerCallbackQuery is called after callback processing
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_webhook_callback_answer_sent():
"""answerCallbackQuery is called with the callback_query_id after processing."""
async with make_app_client() as client:
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
reg_resp = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "email": "answer@example.com", "login": "answeruser"},
)
assert reg_resp.status_code == 201
from backend import config as _cfg
import aiosqlite
async with aiosqlite.connect(_cfg.DB_PATH) as conn:
conn.row_factory = aiosqlite.Row
async with conn.execute("SELECT id FROM registrations LIMIT 1") as cur:
row = await cur.fetchone()
reg_id = row["id"] if row else None
assert reg_id is not None
cb_payload = {
"callback_query": {
"id": "cq_a001",
"data": f"approve:{reg_id}",
"message": {"message_id": 52, "chat": {"id": 5694335584}},
}
}
answer_calls: list[str] = []
async def _capture_answer(callback_query_id):
answer_calls.append(callback_query_id)
with patch("backend.telegram.answer_callback_query", side_effect=_capture_answer):
await client.post("/api/webhook/telegram", json=cb_payload, headers=_WEBHOOK_HEADERS)
assert len(answer_calls) == 1, f"Expected 1 answerCallbackQuery call, got {len(answer_calls)}"
assert answer_calls[0] == "cq_a001"
# ---------------------------------------------------------------------------
# 12. CORS — Authorization header is allowed
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_cors_authorization_header_allowed():
"""CORS preflight request allows Authorization header."""
async with make_app_client() as client:
resp = await client.options(
"/api/auth/register",
headers={
"Origin": "http://localhost:3000",
"Access-Control-Request-Method": "POST",
"Access-Control-Request-Headers": "Authorization",
},
)
assert resp.status_code in (200, 204), f"CORS preflight returned {resp.status_code}"
allow_headers = resp.headers.get("access-control-allow-headers", "")
assert "authorization" in allow_headers.lower(), (
f"Authorization not in Access-Control-Allow-Headers: {allow_headers!r}"
)
# ---------------------------------------------------------------------------
# 13. DB — registrations table exists after init_db
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_registrations_table_created():
"""init_db creates the registrations table with correct schema."""
from tests.conftest import temp_db
from backend import db as _db, config as _cfg
import aiosqlite
with temp_db():
await _db.init_db()
async with aiosqlite.connect(_cfg.DB_PATH) as conn:
async with conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='registrations'"
) as cur:
row = await cur.fetchone()
assert row is not None, "Table 'registrations' not found after init_db()"
# ---------------------------------------------------------------------------
# 14. DB — password_hash uses PBKDF2 '{salt_hex}:{dk_hex}' format
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_password_hash_stored_in_pbkdf2_format():
"""Stored password_hash uses '<salt_hex>:<dk_hex>' PBKDF2 format."""
from backend import config as _cfg
import aiosqlite
async with make_app_client() as client:
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "email": "pbkdf2@example.com", "login": "pbkdf2user"},
)
async with aiosqlite.connect(_cfg.DB_PATH) as conn:
conn.row_factory = aiosqlite.Row
async with conn.execute(
"SELECT password_hash FROM registrations WHERE login = 'pbkdf2user'"
) as cur:
row = await cur.fetchone()
assert row is not None, "Registration not found in DB"
password_hash = row["password_hash"]
assert ":" in password_hash, f"Expected 'salt:hash' format, got {password_hash!r}"
parts = password_hash.split(":")
assert len(parts) == 2, f"Expected exactly one colon separator, got {password_hash!r}"
salt_hex, dk_hex = parts
# salt = os.urandom(16) → 32 hex chars; dk = SHA-256 output (32 bytes) → 64 hex chars
assert len(salt_hex) == 32, f"Expected 32-char salt hex, got {len(salt_hex)}"
assert len(dk_hex) == 64, f"Expected 64-char dk hex (SHA-256), got {len(dk_hex)}"
int(salt_hex, 16) # raises ValueError if not valid hex
int(dk_hex, 16)
Reference in a new issue Copy permalink