RSHB/DFiP_Budget_planing
5
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request 'admin-back: ручки для админки' (#19) from admin-back into test

Browse Source 
Reviewed-on: #19
Reviewed-by: Raykov-MS <RaykovMS@avt.rshb.ru>
This commit is contained in:
tsygankoviva 2026-05-26 17:32:31 +03:00
commit 3f6d59ba06
15 changed files with 753 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
api/src/api/v1/forms.py
View File

@ -111,11 +111,11 @@ async def get_form_sheets(
result=SheetFormTypeResponse(
form_type=form.form_type.code,
sheets=form.form_type.sheet_list,
all_sheets=form.form_type.sheet_list_with_directions,
)
)
@router.get("/{form_id}/sheet/{sheet}")
async def get_sheet(
form_id: int,

120
api/src/api/v1/org_unit.py Normal file
View File

@ -0,0 +1,120 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from src.core.errors import AccessDeniedException
from src.api.v1.deps import get_current_active_user, require_admin
from src.db.models.app_user import AppUser
from src.db.session import get_db
from src.domain.schemas import BaseListResponse, BaseSingleResponse, OrgUnitBaseSchema, OrgUnitListSchema, OrgUnitSchema, OrgUnitUpdateSchema, ResponseBase
from src.services.org_unit_service import OrgUnitService
router = APIRouter(prefix="/org-unit", tags=["ssp"])
@router.get("/")
async def org_unit_list(
offset: int = 0,
limit: int = 100,
load_users_count: bool = False,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(require_admin),
) -> BaseListResponse[OrgUnitListSchema]:
org_service = OrgUnitService(db)
count, units = await org_service.get_list(
user=current_user,
with_count=True,
offset=offset,
limit=limit,
load_users=load_users_count,
)
return BaseListResponse(
result=[
OrgUnitListSchema.model_validate(unit) for unit in units
],
count=count,
)
@router.get("/{org_unit_id}")
async def org_unit_id(
org_unit_id: int,
load_users: bool = False,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(require_admin),
) -> BaseSingleResponse[OrgUnitSchema]:
org_service = OrgUnitService(db)
unit = await org_service.get(
user=current_user,
org_unit_id=org_unit_id,
load_users=load_users,
)
if not unit:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="ССП не найден")
return BaseSingleResponse(
result = OrgUnitSchema.model_validate(unit)
)
@router.post("/")
async def create_ssp(
org_data: OrgUnitBaseSchema,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(require_admin),
) -> BaseSingleResponse[OrgUnitSchema]:
"""Создание ССП/РФ."""
org_unit_service = OrgUnitService(db)
org_unit = await org_unit_service.create(
title=org_data.title,
is_ssp=org_data.is_ssp,
user=current_user,
)
return BaseSingleResponse(
message="ССП/РФ создана",
result=OrgUnitSchema.model_validate(org_unit),
)
@router.patch("/{org_unit_id}")
async def update_ssp(
org_unit_id: int,
org_data: OrgUnitUpdateSchema,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(require_admin),
) -> BaseSingleResponse[OrgUnitSchema]:
"""Редактирование ССП/РФ."""
org_unit_service = OrgUnitService(db)
org_unit = await org_unit_service.update(
org_unit_id=org_unit_id,
user=current_user,
**org_data.model_dump(exclude_unset=True),
)
if not org_unit:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="ССП не найден")
return BaseSingleResponse(
message="ССП/РФ обновлена",
result=OrgUnitSchema.model_validate(org_unit),
)
@router.delete("/{org_unit_id}")
async def delete_ssp(
org_unit_id: int,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(require_admin),
) -> ResponseBase:
"""Удаление ССП/РФ."""
org_unit_service = OrgUnitService(db)
result = await org_unit_service.logical_delete(
org_unit_id=org_unit_id,
user=current_user,
is_active=True,
)
if not result:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="ССП не найден")
return ResponseBase(success=True, message="ССП/РФ удалена")

4
api/src/api/v1/router.py
View File

@ -1,6 +1,7 @@
from fastapi import APIRouter
from src.api.v1 import auth, users, admin, audit, forms, form_phases, projects, export
from src.api.v1 import auth, users, admin, audit, forms, form_phases, projects, export, org_unit
from src.api.v1 import websocket
@ -14,3 +15,4 @@ api_router.include_router(projects.router)
api_router.include_router(form_phases.router)
api_router.include_router(export.router)
api_router.include_router(websocket.router)
api_router.include_router(org_unit.router)

35
api/src/api/v1/users.py
View File

@ -8,7 +8,7 @@ from src.api.v1.deps import (
)
from src.db.session import get_db
from src.db.models.app_user import AppUser
from src.domain.schemas import User as UserSchema
from src.domain.schemas import BaseListResponse, BaseSingleResponse, User as UserSchema, UserAdminListResponse
from src.domain.schemas import (
SSPIDList,
ResponseBase,
@ -29,16 +29,27 @@ async def get_current_user_info(current_user: AppUser = Depends(get_current_acti
return current_user
@router.get("/", response_model=UserListResponse)
@router.get("/")
async def get_users(
skip: int = 0,
limit: int = 100,
load_orgs: bool = False,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(require_admin),
):
) -> BaseListResponse[UserAdminListResponse]:
user_service = UserService(db)
users = await user_service.get_all(current_user, skip=skip, limit=limit)
return UserListResponse(success=True, message="Список пользователей", result=users)
count, users = await user_service.get_all(
current_user,
skip=skip,
limit=limit,
load_orgs=load_orgs,
with_count=True
)
return BaseListResponse(
count=count,
result=[UserAdminListResponse.model_validate(user) for user in users]
)
@router.get("/dfip-many-ssp/", response_model=UserSSPLinkResponse)
@ -88,20 +99,26 @@ async def unset_many_ssp(
)
@router.get("/{user_id}", response_model=UserSchema)
@router.get("/{user_id}")
async def get_user(
user_id: int,
load_orgs: bool = False,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(require_admin),
):
) -> UserAdminListResponse:
user_service = UserService(db)
user = await user_service.get(user_id, current_user)
user = await user_service.get(
user_id=user_id,
current_user=current_user,
load_orgs=load_orgs,
)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Пользователь не найден",
)
return user
return UserAdminListResponse.model_validate(user)
@router.put("/", response_model=UserResponse)

26
api/src/db/models/form_type.py
View File

@ -44,6 +44,32 @@ class FormType(Base):
FormTypeEnum.FORM_4: ["plan", "seq_dfip", "approved", "contract", "booking", "q1", "q2", "q3", "q4", "totals", "contract_summary", "allocation", "reserve", "collegial", "ckk"],
}
_sheet_list_with_directions = {
FormTypeEnum.FORM_1: [
{
"sheet_name": sheet_name,
"direction": None,
} for sheet_name in ["OPER", "SMETA"]
] + [
{
"sheet_name": sheet_name,
"direction": direction
} for sheet_name in ["AHR", "CAP"] for direction in ["Support", "Development"]
]
}
@property
def sheet_list_with_directions(self) -> dict:
return self._sheet_list_with_directions.get(
self.code,
[
{
"sheet_name": sheet_name,
"direction": None,
} for sheet_name in self.sheet_list
]
)
@property
def sheet_list(self):
return self.SHEETS_BY_FORM_TYPE.get(self.code, [])

6
api/src/db/models/org_unit.py
View File

@ -22,6 +22,11 @@ class OrgUnit(Base):
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
is_ssp: Mapped[bool] = mapped_column(Boolean)
@property
def users_count(self) -> int | None:
return len(self.users) if self.users is not None else None
# budget_forms: Mapped[list["BudgetForm"]] = relationship( # type: ignore
# "BudgetForm", back_populates="org_unit"
# )
@ -33,3 +38,4 @@ class OrgUnit(Base):
# )
# vsps: Mapped[list["Vsp"]] = relationship("Vsp", back_populates="org_unit") # type: ignore
user_orgs: Mapped[list["UserOrg"]] = relationship("UserOrg", back_populates="org_unit")
users: Mapped[list["AppUser"]] = relationship("AppUser", secondary="v3.user_org", back_populates="org_units")

119
api/src/domain/schemas.py
View File

@ -1,11 +1,12 @@
from datetime import datetime
import enum
from typing import Any, Dict, Generic, List, Literal, Optional, TypeVar
from sqlalchemy.exc import MissingGreenlet
FormPhaseRole = Literal["DFIP", "EXECUTOR_RF"]
from pydantic import BaseModel, ConfigDict, EmailStr, Field
from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator, model_validator
T = TypeVar("T")
@ -37,6 +38,16 @@ PlacementTypeLiteral = Literal[
]
# def serialize_missing_greenlet(cls, obj):
# result = {}
# for attr in cls.model_fields:
# try:
# result[attr] = getattr(obj, attr)
# except MissingGreenlet:
# result[attr] = None
# return result
class ResponseBase(BaseModel):
success: Optional[bool] = Field(True)
message: Optional[str] = None
@ -68,6 +79,9 @@ class UserBase(BaseModel):
)
role_id: int = 3
model_config = ConfigDict(from_attributes=True)
class UserCreate(UserBase):
password: str | None = Field(default=None, examples=["pass123"])
@ -117,6 +131,34 @@ class User(UserInDB):
pass
class UserAdminListResponse(UserInDB):
org_units: Optional[list["OrgUnitSchema"]] = None
@model_validator(mode='before')
@classmethod
def safe_orm_handler(cls, data: Any) -> Any:
if isinstance(data, dict):
return data
result = {
"org_units": None,
"id": data.id,
"created_at": data.created_at,
"updated_at": data.updated_at,
"is_active": data.is_active,
"email": data.email,
"username": data.username,
"full_name": data.full_name,
"role_id": data.role_id,
}
try:
result["org_units"] = data.org_units
except MissingGreenlet:
pass
return result
class UserResponse(ResponseBase):
result: Optional[UserInDB] = None
@ -172,14 +214,70 @@ class FormTypeSchemaEnum(str, enum.Enum):
FORM_4 = "FORM_4"
class OrgUnitSchema(BaseModel):
id: int
class OrgUnitBaseSchema(BaseModel):
title: str
is_active: bool
is_ssp: bool
class Config:
from_attributes = True
model_config = ConfigDict(from_attributes=True)
class OrgUnitUpdateSchema(OrgUnitBaseSchema):
title: Optional[str] = None
is_ssp: Optional[bool] = None
model_config = ConfigDict(from_attributes=True)
class OrgUnitResponseSchema(OrgUnitBaseSchema):
id: int
is_active: bool
users_count: int | None = None
model_config = ConfigDict(from_attributes=True)
@model_validator(mode='before')
@classmethod
def safe_orm_handler(cls, data: Any) -> Any:
if isinstance(data, dict):
return data
result = {
"users_count": None,
"id": data.id,
"title": data.title,
"is_ssp": data.is_ssp,
"is_ssp": data.is_ssp,
"is_active": data.is_active,
}
try:
result["users_count"] = data.users_count
except MissingGreenlet:
pass
return result
class OrgUnitListSchema(OrgUnitResponseSchema):
pass
class OrgUnitSchema(OrgUnitResponseSchema):
users: list[UserBase] | None = None
@model_validator(mode='before')
@classmethod
def safe_orm_handler(cls, data: Any) -> Any:
if isinstance(data, dict):
return data
result = super().safe_orm_handler(data)
result["users"] = None
try:
result["users"] = data.users
except MissingGreenlet:
pass
return result
class BudgetFormResponse(BaseModel):
id: int
@ -191,13 +289,18 @@ class BudgetFormResponse(BaseModel):
org_unit_id: int
org_unit: OrgUnitSchema | None = None
class Config:
from_attributes = True
model_config = ConfigDict(from_attributes=True)
class FullSheetSchema(BaseModel):
sheet_name: str
direction: str | None = None
class SheetFormTypeResponse(BaseModel):
form_type: FormTypeSchemaEnum
sheets: list[str]
all_sheets: list[FullSheetSchema] | None = None
class DirectionSchemaEnum(str, enum.Enum):

112
api/src/repository/org_unit_repository.py Normal file
View File

@ -0,0 +1,112 @@
from sqlalchemy import func, select, update
from sqlalchemy.ext.asyncio import AsyncSession
from src.db.models.org_unit import OrgUnit
from sqlalchemy.orm import joinedload
class OrgUnitRepository:
def __init__(self, db: AsyncSession):
self.db = db
async def get_list(
self,
offset: int | None = None,
limit: int | None = None,
is_ssp: bool | None = None,
is_active: bool | None = None,
with_count: bool = False,
load_users: bool = False,
) -> list[OrgUnit] | tuple[int, list[OrgUnit]]:
if with_count:
query = select(func.count().over().label("total_count"), OrgUnit)
else:
query = select(OrgUnit)
where = []
if is_ssp is not None:
where.append(OrgUnit.is_ssp == is_ssp)
if is_active is not None:
where.append(OrgUnit.is_active == is_active)
query = query.where(*where)
if load_users:
query = query.options(
joinedload(OrgUnit.users)
)
query = query.order_by(OrgUnit.id)
if offset is not None:
query = query.offset(offset)
if limit is not None:
query = query.limit(limit)
if with_count:
result = (await self.db.execute(query)).unique().all()
if result:
return result[0][0], [res[1] for res in result]
else:
if query.whereclause is not None:
count_query = select(func.count(OrgUnit.id)).where(
query.whereclause
)
return (await self.db.execute(count_query)).scalar(), []
else:
return (await self.db.execute(query)).scalars().unique().all()
async def get(
self,
org_unit_id: int,
load_users: bool = False
) -> OrgUnit | None:
query = select(OrgUnit).where(OrgUnit.id == org_unit_id)
if load_users:
query = query.options(
joinedload(OrgUnit.users)
)
return (await self.db.execute(query)).scalars().first()
async def create(
self,
title: str,
is_ssp: bool,
) -> OrgUnit:
"""Создание SSP."""
ssp = OrgUnit(title=title, is_ssp=is_ssp)
self.db.add(ssp)
await self.db.flush()
# await self.db.refresh(ssp)
return ssp
async def update(
self, org_unit: OrgUnit, **kwargs,
) -> OrgUnit | None:
if not org_unit:
return None
for field, value in kwargs.items():
setattr(org_unit, field, value)
await self.db.flush()
# await self.db.refresh(ssp)
return org_unit
async def logical_delete(self, org_unit_id: int, is_active: bool | None = None) -> bool:
"""Логическое удаление SSP."""
query = update(OrgUnit).where(OrgUnit.id == org_unit_id)
if is_active is not None:
query = query.where(OrgUnit.is_active == is_active)
query = query.values(is_active=False)
res = await self.db.execute(query)
await self.db.flush()
return res.rowcount > 0

54
api/src/repository/user_repository.py
View File

@ -1,8 +1,8 @@
from typing import Optional
from sqlalchemy import select, text, update
from sqlalchemy import delete, func, select, text, update
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from sqlalchemy.orm import joinedload, selectinload
from src.db.models.org_unit import OrgUnit
@ -65,15 +65,47 @@ class UserRepository:
)
async def get_list(self, skip: int = 0, limit: int = 100) -> list[AppUser]:
query = (
select(AppUser)
.where(AppUser.is_active.is_(True))
.offset(skip)
.limit(limit)
.order_by(AppUser.id)
)
return (await self.db.execute(query)).scalars().all()
async def get_list(
self,
skip: int = 0,
limit: int = 100,
with_count: bool = False,
load_orgs: bool = False,
) -> list[AppUser] | tuple[int, list[AppUser]]:
where = [AppUser.is_active.is_(True)]
if with_count:
query = select(func.count().over().label("total_count"), AppUser)
else:
query = select(AppUser)
query = query.where(*where)
if load_orgs:
query = query.options(
joinedload(AppUser.org_units)
)
query = query.order_by(AppUser.id)
if skip is not None:
query = query.offset(skip)
if limit is not None:
query = query.limit(limit)
if with_count:
result = (await self.db.execute(query)).unique().all()
if result:
return result[0][0], [res[1] for res in result]
else:
if query.whereclause is not None:
count_query = select(func.count(AppUser.id)).where(
query.whereclause
)
return (await self.db.execute(count_query)).scalar(), []
else:
return (await self.db.execute(query)).scalars().unique().all()
async def logical_delete(self, user_id: int) -> bool:
query = update(AppUser).where(AppUser.id == user_id).values(is_active=False)

2
api/src/services/budget_form_service.py
View File

@ -48,7 +48,7 @@ class BudgetFormService:
user: AppUser,
load_form_type: bool = False,
load_org: bool = False,
) -> list[BudgetForm] | tuple[int, list[BudgetForm]]:
) -> BudgetForm | None:
if user.role_id == UserRoleEnum.ADMIN:
return await self.bf_repo.get(
budget_form_id=budget_form_id,

102
api/src/services/org_unit_service.py Normal file
View File

@ -0,0 +1,102 @@
from sqlalchemy.ext.asyncio import AsyncSession
from src.core.errors import AccessDeniedException
from src.db.models.app_user import AppUser
from src.db.models.org_unit import OrgUnit
from src.db.models.role import UserRoleEnum
from src.repository.org_unit_repository import OrgUnitRepository
class OrgUnitService:
def __init__(self, db: AsyncSession):
self.db = db
self.org_repo = OrgUnitRepository(db)
async def get_list(
self,
user: AppUser,
offset: int | None = None,
limit: int | None = None,
is_ssp: bool | None = None,
is_active: bool | None = None,
with_count: bool = False,
load_users: bool = False,
) -> list[OrgUnit] | tuple[int, list[OrgUnit]]:
if user.role_id != UserRoleEnum.ADMIN:
return (0, []) if with_count else []
return await self.org_repo.get_list(
offset=offset,
limit=limit,
with_count=with_count,
is_ssp=is_ssp,
is_active=is_active,
load_users=load_users,
)
async def get(
self,
org_unit_id: int,
user: AppUser,
load_users: bool = False,
) -> OrgUnit | None:
if user.role_id != UserRoleEnum.ADMIN:
return None
return await self.org_repo.get(
org_unit_id=org_unit_id,
load_users=load_users,
)
async def create(
self,
title: str,
is_ssp: bool,
user: AppUser,
) -> OrgUnit:
"""Создание нового ССП/РФ."""
if not self._can_create_ssp(user):
raise AccessDeniedException()
return await self.org_repo.create(
title=title,
is_ssp=is_ssp,
)
async def update(
self, org_unit_id: int, user: AppUser, **kwargs
) -> OrgUnit | None:
"""Редактирование ССП/РФ"""
#ToDo: писать логи
if not self._can_edit_ssp(user):
raise AccessDeniedException()
org_unit = await self.org_repo.get(org_unit_id=org_unit_id)
if not org_unit:
return None
return await self.org_repo.update(
org_unit=org_unit, **kwargs,
)
async def logical_delete(
self,
org_unit_id: int,
user: AppUser,
is_active: bool | None = None,
) -> bool:
"""Логическое удаление ССП/РФ"""
if not self._can_delete_ssp(user):
raise AccessDeniedException()
return bool(await self.org_repo.logical_delete(org_unit_id=org_unit_id, is_active=is_active))
def _can_create_ssp(self, user: AppUser) -> bool:
return user.role_id == UserRoleEnum.ADMIN
def _can_edit_ssp(self, user: AppUser) -> bool:
return user.role_id == UserRoleEnum.ADMIN
def _can_delete_ssp(self, user: AppUser) -> bool:
return user.role_id == UserRoleEnum.ADMIN

25
api/src/services/user_service.py
View File

@ -17,15 +17,32 @@ class UserService:
self.db = db
self.user_repo = UserRepository(db)
async def get_all(self, current_user: AppUser, skip: int = 0, limit: int = 100):
async def get_all(
self,
current_user: AppUser,
skip: int = 0,
limit: int = 100,
with_count: bool = False,
load_orgs: bool = False,
) -> list[AppUser] | tuple[int, list[AppUser]]:
if current_user.role_id != UserRoleEnum.ADMIN:
raise AccessDeniedException()
return await self.user_repo.get_list(skip=skip, limit=limit)
return await self.user_repo.get_list(
skip=skip,
limit=limit,
with_count=with_count,
load_orgs=load_orgs,
)
async def get(self, user_id: int, current_user: AppUser | None = None):
async def get(
self,
user_id: int,
current_user: AppUser | None = None,
load_orgs: bool = False,
):
if current_user is not None and current_user.role_id != UserRoleEnum.ADMIN:
raise AccessDeniedException()
return await self.user_repo.get(user_id)
return await self.user_repo.get(user_id=user_id, load_orgs=load_orgs)
async def create_user(self, user_data: UserCreate, creator: AppUser):
if creator.role_id != UserRoleEnum.ADMIN:

39
api/tests/integration/conftest.py
View File

@ -14,23 +14,32 @@ from src.main import app
@pytest.fixture
def client():
loop = asyncio.get_event_loop()
db_ref = SessionLocal()
loop_ref = None
db = SessionLocal()
try:
async def _get_db():
try:
yield db
await db.flush()
finally:
await db.close()
app.dependency_overrides[get_db] = _get_db
with TestClient(app) as test_client:
async def _get_db():
nonlocal loop_ref
loop_ref = asyncio.get_running_loop()
try:
yield db_ref
await db_ref.flush()
except:
await db_ref.rollback()
raise
app.dependency_overrides[get_db] = _get_db
with TestClient(app) as test_client:
try:
yield test_client
app.dependency_overrides.clear()
finally:
loop.run_until_complete(db.rollback())
loop.run_until_complete(db.close())
finally:
if db_ref is not None and loop_ref:
async def cleanup():
await db_ref.rollback()
await db_ref.close()
future = asyncio.run_coroutine_threadsafe(cleanup(), loop_ref)
future.result()
app.dependency_overrides.clear()
@pytest.fixture

111
api/tests/integration/test_org_units_integration.py Normal file
View File

@ -0,0 +1,111 @@
import pydantic_core
import pytest
def test_org_units_list_smoke(client, admin_tokens, auth_headers):
response = client.get("/api/v1/org-unit/", headers=auth_headers(admin_tokens))
assert response.status_code == 200
payload = response.json()
assert "result" in payload
assert "count" in payload
assert isinstance(payload["result"], list)
def test_org_units_list_with_users_count(client, admin_tokens, auth_headers):
response = client.get(
"/api/v1/org-unit/?load_users_count=true",
headers=auth_headers(admin_tokens),
)
assert response.status_code == 200
payload = response.json()
assert "result" in payload
assert isinstance(payload["result"], list)
if payload["result"]:
assert "users_count" in payload["result"][0]
def test_org_units_get(client, admin_tokens, auth_headers):
response = client.get("/api/v1/org-unit/1", headers=auth_headers(admin_tokens))
assert response.status_code == 200
payload = response.json()
assert "result" in payload
assert payload["result"]["id"] == 1
def test_org_units_get_not_found(client, admin_tokens, auth_headers):
response = client.get("/api/v1/org-unit/99999", headers=auth_headers(admin_tokens))
assert response.status_code == 404
# def test_org_units_create_smoke(client, admin_tokens, auth_headers):
# suffix = uuid.uuid4().hex[:8]
# title = f"Test Org {suffix}"
# response = client.post(
# "/api/v1/org-unit/",
# json={"title": title, "is_ssp": True},
# headers=auth_headers(admin_tokens),
# )
# assert response.status_code == 200
# payload = response.json()
# assert payload["result"]["title"] == title
# assert payload["result"]["is_ssp"] is True
# def test_org_units_create_non_ssp(client, admin_tokens, auth_headers):
# suffix = uuid.uuid4().hex[:8]
# title = f"Test RF {suffix}"
# response = client.post(
# "/api/v1/org-unit/",
# json={"title": title, "is_ssp": False},
# headers=auth_headers(admin_tokens),
# )
# assert response.status_code == 200
# payload = response.json()
# assert payload["result"]["title"] == title
# assert payload["result"]["is_ssp"] is False
# def test_org_units_update_smoke(client, admin_tokens, auth_headers):
# new_title = f"Updated_{uuid.uuid4().hex[:8]}"
# response = client.patch(
# "/api/v1/org-unit/1",
# json={"title": new_title},
# headers=auth_headers(admin_tokens),
# )
# assert response.status_code == 200
# payload = response.json()
# assert payload["result"]["title"] == new_title
# assert payload["result"]["id"] == 1
def test_org_units_update_not_found_error_shape(client, admin_tokens, auth_headers):
response = client.patch(
"/api/v1/org-unit/99999",
json={"title": "No Matter"},
headers=auth_headers(admin_tokens),
)
assert response.status_code == 404
payload = response.json()
assert isinstance(payload, dict)
assert "code" in payload
assert "message" in payload
def test_org_units_delete_not_found_error_shape(client, admin_tokens, auth_headers):
response = client.delete(
"/api/v1/org-unit/99999", headers=auth_headers(admin_tokens)
)
assert response.status_code == 404
payload = response.json()
assert isinstance(payload, dict)
assert "code" in payload
assert "message" in payload
def test_org_units_access_denied_for_non_admin(client, isp_tokens, auth_headers):
response = client.get("/api/v1/org-unit/", headers=auth_headers(isp_tokens))
assert response.status_code == 403
payload = response.json()
assert isinstance(payload, dict)
assert "code" in payload
assert "message" in payload

46
api/tests/integration/test_users_api_smoke.py
View File

@ -6,6 +6,52 @@ def test_users_me(client, admin_tokens, auth_headers):
assert response.status_code == 200
assert response.json()["username"] == "admin"
def test_users_list(client, admin_tokens, auth_headers):
response = client.get("/api/v1/users/", headers=auth_headers(admin_tokens))
assert response.status_code == 200
result = response.json()
assert "result" in result
assert isinstance(result["result"], list)
def test_users_org_list(client, admin_tokens, auth_headers):
response = client.get("/api/v1/users/?load_orgs=true", headers=auth_headers(admin_tokens))
assert response.status_code == 200
result = response.json()
assert "result" in result
assert isinstance(result["result"], list)
assert len(result["result"]) > 0
assert "org_units" in result["result"][0]
assert len(result["result"][0]["org_units"]) > 0
def test_users_get(client, admin_tokens, auth_headers):
response = client.get("/api/v1/users/me", headers=auth_headers(admin_tokens))
assert response.status_code == 200
user = response.json()
assert "id" in user
user_id = user["id"]
response = client.get(f"/api/v1/users/{user_id}", headers=auth_headers(admin_tokens))
assert response.status_code == 200
def test_user_get_org_list(client, admin_tokens, auth_headers):
response = client.get("/api/v1/users/me", headers=auth_headers(admin_tokens))
assert response.status_code == 200
user = response.json()
assert "id" in user
user_id = user["id"]
response = client.get(f"/api/v1/users/{user_id}?load_orgs=true", headers=auth_headers(admin_tokens))
assert response.status_code == 200
result = response.json()
assert "org_units" in result
assert len(result["org_units"]) > 0
# Задокуменировано что бы не было переполнения бд, так как юзер удаляется только логически
# def test_users_create_smoke(client, admin_tokens, auth_headers):
# suffix = uuid.uuid4().hex[:8]