888 lines
35 KiB
Python
888 lines
35 KiB
Python
"""
|
|
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")
|
|
os.environ.setdefault("ADMIN_CHAT_ID", "5694335584")
|
|
|
|
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 real HTTP sendMessage to ADMIN_CHAT_ID with correct login/email."""
|
|
from backend import config as _cfg
|
|
|
|
captured: list[dict] = []
|
|
async with make_app_client(capture_send_requests=captured) as client:
|
|
resp = await client.post("/api/auth/register", json=_VALID_PAYLOAD)
|
|
assert resp.status_code == 201
|
|
await asyncio.sleep(0)
|
|
|
|
admin_chat_id = str(_cfg.ADMIN_CHAT_ID)
|
|
admin_msgs = [r for r in captured if str(r.get("chat_id")) == admin_chat_id]
|
|
assert len(admin_msgs) >= 1, (
|
|
f"Expected sendMessage to ADMIN_CHAT_ID={admin_chat_id!r}, captured: {captured}"
|
|
)
|
|
text = admin_msgs[0].get("text", "")
|
|
assert _VALID_PAYLOAD["login"] in text, f"Expected login in text: {text!r}"
|
|
assert _VALID_PAYLOAD["email"] in text, f"Expected email in text: {text!r}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 15. State machine — повторное нажатие approve на уже approved
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_webhook_callback_double_approve_does_not_send_push():
|
|
"""Second approve on already-approved registration must NOT fire push."""
|
|
push_sub = {
|
|
"endpoint": "https://fcm.googleapis.com/fcm/send/test2",
|
|
"keys": {"p256dh": "BQDEF", "auth": "abc"},
|
|
}
|
|
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": "double@example.com", "login": "doubleuser", "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_d1",
|
|
"data": f"approve:{reg_id}",
|
|
"message": {"message_id": 60, "chat": {"id": 5694335584}},
|
|
}
|
|
}
|
|
|
|
# First approve — should succeed
|
|
await client.post("/api/webhook/telegram", json=cb_payload, headers=_WEBHOOK_HEADERS)
|
|
await asyncio.sleep(0)
|
|
|
|
# Second approve — push must NOT be fired
|
|
push_calls: list = []
|
|
|
|
async def _capture_push(sub_json, title, body):
|
|
push_calls.append(sub_json)
|
|
|
|
cb_payload2 = {**cb_payload, "callback_query": {**cb_payload["callback_query"], "id": "cq_d2"}}
|
|
with patch("backend.push.send_push", side_effect=_capture_push):
|
|
await client.post("/api/webhook/telegram", json=cb_payload2, headers=_WEBHOOK_HEADERS)
|
|
await asyncio.sleep(0)
|
|
|
|
assert len(push_calls) == 0, f"Second approve must not fire push, got {len(push_calls)} calls"
|
|
|
|
# Also verify status is still 'approved'
|
|
from backend import db as _db
|
|
# Can't check here as client context is closed; DB assertion was covered by state machine logic
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_webhook_callback_double_approve_status_stays_approved():
|
|
"""Status remains 'approved' after a second approve callback."""
|
|
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, "email": "stay@example.com", "login": "stayuser"},
|
|
)
|
|
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 = {
|
|
"callback_query": {
|
|
"id": "cq_s1",
|
|
"data": f"approve:{reg_id}",
|
|
"message": {"message_id": 70, "chat": {"id": 5694335584}},
|
|
}
|
|
}
|
|
await client.post("/api/webhook/telegram", json=cb, headers=_WEBHOOK_HEADERS)
|
|
await asyncio.sleep(0)
|
|
|
|
cb2 = {**cb, "callback_query": {**cb["callback_query"], "id": "cq_s2"}}
|
|
await client.post("/api/webhook/telegram", json=cb2, headers=_WEBHOOK_HEADERS)
|
|
await asyncio.sleep(0)
|
|
|
|
reg = await _db.get_registration(reg_id)
|
|
assert reg["status"] == "approved", f"Expected 'approved', got {reg['status']!r}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 16. State machine — approve после reject
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_webhook_callback_approve_after_reject_status_stays_rejected():
|
|
"""Approve after reject must NOT change status — remains 'rejected'."""
|
|
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, "email": "artest@example.com", "login": "artestuser"},
|
|
)
|
|
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
|
|
|
|
# First: reject
|
|
rej_cb = {
|
|
"callback_query": {
|
|
"id": "cq_ar1",
|
|
"data": f"reject:{reg_id}",
|
|
"message": {"message_id": 80, "chat": {"id": 5694335584}},
|
|
}
|
|
}
|
|
await client.post("/api/webhook/telegram", json=rej_cb, headers=_WEBHOOK_HEADERS)
|
|
await asyncio.sleep(0)
|
|
|
|
# Then: approve — must be ignored
|
|
push_calls: list = []
|
|
|
|
async def _capture_push(sub_json, title, body):
|
|
push_calls.append(sub_json)
|
|
|
|
app_cb = {
|
|
"callback_query": {
|
|
"id": "cq_ar2",
|
|
"data": f"approve:{reg_id}",
|
|
"message": {"message_id": 81, "chat": {"id": 5694335584}},
|
|
}
|
|
}
|
|
with patch("backend.push.send_push", side_effect=_capture_push):
|
|
await client.post("/api/webhook/telegram", json=app_cb, headers=_WEBHOOK_HEADERS)
|
|
await asyncio.sleep(0)
|
|
|
|
reg = await _db.get_registration(reg_id)
|
|
assert reg["status"] == "rejected", f"Expected 'rejected', got {reg['status']!r}"
|
|
|
|
assert len(push_calls) == 0, f"Approve after reject must not fire push, got {len(push_calls)}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 17. Rate limiting — 4th request returns 429
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auth_register_rate_limit_fourth_request_returns_429():
|
|
"""4th registration request from same IP within the window returns 429."""
|
|
async with make_app_client() as client:
|
|
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
|
|
for i in range(3):
|
|
r = await client.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"email": f"ratetest{i}@example.com",
|
|
"login": f"ratetest{i}",
|
|
"password": "strongpassword123",
|
|
},
|
|
)
|
|
assert r.status_code == 201, f"Request {i+1} should succeed, got {r.status_code}"
|
|
|
|
# 4th request — must be rate-limited
|
|
r4 = await client.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"email": "ratetest4@example.com",
|
|
"login": "ratetest4",
|
|
"password": "strongpassword123",
|
|
},
|
|
)
|
|
|
|
assert r4.status_code == 429, f"Expected 429 on 4th request, got {r4.status_code}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 18. VAPID public key endpoint /api/push/public-key
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_vapid_public_key_new_endpoint_returns_200():
|
|
"""GET /api/push/public-key returns 200 with vapid_public_key field."""
|
|
async with make_app_client() as client:
|
|
resp = await client.get("/api/push/public-key")
|
|
|
|
assert resp.status_code == 200, f"Expected 200, got {resp.status_code}: {resp.text}"
|
|
body = resp.json()
|
|
assert "vapid_public_key" in body, f"Expected 'vapid_public_key' in response, got {body}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 19. Password max length — 129 chars → 422
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auth_register_422_password_too_long():
|
|
"""Password of 129 characters returns 422."""
|
|
async with make_app_client() as client:
|
|
resp = await client.post(
|
|
"/api/auth/register",
|
|
json={**_VALID_PAYLOAD, "password": "a" * 129},
|
|
)
|
|
assert resp.status_code == 422, f"Expected 422 on 129-char password, got {resp.status_code}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 20. Login max length — 31 chars → 422
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auth_register_422_login_too_long():
|
|
"""Login of 31 characters returns 422."""
|
|
async with make_app_client() as client:
|
|
resp = await client.post(
|
|
"/api/auth/register",
|
|
json={**_VALID_PAYLOAD, "login": "a" * 31},
|
|
)
|
|
assert resp.status_code == 422, f"Expected 422 on 31-char login, got {resp.status_code}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 21. Empty body — POST /api/auth/register with {} → 422
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auth_register_422_empty_body():
|
|
"""Empty JSON body returns 422."""
|
|
async with make_app_client() as client:
|
|
resp = await client.post("/api/auth/register", json={})
|
|
assert resp.status_code == 422, f"Expected 422 on empty body, got {resp.status_code}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 22. Malformed callback_data — no colon → ok:True without crash
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_webhook_callback_malformed_data_no_colon_returns_ok():
|
|
"""callback_query with data='garbage' (no colon) returns ok:True gracefully."""
|
|
async with make_app_client() as client:
|
|
cb_payload = {
|
|
"callback_query": {
|
|
"id": "cq_mal1",
|
|
"data": "garbage",
|
|
"message": {"message_id": 90, "chat": {"id": 5694335584}},
|
|
}
|
|
}
|
|
resp = await client.post(
|
|
"/api/webhook/telegram", json=cb_payload, headers=_WEBHOOK_HEADERS
|
|
)
|
|
|
|
assert resp.status_code == 200, f"Expected 200, got {resp.status_code}"
|
|
assert resp.json() == {"ok": True}, f"Expected ok:True, got {resp.json()}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 23. Non-numeric reg_id — data='approve:abc' → ok:True without crash
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_webhook_callback_non_numeric_reg_id_returns_ok():
|
|
"""callback_query with data='approve:abc' (non-numeric reg_id) returns ok:True."""
|
|
async with make_app_client() as client:
|
|
cb_payload = {
|
|
"callback_query": {
|
|
"id": "cq_nan1",
|
|
"data": "approve:abc",
|
|
"message": {"message_id": 91, "chat": {"id": 5694335584}},
|
|
}
|
|
}
|
|
resp = await client.post(
|
|
"/api/webhook/telegram", json=cb_payload, headers=_WEBHOOK_HEADERS
|
|
)
|
|
|
|
assert resp.status_code == 200, f"Expected 200, got {resp.status_code}"
|
|
assert resp.json() == {"ok": True}, f"Expected ok:True, got {resp.json()}"
|