234e830476
Проверки §7 по колонкам, допуск, abs_terms, M-9, граница округления; граничные случаи маппинга колонок и уверенности; шапка, смешанные снимки, статус при задании needs_check, предупреждение о повторе формы и идемпотентный повторный маппинг форм; редакция формы, пара ББ/ОФР, 403 без права, подтверждение формы без шапки и без дат; комментарий без видимости ЮЛ; справочник пользователей; documents_count; CSV чек-листа. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
87 lines
4.1 KiB
Python
87 lines
4.1 KiB
Python
"""Краткий справочник пользователей (spec users-admin «Краткий справочник пользователей»; D-12)."""
|
|
|
|
import uuid
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.api.endpoints.users import MAX_BRIEF_IDS
|
|
from src.models.access import AppUser
|
|
from src.models.directories import RegionalOffice
|
|
from tests.conftest import AuthClientFactory
|
|
from tests.factories import make_user
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
HTTP_OK = 200
|
|
HTTP_UNAUTHORIZED = 401
|
|
HTTP_UNPROCESSABLE = 422
|
|
KA_RF = "ka_rf@armka.local"
|
|
|
|
|
|
async def test_brief_for_non_admin(seeded: AsyncSession, auth_client: AuthClientFactory) -> None:
|
|
users = (await seeded.execute(select(AppUser).order_by(AppUser.email).limit(2))).scalars().all()
|
|
unknown = uuid.uuid4()
|
|
client = await auth_client(KA_RF) # без USERS_ADMIN
|
|
ids = ",".join([*(str(user.id) for user in users), str(unknown)])
|
|
response = await client.get("/users/brief", params={"ids": ids})
|
|
assert response.status_code == HTTP_OK
|
|
body = {item["id"]: item for item in response.json()}
|
|
assert set(body) == {str(user.id) for user in users}
|
|
assert all(set(item) == {"id", "full_name", "is_active"} for item in body.values())
|
|
|
|
|
|
async def test_brief_limits(seeded: AsyncSession, auth_client: AuthClientFactory) -> None: # noqa: ARG001
|
|
client = await auth_client(KA_RF)
|
|
too_many = ",".join(str(uuid.uuid4()) for _ in range(MAX_BRIEF_IDS + 1))
|
|
response = await client.get("/users/brief", params={"ids": too_many})
|
|
assert response.status_code == HTTP_UNPROCESSABLE and response.json()["detail"]["code"] == "TOO_MANY_IDS"
|
|
assert (await client.get("/users/brief", params={"ids": "not-a-uuid"})).status_code == HTTP_UNPROCESSABLE
|
|
|
|
|
|
async def test_brief_without_token_is_401(app: FastAPI, seeded: AsyncSession) -> None: # noqa: ARG001
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test/api/v1") as anonymous:
|
|
response = await anonymous.get("/users/brief", params={"ids": str(uuid.uuid4())})
|
|
assert response.status_code == HTTP_UNAUTHORIZED
|
|
|
|
|
|
async def test_brief_includes_inactive_user(seeded: AsyncSession, auth_client: AuthClientFactory) -> None:
|
|
office = (await seeded.execute(select(RegionalOffice).limit(1))).scalar_one()
|
|
gone = await make_user(
|
|
seeded, email="gone@armka.local", office=office, full_name="Уволенный Сотрудник", is_active=False
|
|
)
|
|
client = await auth_client(KA_RF)
|
|
response = await client.get("/users/brief", params={"ids": str(gone.id)})
|
|
assert response.status_code == HTTP_OK
|
|
assert response.json() == [{"id": str(gone.id), "full_name": "Уволенный Сотрудник", "is_active": False}]
|
|
|
|
|
|
async def test_brief_invalid_uuid_code(seeded: AsyncSession, auth_client: AuthClientFactory) -> None: # noqa: ARG001
|
|
client = await auth_client(KA_RF)
|
|
response = await client.get("/users/brief", params={"ids": f"{uuid.uuid4()},not-a-uuid"})
|
|
assert response.status_code == HTTP_UNPROCESSABLE
|
|
assert response.json()["detail"]["code"] == "INVALID_IDS"
|
|
|
|
|
|
@pytest.mark.parametrize("ids", ["", " ", ",", " , ,"])
|
|
async def test_brief_empty_ids_is_empty_list(
|
|
seeded: AsyncSession,
|
|
auth_client: AuthClientFactory,
|
|
ids: str, # noqa: ARG001
|
|
) -> None:
|
|
client = await auth_client(KA_RF)
|
|
response = await client.get("/users/brief", params={"ids": ids})
|
|
assert response.status_code == HTTP_OK and response.json() == []
|
|
|
|
|
|
async def test_brief_duplicates_and_trailing_comma(seeded: AsyncSession, auth_client: AuthClientFactory) -> None:
|
|
user = (await seeded.execute(select(AppUser).where(AppUser.email == KA_RF))).scalar_one()
|
|
client = await auth_client(KA_RF)
|
|
response = await client.get("/users/brief", params={"ids": f" {user.id} ,{user.id},{str(user.id).upper()},"})
|
|
assert response.status_code == HTTP_OK
|
|
assert [item["id"] for item in response.json()] == [str(user.id)] # дубли схлопываются, хвостовая запятая допустима
|