test: add test suite with 65 tests, 73% coverage
- Unit tests: password hashing, notification providers, webhook field parsing - Integration tests: auth routes (register/login/confirm-email/logout), invite flow, Evotor webhooks (/user/create, /user/verify, /user/token), admin panel (access control, activate/suspend/delete/reset-password) - conftest: SQLite in-memory engine, transactional sessions, factory-boy factories (UserFactory with UserRoleEnum variants) - Fix bcrypt: replace passlib (broken on Python 3.14 + bcrypt 5.x) with direct bcrypt calls; drop passlib from requirements.txt - Fix datetime.utcnow() deprecation across routes and tests - Fix Jinja2 TemplateResponse signature (request as first positional arg) - Add coverage config to pyproject.toml Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
mguschin
196
tests/test_routes_evotor_webhooks.py
Normal file
196
tests/test_routes_evotor_webhooks.py
Normal file
@@ -0,0 +1,196 @@
|
||||
"""Integration tests for Evotor webhook endpoints."""
|
||||
import json
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from web.models.connections import EvotorConnection
|
||||
from web.models.user import User, UserStatusEnum
|
||||
|
||||
|
||||
WEBHOOK_SECRET = "test-secret-abc"
|
||||
|
||||
|
||||
def auth_headers(secret=WEBHOOK_SECRET):
|
||||
return {"Authorization": f"Bearer {secret}"}
|
||||
|
||||
|
||||
# ── /user/create ──────────────────────────────────────────────────────────────
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("web.routes.evotor_webhooks.send_email_task")
|
||||
async def test_user_create_new_user(mock_task, client, override_db, monkeypatch):
|
||||
monkeypatch.setattr("web.routes.evotor_webhooks.settings.EVOTOR_WEBHOOK_SECRET", WEBHOOK_SECRET)
|
||||
monkeypatch.setattr("web.routes.evotor_webhooks.settings.INVITE_EXPIRE_HOURS", 48)
|
||||
monkeypatch.setattr("web.routes.evotor_webhooks.settings.BASE_URL", "http://test")
|
||||
|
||||
payload = {
|
||||
"userId": "evo-001",
|
||||
"customField": json.dumps({"email": "newuser@test.com", "phone": "+79001234501"}),
|
||||
}
|
||||
resp = await client.post("/user/create", json=payload, headers=auth_headers())
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["userId"] == "evo-001"
|
||||
assert "token" in data
|
||||
assert len(data["token"]) > 10
|
||||
|
||||
user = override_db.query(User).filter(User.evotor_user_id == "evo-001").first()
|
||||
assert user is not None
|
||||
assert user.status == UserStatusEnum.pending
|
||||
assert user.invite_token is not None
|
||||
assert user.password_hash is None
|
||||
|
||||
conn = override_db.query(EvotorConnection).filter(
|
||||
EvotorConnection.evotor_user_id == "evo-001"
|
||||
).first()
|
||||
assert conn is not None
|
||||
assert conn.api_token == data["token"]
|
||||
|
||||
mock_task.delay.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("web.routes.evotor_webhooks.send_email_task")
|
||||
async def test_user_create_links_existing_user_by_email(mock_task, client, override_db, user_factory, monkeypatch):
|
||||
monkeypatch.setattr("web.routes.evotor_webhooks.settings.EVOTOR_WEBHOOK_SECRET", WEBHOOK_SECRET)
|
||||
monkeypatch.setattr("web.routes.evotor_webhooks.settings.INVITE_EXPIRE_HOURS", 48)
|
||||
monkeypatch.setattr("web.routes.evotor_webhooks.settings.BASE_URL", "http://test")
|
||||
|
||||
existing = user_factory.create(email="existing@test.com")
|
||||
assert existing.evotor_user_id is None
|
||||
|
||||
payload = {
|
||||
"userId": "evo-link-001",
|
||||
"customField": json.dumps({"email": "existing@test.com"}),
|
||||
}
|
||||
resp = await client.post("/user/create", json=payload, headers=auth_headers())
|
||||
assert resp.status_code == 200
|
||||
|
||||
override_db.refresh(existing)
|
||||
assert existing.evotor_user_id == "evo-link-001"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_user_create_wrong_secret(client, monkeypatch):
|
||||
monkeypatch.setattr("web.routes.evotor_webhooks.settings.EVOTOR_WEBHOOK_SECRET", WEBHOOK_SECRET)
|
||||
resp = await client.post("/user/create", json={"userId": "x"}, headers={"Authorization": "Bearer wrong"})
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_user_create_missing_user_id(client, monkeypatch):
|
||||
monkeypatch.setattr("web.routes.evotor_webhooks.settings.EVOTOR_WEBHOOK_SECRET", "")
|
||||
resp = await client.post("/user/create", json={"customField": "{}"})
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("web.routes.evotor_webhooks.send_email_task")
|
||||
async def test_user_create_no_secret_dev_mode(mock_task, client, override_db, monkeypatch):
|
||||
monkeypatch.setattr("web.routes.evotor_webhooks.settings.EVOTOR_WEBHOOK_SECRET", "")
|
||||
monkeypatch.setattr("web.routes.evotor_webhooks.settings.INVITE_EXPIRE_HOURS", 48)
|
||||
monkeypatch.setattr("web.routes.evotor_webhooks.settings.BASE_URL", "http://test")
|
||||
|
||||
resp = await client.post("/user/create", json={"userId": "evo-dev-001"})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["userId"] == "evo-dev-001"
|
||||
|
||||
|
||||
# ── /user/verify ──────────────────────────────────────────────────────────────
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_user_verify_success(client, override_db, user_factory, monkeypatch):
|
||||
monkeypatch.setattr("web.routes.evotor_webhooks.settings.EVOTOR_WEBHOOK_SECRET", WEBHOOK_SECRET)
|
||||
user = user_factory.create(evotor_user_id="evo-verify-001")
|
||||
conn = EvotorConnection(
|
||||
user_id=user.id,
|
||||
evotor_user_id="evo-verify-001",
|
||||
access_token="evotor-access-token",
|
||||
api_token="my-api-token-xyz",
|
||||
)
|
||||
override_db.add(conn)
|
||||
override_db.commit()
|
||||
|
||||
resp = await client.post("/user/verify", json={
|
||||
"userId": "evo-verify-001",
|
||||
"username": user.email,
|
||||
"password": "testpass123",
|
||||
}, headers=auth_headers())
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["userId"] == "evo-verify-001"
|
||||
assert "token" in data
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_user_verify_wrong_password(client, user_factory, monkeypatch):
|
||||
monkeypatch.setattr("web.routes.evotor_webhooks.settings.EVOTOR_WEBHOOK_SECRET", WEBHOOK_SECRET)
|
||||
user = user_factory.create()
|
||||
resp = await client.post("/user/verify", json={
|
||||
"userId": "x",
|
||||
"username": user.email,
|
||||
"password": "wrongpass",
|
||||
}, headers=auth_headers())
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_user_verify_suspended(client, user_factory, monkeypatch):
|
||||
monkeypatch.setattr("web.routes.evotor_webhooks.settings.EVOTOR_WEBHOOK_SECRET", WEBHOOK_SECRET)
|
||||
user = user_factory.create(status=UserStatusEnum.suspended)
|
||||
resp = await client.post("/user/verify", json={
|
||||
"userId": "x",
|
||||
"username": user.email,
|
||||
"password": "testpass123",
|
||||
}, headers=auth_headers())
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_user_verify_no_password_hash(client, user_factory, monkeypatch):
|
||||
monkeypatch.setattr("web.routes.evotor_webhooks.settings.EVOTOR_WEBHOOK_SECRET", WEBHOOK_SECRET)
|
||||
user = user_factory.create(password_hash=None)
|
||||
resp = await client.post("/user/verify", json={
|
||||
"userId": "x",
|
||||
"username": user.email,
|
||||
"password": "anything",
|
||||
}, headers=auth_headers())
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
# ── /user/token ───────────────────────────────────────────────────────────────
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_user_token_updates_connection(client, override_db, user_factory, monkeypatch):
|
||||
monkeypatch.setattr("web.routes.evotor_webhooks.settings.EVOTOR_WEBHOOK_SECRET", WEBHOOK_SECRET)
|
||||
user = user_factory.create(evotor_user_id="evo-token-001")
|
||||
old_conn = EvotorConnection(
|
||||
user_id=user.id,
|
||||
evotor_user_id="evo-token-001",
|
||||
access_token="old-token",
|
||||
api_token="api-tok",
|
||||
)
|
||||
override_db.add(old_conn)
|
||||
override_db.commit()
|
||||
|
||||
resp = await client.post("/user/token", json={
|
||||
"userId": "evo-token-001",
|
||||
"token": "new-evotor-token-xyz",
|
||||
}, headers=auth_headers())
|
||||
assert resp.status_code == 200
|
||||
assert resp.json() == {}
|
||||
|
||||
override_db.refresh(old_conn)
|
||||
assert old_conn.access_token == "new-evotor-token-xyz"
|
||||
assert old_conn.is_online is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_user_token_unknown_user(client, monkeypatch):
|
||||
monkeypatch.setattr("web.routes.evotor_webhooks.settings.EVOTOR_WEBHOOK_SECRET", WEBHOOK_SECRET)
|
||||
resp = await client.post("/user/token", json={
|
||||
"userId": "does-not-exist",
|
||||
"token": "some-token",
|
||||
}, headers=auth_headers())
|
||||
assert resp.status_code == 404
|
||||
Reference in New Issue
Block a user