Add audit read api and remove python audit writes
This commit is contained in:
Raykov-MS
parent
5cd7232eb2
commit
bea8eb6240
42
api/src/api/v1/audit.py
Normal file
42
api/src/api/v1/audit.py
Normal file
@ -0,0 +1,42 @@
|
||||
from fastapi import APIRouter, Depends, status
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.api.v1.deps import require_admin
|
||||
from src.db.session import get_db
|
||||
# from src.domain.models import Users
|
||||
from src.db.models.app_user import AppUser
|
||||
from src.domain.schemas import AuditLogListResponse, AuditLogQueryParams
|
||||
from src.services.auditlog_service import AuditLogService
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get(
|
||||
"/audit-logs",
|
||||
response_model=AuditLogListResponse,
|
||||
status_code=status.HTTP_200_OK,
|
||||
summary="Получение журнала аудита",
|
||||
description="Возвращает список записей аудита с возможностью фильтрации. Доступно только администраторам.",
|
||||
)
|
||||
async def get_audit_logs(
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: AppUser = Depends(require_admin),
|
||||
params: AuditLogQueryParams = Depends(),
|
||||
):
|
||||
offset = (params.page - 1) * params.limit
|
||||
audit_service = AuditLogService(db)
|
||||
|
||||
logs = await audit_service.get_all(
|
||||
user=current_user,
|
||||
limit=params.limit,
|
||||
offset=offset,
|
||||
user_id=params.user_id,
|
||||
org_unit_id=params.org_unit_id,
|
||||
task_id=params.task_id,
|
||||
form_id=params.form_id,
|
||||
event_type=params.event_type,
|
||||
date_from=params.date_from,
|
||||
date_to=params.date_to,
|
||||
)
|
||||
result = [audit_service.orm_log_to_response(log) for log in logs]
|
||||
return AuditLogListResponse(result=result)
|
||||
@ -1,10 +1,11 @@
|
||||
from fastapi import APIRouter
|
||||
|
||||
from src.api.v1 import auth, users, admin, forms, projects
|
||||
from src.api.v1 import auth, users, admin, audit, forms, projects
|
||||
|
||||
api_router = APIRouter()
|
||||
api_router.include_router(auth.router)
|
||||
api_router.include_router(users.router)
|
||||
api_router.include_router(admin.router)
|
||||
api_router.include_router(audit.router)
|
||||
api_router.include_router(forms.router)
|
||||
api_router.include_router(projects.router)
|
||||
@ -105,6 +105,11 @@ class Settings(BaseSettings):
|
||||
description="Размер батча для очистки аудита",
|
||||
alias="AUDIT_LOG_CLEANUP_BATCH_SIZE",
|
||||
)
|
||||
AUDIT_LOG_CLEANUP_INTERVAL_SECONDS: int = Field(
|
||||
default=86400,
|
||||
description="Интервал автозапуска очистки аудита в секундах",
|
||||
alias="AUDIT_LOG_CLEANUP_INTERVAL_SECONDS",
|
||||
)
|
||||
|
||||
model_config = SettingsConfigDict(
|
||||
env_file=".env",
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
from datetime import datetime
|
||||
import enum
|
||||
from typing import Any, Generic, Literal, Optional, TypeVar
|
||||
from typing import Any, Dict, Generic, List, Literal, Optional, TypeVar
|
||||
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, EmailStr, Field
|
||||
@ -250,3 +250,49 @@ class AddLineSchema(BaseModel):
|
||||
justification: Optional[str] = None
|
||||
contract_number: Optional[str] = None
|
||||
contract_end_date: Optional[datetime] = None
|
||||
|
||||
|
||||
class AuditLogBase(BaseModel):
|
||||
"""Базовая схема записи аудита (единый формат вывода как у auditlog)."""
|
||||
|
||||
entity: str = Field(..., description="Тип сущности")
|
||||
entity_id: Optional[int] = Field(None, description="ID сущности")
|
||||
action: str = Field(..., description="Действие")
|
||||
payload_json: Optional[Dict[str, Any]] = None
|
||||
|
||||
|
||||
class AuditLogInDB(AuditLogBase):
|
||||
"""Схема записи аудита в базе данных."""
|
||||
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
id: int = Field(..., description="ID записи")
|
||||
user_id: Optional[int] = Field(None, description="ID пользователя")
|
||||
at: datetime = Field(..., description="Дата/время события")
|
||||
|
||||
|
||||
class AuditLog(AuditLogInDB):
|
||||
"""Схема записи аудита для ответа API."""
|
||||
|
||||
user: Optional[User] = None
|
||||
|
||||
|
||||
class AuditLogListResponse(ResponseBase):
|
||||
"""Схема всех записей аудита для ответа API."""
|
||||
|
||||
result: List[AuditLog] = Field(..., description="Вывод записей аудита")
|
||||
|
||||
|
||||
class AuditLogQueryParams(BaseModel):
|
||||
"""Query-параметры для фильтрации журнала аудита."""
|
||||
|
||||
page: int = Field(1, ge=1, description="Номер страницы")
|
||||
limit: int = Field(
|
||||
20, ge=1, le=100, description="Количество записей на странице"
|
||||
)
|
||||
user_id: Optional[int] = Field(None, description="ID пользователя")
|
||||
org_unit_id: Optional[int] = Field(None, description="ID ССП")
|
||||
task_id: Optional[int] = Field(None, description="ID задачи")
|
||||
form_id: Optional[int] = Field(None, description="ID формы")
|
||||
event_type: Optional[str] = Field(None, description="Тип события")
|
||||
date_from: Optional[datetime] = Field(None, description="Дата начала (ISO 8601)")
|
||||
date_to: Optional[datetime] = Field(None, description="Дата окончания (ISO 8601)")
|
||||
|
||||
@ -1,6 +1,8 @@
|
||||
import asyncio
|
||||
import os
|
||||
from contextlib import asynccontextmanager
|
||||
from datetime import datetime, timezone
|
||||
from contextlib import suppress
|
||||
|
||||
from fastapi import FastAPI, Response
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
@ -27,8 +29,72 @@ if not settings.DEBUG:
|
||||
)
|
||||
|
||||
|
||||
_AUDIT_CLEANUP_LOCK_KEY = 21987431
|
||||
|
||||
|
||||
async def _cleanup_audit_log_once() -> int:
|
||||
if "postgresql" not in settings.DATABASE_URL:
|
||||
return 0
|
||||
|
||||
total_deleted = 0
|
||||
async with engine.begin() as conn:
|
||||
lock_ok = (
|
||||
await conn.execute(
|
||||
text("SELECT pg_try_advisory_lock(:k)"),
|
||||
{"k": _AUDIT_CLEANUP_LOCK_KEY},
|
||||
)
|
||||
).scalar_one()
|
||||
if not lock_ok:
|
||||
return 0
|
||||
try:
|
||||
while True:
|
||||
deleted = (
|
||||
await conn.execute(
|
||||
text(
|
||||
"""
|
||||
WITH doomed AS (
|
||||
SELECT id
|
||||
FROM v3.audit_log
|
||||
WHERE event_dt < now() - make_interval(days => :retention_days)
|
||||
ORDER BY id
|
||||
LIMIT :batch_size
|
||||
)
|
||||
DELETE FROM v3.audit_log a
|
||||
USING doomed d
|
||||
WHERE a.id = d.id
|
||||
RETURNING a.id
|
||||
"""
|
||||
),
|
||||
{
|
||||
"retention_days": settings.AUDIT_LOG_RETENTION_DAYS,
|
||||
"batch_size": settings.AUDIT_LOG_CLEANUP_BATCH_SIZE,
|
||||
},
|
||||
)
|
||||
).rowcount
|
||||
if not deleted:
|
||||
break
|
||||
total_deleted += deleted
|
||||
finally:
|
||||
await conn.execute(
|
||||
text("SELECT pg_advisory_unlock(:k)"),
|
||||
{"k": _AUDIT_CLEANUP_LOCK_KEY},
|
||||
)
|
||||
return total_deleted
|
||||
|
||||
|
||||
async def _audit_cleanup_loop() -> None:
|
||||
# Первый прогон сразу после старта, дальше — по интервалу.
|
||||
while True:
|
||||
try:
|
||||
await _cleanup_audit_log_once()
|
||||
except Exception:
|
||||
pass
|
||||
await asyncio.sleep(max(60, settings.AUDIT_LOG_CLEANUP_INTERVAL_SECONDS))
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
cleanup_task: asyncio.Task | None = None
|
||||
if not settings.DEBUG:
|
||||
await create_tables()
|
||||
ProtectedSettings(
|
||||
@ -38,9 +104,14 @@ async def lifespan(app: FastAPI):
|
||||
APP_NAME=settings.APP_NAME,
|
||||
),
|
||||
)
|
||||
cleanup_task = asyncio.create_task(_audit_cleanup_loop())
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
if cleanup_task:
|
||||
cleanup_task.cancel()
|
||||
with suppress(asyncio.CancelledError):
|
||||
await cleanup_task
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
|
||||
68
api/src/repository/auditlog_repository.py
Normal file
68
api/src/repository/auditlog_repository.py
Normal file
@ -0,0 +1,68 @@
|
||||
from datetime import datetime
|
||||
from typing import Iterable, Optional
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
# from sqlalchemy.orm import selectinload
|
||||
|
||||
# from app.domain.models import AuditLog
|
||||
from src.db.models.audit_log import AuditLog
|
||||
|
||||
|
||||
class AuditLogRepository:
|
||||
"""Репозиторий для работы с журналом аудита."""
|
||||
|
||||
def __init__(self, db: AsyncSession):
|
||||
self.db = db
|
||||
|
||||
async def get(self, audit_log_id: int) -> Optional[AuditLog]:
|
||||
"""Получение записи аудита по ID."""
|
||||
return (
|
||||
(
|
||||
await self.db.execute(
|
||||
select(AuditLog).where(AuditLog.id == audit_log_id).limit(1)
|
||||
)
|
||||
)
|
||||
.scalars()
|
||||
.first()
|
||||
)
|
||||
|
||||
async def get_all(
|
||||
self,
|
||||
limit: int | None = None,
|
||||
offset: int | None = None,
|
||||
user_id: int | None = None,
|
||||
org_unit_id: int | None = None,
|
||||
task_id: int | None = None,
|
||||
form_id: int | None = None,
|
||||
event_type: str | None = None,
|
||||
date_from: datetime | None = None,
|
||||
date_to: datetime | None = None,
|
||||
) -> Iterable[AuditLog]:
|
||||
"""Получение всех записей аудита с опциональной фильтрацией."""
|
||||
query = select(AuditLog)
|
||||
|
||||
if user_id is not None:
|
||||
query = query.where(AuditLog.user_id == user_id)
|
||||
if org_unit_id is not None:
|
||||
query = query.where(AuditLog.org_unit_id == org_unit_id)
|
||||
if task_id is not None:
|
||||
query = query.where(AuditLog.task_id == task_id)
|
||||
if form_id is not None:
|
||||
query = query.where(AuditLog.form_id == form_id)
|
||||
if event_type is not None:
|
||||
query = query.where(AuditLog.event_type == event_type)
|
||||
if date_from is not None:
|
||||
query = query.where(AuditLog.event_dt >= date_from)
|
||||
if date_to is not None:
|
||||
query = query.where(AuditLog.event_dt <= date_to)
|
||||
|
||||
query = query.order_by(AuditLog.event_dt.desc())
|
||||
|
||||
if limit is not None:
|
||||
query = query.limit(limit)
|
||||
if offset is not None:
|
||||
query = query.offset(offset)
|
||||
|
||||
return (await self.db.execute(query)).scalars().all()
|
||||
|
||||
96
api/src/services/auditlog_service.py
Normal file
96
api/src/services/auditlog_service.py
Normal file
@ -0,0 +1,96 @@
|
||||
from datetime import datetime
|
||||
from typing import Any, Dict, Iterable, Optional
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.core.errors import AccessDeniedException
|
||||
|
||||
# from src.domain.models import AuditLog, UserRole, Users
|
||||
from src.db.models.audit_log import AuditLog
|
||||
from src.db.models.role import UserRoleEnum
|
||||
from src.db.models.app_user import AppUser
|
||||
|
||||
from src.repository.auditlog_repository import AuditLogRepository
|
||||
|
||||
from src.domain.schemas import AuditLog as AuditLogSchema
|
||||
|
||||
|
||||
class AuditLogService:
|
||||
"""Сервис для работы с журналом аудита."""
|
||||
|
||||
def __init__(self, db: AsyncSession):
|
||||
self.db = db
|
||||
self.audit_repo = AuditLogRepository(db)
|
||||
|
||||
async def get(self, audit_log_id: int, user: AppUser) -> Optional[AuditLog]:
|
||||
"""Получение записи аудита по ID."""
|
||||
if not self._can_view_audit_logs(user):
|
||||
raise AccessDeniedException(
|
||||
"Недостаточно прав для просмотра журнала аудита"
|
||||
)
|
||||
|
||||
return await self.audit_repo.get(audit_log_id)
|
||||
|
||||
async def get_all(
|
||||
self,
|
||||
user: AppUser,
|
||||
limit: int | None = None,
|
||||
offset: int | None = None,
|
||||
user_id: int | None = None,
|
||||
org_unit_id: int | None = None,
|
||||
task_id: int | None = None,
|
||||
form_id: int | None = None,
|
||||
event_type: str | None = None,
|
||||
date_from: datetime | None = None,
|
||||
date_to: datetime | None = None,
|
||||
) -> Iterable[AuditLog]:
|
||||
"""Получение всех записей аудита с опциональной фильтрацией."""
|
||||
if not self._can_view_audit_logs(user):
|
||||
raise AccessDeniedException(
|
||||
"Недостаточно прав для просмотра журнала аудита"
|
||||
)
|
||||
|
||||
return await self.audit_repo.get_all(
|
||||
limit=limit,
|
||||
offset=offset,
|
||||
user_id=user_id,
|
||||
org_unit_id=org_unit_id,
|
||||
task_id=task_id,
|
||||
form_id=form_id,
|
||||
event_type=event_type,
|
||||
date_from=date_from,
|
||||
date_to=date_to,
|
||||
)
|
||||
|
||||
def _can_view_audit_logs(self, user: AppUser) -> bool:
|
||||
"""Проверяет, может ли пользователь просматривать журнал аудита."""
|
||||
return user.role_id == UserRoleEnum.ADMIN.value
|
||||
|
||||
def orm_log_to_response(self, log: Any) -> AuditLogSchema:
|
||||
"""Маппинг записи ORM audit_log в формат ответа API (entity, entity_id, action, at, payload_json).
|
||||
model_validate(orm) не подходит: в БД поля event_dt/event/event_type/event_data, в API — at/action/entity/entity_id; entity и entity_id из event_data JSON.
|
||||
"""
|
||||
event_data: Optional[Dict[str, Any]] = getattr(log, "event_data", None) or {}
|
||||
entity = (
|
||||
event_data.get("entity_type") if isinstance(event_data, dict) else None
|
||||
) or getattr(log, "event_type", "unknown")
|
||||
entity_id = (
|
||||
event_data.get("entity_id") if isinstance(event_data, dict) else None
|
||||
)
|
||||
if entity_id is not None and not isinstance(entity_id, int):
|
||||
try:
|
||||
entity_id = int(entity_id)
|
||||
except (TypeError, ValueError):
|
||||
entity_id = None
|
||||
action = getattr(log, "event", "")
|
||||
at = getattr(log, "event_dt", None)
|
||||
return AuditLogSchema(
|
||||
entity=entity,
|
||||
entity_id=entity_id,
|
||||
action=action,
|
||||
payload_json=event_data if isinstance(event_data, dict) else None,
|
||||
id=log.id,
|
||||
user_id=getattr(log, "user_id", None),
|
||||
at=at,
|
||||
user=getattr(log, "user", None),
|
||||
)
|
||||
@ -47,3 +47,11 @@ def admin_tokens(client, admin_password: str):
|
||||
)
|
||||
assert response.status_code == 200
|
||||
return response.json()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_headers():
|
||||
def _build(tokens: dict) -> dict:
|
||||
return {"Authorization": f"Bearer {tokens['access_token']}"}
|
||||
|
||||
return _build
|
||||
|
||||
66
api/tests/integration/test_audit_api_integration.py
Normal file
66
api/tests/integration/test_audit_api_integration.py
Normal file
@ -0,0 +1,66 @@
|
||||
import uuid
|
||||
|
||||
|
||||
def _create_executor_and_tokens(client, admin_tokens, auth_headers) -> tuple[int, dict]:
|
||||
suffix = uuid.uuid4().hex[:8]
|
||||
username = f"audit_exec_{suffix}"
|
||||
password = "pass123"
|
||||
created = client.put(
|
||||
"/api/v1/users/",
|
||||
json={
|
||||
"email": f"{username}@example.com",
|
||||
"username": username,
|
||||
"password": password,
|
||||
"full_name": "Audit Executor",
|
||||
"role_id": 2,
|
||||
},
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert created.status_code == 200
|
||||
user_id = created.json()["result"]["id"]
|
||||
|
||||
login = client.post(
|
||||
"/api/v1/auth/login",
|
||||
json={"username": username, "password": password},
|
||||
)
|
||||
assert login.status_code == 200
|
||||
return user_id, login.json()
|
||||
|
||||
|
||||
def test_audit_logs_admin_smoke(client, admin_tokens, auth_headers):
|
||||
response = client.get("/api/v1/audit-logs", headers=auth_headers(admin_tokens))
|
||||
|
||||
assert response.status_code == 200
|
||||
payload = response.json()
|
||||
assert "result" in payload
|
||||
assert isinstance(payload["result"], list)
|
||||
|
||||
|
||||
def test_audit_logs_requires_auth(client):
|
||||
response = client.get("/api/v1/audit-logs")
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
def test_audit_logs_forbidden_for_non_admin(client, admin_tokens, auth_headers):
|
||||
user_id, executor_tokens = _create_executor_and_tokens(
|
||||
client, admin_tokens, auth_headers
|
||||
)
|
||||
try:
|
||||
response = client.get(
|
||||
"/api/v1/audit-logs",
|
||||
headers=auth_headers(executor_tokens),
|
||||
)
|
||||
assert response.status_code == 403
|
||||
finally:
|
||||
client.delete(
|
||||
f"/api/v1/users/{user_id}",
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
|
||||
|
||||
def test_audit_logs_query_validation(client, admin_tokens, auth_headers):
|
||||
response = client.get(
|
||||
"/api/v1/audit-logs?limit=101",
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert response.status_code == 422
|
||||
@ -3,12 +3,8 @@ import uuid
|
||||
import pytest
|
||||
|
||||
|
||||
def _auth_headers(tokens: dict) -> dict:
|
||||
return {"Authorization": f"Bearer {tokens['access_token']}"}
|
||||
|
||||
|
||||
def _pick_form_and_sheet(client, admin_tokens) -> tuple[int, str, str | None]:
|
||||
response = client.get("/api/v1/form/", headers=_auth_headers(admin_tokens))
|
||||
def _pick_form_and_sheet(client, admin_tokens, auth_headers) -> tuple[int, str, str | None]:
|
||||
response = client.get("/api/v1/form/", headers=auth_headers(admin_tokens))
|
||||
assert response.status_code == 200
|
||||
forms = response.json().get("result") or []
|
||||
|
||||
@ -20,7 +16,7 @@ def _pick_form_and_sheet(client, admin_tokens) -> tuple[int, str, str | None]:
|
||||
|
||||
sheets_response = client.get(
|
||||
f"/api/v1/form/{form_id}/sheets",
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
if sheets_response.status_code != 200:
|
||||
continue
|
||||
@ -36,9 +32,9 @@ def _pick_form_and_sheet(client, admin_tokens) -> tuple[int, str, str | None]:
|
||||
|
||||
|
||||
def _pick_form_with_sheet(
|
||||
client, admin_tokens, target_sheet: str, form_type_code: str | None = None
|
||||
client, admin_tokens, auth_headers, target_sheet: str, form_type_code: str | None = None
|
||||
) -> tuple[int, str]:
|
||||
response = client.get("/api/v1/form/", headers=_auth_headers(admin_tokens))
|
||||
response = client.get("/api/v1/form/", headers=auth_headers(admin_tokens))
|
||||
assert response.status_code == 200
|
||||
forms = response.json().get("result") or []
|
||||
for form in forms:
|
||||
@ -49,7 +45,7 @@ def _pick_form_with_sheet(
|
||||
continue
|
||||
sheets_response = client.get(
|
||||
f"/api/v1/form/{form_id}/sheets",
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
if sheets_response.status_code != 200:
|
||||
continue
|
||||
@ -59,7 +55,9 @@ def _pick_form_with_sheet(
|
||||
pytest.skip(f"Не найдена форма с листом {target_sheet}")
|
||||
|
||||
|
||||
def _pick_input_line_id(client, admin_tokens, form_id: int, sheet: str, direction: str | None) -> int:
|
||||
def _pick_input_line_id(
|
||||
client, admin_tokens, auth_headers, form_id: int, sheet: str, direction: str | None
|
||||
) -> int:
|
||||
url = f"/api/v1/form/{form_id}/sheet/{sheet}"
|
||||
params = []
|
||||
if direction:
|
||||
@ -67,7 +65,7 @@ def _pick_input_line_id(client, admin_tokens, form_id: int, sheet: str, directio
|
||||
if params:
|
||||
url = f"{url}?{'&'.join(params)}"
|
||||
|
||||
response = client.get(url, headers=_auth_headers(admin_tokens))
|
||||
response = client.get(url, headers=auth_headers(admin_tokens))
|
||||
assert response.status_code == 200
|
||||
rows = response.json().get("result") or []
|
||||
for row in rows:
|
||||
@ -78,13 +76,13 @@ def _pick_input_line_id(client, admin_tokens, form_id: int, sheet: str, directio
|
||||
pytest.skip("Не найден INPUT line_id для проверки upd_form_cell")
|
||||
|
||||
|
||||
def test_backend_calls_v_form_view_via_sheet_endpoint(client, admin_tokens):
|
||||
form_id, sheet, direction = _pick_form_and_sheet(client, admin_tokens)
|
||||
def test_backend_calls_v_form_view_via_sheet_endpoint(client, admin_tokens, auth_headers):
|
||||
form_id, sheet, direction = _pick_form_and_sheet(client, admin_tokens, auth_headers)
|
||||
url = f"/api/v1/form/{form_id}/sheet/{sheet}"
|
||||
if direction:
|
||||
url = f"{url}?direction={direction}"
|
||||
|
||||
response = client.get(url, headers=_auth_headers(admin_tokens))
|
||||
response = client.get(url, headers=auth_headers(admin_tokens))
|
||||
assert response.status_code == 200
|
||||
payload = response.json()
|
||||
assert isinstance(payload.get("result"), list)
|
||||
@ -94,9 +92,13 @@ def test_backend_calls_v_form_view_via_sheet_endpoint(client, admin_tokens):
|
||||
assert "data" in first_row
|
||||
|
||||
|
||||
def test_backend_calls_upd_form_cell_and_maps_sql_validation_error(client, admin_tokens):
|
||||
form_id, sheet, direction = _pick_form_and_sheet(client, admin_tokens)
|
||||
line_id = _pick_input_line_id(client, admin_tokens, form_id, sheet, direction)
|
||||
def test_backend_calls_upd_form_cell_and_maps_sql_validation_error(
|
||||
client, admin_tokens, auth_headers
|
||||
):
|
||||
form_id, sheet, direction = _pick_form_and_sheet(client, admin_tokens, auth_headers)
|
||||
line_id = _pick_input_line_id(
|
||||
client, admin_tokens, auth_headers, form_id, sheet, direction
|
||||
)
|
||||
|
||||
url = f"/api/v1/form/{form_id}/sheet/{sheet}/cell"
|
||||
if direction:
|
||||
@ -105,7 +107,7 @@ def test_backend_calls_upd_form_cell_and_maps_sql_validation_error(client, admin
|
||||
response = client.patch(
|
||||
url,
|
||||
json={"line_id": line_id, "column": "bad.scope", "value": 1},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
|
||||
# SQL-функция отдает prefixed validation error.
|
||||
@ -116,11 +118,13 @@ def test_backend_calls_upd_form_cell_and_maps_sql_validation_error(client, admin
|
||||
assert isinstance(payload["message"], str)
|
||||
|
||||
|
||||
def test_backend_calls_form3_add_project_and_add_line_functions(client, admin_tokens):
|
||||
def test_backend_calls_form3_add_project_and_add_line_functions(
|
||||
client, admin_tokens, auth_headers
|
||||
):
|
||||
create = client.post(
|
||||
"/api/v1/projects",
|
||||
json={"name": f"ITEST_DB_FUNC_{uuid.uuid4().hex[:6]}", "year": 2026, "branch_id": 1},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert create.status_code == 200
|
||||
project_payload = create.json()
|
||||
@ -132,18 +136,18 @@ def test_backend_calls_form3_add_project_and_add_line_functions(client, admin_to
|
||||
add_line = client.post(
|
||||
f"/api/v1/projects/{project_id}/report/2026/LIMIT/line",
|
||||
json={"expense_item_id": 1},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert add_line.status_code == 200
|
||||
rows = add_line.json()
|
||||
assert isinstance(rows, list)
|
||||
|
||||
|
||||
def test_forms_validation_invalid_sections_returns_400(client, admin_tokens):
|
||||
form_id, sheet = _pick_form_with_sheet(client, admin_tokens, "AHR")
|
||||
def test_forms_validation_invalid_sections_returns_400(client, admin_tokens, auth_headers):
|
||||
form_id, sheet = _pick_form_with_sheet(client, admin_tokens, auth_headers, "AHR")
|
||||
response = client.get(
|
||||
f"/api/v1/form/{form_id}/sheet/{sheet}?sections=bad_section",
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert response.status_code == 400
|
||||
payload = response.json()
|
||||
@ -151,22 +155,26 @@ def test_forms_validation_invalid_sections_returns_400(client, admin_tokens):
|
||||
assert "message" in payload
|
||||
|
||||
|
||||
def test_forms_validation_empty_sections_csv_is_ignored(client, admin_tokens):
|
||||
form_id, sheet, direction = _pick_form_and_sheet(client, admin_tokens)
|
||||
def test_forms_validation_empty_sections_csv_is_ignored(client, admin_tokens, auth_headers):
|
||||
form_id, sheet, direction = _pick_form_and_sheet(client, admin_tokens, auth_headers)
|
||||
url = f"/api/v1/form/{form_id}/sheet/{sheet}?sections= , , "
|
||||
if direction:
|
||||
url += f"&direction={direction}"
|
||||
response = client.get(url, headers=_auth_headers(admin_tokens))
|
||||
response = client.get(url, headers=auth_headers(admin_tokens))
|
||||
assert response.status_code == 200
|
||||
payload = response.json()
|
||||
assert isinstance(payload.get("result"), list)
|
||||
|
||||
|
||||
def test_forms_validation_direction_required_for_form1_ahr(client, admin_tokens):
|
||||
form_id, sheet = _pick_form_with_sheet(client, admin_tokens, "AHR", form_type_code="FORM_1")
|
||||
def test_forms_validation_direction_required_for_form1_ahr(
|
||||
client, admin_tokens, auth_headers
|
||||
):
|
||||
form_id, sheet = _pick_form_with_sheet(
|
||||
client, admin_tokens, auth_headers, "AHR", form_type_code="FORM_1"
|
||||
)
|
||||
response = client.get(
|
||||
f"/api/v1/form/{form_id}/sheet/{sheet}",
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert response.status_code == 400
|
||||
payload = response.json()
|
||||
@ -175,20 +183,26 @@ def test_forms_validation_direction_required_for_form1_ahr(client, admin_tokens)
|
||||
assert "direction" in payload["message"].lower()
|
||||
|
||||
|
||||
def test_forms_validation_direction_case_sensitive(client, admin_tokens):
|
||||
form_id, sheet = _pick_form_with_sheet(client, admin_tokens, "AHR", form_type_code="FORM_1")
|
||||
def test_forms_validation_direction_case_sensitive(client, admin_tokens, auth_headers):
|
||||
form_id, sheet = _pick_form_with_sheet(
|
||||
client, admin_tokens, auth_headers, "AHR", form_type_code="FORM_1"
|
||||
)
|
||||
response = client.get(
|
||||
f"/api/v1/form/{form_id}/sheet/{sheet}?direction=support",
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert response.status_code == 422
|
||||
|
||||
|
||||
def test_forms_validation_direction_forbidden_on_non_form1(client, admin_tokens):
|
||||
form_id, sheet = _pick_form_with_sheet(client, admin_tokens, "AHR", form_type_code="FORM_2")
|
||||
def test_forms_validation_direction_forbidden_on_non_form1(
|
||||
client, admin_tokens, auth_headers
|
||||
):
|
||||
form_id, sheet = _pick_form_with_sheet(
|
||||
client, admin_tokens, auth_headers, "AHR", form_type_code="FORM_2"
|
||||
)
|
||||
response = client.get(
|
||||
f"/api/v1/form/{form_id}/sheet/{sheet}?direction=Support",
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert response.status_code == 400
|
||||
payload = response.json()
|
||||
@ -197,17 +211,17 @@ def test_forms_validation_direction_forbidden_on_non_form1(client, admin_tokens)
|
||||
assert "direction" in payload["message"].lower()
|
||||
|
||||
|
||||
def test_form3_validation_invalid_sections_returns_400(client, admin_tokens):
|
||||
def test_form3_validation_invalid_sections_returns_400(client, admin_tokens, auth_headers):
|
||||
create = client.post(
|
||||
"/api/v1/projects",
|
||||
json={"name": f"ITEST_F3_SECT_{uuid.uuid4().hex[:6]}", "year": 2026, "branch_id": 1},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert create.status_code == 200
|
||||
project_id = int(create.json()["project_id"])
|
||||
response = client.get(
|
||||
f"/api/v1/projects/{project_id}/report/2026/LIMIT?sections=q1,q5",
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert response.status_code == 400
|
||||
payload = response.json()
|
||||
@ -215,44 +229,44 @@ def test_form3_validation_invalid_sections_returns_400(client, admin_tokens):
|
||||
assert "message" in payload
|
||||
|
||||
|
||||
def test_form3_validation_empty_sections_csv_is_ignored(client, admin_tokens):
|
||||
def test_form3_validation_empty_sections_csv_is_ignored(client, admin_tokens, auth_headers):
|
||||
create = client.post(
|
||||
"/api/v1/projects",
|
||||
json={"name": f"ITEST_F3_EMPTY_{uuid.uuid4().hex[:6]}", "year": 2026, "branch_id": 1},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert create.status_code == 200
|
||||
project_id = int(create.json()["project_id"])
|
||||
response = client.get(
|
||||
f"/api/v1/projects/{project_id}/report/2026/LIMIT?sections= , , ",
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert isinstance(response.json().get("result"), list)
|
||||
|
||||
|
||||
def test_form3_validation_duplicate_sections_is_allowed(client, admin_tokens):
|
||||
def test_form3_validation_duplicate_sections_is_allowed(client, admin_tokens, auth_headers):
|
||||
create = client.post(
|
||||
"/api/v1/projects",
|
||||
json={"name": f"ITEST_F3_DUP_{uuid.uuid4().hex[:6]}", "year": 2026, "branch_id": 1},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert create.status_code == 200
|
||||
project_id = int(create.json()["project_id"])
|
||||
response = client.get(
|
||||
f"/api/v1/projects/{project_id}/report/2026/LIMIT?sections=q1,q1",
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert isinstance(response.json().get("result"), list)
|
||||
|
||||
|
||||
def test_form3_add_project_invalid_name_returns_422(client, admin_tokens):
|
||||
def test_form3_add_project_invalid_name_returns_422(client, admin_tokens, auth_headers):
|
||||
# Пробел запрещён regex-правилом AddProjectBody.
|
||||
response = client.post(
|
||||
"/api/v1/projects",
|
||||
json={"name": "INVALID NAME", "year": 2026, "branch_id": 1},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert response.status_code == 422
|
||||
|
||||
@ -271,10 +285,10 @@ def test_health_and_ready_endpoints(client):
|
||||
assert "mv_expense_item_tree" in ready_payload
|
||||
|
||||
|
||||
def test_admin_refresh_tree_requires_admin_role(client, admin_tokens):
|
||||
def test_admin_refresh_tree_requires_admin_role(client, admin_tokens, auth_headers):
|
||||
admin_resp = client.post(
|
||||
"/api/v1/admin/refresh-tree",
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert admin_resp.status_code in (200, 503)
|
||||
|
||||
@ -288,7 +302,7 @@ def test_admin_refresh_tree_requires_admin_role(client, admin_tokens):
|
||||
"full_name": "Integration User",
|
||||
"role_id": 2,
|
||||
},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
if create_user.status_code != 200:
|
||||
pytest.skip("Не удалось создать non-admin пользователя в текущей БД")
|
||||
@ -300,10 +314,17 @@ def test_admin_refresh_tree_requires_admin_role(client, admin_tokens):
|
||||
|
||||
non_admin_resp = client.post(
|
||||
"/api/v1/admin/refresh-tree",
|
||||
headers=_auth_headers(user_tokens),
|
||||
headers=auth_headers(user_tokens),
|
||||
)
|
||||
assert non_admin_resp.status_code == 403
|
||||
|
||||
# Что бы не было переполнения бд
|
||||
deleted = client.delete(
|
||||
f"/api/v1/users/{create_user.json()['result']['id']}",
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert deleted.status_code == 200
|
||||
|
||||
|
||||
def test_admin_refresh_tree_requires_auth(client):
|
||||
response = client.post("/api/v1/admin/refresh-tree")
|
||||
|
||||
@ -1,12 +1,8 @@
|
||||
import pytest
|
||||
|
||||
|
||||
def _auth_headers(tokens: dict) -> dict:
|
||||
return {"Authorization": f"Bearer {tokens['access_token']}"}
|
||||
|
||||
|
||||
def test_forms_list_smoke(client, admin_tokens):
|
||||
response = client.get("/api/v1/form/", headers=_auth_headers(admin_tokens))
|
||||
def test_forms_list_smoke(client, admin_tokens, auth_headers):
|
||||
response = client.get("/api/v1/form/", headers=auth_headers(admin_tokens))
|
||||
assert response.status_code == 200
|
||||
payload = response.json()
|
||||
assert "result" in payload
|
||||
@ -14,9 +10,9 @@ def test_forms_list_smoke(client, admin_tokens):
|
||||
assert isinstance(payload["result"], list)
|
||||
|
||||
|
||||
def test_forms_sheets_not_found_error_shape(client, admin_tokens):
|
||||
def test_forms_sheets_not_found_error_shape(client, admin_tokens, auth_headers):
|
||||
response = client.get(
|
||||
"/api/v1/form/999/sheets", headers=_auth_headers(admin_tokens)
|
||||
"/api/v1/form/999/sheets", headers=auth_headers(admin_tokens)
|
||||
)
|
||||
assert response.status_code == 404
|
||||
payload = response.json()
|
||||
@ -25,9 +21,9 @@ def test_forms_sheets_not_found_error_shape(client, admin_tokens):
|
||||
assert isinstance(payload["message"], str)
|
||||
|
||||
|
||||
def test_forms_sheet_get_not_found_error_shape(client, admin_tokens):
|
||||
def test_forms_sheet_get_not_found_error_shape(client, admin_tokens, auth_headers):
|
||||
response = client.get(
|
||||
"/api/v1/form/999/sheet/AHR", headers=_auth_headers(admin_tokens)
|
||||
"/api/v1/form/999/sheet/AHR", headers=auth_headers(admin_tokens)
|
||||
)
|
||||
assert response.status_code == 404
|
||||
payload = response.json()
|
||||
@ -35,11 +31,11 @@ def test_forms_sheet_get_not_found_error_shape(client, admin_tokens):
|
||||
assert "message" in payload
|
||||
|
||||
|
||||
def test_forms_cell_update_not_found_error_shape(client, admin_tokens):
|
||||
def test_forms_cell_update_not_found_error_shape(client, admin_tokens, auth_headers):
|
||||
response = client.patch(
|
||||
"/api/v1/form/999/sheet/AHR/cell",
|
||||
json={"line_id": 1, "column": "q1.m1", "value": 100},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert response.status_code == 404
|
||||
payload = response.json()
|
||||
@ -47,7 +43,7 @@ def test_forms_cell_update_not_found_error_shape(client, admin_tokens):
|
||||
assert "message" in payload
|
||||
|
||||
|
||||
def test_forms_cell_update(client, admin_tokens):
|
||||
def test_forms_cell_update(client, admin_tokens, auth_headers):
|
||||
response = client.patch(
|
||||
"/api/v1/form/1/sheet/AHR/cell?direction=Support§ions=plan,contract_summary",
|
||||
json={
|
||||
@ -55,7 +51,7 @@ def test_forms_cell_update(client, admin_tokens):
|
||||
"column": "plan.q1",
|
||||
"value": 225
|
||||
},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert response.status_code == 200
|
||||
payload = response.json()
|
||||
@ -63,11 +59,11 @@ def test_forms_cell_update(client, admin_tokens):
|
||||
assert "result" in payload
|
||||
|
||||
|
||||
def test_forms_cells_update_not_found_error_shape(client, admin_tokens):
|
||||
def test_forms_cells_update_not_found_error_shape(client, admin_tokens, auth_headers):
|
||||
response = client.patch(
|
||||
"/api/v1/form/999/sheet/AHR/cells",
|
||||
json={"changes": [{"line_id": 1, "column": "q1.m1", "value": 100}]},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert response.status_code == 404
|
||||
payload = response.json()
|
||||
@ -75,7 +71,7 @@ def test_forms_cells_update_not_found_error_shape(client, admin_tokens):
|
||||
assert "message" in payload
|
||||
|
||||
|
||||
def test_forms_cells_update(client, admin_tokens):
|
||||
def test_forms_cells_update(client, admin_tokens, auth_headers):
|
||||
response = client.patch(
|
||||
"/api/v1/form/1/sheet/AHR/cells?direction=Support§ions=plan,contract_summary",
|
||||
json={
|
||||
@ -87,7 +83,7 @@ def test_forms_cells_update(client, admin_tokens):
|
||||
}
|
||||
],
|
||||
},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert response.status_code == 200
|
||||
payload = response.json()
|
||||
@ -95,11 +91,11 @@ def test_forms_cells_update(client, admin_tokens):
|
||||
assert "result" in payload
|
||||
|
||||
|
||||
def test_forms_add_line_not_found_error_shape(client, admin_tokens):
|
||||
def test_forms_add_line_not_found_error_shape(client, admin_tokens, auth_headers):
|
||||
response = client.post(
|
||||
"/api/v1/form/999/sheet/AHR/line",
|
||||
json={},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert response.status_code == 404
|
||||
payload = response.json()
|
||||
@ -107,14 +103,14 @@ def test_forms_add_line_not_found_error_shape(client, admin_tokens):
|
||||
assert "message" in payload
|
||||
|
||||
|
||||
def test_forms_add_line(client, admin_tokens):
|
||||
def test_forms_add_line(client, admin_tokens, auth_headers):
|
||||
response = client.post(
|
||||
"/api/v1/form/1/sheet/AHR/line?direction=Support§ions=plan,contract_summary",
|
||||
json={
|
||||
"expense_item_id": 120,
|
||||
"direction": "Support"
|
||||
},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert response.status_code == 200
|
||||
payload = response.json()
|
||||
@ -122,10 +118,10 @@ def test_forms_add_line(client, admin_tokens):
|
||||
assert "result" in payload
|
||||
|
||||
|
||||
def test_forms_delete_line_not_found_error_shape(client, admin_tokens):
|
||||
def test_forms_delete_line_not_found_error_shape(client, admin_tokens, auth_headers):
|
||||
response = client.delete(
|
||||
"/api/v1/form/999/sheet/AHR/line/1",
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert response.status_code == 404
|
||||
payload = response.json()
|
||||
|
||||
@ -3,12 +3,8 @@ import uuid
|
||||
import pytest
|
||||
|
||||
|
||||
def _auth_headers(tokens: dict) -> dict:
|
||||
return {"Authorization": f"Bearer {tokens['access_token']}"}
|
||||
|
||||
|
||||
def test_projects_list_smoke(client, admin_tokens):
|
||||
response = client.get("/api/v1/projects", headers=_auth_headers(admin_tokens))
|
||||
def test_projects_list_smoke(client, admin_tokens, auth_headers):
|
||||
response = client.get("/api/v1/projects", headers=auth_headers(admin_tokens))
|
||||
assert response.status_code == 200
|
||||
payload = response.json()
|
||||
assert "result" in payload
|
||||
@ -16,24 +12,24 @@ def test_projects_list_smoke(client, admin_tokens):
|
||||
assert isinstance(payload["result"], list)
|
||||
|
||||
|
||||
def _create_project(client, admin_tokens) -> tuple[int, str]:
|
||||
def _create_project(client, admin_tokens, auth_headers) -> tuple[int, str]:
|
||||
name = f"PT_{uuid.uuid4().hex[:10]}"
|
||||
response = client.post(
|
||||
"/api/v1/projects",
|
||||
json={"name": name, "year": 2026, "branch_id": 1},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert response.status_code == 200
|
||||
payload = response.json()
|
||||
return payload["project_id"], name
|
||||
|
||||
|
||||
def _add_line(client, admin_tokens, project_id: int) -> int:
|
||||
def _add_line(client, admin_tokens, auth_headers, project_id: int) -> int:
|
||||
for expense_item_id in range(1, 80):
|
||||
response = client.post(
|
||||
f"/api/v1/projects/{project_id}/report/2026/LIMIT/line",
|
||||
json={"expense_item_id": expense_item_id},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
if response.status_code != 200:
|
||||
continue
|
||||
@ -52,22 +48,22 @@ def _add_line(client, admin_tokens, project_id: int) -> int:
|
||||
pytest.skip("Не удалось подобрать expense_item_id для add_form3_line в текущей БД")
|
||||
|
||||
|
||||
def test_projects_write_smoke(client, admin_tokens):
|
||||
project_id, name = _create_project(client, admin_tokens)
|
||||
def test_projects_write_smoke(client, admin_tokens, auth_headers):
|
||||
project_id, name = _create_project(client, admin_tokens, auth_headers)
|
||||
|
||||
upd_project_response = client.patch(
|
||||
f"/api/v1/project/{project_id}",
|
||||
json={"column": "name", "value": f"{name}_U"},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert upd_project_response.status_code == 200
|
||||
|
||||
line_id = _add_line(client, admin_tokens, project_id)
|
||||
line_id = _add_line(client, admin_tokens, auth_headers, project_id)
|
||||
|
||||
upd_cell_response = client.patch(
|
||||
f"/api/v1/projects/{project_id}/report/2026/LIMIT/cell",
|
||||
json={"line_id": line_id, "column": "q1.m1", "value": 100},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert upd_cell_response.status_code == 200
|
||||
assert isinstance(upd_cell_response.json(), list)
|
||||
@ -80,26 +76,26 @@ def test_projects_write_smoke(client, admin_tokens):
|
||||
{"line_id": line_id, "column": "q1.m3", "value": 300},
|
||||
]
|
||||
},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert upd_cells_response.status_code == 200
|
||||
assert isinstance(upd_cells_response.json(), list)
|
||||
|
||||
del_line_response = client.delete(
|
||||
f"/api/v1/projects/{project_id}/report/2026/LIMIT/line/{line_id}",
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert del_line_response.status_code == 200
|
||||
|
||||
|
||||
def test_projects_write_invalid_report_type_error_shape(client, admin_tokens):
|
||||
project_id, _ = _create_project(client, admin_tokens)
|
||||
line_id = _add_line(client, admin_tokens, project_id)
|
||||
def test_projects_write_invalid_report_type_error_shape(client, admin_tokens, auth_headers):
|
||||
project_id, _ = _create_project(client, admin_tokens, auth_headers)
|
||||
line_id = _add_line(client, admin_tokens, auth_headers, project_id)
|
||||
|
||||
response = client.patch(
|
||||
f"/api/v1/projects/{project_id}/report/2026/BAD_TYPE/cell",
|
||||
json={"line_id": line_id, "column": "q1.m1", "value": 100},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert response.status_code == 400
|
||||
payload = response.json()
|
||||
@ -108,14 +104,14 @@ def test_projects_write_invalid_report_type_error_shape(client, admin_tokens):
|
||||
assert isinstance(payload["message"], str)
|
||||
|
||||
|
||||
def test_projects_write_report_not_found_error_shape(client, admin_tokens):
|
||||
project_id, _ = _create_project(client, admin_tokens)
|
||||
line_id = _add_line(client, admin_tokens, project_id)
|
||||
def test_projects_write_report_not_found_error_shape(client, admin_tokens, auth_headers):
|
||||
project_id, _ = _create_project(client, admin_tokens, auth_headers)
|
||||
line_id = _add_line(client, admin_tokens, auth_headers, project_id)
|
||||
|
||||
response = client.patch(
|
||||
f"/api/v1/projects/{project_id}/report/2099/LIMIT/cell",
|
||||
json={"line_id": line_id, "column": "q1.m1", "value": 100},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert response.status_code == 400
|
||||
payload = response.json()
|
||||
|
||||
@ -1,29 +1,22 @@
|
||||
import uuid
|
||||
|
||||
|
||||
import uuid
|
||||
|
||||
|
||||
def _auth_headers(tokens: dict) -> dict:
|
||||
return {"Authorization": f"Bearer {tokens['access_token']}"}
|
||||
|
||||
|
||||
def test_users_me(client, admin_tokens):
|
||||
response = client.get("/api/v1/users/me", headers=_auth_headers(admin_tokens))
|
||||
def test_users_me(client, admin_tokens, auth_headers):
|
||||
response = client.get("/api/v1/users/me", headers=auth_headers(admin_tokens))
|
||||
assert response.status_code == 200
|
||||
assert response.json()["username"] == "admin"
|
||||
|
||||
|
||||
def test_users_create_smoke(client, admin_tokens):
|
||||
def test_users_create_smoke(client, admin_tokens, auth_headers):
|
||||
suffix = uuid.uuid4().hex[:8]
|
||||
username = f"user_{suffix}"
|
||||
create_payload = {
|
||||
"email": f"{username}@example.com",
|
||||
"username": username,
|
||||
"password": "pass123",
|
||||
"full_name": "User One",
|
||||
"role_id": 2,
|
||||
}
|
||||
# create_payload = {
|
||||
# "email": f"{username}@example.com",
|
||||
# "username": username,
|
||||
# "password": "pass123",
|
||||
# "full_name": "User One",
|
||||
# "role_id": 2,
|
||||
# }
|
||||
created = client.put(
|
||||
"/api/v1/users/",
|
||||
json={
|
||||
@ -33,7 +26,14 @@ def test_users_create_smoke(client, admin_tokens):
|
||||
"full_name": "User One",
|
||||
"role_id": 2,
|
||||
},
|
||||
headers=_auth_headers(admin_tokens),
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert created.status_code == 200
|
||||
assert created.json()["result"]["username"] == username
|
||||
|
||||
# Что бы не было переполнения бд
|
||||
deleted = client.delete(
|
||||
f"/api/v1/users/{created.json()['result']['id']}",
|
||||
headers=auth_headers(admin_tokens),
|
||||
)
|
||||
assert deleted.status_code == 200
|
||||
|
||||
68
api/tests/unit/test_auditlog_repository.py
Normal file
68
api/tests/unit/test_auditlog_repository.py
Normal file
@ -0,0 +1,68 @@
|
||||
from datetime import datetime, timezone
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.repository.auditlog_repository import AuditLogRepository
|
||||
|
||||
|
||||
def _result_with_scalars(items):
|
||||
result = MagicMock()
|
||||
scalars = MagicMock()
|
||||
scalars.first.return_value = items[0] if items else None
|
||||
scalars.all.return_value = items
|
||||
result.scalars.return_value = scalars
|
||||
return result
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_db():
|
||||
return AsyncMock(spec=AsyncSession)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def repository(mock_db):
|
||||
return AuditLogRepository(mock_db)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_returns_audit_log(repository, mock_db):
|
||||
row = MagicMock(id=10, user_id=1, event_type="WRITE")
|
||||
mock_db.execute.return_value = _result_with_scalars([row])
|
||||
|
||||
result = await repository.get(10)
|
||||
|
||||
assert result is row
|
||||
mock_db.execute.assert_awaited_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_returns_none_for_missing_id(repository, mock_db):
|
||||
mock_db.execute.return_value = _result_with_scalars([])
|
||||
|
||||
result = await repository.get(999999)
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_all_applies_filters(repository, mock_db):
|
||||
row = MagicMock(id=11, user_id=2, org_unit_id=7, event_type="UPDATE")
|
||||
mock_db.execute.return_value = _result_with_scalars([row])
|
||||
|
||||
result = await repository.get_all(
|
||||
limit=10,
|
||||
offset=5,
|
||||
user_id=2,
|
||||
org_unit_id=7,
|
||||
task_id=3,
|
||||
form_id=4,
|
||||
event_type="UPDATE",
|
||||
date_from=datetime(2026, 1, 1, tzinfo=timezone.utc),
|
||||
date_to=datetime(2026, 12, 31, tzinfo=timezone.utc),
|
||||
)
|
||||
|
||||
assert result == [row]
|
||||
mock_db.execute.assert_awaited_once()
|
||||
|
||||
91
api/tests/unit/test_auditlog_service.py
Normal file
91
api/tests/unit/test_auditlog_service.py
Normal file
@ -0,0 +1,91 @@
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from src.core.errors import AccessDeniedException
|
||||
from src.db.models.role import UserRoleEnum
|
||||
from src.services.auditlog_service import AuditLogService
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def service_with_mock_repo(monkeypatch):
|
||||
mock_repo = MagicMock()
|
||||
mock_repo.get = AsyncMock()
|
||||
mock_repo.get_all = AsyncMock()
|
||||
|
||||
repo_cls = MagicMock(return_value=mock_repo)
|
||||
monkeypatch.setattr("src.services.auditlog_service.AuditLogRepository", repo_cls)
|
||||
|
||||
service = AuditLogService(AsyncMock())
|
||||
return service, mock_repo
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def admin_user():
|
||||
return SimpleNamespace(
|
||||
id=1,
|
||||
role_id=UserRoleEnum.ADMIN.value,
|
||||
email="admin@example.com",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def non_admin_user():
|
||||
return SimpleNamespace(
|
||||
id=2,
|
||||
role_id=UserRoleEnum.EXECUTOR_DFIP.value,
|
||||
email="executor@example.com",
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_all_allowed_for_admin(service_with_mock_repo, admin_user):
|
||||
service, mock_repo = service_with_mock_repo
|
||||
mock_row = SimpleNamespace(id=10)
|
||||
mock_repo.get_all.return_value = [mock_row]
|
||||
|
||||
result = await service.get_all(admin_user, org_unit_id=77, event_type="form")
|
||||
|
||||
assert result == [mock_row]
|
||||
mock_repo.get_all.assert_awaited_once_with(
|
||||
limit=None,
|
||||
offset=None,
|
||||
user_id=None,
|
||||
org_unit_id=77,
|
||||
task_id=None,
|
||||
form_id=None,
|
||||
event_type="form",
|
||||
date_from=None,
|
||||
date_to=None,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_all_denied_for_non_admin(service_with_mock_repo, non_admin_user):
|
||||
service, mock_repo = service_with_mock_repo
|
||||
|
||||
with pytest.raises(AccessDeniedException):
|
||||
await service.get_all(non_admin_user)
|
||||
|
||||
mock_repo.get_all.assert_not_called()
|
||||
|
||||
|
||||
def test_orm_log_to_response_maps_entity_and_entity_id(service_with_mock_repo):
|
||||
service, _ = service_with_mock_repo
|
||||
log = SimpleNamespace(
|
||||
id=55,
|
||||
user_id=7,
|
||||
event="UPDATE",
|
||||
event_type="fallback_entity",
|
||||
event_dt="2026-05-19T12:00:00Z",
|
||||
event_data={"entity_type": "form_line", "entity_id": "42", "x": 1},
|
||||
user=None,
|
||||
)
|
||||
|
||||
dto = service.orm_log_to_response(log)
|
||||
|
||||
assert dto.entity == "form_line"
|
||||
assert dto.entity_id == 42
|
||||
assert dto.action == "UPDATE"
|
||||
assert dto.user_id == 7
|
||||
Loading…
x
Reference in New Issue
Block a user