182 lines
7.4 KiB
Python
182 lines
7.4 KiB
Python
|
|
"""Integration tests for admin panel routes."""
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from web.models.user import User, UserRoleEnum, UserStatusEnum
|
||
|
|
|
||
|
|
|
||
|
|
def _set_session(client, user_id: int):
|
||
|
|
"""Inject a session cookie so the client appears logged in as user_id."""
|
||
|
|
client.cookies.set("session", "") # will be overwritten by actual login
|
||
|
|
# We inject directly into the app's session via a helper request
|
||
|
|
# The simplest approach: use the login endpoint to set the real session cookie
|
||
|
|
return user_id
|
||
|
|
|
||
|
|
|
||
|
|
async def _login(client, user):
|
||
|
|
"""Log in as user via the /login endpoint to get a real session cookie."""
|
||
|
|
resp = await client.post("/login", data={
|
||
|
|
"email": user.email,
|
||
|
|
"password": "testpass123",
|
||
|
|
}, follow_redirects=False)
|
||
|
|
assert resp.status_code == 303, f"Login failed: {resp.text}"
|
||
|
|
|
||
|
|
|
||
|
|
# ── Access control ────────────────────────────────────────────────────────────
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_admin_users_requires_auth(client):
|
||
|
|
resp = await client.get("/admin/users", follow_redirects=False)
|
||
|
|
# Unauthenticated → redirect to login
|
||
|
|
assert resp.status_code in (302, 303, 307)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_admin_users_requires_admin_role(client, active_user):
|
||
|
|
await _login(client, active_user)
|
||
|
|
resp = await client.get("/admin/users", follow_redirects=False)
|
||
|
|
# Regular user → redirect (not admin)
|
||
|
|
assert resp.status_code in (302, 303, 307)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_admin_users_accessible_by_admin(client, admin_user):
|
||
|
|
await _login(client, admin_user)
|
||
|
|
resp = await client.get("/admin/users")
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert "Пользователи" in resp.text
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_admin_users_accessible_by_system(client, system_user):
|
||
|
|
await _login(client, system_user)
|
||
|
|
resp = await client.get("/admin/users")
|
||
|
|
assert resp.status_code == 200
|
||
|
|
|
||
|
|
|
||
|
|
# ── User list + filters ───────────────────────────────────────────────────────
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_admin_users_shows_all_users(client, admin_user, user_factory):
|
||
|
|
extra = user_factory.create(email="findme@test.com")
|
||
|
|
await _login(client, admin_user)
|
||
|
|
resp = await client.get("/admin/users")
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert extra.email in resp.text
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_admin_users_search_filter(client, admin_user, user_factory):
|
||
|
|
target = user_factory.create(email="searchable@test.com")
|
||
|
|
await _login(client, admin_user)
|
||
|
|
resp = await client.get("/admin/users?search=searchable")
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert target.email in resp.text
|
||
|
|
|
||
|
|
|
||
|
|
# ── User detail ───────────────────────────────────────────────────────────────
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_admin_user_detail(client, admin_user, active_user):
|
||
|
|
await _login(client, admin_user)
|
||
|
|
resp = await client.get(f"/admin/users/{active_user.id}")
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert active_user.email in resp.text
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_admin_user_detail_not_found(client, admin_user):
|
||
|
|
await _login(client, admin_user)
|
||
|
|
resp = await client.get("/admin/users/99999", follow_redirects=False)
|
||
|
|
assert resp.status_code in (302, 303, 307)
|
||
|
|
|
||
|
|
|
||
|
|
# ── Activate / suspend ────────────────────────────────────────────────────────
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_admin_activate_user(client, admin_user, user_factory, override_db):
|
||
|
|
target = user_factory.create(status=UserStatusEnum.suspended)
|
||
|
|
await _login(client, admin_user)
|
||
|
|
resp = await client.post(f"/admin/users/{target.id}/activate", follow_redirects=False)
|
||
|
|
assert resp.status_code == 303
|
||
|
|
override_db.refresh(target)
|
||
|
|
assert target.status == UserStatusEnum.active
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_admin_suspend_user(client, admin_user, active_user, override_db):
|
||
|
|
await _login(client, admin_user)
|
||
|
|
resp = await client.post(f"/admin/users/{active_user.id}/suspend", follow_redirects=False)
|
||
|
|
assert resp.status_code == 303
|
||
|
|
override_db.refresh(active_user)
|
||
|
|
assert active_user.status == UserStatusEnum.suspended
|
||
|
|
|
||
|
|
|
||
|
|
# ── Delete (system only) ──────────────────────────────────────────────────────
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_admin_delete_user_by_system(client, system_user, user_factory, override_db):
|
||
|
|
target = user_factory.create()
|
||
|
|
target_id = target.id
|
||
|
|
await _login(client, system_user)
|
||
|
|
resp = await client.post(f"/admin/users/{target_id}/delete", follow_redirects=False)
|
||
|
|
assert resp.status_code == 303
|
||
|
|
assert override_db.get(User, target_id) is None
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_admin_delete_blocked_for_admin_role(client, admin_user, active_user, override_db):
|
||
|
|
target_id = active_user.id
|
||
|
|
await _login(client, admin_user)
|
||
|
|
resp = await client.post(f"/admin/users/{target_id}/delete", follow_redirects=False)
|
||
|
|
assert resp.status_code == 303
|
||
|
|
# Admin cannot delete — user still exists
|
||
|
|
assert override_db.get(User, target_id) is not None
|
||
|
|
|
||
|
|
|
||
|
|
# ── Reset password / send invite ──────────────────────────────────────────────
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_admin_reset_password_generates_token(client, admin_user, active_user, override_db):
|
||
|
|
from unittest.mock import patch
|
||
|
|
with patch("web.routes.admin.send_email_task") as mock_task:
|
||
|
|
await _login(client, admin_user)
|
||
|
|
resp = await client.post(
|
||
|
|
f"/admin/users/{active_user.id}/reset-password", follow_redirects=False
|
||
|
|
)
|
||
|
|
assert resp.status_code == 303
|
||
|
|
override_db.refresh(active_user)
|
||
|
|
assert active_user.password_reset_token is not None
|
||
|
|
mock_task.delay.assert_called_once()
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_admin_send_invite(client, admin_user, active_user, override_db):
|
||
|
|
from unittest.mock import patch
|
||
|
|
with patch("web.routes.admin.send_email_task") as mock_task:
|
||
|
|
await _login(client, admin_user)
|
||
|
|
resp = await client.post(
|
||
|
|
f"/admin/users/{active_user.id}/send-invite", follow_redirects=False
|
||
|
|
)
|
||
|
|
assert resp.status_code == 303
|
||
|
|
override_db.refresh(active_user)
|
||
|
|
assert active_user.invite_token is not None
|
||
|
|
mock_task.delay.assert_called_once()
|
||
|
|
|
||
|
|
|
||
|
|
# ── Roles page (system only) ──────────────────────────────────────────────────
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_admin_roles_accessible_by_system(client, system_user):
|
||
|
|
await _login(client, system_user)
|
||
|
|
resp = await client.get("/admin/roles")
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert "Роли" in resp.text
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_admin_roles_blocked_for_admin(client, admin_user):
|
||
|
|
await _login(client, admin_user)
|
||
|
|
resp = await client.get("/admin/roles", follow_redirects=False)
|
||
|
|
# Admin is redirected away from roles page
|
||
|
|
assert resp.status_code in (302, 303, 307)
|