d241c07f54
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
629 lines
32 KiB
Python
629 lines
32 KiB
Python
"""API администрирования `/admin/*` (spec users-admin, audit-trail; design D-8; Review Focus 5 фазы 2).
|
||
|
||
Пользователи (создание, изменение, деактивация), назначение ролей, матрица «роль × право», аудит и защита `LAST_ADMIN`:
|
||
последнее право `USERS_ADMIN` у активных пользователей нельзя снять ни деактивацией, ни заменой ролей, ни снятием
|
||
права с роли. Новые пользователи — с доменом `example.com`: `EmailStr` не принимает зарезервированный `.local`
|
||
dev-пользователей.
|
||
"""
|
||
|
||
import asyncio
|
||
import uuid
|
||
from collections.abc import Sequence
|
||
from typing import Any
|
||
|
||
import httpx
|
||
import pytest
|
||
from sqlalchemy import func, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||
|
||
from src.api.dependencies.audit import RequestAudit
|
||
from src.common.exceptions import DomainError
|
||
from src.models.access import AppUser, Permission, Role, RolePermission, UserRole
|
||
from src.models.audit_ex import AuditEvent
|
||
from src.models.directories import RegionalOffice
|
||
from src.models.enums import AuditAction, AuditSection
|
||
from src.repositories import users as users_repo
|
||
from src.services.audit import AuditService
|
||
from src.services.users_admin import UsersAdminService
|
||
from tests.conftest import AuthClientFactory
|
||
from tests.factories import make_user
|
||
|
||
HTTP_OK = 200
|
||
HTTP_CREATED = 201
|
||
HTTP_FORBIDDEN = 403
|
||
HTTP_NOT_FOUND = 404
|
||
HTTP_CONFLICT = 409
|
||
HTTP_UNPROCESSABLE = 422
|
||
|
||
ADMIN = "admin@armka.local"
|
||
USERS = "/admin/users"
|
||
ROSTOV = "Ростовский региональный филиал"
|
||
HEAD_OFFICE = "Головной офис"
|
||
PERMISSIONS_TOTAL = 16 # seed: 16 кодов прав (§10)
|
||
EXPERT_PERMISSIONS = ["CONSOLIDATION_VIEW", "DIRECTORY_VIEW", "DOCUMENT_VIEW", "EX_VIEW", "KFO_VIEW"]
|
||
|
||
|
||
async def _office(session: AsyncSession, name: str) -> RegionalOffice:
|
||
return (await session.execute(select(RegionalOffice).where(RegionalOffice.name == name))).scalar_one()
|
||
|
||
|
||
async def _user_id(session: AsyncSession, email: str) -> uuid.UUID:
|
||
return (await session.execute(select(AppUser.id).where(AppUser.email == email))).scalar_one()
|
||
|
||
|
||
async def _role_codes(session: AsyncSession, user_id: uuid.UUID) -> list[str]:
|
||
stmt = select(Role.code).join(UserRole, UserRole.role_id == Role.id).where(UserRole.user_id == user_id)
|
||
return sorted((await session.execute(stmt)).scalars().all())
|
||
|
||
|
||
async def _events(session: AsyncSession, action: AuditAction) -> Sequence[AuditEvent]:
|
||
stmt = select(AuditEvent).where(AuditEvent.action == action).order_by(AuditEvent.occurred_at, AuditEvent.id)
|
||
return (await session.execute(stmt)).scalars().all()
|
||
|
||
|
||
async def _create(client: httpx.AsyncClient, **body: Any) -> dict[str, Any]:
|
||
response = await client.post(USERS, json=body)
|
||
assert response.status_code == HTTP_CREATED, response.text
|
||
created: dict[str, Any] = response.json()
|
||
return created
|
||
|
||
|
||
def _detail_code(response: httpx.Response) -> str:
|
||
code: str = response.json()["detail"]["code"]
|
||
return code
|
||
|
||
|
||
# --- Пользователи ------------------------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_create_user_with_role_gives_role_permissions_in_me(
|
||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||
) -> None:
|
||
admin = await auth_client(ADMIN)
|
||
rostov = await _office(seeded, ROSTOV)
|
||
created = await _create(
|
||
admin,
|
||
email=" Ivan.Petrov@Example.com ",
|
||
full_name="Иван Петров",
|
||
regional_office_id=str(rostov.id),
|
||
employee_no="E-1",
|
||
business_segment="DKB",
|
||
roles=["KA_RF"],
|
||
)
|
||
assert created["email"] == "Ivan.Petrov@example.com" # регистр локальной части сохранён (EmailStr: домен строчными)
|
||
assert created["roles"] == ["KA_RF"]
|
||
assert created["is_active"] is True
|
||
assert created["regional_office"] == {
|
||
"id": str(rostov.id),
|
||
"name": ROSTOV,
|
||
"is_head_office": False,
|
||
"is_active": True,
|
||
}
|
||
assert set(created) == {
|
||
"id",
|
||
"email",
|
||
"full_name",
|
||
"employee_no",
|
||
"regional_office",
|
||
"business_segment",
|
||
"is_active",
|
||
"roles",
|
||
"created_at",
|
||
"updated_at",
|
||
}
|
||
user_id = uuid.UUID(created["id"])
|
||
assert await _role_codes(seeded, user_id) == ["KA_RF"]
|
||
user = await seeded.get(AppUser, user_id)
|
||
assert user is not None and user.external_id is None and user.created_at is not None
|
||
|
||
events = await _events(seeded, AuditAction.USER_CHANGE)
|
||
assert len(events) == 1
|
||
event = events[0]
|
||
assert (event.section, event.object_type, event.object_id) == (AuditSection.USERS, "app_user", user_id)
|
||
assert event.user_id == await _user_id(seeded, ADMIN)
|
||
assert event.after_value == {
|
||
"email": "Ivan.Petrov@example.com",
|
||
"full_name": "Иван Петров",
|
||
"employee_no": "E-1",
|
||
"regional_office_id": str(rostov.id),
|
||
"business_segment": "DKB",
|
||
"roles": ["KA_RF"],
|
||
}
|
||
|
||
reference = (await (await auth_client("ka_rf@armka.local")).get("/me")).json()
|
||
me = await (await auth_client("ivan.petrov@example.com")).get("/me")
|
||
assert me.status_code == HTTP_OK, me.text
|
||
assert me.json()["roles"] == ["KA_RF"]
|
||
assert me.json()["permissions"] == reference["permissions"]
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_create_user_email_in_other_case_conflicts(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
admin = await auth_client(ADMIN)
|
||
office_id = str((await _office(seeded, HEAD_OFFICE)).id)
|
||
await _create(admin, email="anna@example.com", full_name="Анна", regional_office_id=office_id)
|
||
response = await admin.post(
|
||
USERS, json={"email": "ANNA@Example.COM", "full_name": "Анна 2", "regional_office_id": office_id}
|
||
)
|
||
assert response.status_code == HTTP_CONFLICT
|
||
assert _detail_code(response) == "EMAIL_EXISTS"
|
||
total = (await seeded.execute(select(func.count()).where(func.lower(AppUser.email) == "anna@example.com"))).scalar()
|
||
assert total == 1
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_create_user_validation(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
admin = await auth_client(ADMIN)
|
||
office_id = str((await _office(seeded, HEAD_OFFICE)).id)
|
||
base = {"email": "v@example.com", "full_name": "В", "regional_office_id": office_id}
|
||
|
||
unknown_role = await admin.post(USERS, json={**base, "roles": ["KA_RF", "NO_SUCH_ROLE"]})
|
||
assert unknown_role.status_code == HTTP_UNPROCESSABLE
|
||
assert unknown_role.json()["detail"]["code"] == "UNKNOWN_ROLE"
|
||
assert unknown_role.json()["detail"]["roles"] == ["NO_SUCH_ROLE"]
|
||
|
||
unknown_office = await admin.post(USERS, json={**base, "regional_office_id": str(uuid.uuid4())})
|
||
assert unknown_office.status_code == HTTP_UNPROCESSABLE
|
||
assert _detail_code(unknown_office) == "OFFICE_NOT_FOUND"
|
||
|
||
for body in (
|
||
{**base, "external_id": "raisa-1"}, # external_id через API не задаётся
|
||
{**base, "email": "not-an-email"},
|
||
{**base, "full_name": " "},
|
||
):
|
||
response = await admin.post(USERS, json=body)
|
||
assert response.status_code == HTTP_UNPROCESSABLE, body
|
||
assert (await seeded.execute(select(AppUser).where(AppUser.email == "v@example.com"))).first() is None
|
||
assert await _events(seeded, AuditAction.USER_CHANGE) == []
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_list_users_search_filter_and_page(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
office = await _office(seeded, ROSTOV)
|
||
await make_user(seeded, email="Zoya_Search@example.com", office=office, roles=["KA_RF", "EXPERT"], full_name="Зоя")
|
||
await make_user(seeded, email="off@example.com", office=office, full_name="Зоя Отключённая", is_active=False)
|
||
await make_user(seeded, email="zoyaXs@example.com", office=office) # совпал бы с `zoya_s`, если `_` — шаблон LIKE
|
||
admin = await auth_client(ADMIN)
|
||
|
||
everyone = (await admin.get(USERS)).json()
|
||
assert set(everyone) == {"items", "total", "limit", "offset"}
|
||
assert everyone["total"] == len(everyone["items"]) >= 10 # noqa: PLR2004 - 7 dev-пользователей + 3 новых
|
||
|
||
by_email = (await admin.get(USERS, params={"q": "zoya_s"})).json() # `_` ищется буквально, без регистра
|
||
assert [item["email"] for item in by_email["items"]] == ["Zoya_Search@example.com"] # `zoyaXs@…` не найден
|
||
assert by_email["items"][0]["roles"] == ["EXPERT", "KA_RF"]
|
||
assert by_email["items"][0]["regional_office"]["name"] == ROSTOV
|
||
|
||
by_name = (await admin.get(USERS, params={"q": "зоя"})).json()
|
||
assert by_name["total"] == 2 # noqa: PLR2004
|
||
inactive = (await admin.get(USERS, params={"q": "зоя", "is_active": "false"})).json()
|
||
assert [item["email"] for item in inactive["items"]] == ["off@example.com"]
|
||
assert inactive["items"][0]["roles"] == []
|
||
|
||
page = (await admin.get(USERS, params={"limit": 2, "offset": 1})).json()
|
||
assert (len(page["items"]), page["limit"], page["offset"], page["total"]) == (2, 2, 1, everyone["total"])
|
||
assert page["items"] == everyone["items"][1:3]
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_patch_user_fields_audit_and_noop(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
rostov = await _office(seeded, ROSTOV)
|
||
head = await _office(seeded, HEAD_OFFICE)
|
||
user = await make_user(seeded, email="p@example.com", office=rostov, roles=["KA_RF"], full_name="Пётр")
|
||
admin = await auth_client(ADMIN)
|
||
|
||
response = await admin.patch(
|
||
f"{USERS}/{user.id}", json={"full_name": "Пётр Петров", "regional_office_id": str(head.id)}
|
||
)
|
||
assert response.status_code == HTTP_OK, response.text
|
||
body = response.json()
|
||
assert body["full_name"] == "Пётр Петров"
|
||
assert body["regional_office"]["id"] == str(head.id)
|
||
assert body["roles"] == ["KA_RF"]
|
||
events = await _events(seeded, AuditAction.USER_CHANGE)
|
||
assert len(events) == 1
|
||
assert (events[0].section, events[0].object_type, events[0].object_id) == (AuditSection.USERS, "app_user", user.id)
|
||
assert events[0].before_value == {"full_name": "Пётр", "regional_office_id": str(rostov.id)}
|
||
assert events[0].after_value == {"full_name": "Пётр Петров", "regional_office_id": str(head.id)}
|
||
assert events[0].changed_field is None
|
||
|
||
single = await admin.patch(f"{USERS}/{user.id}", json={"employee_no": "E-7"})
|
||
assert single.status_code == HTTP_OK
|
||
last = (await _events(seeded, AuditAction.USER_CHANGE))[-1]
|
||
assert (last.changed_field, last.before_value, last.after_value) == (
|
||
"employee_no",
|
||
{"employee_no": None},
|
||
{"employee_no": "E-7"},
|
||
)
|
||
|
||
noop = await admin.patch(f"{USERS}/{user.id}", json={"full_name": "Пётр Петров", "employee_no": "E-7"})
|
||
assert noop.status_code == HTTP_OK
|
||
assert len(await _events(seeded, AuditAction.USER_CHANGE)) == 2 # noqa: PLR2004 - no-op не пишет аудит
|
||
|
||
for bad in ({"email": "x@example.com"}, {"full_name": None}, {"is_active": None}, {"regional_office_id": None}):
|
||
assert (await admin.patch(f"{USERS}/{user.id}", json=bad)).status_code == HTTP_UNPROCESSABLE, bad
|
||
office_missing = await admin.patch(f"{USERS}/{user.id}", json={"regional_office_id": str(uuid.uuid4())})
|
||
assert office_missing.status_code == HTTP_UNPROCESSABLE
|
||
assert _detail_code(office_missing) == "OFFICE_NOT_FOUND"
|
||
assert (await admin.patch(f"{USERS}/{uuid.uuid4()}", json={"full_name": "Х"})).status_code == HTTP_NOT_FOUND
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_deactivated_user_loses_access(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
user_id = await _user_id(seeded, "expert@armka.local")
|
||
expert = await auth_client("expert@armka.local")
|
||
admin = await auth_client(ADMIN)
|
||
response = await admin.patch(f"{USERS}/{user_id}", json={"is_active": False})
|
||
assert response.status_code == HTTP_OK
|
||
assert response.json()["is_active"] is False
|
||
assert (await expert.get("/me")).status_code == HTTP_FORBIDDEN
|
||
event = (await _events(seeded, AuditAction.USER_CHANGE))[-1]
|
||
assert (event.changed_field, event.before_value, event.after_value) == (
|
||
"is_active",
|
||
{"is_active": True},
|
||
{"is_active": False},
|
||
)
|
||
|
||
|
||
# --- Назначение ролей --------------------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_put_roles_replaces_set_with_audit(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
user = await make_user(seeded, email="r@example.com", office=await _office(seeded, ROSTOV), roles=["KA_RF"])
|
||
admin = await auth_client(ADMIN)
|
||
|
||
response = await admin.put(f"{USERS}/{user.id}/roles", json={"roles": ["KA_GO", "EXPERT", "KA_GO"]})
|
||
assert response.status_code == HTTP_OK, response.text
|
||
assert response.json()["roles"] == ["EXPERT", "KA_GO"]
|
||
assert await _role_codes(seeded, user.id) == ["EXPERT", "KA_GO"]
|
||
rows = (await seeded.execute(select(func.count()).where(UserRole.user_id == user.id))).scalar()
|
||
assert rows == 2 # noqa: PLR2004
|
||
|
||
events = await _events(seeded, AuditAction.USER_ROLE_CHANGE)
|
||
assert len(events) == 1
|
||
event = events[0]
|
||
assert (event.section, event.object_type, event.object_id) == (AuditSection.USERS, "app_user", user.id)
|
||
assert event.before_value == {"roles": ["KA_RF"]}
|
||
assert event.after_value == {"roles": ["KA_GO", "EXPERT"]}
|
||
|
||
same = await admin.put(f"{USERS}/{user.id}/roles", json={"roles": ["EXPERT", "KA_GO"]})
|
||
assert same.status_code == HTTP_OK
|
||
assert len(await _events(seeded, AuditAction.USER_ROLE_CHANGE)) == 1 # тот же набор — без аудита
|
||
|
||
cleared = await admin.put(f"{USERS}/{user.id}/roles", json={"roles": []})
|
||
assert cleared.status_code == HTTP_OK
|
||
assert await _role_codes(seeded, user.id) == []
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_put_roles_rejects_unknown_and_inactive_role(
|
||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||
) -> None:
|
||
user = await make_user(seeded, email="u@example.com", office=await _office(seeded, ROSTOV), roles=["KA_RF"])
|
||
seeded.add(Role(code="RETIRED", name="Упразднённая", is_system=False, is_active=False))
|
||
await seeded.flush()
|
||
admin = await auth_client(ADMIN)
|
||
|
||
for codes, bad in ((["KA_GO", "NOPE"], ["NOPE"]), (["RETIRED"], ["RETIRED"])):
|
||
response = await admin.put(f"{USERS}/{user.id}/roles", json={"roles": codes})
|
||
assert response.status_code == HTTP_UNPROCESSABLE
|
||
assert response.json()["detail"]["code"] == "UNKNOWN_ROLE"
|
||
assert response.json()["detail"]["roles"] == bad
|
||
assert await _role_codes(seeded, user.id) == ["KA_RF"]
|
||
missing = await admin.put(f"{USERS}/{uuid.uuid4()}/roles", json={"roles": ["KA_GO"]})
|
||
assert missing.status_code == HTTP_NOT_FOUND
|
||
|
||
|
||
# --- LAST_ADMIN --------------------------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_last_admin_cannot_drop_own_admin_role(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
admin_id = await _user_id(seeded, ADMIN)
|
||
admin = await auth_client(ADMIN)
|
||
response = await admin.put(f"{USERS}/{admin_id}/roles", json={"roles": ["EXPERT"]})
|
||
assert response.status_code == HTTP_CONFLICT
|
||
assert _detail_code(response) == "LAST_ADMIN"
|
||
assert await _role_codes(seeded, admin_id) == ["ADMIN"]
|
||
assert await _events(seeded, AuditAction.USER_ROLE_CHANGE) == []
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_inactive_user_or_inactive_role_is_not_an_admin_source(
|
||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||
) -> None:
|
||
head = await _office(seeded, HEAD_OFFICE)
|
||
await make_user(seeded, email="sleeping.admin@example.com", office=head, roles=["ADMIN"], is_active=False)
|
||
retired = Role(code="RETIRED_ADMIN", name="Бывшая админская", is_system=False, is_active=False)
|
||
seeded.add(retired)
|
||
await seeded.flush()
|
||
users_admin = (await seeded.execute(select(Permission.id).where(Permission.code == "USERS_ADMIN"))).scalar_one()
|
||
seeded.add(RolePermission(role_id=retired.id, permission_id=users_admin))
|
||
await make_user(seeded, email="retired.admin@example.com", office=head, roles=["RETIRED_ADMIN"])
|
||
|
||
admin_id = await _user_id(seeded, ADMIN)
|
||
admin = await auth_client(ADMIN)
|
||
response = await admin.put(f"{USERS}/{admin_id}/roles", json={"roles": ["EXPERT"]})
|
||
assert response.status_code == HTTP_CONFLICT
|
||
assert _detail_code(response) == "LAST_ADMIN"
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_second_admin_allows_dropping_own_admin_role(
|
||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||
) -> None:
|
||
await make_user(
|
||
seeded, email="second.admin@example.com", office=await _office(seeded, HEAD_OFFICE), roles=["ADMIN"]
|
||
)
|
||
admin_id = await _user_id(seeded, ADMIN)
|
||
admin = await auth_client(ADMIN)
|
||
response = await admin.put(f"{USERS}/{admin_id}/roles", json={"roles": ["EXPERT"]})
|
||
assert response.status_code == HTTP_OK, response.text
|
||
assert await _role_codes(seeded, admin_id) == ["EXPERT"]
|
||
assert (await admin.get("/admin/roles")).status_code == HTTP_FORBIDDEN # права пересчитаны на следующем запросе
|
||
|
||
|
||
async def _grant_users_admin(session: AsyncSession, role_code: str, *, is_active: bool = True) -> None:
|
||
"""Даёт роли право `USERS_ADMIN` напрямую в БД (и задаёт `is_active` роли) — как после правки матрицы прав."""
|
||
role = (await session.execute(select(Role).where(Role.code == role_code))).scalar_one()
|
||
permission_id = (await session.execute(select(Permission.id).where(Permission.code == "USERS_ADMIN"))).scalar_one()
|
||
session.add(RolePermission(role_id=role.id, permission_id=permission_id))
|
||
role.is_active = is_active
|
||
await session.flush()
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_sole_admin_moved_to_active_role_with_users_admin_stays_admin(
|
||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||
) -> None:
|
||
"""Ветка `kept_roles`: оставшаяся активная роль даёт `USERS_ADMIN` — другого администратора не требуется (T12.1)."""
|
||
await _grant_users_admin(seeded, "EXPERT")
|
||
admin_id = await _user_id(seeded, ADMIN)
|
||
admin = await auth_client(ADMIN)
|
||
response = await admin.put(f"{USERS}/{admin_id}/roles", json={"roles": ["EXPERT"]})
|
||
assert response.status_code == HTTP_OK, response.text
|
||
assert await _role_codes(seeded, admin_id) == ["EXPERT"]
|
||
assert (await admin.get(USERS)).status_code == HTTP_OK # право USERS_ADMIN сохранено через EXPERT
|
||
assert len(await _events(seeded, AuditAction.USER_ROLE_CHANGE)) == 1
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_sole_admin_cannot_be_moved_to_inactive_role_with_users_admin(
|
||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||
) -> None:
|
||
"""Неактивная роль с `USERS_ADMIN` не даёт права и не назначается: набор ролей и администратор не меняются (T12.1).
|
||
|
||
Сервис отсекает её раньше проверки `LAST_ADMIN` — 422 `UNKNOWN_ROLE`; сам фильтр `is_active` в `kept_roles`-проверке
|
||
(`roles_grant_permission`) защитный и до неё через API недостижим, поэтому проверяется отдельно.
|
||
"""
|
||
await _grant_users_admin(seeded, "EXPERT", is_active=False)
|
||
admin_id = await _user_id(seeded, ADMIN)
|
||
admin = await auth_client(ADMIN)
|
||
response = await admin.put(f"{USERS}/{admin_id}/roles", json={"roles": ["EXPERT"]})
|
||
assert response.status_code == HTTP_UNPROCESSABLE, response.text
|
||
assert _detail_code(response) == "UNKNOWN_ROLE"
|
||
assert await _role_codes(seeded, admin_id) == ["ADMIN"]
|
||
assert await _events(seeded, AuditAction.USER_ROLE_CHANGE) == []
|
||
assert (await admin.get(USERS)).status_code == HTTP_OK
|
||
|
||
roles = {
|
||
role.code: role.id
|
||
for role in (await seeded.execute(select(Role).where(Role.code.in_(["ADMIN", "EXPERT"])))).scalars()
|
||
}
|
||
assert await users_repo.roles_grant_permission(seeded, {roles["ADMIN"]}, "USERS_ADMIN") is True
|
||
assert await users_repo.roles_grant_permission(seeded, {roles["EXPERT"]}, "USERS_ADMIN") is False # неактивная роль
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_last_admin_cannot_deactivate_self(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
"""Review Focus 5: последний активный с `USERS_ADMIN` не может деактивировать себя; при втором — может."""
|
||
admin_id = await _user_id(seeded, ADMIN)
|
||
admin = await auth_client(ADMIN)
|
||
response = await admin.patch(f"{USERS}/{admin_id}", json={"is_active": False, "full_name": "Не применится"})
|
||
assert response.status_code == HTTP_CONFLICT
|
||
assert _detail_code(response) == "LAST_ADMIN"
|
||
user = await seeded.get(AppUser, admin_id)
|
||
assert user is not None
|
||
await seeded.refresh(user)
|
||
assert user.is_active is True and user.full_name != "Не применится"
|
||
assert await _events(seeded, AuditAction.USER_CHANGE) == []
|
||
|
||
await make_user(
|
||
seeded, email="second.admin@example.com", office=await _office(seeded, HEAD_OFFICE), roles=["ADMIN"]
|
||
)
|
||
allowed = await admin.patch(f"{USERS}/{admin_id}", json={"is_active": False})
|
||
assert allowed.status_code == HTTP_OK, allowed.text
|
||
assert allowed.json()["is_active"] is False
|
||
assert (await admin.get(USERS)).status_code == HTTP_FORBIDDEN # отключённый пользователь — 403
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_deactivating_other_admin_respects_last_admin(
|
||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||
) -> None:
|
||
second = await make_user(
|
||
seeded, email="second.admin@example.com", office=await _office(seeded, HEAD_OFFICE), roles=["ADMIN"]
|
||
)
|
||
admin = await auth_client(ADMIN)
|
||
second_client = await auth_client("second.admin@example.com")
|
||
assert (await admin.patch(f"{USERS}/{second.id}", json={"is_active": False})).status_code == HTTP_OK
|
||
# второй отключён — первый снова последний
|
||
admin_id = await _user_id(seeded, ADMIN)
|
||
response = await admin.patch(f"{USERS}/{admin_id}", json={"is_active": False})
|
||
assert response.status_code == HTTP_CONFLICT
|
||
assert _detail_code(response) == "LAST_ADMIN"
|
||
assert (await second_client.get(USERS)).status_code == HTTP_FORBIDDEN
|
||
|
||
|
||
# --- Матрица прав ------------------------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_roles_and_permissions_listing(auth_client: AuthClientFactory) -> None:
|
||
admin = await auth_client(ADMIN)
|
||
permissions = (await admin.get("/admin/permissions")).json()
|
||
assert len(permissions) == PERMISSIONS_TOTAL
|
||
codes = [item["code"] for item in permissions]
|
||
assert codes == sorted(codes) and "USERS_ADMIN" in codes
|
||
users_admin = next(item for item in permissions if item["code"] == "USERS_ADMIN")
|
||
assert set(users_admin) == {"code", "object_type", "action", "name", "description"}
|
||
assert (users_admin["object_type"], users_admin["action"]) == ("USERS", "ADMIN")
|
||
|
||
roles = {item["code"]: item for item in (await admin.get("/admin/roles")).json()}
|
||
assert {"ADMIN", "EXPERT", "KA_RF", "KA_GO"} <= set(roles)
|
||
assert roles["EXPERT"]["permissions"] == EXPERT_PERMISSIONS
|
||
assert "USERS_ADMIN" in roles["ADMIN"]["permissions"]
|
||
assert set(roles["ADMIN"]) == {"code", "name", "description", "is_system", "is_active", "permissions"}
|
||
assert roles["ADMIN"]["is_system"] is True
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_role_permissions_change_reaches_me_and_audit(
|
||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||
) -> None:
|
||
admin = await auth_client(ADMIN)
|
||
expert = await auth_client("expert@armka.local")
|
||
assert "EXPORT" not in (await expert.get("/me")).json()["permissions"]
|
||
|
||
response = await admin.put("/admin/roles/EXPERT/permissions", json={"permissions": [*EXPERT_PERMISSIONS, "EXPORT"]})
|
||
assert response.status_code == HTTP_OK, response.text
|
||
assert response.json()["permissions"] == sorted([*EXPERT_PERMISSIONS, "EXPORT"])
|
||
|
||
role = (await seeded.execute(select(Role).where(Role.code == "EXPERT"))).scalar_one()
|
||
row = (
|
||
await seeded.execute(
|
||
select(RolePermission)
|
||
.join(Permission, Permission.id == RolePermission.permission_id)
|
||
.where(RolePermission.role_id == role.id, Permission.code == "EXPORT")
|
||
)
|
||
).scalar_one()
|
||
admin_id = await _user_id(seeded, ADMIN)
|
||
assert row.granted_by == admin_id and row.granted_at is not None
|
||
assert "EXPORT" in (await expert.get("/me")).json()["permissions"]
|
||
|
||
events = await _events(seeded, AuditAction.ROLE_PERMISSION_CHANGE)
|
||
assert len(events) == 1
|
||
event = events[0]
|
||
assert (event.section, event.object_type, event.object_id, event.user_id) == (
|
||
AuditSection.ACCESS,
|
||
"role",
|
||
role.id,
|
||
admin_id,
|
||
)
|
||
assert event.before_value == {"permissions": EXPERT_PERMISSIONS}
|
||
assert event.after_value == {"permissions": sorted([*EXPERT_PERMISSIONS, "EXPORT"])}
|
||
|
||
same = await admin.put("/admin/roles/EXPERT/permissions", json={"permissions": [*EXPERT_PERMISSIONS, "EXPORT"]})
|
||
assert same.status_code == HTTP_OK
|
||
assert len(await _events(seeded, AuditAction.ROLE_PERMISSION_CHANGE)) == 1
|
||
|
||
shrink = await admin.put("/admin/roles/EXPERT/permissions", json={"permissions": ["DIRECTORY_VIEW"]})
|
||
assert shrink.status_code == HTTP_OK
|
||
assert (await expert.get("/me")).json()["permissions"] == ["DIRECTORY_VIEW"]
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_role_permissions_validation_and_last_admin(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
admin = await auth_client(ADMIN)
|
||
roles = {item["code"]: item for item in (await admin.get("/admin/roles")).json()}
|
||
without_users_admin = [code for code in roles["ADMIN"]["permissions"] if code != "USERS_ADMIN"]
|
||
|
||
last = await admin.put("/admin/roles/ADMIN/permissions", json={"permissions": without_users_admin})
|
||
assert last.status_code == HTTP_CONFLICT
|
||
assert _detail_code(last) == "LAST_ADMIN"
|
||
|
||
unknown = await admin.put("/admin/roles/EXPERT/permissions", json={"permissions": ["KFO_VIEW", "FLY"]})
|
||
assert unknown.status_code == HTTP_UNPROCESSABLE
|
||
assert unknown.json()["detail"]["code"] == "UNKNOWN_PERMISSION"
|
||
assert unknown.json()["detail"]["permissions"] == ["FLY"]
|
||
missing = await admin.put("/admin/roles/NO_SUCH/permissions", json={"permissions": []})
|
||
assert missing.status_code == HTTP_NOT_FOUND
|
||
assert await _events(seeded, AuditAction.ROLE_PERMISSION_CHANGE) == []
|
||
|
||
# USERS_ADMIN есть у другой роли активного пользователя — снять с ADMIN можно
|
||
await admin.put("/admin/roles/EXPERT/permissions", json={"permissions": [*EXPERT_PERMISSIONS, "USERS_ADMIN"]})
|
||
allowed = await admin.put("/admin/roles/ADMIN/permissions", json={"permissions": without_users_admin})
|
||
assert allowed.status_code == HTTP_OK, allowed.text
|
||
assert "USERS_ADMIN" not in allowed.json()["permissions"]
|
||
|
||
|
||
# --- Права доступа -----------------------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_admin_routes_forbidden_without_users_admin(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
client = await auth_client("ka_go@armka.local")
|
||
user_id = await _user_id(seeded, "ka_rf@armka.local")
|
||
calls = [
|
||
("GET", USERS, None),
|
||
("POST", USERS, {"email": "x@example.com", "full_name": "Х", "regional_office_id": str(uuid.uuid4())}),
|
||
("PATCH", f"{USERS}/{user_id}", {"full_name": "Х"}),
|
||
("PATCH", f"{USERS}/{uuid.uuid4()}", {"full_name": "Х"}),
|
||
("PUT", f"{USERS}/{user_id}/roles", {"roles": ["ADMIN"]}),
|
||
("GET", "/admin/roles", None),
|
||
("GET", "/admin/permissions", None),
|
||
("PUT", "/admin/roles/KA_GO/permissions", {"permissions": ["USERS_ADMIN"]}),
|
||
("PUT", "/admin/roles/NO_SUCH/permissions", {"permissions": []}),
|
||
("PATCH", f"{USERS}/not-a-uuid", {"bad": 1}),
|
||
]
|
||
for method, path, body in calls:
|
||
response = await client.request(method, path, json=body)
|
||
assert response.status_code == HTTP_FORBIDDEN, (method, path, response.text)
|
||
assert response.json() == {"detail": "недостаточно прав: требуется USERS_ADMIN"}
|
||
assert await _role_codes(seeded, user_id) == ["KA_RF"]
|
||
|
||
|
||
# --- Гонка двух администраторов ----------------------------------------------------------------------------------
|
||
|
||
|
||
async def _remove_admin_role(maker: async_sessionmaker[AsyncSession], actor_id: uuid.UUID, target_id: uuid.UUID) -> str:
|
||
"""Отдельная транзакция «как запрос»: `actor_id` заменяет роли `target_id` на `[EXPERT]`; итог — `ok` или код."""
|
||
try:
|
||
async with maker() as session, session.begin():
|
||
audit = RequestAudit(service=AuditService(session), user_id=actor_id, ip_address=None, user_agent=None)
|
||
await UsersAdminService(session, audit, actor_id).set_roles(target_id, ["EXPERT"])
|
||
except DomainError as error:
|
||
return error.code
|
||
return "ok"
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_concurrent_mutual_admin_removal_keeps_one_admin(
|
||
committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Два администратора в параллельных транзакциях снимают ADMIN друг с друга: успевает ровно один (D-8).
|
||
|
||
Без блокировки обе транзакции насчитали бы «другого» администратора и обе зафиксировались бы — ноль администраторов.
|
||
Проверка на уровне сервиса: обе транзакции стартуют одновременно, без предварительной аутентификации запроса.
|
||
Барьер перед `lock_permission` гарантирует, что обе транзакции уже прочитали своих пользователей и дошли до проверки
|
||
до того, как любая из них зафиксируется (без барьера первая могла бы успеть целиком и гонки бы не было).
|
||
"""
|
||
barrier = asyncio.Barrier(2)
|
||
original_lock = users_repo.lock_permission
|
||
|
||
async def lock_after_barrier(session: AsyncSession, code: str) -> None:
|
||
async with asyncio.timeout(10):
|
||
await barrier.wait()
|
||
await original_lock(session, code)
|
||
|
||
monkeypatch.setattr(users_repo, "lock_permission", lock_after_barrier)
|
||
|
||
async with committed_sessionmaker() as session, session.begin():
|
||
first_id = await _user_id(session, ADMIN)
|
||
second = await make_user(
|
||
session, email="second.admin@example.com", office=await _office(session, HEAD_OFFICE), roles=["ADMIN"]
|
||
)
|
||
second_id = second.id
|
||
|
||
results = await asyncio.gather(
|
||
_remove_admin_role(committed_sessionmaker, first_id, second_id),
|
||
_remove_admin_role(committed_sessionmaker, second_id, first_id),
|
||
)
|
||
assert sorted(results) == ["LAST_ADMIN", "ok"]
|
||
|
||
async with committed_sessionmaker() as session:
|
||
remaining = await users_repo.count_active_users_with_permission(session, "USERS_ADMIN")
|
||
assert remaining == 1
|