tests/modules/auth/test_admin_users.py - master - systemsstrategyy/Treker

tests/modules/auth/test_admin_users.py
178 строк7 KB
SystemsStrategy
Синхронизация с актуальной линией разработки (август 2026)
06 авг 2026, 12:52
06 авг 2026, 12:5219af6aa
"""HTTP-тесты admin-управления юзерами: GET /auth/users/manage + PATCH /auth/users/{uid}.

Только admin (permission `user:manage`) может смотреть расширенный список юзеров
(с ролью и is_active) и менять роль/активность других юзеров своей организации.

Бизнес-правила, которые здесь фиксируются:
- admin НЕ может менять свой аккаунт (self-lockout guard) → 400;
- юзера чужой организации не видно (tenant-isolation) → 404, не 403;
- не-admin → 403.
"""

from __future__ import annotations

from datetime import UTC, datetime
from typing import Any

import pytest
from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

ADMIN_PAYLOAD = {
    "email": "owner@example.ru",
    "password": "supersecret-12345",
    "full_name": "Владелец Орг",
    "organization_name": "strategiya-sistem",
    "consent_accepted": True,
}


async def _seed_user(
    db: AsyncSession,
    organization_id: int,
    *,
    email: str,
    role: str = "member",
    is_active: bool = True,
) -> Any:
    """Создаёт доп. юзера прямым ORM-insert (в тестах допустимо)."""
    from api.modules.auth.models import User

    user = User(
        organization_id=organization_id,
        email=email,
        password_hash="$2b$12$dummybcryptdummybcryptdummybcryptdummybcrypt",  # noqa: S106
        full_name="Коллега",
        role=role,
        is_active=is_active,
        email_verified_at=datetime.now(UTC),
    )
    db.add(user)
    await db.flush()
    return user


async def _admin_org_id(db: AsyncSession) -> int:
    """organization_id зарегистрированного admin'а."""
    from api.modules.auth.models import User

    return await db.scalar(select(User.organization_id).where(User.email == ADMIN_PAYLOAD["email"]))


@pytest.mark.asyncio
async def test_list_manage_returns_role_and_active(
    auth_client: AsyncClient, db_session: AsyncSession, register_verify_login
) -> None:
    """GET /auth/users/manage отдаёт role и is_active (для admin-таблицы)."""
    await register_verify_login(auth_client, ADMIN_PAYLOAD)
    org_id = await _admin_org_id(db_session)
    await _seed_user(db_session, org_id, email="member@example.ru", role="member")

    resp = await auth_client.get("/api/v1/auth/users/manage")

    assert resp.status_code == 200, resp.text
    users = {u["email"]: u for u in resp.json()}
    assert users[ADMIN_PAYLOAD["email"]]["role"] == "admin"
    assert users["member@example.ru"]["role"] == "member"
    assert users["member@example.ru"]["is_active"] is True


@pytest.mark.asyncio
async def test_patch_changes_role(
    auth_client: AsyncClient, db_session: AsyncSession, register_verify_login
) -> None:
    """admin меняет роль другого юзера → 200 + обновлённая роль."""
    await register_verify_login(auth_client, ADMIN_PAYLOAD)
    org_id = await _admin_org_id(db_session)
    member = await _seed_user(db_session, org_id, email="m@example.ru", role="member")

    resp = await auth_client.patch(f"/api/v1/auth/users/{member.uid}", json={"role": "teamlead"})

    assert resp.status_code == 200, resp.text
    assert resp.json()["role"] == "teamlead"


@pytest.mark.asyncio
async def test_patch_deactivates_user(
    auth_client: AsyncClient, db_session: AsyncSession, register_verify_login
) -> None:
    """admin деактивирует юзера → is_active=False."""
    await register_verify_login(auth_client, ADMIN_PAYLOAD)
    org_id = await _admin_org_id(db_session)
    member = await _seed_user(db_session, org_id, email="m2@example.ru")

    resp = await auth_client.patch(f"/api/v1/auth/users/{member.uid}", json={"is_active": False})

    assert resp.status_code == 200, resp.text
    assert resp.json()["is_active"] is False


@pytest.mark.asyncio
async def test_patch_self_forbidden(
    auth_client: AsyncClient, db_session: AsyncSession, register_verify_login
) -> None:
    """admin НЕ может менять свой аккаунт (self-lockout guard) → 400."""
    await register_verify_login(auth_client, ADMIN_PAYLOAD)
    me = (await auth_client.get("/api/v1/auth/me")).json()

    resp = await auth_client.patch(f"/api/v1/auth/users/{me['uid']}", json={"role": "member"})

    assert resp.status_code == 400, resp.text


@pytest.mark.asyncio
async def test_patch_cross_org_returns_404(
    auth_client: AsyncClient, db_session: AsyncSession, register_verify_login
) -> None:
    """Юзер чужой организации не виден admin'у — 404 (не 403, чтобы не
    подтверждать существование чужого аккаунта)."""
    await register_verify_login(auth_client, ADMIN_PAYLOAD)

    from api.modules.auth.models import Organization

    other_org = Organization(name="Other", slug="other-org")
    db_session.add(other_org)
    await db_session.flush()
    outsider = await _seed_user(db_session, other_org.id, email="outsider@other.ru", role="member")

    resp = await auth_client.patch(f"/api/v1/auth/users/{outsider.uid}", json={"role": "admin"})

    assert resp.status_code == 404, resp.text


@pytest.mark.asyncio
async def test_patch_requires_admin(
    auth_client: AsyncClient, db_session: AsyncSession, register_verify_login
) -> None:
    """Не-admin (member) не может управлять юзерами → 403."""
    # Регистрируем admin, затем «понижаем» залогиненного актора до member —
    # имитируем member'а, ходящего на admin-эндпоинт.
    await register_verify_login(auth_client, ADMIN_PAYLOAD)
    from api.modules.auth.models import User

    actor = await db_session.scalar(select(User).where(User.email == ADMIN_PAYLOAD["email"]))
    target = await _seed_user(db_session, actor.organization_id, email="t@example.ru")
    actor.role = "member"
    await db_session.flush()

    resp = await auth_client.patch(f"/api/v1/auth/users/{target.uid}", json={"role": "teamlead"})

    assert resp.status_code == 403, resp.text


@pytest.mark.asyncio
async def test_list_manage_requires_admin(
    auth_client: AsyncClient, db_session: AsyncSession, register_verify_login
) -> None:
    """Не-admin не видит admin-список юзеров → 403."""
    await register_verify_login(auth_client, ADMIN_PAYLOAD)
    from api.modules.auth.models import User

    actor = await db_session.scalar(select(User).where(User.email == ADMIN_PAYLOAD["email"]))
    actor.role = "member"
    await db_session.flush()

    resp = await auth_client.get("/api/v1/auth/users/manage")

    assert resp.status_code == 403, resp.text