feat(directories): импорт выгрузки ЕЦП — сервис, эндпоинт, аудит, файл в S3
POST /api/v1/directories/import (DIRECTORY_ADMIN, multipart: file, dry_run=true, on_existing=skip):
снимок справочника тремя запросами (load_snapshot), план ядра §13.2, применение пакетами по типу
операции; UpdateGroup/UpdateEntity — apply_group_update/apply_entity_update (R-14) с description
«импорт ЕЦП». dry_run — явный session.rollback(), без аудита и S3. Боевой прогон —
audit_event(DIRECTORY_IMPORT, after_value = отчёт) и файл uploads/imports/{event_id}/{имя} +
s3_object(uploads). IMPORT_MAX_BYTES (20 МБ) → 413; ошибки файла → 422 MISSING_COLUMNS/UNSUPPORTED_FILE.
Тест dry_run — на отдельной БД с сессией запроса как в проде (SAVEPOINT-обёртка db_session
при rollback снимает и seed теста).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
This commit is contained in:
Andrey Epifancev
parent
e373dfc187
commit
c9c3b69076
@ -46,3 +46,6 @@ NATS_ENABLED=true
|
||||
OUTBOX_POLL_SECONDS=5
|
||||
ACK_TIMEOUT_SECONDS=30
|
||||
OUTBOX_MAX_REPUBLISH=3
|
||||
|
||||
# Импорт справочников из выгрузки ЕЦП CRM: предельный размер файла в байтах (20 МБ), больше — ответ 413
|
||||
IMPORT_MAX_BYTES=20971520
|
||||
|
||||
@ -40,3 +40,6 @@ NATS_ENABLED=false
|
||||
OUTBOX_POLL_SECONDS=5
|
||||
ACK_TIMEOUT_SECONDS=30
|
||||
OUTBOX_MAX_REPUBLISH=3
|
||||
|
||||
# Импорт справочников: предельный размер файла (20 МБ), как в .env.example
|
||||
IMPORT_MAX_BYTES=20971520
|
||||
|
||||
@ -2,14 +2,16 @@
|
||||
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import Depends
|
||||
from fastapi import Depends, Request
|
||||
|
||||
from src.api.dependencies.audit import RequestAudit, get_request_audit
|
||||
from src.api.dependencies.auth import Principal, current_user
|
||||
from src.api.dependencies.db import SessionDep
|
||||
from src.core.config import Settings, settings
|
||||
from src.domain.policy import VisibilityScope, scope_of
|
||||
from src.services.aliases import AliasService
|
||||
from src.services.company_groups import CompanyGroupService
|
||||
from src.services.directory_import import DirectoryImportService
|
||||
from src.services.legal_entities import LegalEntityService
|
||||
from src.services.related_groups import RelatedGroupService
|
||||
|
||||
@ -61,3 +63,24 @@ def get_related_group_service(
|
||||
|
||||
|
||||
RelatedGroupServiceDep = Annotated[RelatedGroupService, Depends(get_related_group_service)]
|
||||
|
||||
|
||||
def get_app_settings() -> Settings:
|
||||
"""Настройки процесса (`src.core.config.settings`); зависимость — чтобы тест мог подменить."""
|
||||
return settings
|
||||
|
||||
|
||||
SettingsDep = Annotated[Settings, Depends(get_app_settings)]
|
||||
|
||||
|
||||
def get_directory_import_service(
|
||||
request: Request,
|
||||
session: SessionDep,
|
||||
audit: Annotated[RequestAudit, Depends(get_request_audit)],
|
||||
app_settings: SettingsDep,
|
||||
) -> DirectoryImportService:
|
||||
"""Сервис импорта; S3 — `app.state.storage` (`None`, если S3 недоступно или не подключено)."""
|
||||
return DirectoryImportService(session, audit, getattr(request.app.state, "storage", None), app_settings)
|
||||
|
||||
|
||||
DirectoryImportServiceDep = Annotated[DirectoryImportService, Depends(get_directory_import_service)]
|
||||
|
||||
@ -2,10 +2,11 @@
|
||||
|
||||
from fastapi import APIRouter
|
||||
|
||||
from src.api.endpoints.directories import company_groups, legal_entities, regional_offices, related_groups
|
||||
from src.api.endpoints.directories import company_groups, import_, legal_entities, regional_offices, related_groups
|
||||
|
||||
router = APIRouter(prefix="/directories", tags=["directories"])
|
||||
router.include_router(regional_offices.router)
|
||||
router.include_router(company_groups.router)
|
||||
router.include_router(legal_entities.router)
|
||||
router.include_router(related_groups.router)
|
||||
router.include_router(import_.router)
|
||||
|
||||
45
api/src/api/endpoints/directories/import_.py
Normal file
45
api/src/api/endpoints/directories/import_.py
Normal file
@ -0,0 +1,45 @@
|
||||
"""`POST /directories/import` — импорт выгрузки ЕЦП CRM (spec directory-import; `05-domain-rules.md` §13).
|
||||
|
||||
multipart: `file` (xlsx/csv), `dry_run` (по умолчанию `true`), `on_existing` ∈ {`skip`, `update`} (по умолчанию
|
||||
`skip`); право `DIRECTORY_ADMIN`. Размер больше `IMPORT_MAX_BYTES` — 413 без чтения файла в память целиком.
|
||||
"""
|
||||
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import APIRouter, Depends, File, Form, UploadFile
|
||||
|
||||
from src.api.dependencies.auth import require
|
||||
from src.api.dependencies.directories import DirectoryImportServiceDep, SettingsDep
|
||||
from src.domain.directory_import import OnExisting
|
||||
from src.schemas.directory_import import ImportReportOut
|
||||
from src.services.directory_import import check_size
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
async def _read_limited(file: UploadFile, limit: int) -> bytes:
|
||||
"""Содержимое загрузки не больше `limit` байт: размер из multipart, если известен, иначе чтение `limit + 1`."""
|
||||
if file.size is not None:
|
||||
check_size(file.size, limit)
|
||||
content = await file.read(limit + 1)
|
||||
check_size(len(content), limit)
|
||||
return content
|
||||
|
||||
|
||||
@router.post(
|
||||
"/import",
|
||||
response_model=ImportReportOut,
|
||||
dependencies=[Depends(require("DIRECTORY_ADMIN"))],
|
||||
summary="Импорт выгрузки ЕЦП CRM",
|
||||
)
|
||||
async def import_directory(
|
||||
file: Annotated[UploadFile, File(description="Выгрузка ЕЦП CRM: xlsx (первый лист) или csv (UTF-8, `;`/`,`)")],
|
||||
service: DirectoryImportServiceDep,
|
||||
settings: SettingsDep,
|
||||
dry_run: Annotated[bool, Form(description="Пробный прогон: отчёт без записи в БД, аудит и S3")] = True,
|
||||
on_existing: Annotated[OnExisting, Form(description="Существующие ГК/ЮЛ: оставить или обновить")] = OnExisting.skip,
|
||||
) -> ImportReportOut:
|
||||
"""Отчёт: счётчики и построчные коды §13.2; при `dry_run = false` — запись в справочник, аудит и файл в S3."""
|
||||
content = await _read_limited(file, settings.IMPORT_MAX_BYTES)
|
||||
report = await service.run(filename=file.filename or "", content=content, dry_run=dry_run, on_existing=on_existing)
|
||||
return ImportReportOut.model_validate(report.to_dict())
|
||||
@ -76,6 +76,9 @@ class Settings(BaseSettings):
|
||||
3 # число повторных публикаций до перевода в dlq (0 — сразу dlq без повторов)
|
||||
)
|
||||
|
||||
# Импорт справочников из выгрузки ЕЦП (§13.3): предельный размер файла, больше — 413
|
||||
IMPORT_MAX_BYTES: PositiveInt = 20 * 1024 * 1024
|
||||
|
||||
model_config = SettingsConfigDict(
|
||||
extra="ignore",
|
||||
env_file=os.environ.get("ENV_FILE", ".env"),
|
||||
|
||||
@ -9,9 +9,10 @@ import datetime
|
||||
import uuid
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import Select, Sequence, delete, func, or_, select
|
||||
from sqlalchemy import Select, Sequence, and_, delete, func, or_, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.domain.directory_import import DirectorySnapshot, EntitySnap, GroupSnap, OfficeSnap
|
||||
from src.domain.policy import VisibilityScope, company_groups_visible, legal_entities_visible
|
||||
from src.models.directories import (
|
||||
CompanyGroup,
|
||||
@ -370,3 +371,73 @@ async def get_regional_office(session: AsyncSession, office_id: uuid.UUID) -> Re
|
||||
async def list_regional_offices(session: AsyncSession) -> list[RegionalOffice]:
|
||||
"""Все филиалы по имени."""
|
||||
return list((await session.execute(select(RegionalOffice).order_by(RegionalOffice.name))).scalars().all())
|
||||
|
||||
|
||||
async def load_snapshot(session: AsyncSession) -> DirectorySnapshot:
|
||||
"""Снимок справочника для плана импорта (§13.2) тремя запросами, без фильтра видимости (импорт — администратор).
|
||||
|
||||
Филиалы — по имени. ГК — только с `crm_id` (импорт сопоставляет по `iD ГК`), включая архивные: иначе план создал
|
||||
бы ГК с занятым `crm_id`; ГСК/ГСЗ ГК — первая по `code` группа каждого вида. ЮЛ — по ИНН с текущими членствами
|
||||
(`is_current`) каждого вида.
|
||||
"""
|
||||
offices = {
|
||||
office.name: OfficeSnap(id=office.id, name=office.name)
|
||||
for office in (await session.execute(select(RegionalOffice))).scalars()
|
||||
}
|
||||
|
||||
groups: dict[str, GroupSnap] = {}
|
||||
group_stmt = (
|
||||
select(
|
||||
CompanyGroup.id,
|
||||
CompanyGroup.crm_id,
|
||||
CompanyGroup.name,
|
||||
CompanyGroup.base_regional_office_id,
|
||||
RelatedGroup.id,
|
||||
RelatedGroup.kind,
|
||||
)
|
||||
.outerjoin(RelatedGroup, RelatedGroup.company_group_id == CompanyGroup.id)
|
||||
.where(CompanyGroup.crm_id.is_not(None))
|
||||
.order_by(CompanyGroup.id, RelatedGroup.code)
|
||||
)
|
||||
for group_id, crm_id, name, office_id, related_id, kind in (await session.execute(group_stmt)).all():
|
||||
snap = groups.get(crm_id)
|
||||
if snap is None:
|
||||
snap = GroupSnap(id=group_id, crm_id=crm_id, name=name, office_id=office_id, gsk_id=None, gsz_id=None)
|
||||
groups[crm_id] = snap
|
||||
if kind is RelatedGroupKind.gsk and snap.gsk_id is None:
|
||||
snap.gsk_id = related_id
|
||||
elif kind is RelatedGroupKind.gsz and snap.gsz_id is None:
|
||||
snap.gsz_id = related_id
|
||||
|
||||
entities: dict[str, EntitySnap] = {}
|
||||
entity_stmt = select(
|
||||
LegalEntity.id,
|
||||
LegalEntity.inn,
|
||||
LegalEntity.name,
|
||||
LegalEntity.entity_type,
|
||||
LegalEntity.company_group_id,
|
||||
RelatedGroupMember.related_group_id,
|
||||
RelatedGroupMember.kind,
|
||||
).outerjoin(
|
||||
RelatedGroupMember,
|
||||
and_(RelatedGroupMember.legal_entity_id == LegalEntity.id, RelatedGroupMember.is_current.is_(True)),
|
||||
)
|
||||
for entity_id, inn, name, entity_type, group_id, related_id, kind in (await session.execute(entity_stmt)).all():
|
||||
entity = entities.get(inn)
|
||||
if entity is None:
|
||||
entity = EntitySnap(
|
||||
id=entity_id,
|
||||
inn=inn,
|
||||
name=name,
|
||||
entity_type=entity_type,
|
||||
group_id=group_id,
|
||||
gsk_group_id=None,
|
||||
gsz_group_id=None,
|
||||
)
|
||||
entities[inn] = entity
|
||||
if kind is RelatedGroupKind.gsk:
|
||||
entity.gsk_group_id = related_id
|
||||
elif kind is RelatedGroupKind.gsz:
|
||||
entity.gsz_group_id = related_id
|
||||
|
||||
return DirectorySnapshot(offices_by_name=offices, groups_by_crm=groups, entities_by_inn=entities)
|
||||
|
||||
30
api/src/schemas/directory_import.py
Normal file
30
api/src/schemas/directory_import.py
Normal file
@ -0,0 +1,30 @@
|
||||
"""Схема ответа импорта справочников (spec directory-import): отчёт = `ImportReport.to_dict()` (§13.3)."""
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from src.domain.directory_import import ImportCode
|
||||
|
||||
|
||||
class RowResultOut(BaseModel):
|
||||
"""Строка отчёта: номер строки файла, ИНН, код §13.2 (итоговый или сопутствующее событие) и пояснение."""
|
||||
|
||||
row: int
|
||||
inn: str
|
||||
code: ImportCode
|
||||
message: str
|
||||
|
||||
|
||||
class ImportReportOut(BaseModel):
|
||||
"""Счётчики и построчный список; `errors` — число пропущенных строк. Тот же объект — в `audit_event.after_value`."""
|
||||
|
||||
rows: int
|
||||
groups_created: int
|
||||
groups_updated: int
|
||||
groups_skipped: int
|
||||
entities_created: int
|
||||
entities_updated: int
|
||||
entities_skipped: int
|
||||
memberships_created: int
|
||||
offices_created: int
|
||||
errors: int
|
||||
lines: list[RowResultOut]
|
||||
265
api/src/services/directory_import.py
Normal file
265
api/src/services/directory_import.py
Normal file
@ -0,0 +1,265 @@
|
||||
"""Сервис импорта справочников из выгрузки ЕЦП CRM (spec directory-import, audit-trail; `05-domain-rules.md` §13.3).
|
||||
|
||||
Чистое ядро (`src.domain.directory_import`) читает файл и строит план над снимком справочника; сервис применяет
|
||||
операции плана в транзакции запроса (`SessionDep`). Ошибки строк импорт не прерывают (они — коды отчёта), ошибки файла —
|
||||
422 до любых записей.
|
||||
|
||||
- `dry_run = true`: план применяется и откатывается явным `session.rollback()` — отчёт тот же, что у боевого прогона;
|
||||
ни `audit_event(DIRECTORY_IMPORT)`, ни файла в S3. Коды ГК/ГСК/ГСЗ из последовательностей при этом расходуются
|
||||
(последовательности PostgreSQL не транзакционны) — дыры в нумерации допустимы (Review Focus 4).
|
||||
- `dry_run = false`: `audit_event(DIRECTORY_IMPORT, after_value = отчёт, description = имя файла)`, файл — в S3
|
||||
`uploads/imports/{audit_event_id}/{имя}` с записью `s3_object(bucket_role = uploads, created_by = user:{id})`.
|
||||
Без S3 (dev, `app.state.storage is None`) — предупреждение в лог, `s3_object` не создаётся.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import datetime
|
||||
import hashlib
|
||||
import uuid
|
||||
from collections import defaultdict
|
||||
from pathlib import PurePosixPath
|
||||
from typing import Any, Protocol
|
||||
|
||||
from fastapi import HTTPException, status
|
||||
from loguru import logger
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.api.dependencies.audit import RequestAudit
|
||||
from src.common.clock import utc_now
|
||||
from src.common.exceptions import DomainValidationError
|
||||
from src.core.config import Settings
|
||||
from src.domain.directory_import import (
|
||||
AddMember,
|
||||
CreateEntity,
|
||||
CreateGroup,
|
||||
CreateOffice,
|
||||
CreateRelatedGroup,
|
||||
FileFormatError,
|
||||
ImportReport,
|
||||
OnExisting,
|
||||
Operation,
|
||||
UpdateEntity,
|
||||
UpdateGroup,
|
||||
plan_import,
|
||||
read_rows,
|
||||
)
|
||||
from src.models.directories import (
|
||||
COMPANY_GROUP_CODE_SEQ,
|
||||
RELATED_GROUP_CODE_SEQ,
|
||||
CompanyGroup,
|
||||
LegalEntity,
|
||||
RegionalOffice,
|
||||
RelatedGroup,
|
||||
RelatedGroupMember,
|
||||
)
|
||||
from src.models.enums import AuditAction, AuditSection, S3BucketRole
|
||||
from src.models.files import S3Object
|
||||
from src.repositories import directories as repo
|
||||
from src.services.company_groups import COMPANY_GROUP_CODE_PREFIX, apply_group_update
|
||||
from src.services.legal_entities import apply_entity_update
|
||||
from src.services.related_groups import CODE_PREFIXES
|
||||
|
||||
OBJECT_TYPE = "directory_import"
|
||||
IMPORT_DESCRIPTION = "импорт ЕЦП" # `description` событий изменения ГК/ЮЛ, сделанных импортом
|
||||
S3_PREFIX = "uploads/imports"
|
||||
DEFAULT_FILENAME = "import"
|
||||
CONTENT_TYPES = {
|
||||
"xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||||
"csv": "text/csv",
|
||||
}
|
||||
MEGABYTE = 1024 * 1024
|
||||
|
||||
|
||||
class ObjectStorage(Protocol):
|
||||
"""Что импорту нужно от S3 (`src.core.clients.S3Storage`): бакет и запись объекта."""
|
||||
|
||||
bucket: str
|
||||
|
||||
async def put_object(self, obj_name: str, file_data: bytes) -> Any: ...
|
||||
|
||||
|
||||
def check_size(size: int, limit: int) -> None:
|
||||
"""413, если файл больше `IMPORT_MAX_BYTES`."""
|
||||
if size > limit:
|
||||
raise HTTPException(status_code=status.HTTP_413_CONTENT_TOO_LARGE, detail=f"файл больше {limit // MEGABYTE} МБ")
|
||||
|
||||
|
||||
def safe_filename(filename: str | None) -> str:
|
||||
"""Имя файла без пути (клиент может прислать `C:\\…\\x.xlsx` или `../x.csv`): для формата, аудита и ключа S3."""
|
||||
name = PurePosixPath((filename or "").replace("\\", "/")).name.strip()
|
||||
return name if name not in {"", ".", ".."} else DEFAULT_FILENAME
|
||||
|
||||
|
||||
class DirectoryImportService:
|
||||
"""Импорт выгрузки ЕЦП от имени администратора справочников (право `DIRECTORY_ADMIN` — на маршруте)."""
|
||||
|
||||
def __init__(
|
||||
self, session: AsyncSession, audit: RequestAudit, storage: ObjectStorage | None, settings: Settings
|
||||
) -> None:
|
||||
self._session = session
|
||||
self._audit = audit
|
||||
self._storage = storage
|
||||
self._settings = settings
|
||||
|
||||
async def run(self, *, filename: str, content: bytes, dry_run: bool, on_existing: OnExisting) -> ImportReport:
|
||||
"""Прочитать файл, построить план над снимком справочника, применить; при `dry_run` — откатить."""
|
||||
check_size(len(content), self._settings.IMPORT_MAX_BYTES)
|
||||
name = safe_filename(filename)
|
||||
try:
|
||||
# Разбор xlsx на тысячах строк — сотни миллисекунд CPU: не блокируем цикл событий
|
||||
rows = await asyncio.to_thread(read_rows, name, content)
|
||||
except FileFormatError as error:
|
||||
raise DomainValidationError(error.code, error.message, missing=error.missing) from error
|
||||
snapshot = await repo.load_snapshot(self._session)
|
||||
plan = plan_import(rows, snapshot, on_existing)
|
||||
now = utc_now()
|
||||
await self._apply(plan.operations, now)
|
||||
if dry_run:
|
||||
# Транзакция запроса откатывается целиком; `SessionDep` затем ничего не фиксирует
|
||||
await self._session.rollback()
|
||||
return plan.report
|
||||
await self._record(name, content, plan.report, now)
|
||||
return plan.report
|
||||
|
||||
async def _apply(self, operations: list[Operation], now: datetime.datetime) -> None:
|
||||
"""Операции плана пакетами по типу в порядке зависимостей FK; `flush` после каждого пакета."""
|
||||
batches: defaultdict[type, list[Any]] = defaultdict(list)
|
||||
for operation in operations:
|
||||
batches[type(operation)].append(operation)
|
||||
user_id = self._audit.user_id
|
||||
session = self._session
|
||||
|
||||
session.add_all(
|
||||
RegionalOffice(id=op.id, name=op.name, is_head_office=False, is_active=True) for op in batches[CreateOffice]
|
||||
)
|
||||
await session.flush()
|
||||
|
||||
session.add_all(
|
||||
[
|
||||
CompanyGroup(
|
||||
id=op.id,
|
||||
code=await repo.next_code(session, COMPANY_GROUP_CODE_SEQ, COMPANY_GROUP_CODE_PREFIX),
|
||||
crm_id=op.crm_id,
|
||||
name=op.name,
|
||||
base_regional_office_id=op.office_id,
|
||||
valid_from=now,
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
updated_by=user_id,
|
||||
)
|
||||
for op in batches[CreateGroup]
|
||||
]
|
||||
)
|
||||
await session.flush()
|
||||
|
||||
await self._update_groups(batches[UpdateGroup], now)
|
||||
|
||||
session.add_all(
|
||||
LegalEntity(
|
||||
id=op.id,
|
||||
inn=op.inn,
|
||||
entity_type=op.entity_type,
|
||||
name=op.name,
|
||||
company_group_id=op.group_id,
|
||||
regional_office_id=op.office_id,
|
||||
valid_from=now,
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
updated_by=user_id,
|
||||
)
|
||||
for op in batches[CreateEntity]
|
||||
)
|
||||
await session.flush()
|
||||
|
||||
await self._update_entities(batches[UpdateEntity], now)
|
||||
|
||||
session.add_all(
|
||||
[
|
||||
RelatedGroup(
|
||||
id=op.id,
|
||||
code=await repo.next_code(session, RELATED_GROUP_CODE_SEQ, CODE_PREFIXES[op.kind]),
|
||||
company_group_id=op.group_id,
|
||||
kind=op.kind,
|
||||
name=op.name,
|
||||
)
|
||||
for op in batches[CreateRelatedGroup]
|
||||
]
|
||||
)
|
||||
await session.flush()
|
||||
|
||||
session.add_all(
|
||||
RelatedGroupMember(
|
||||
related_group_id=op.related_group_id,
|
||||
legal_entity_id=op.entity_id,
|
||||
kind=op.kind,
|
||||
valid_from=now,
|
||||
valid_to=None,
|
||||
is_current=True,
|
||||
created_by=user_id,
|
||||
)
|
||||
for op in batches[AddMember]
|
||||
)
|
||||
await session.flush()
|
||||
|
||||
async def _update_groups(self, operations: list[UpdateGroup], now: datetime.datetime) -> None:
|
||||
"""`UpdateGroup` — той же функцией, что PATCH ГК (R-14): снимок истории и аудит изменения ГК."""
|
||||
if not operations:
|
||||
return
|
||||
stmt = select(CompanyGroup).where(CompanyGroup.id.in_([op.id for op in operations]))
|
||||
groups = {group.id: group for group in (await self._session.execute(stmt)).scalars()}
|
||||
for op in operations:
|
||||
changes: dict[str, Any] = {}
|
||||
if op.name is not None:
|
||||
changes["name"] = op.name
|
||||
if op.office_id is not None:
|
||||
changes["base_regional_office_id"] = op.office_id
|
||||
await apply_group_update(
|
||||
self._session, self._audit, groups[op.id], changes, now=now, description=IMPORT_DESCRIPTION
|
||||
)
|
||||
|
||||
async def _update_entities(self, operations: list[UpdateEntity], now: datetime.datetime) -> None:
|
||||
"""`UpdateEntity` — той же функцией, что PATCH ЮЛ (R-14): снимок `legal_entity_history(UPDATE)` и аудит."""
|
||||
if not operations:
|
||||
return
|
||||
stmt = select(LegalEntity).where(LegalEntity.id.in_([op.id for op in operations]))
|
||||
entities = {entity.id: entity for entity in (await self._session.execute(stmt)).scalars()}
|
||||
for op in operations:
|
||||
changes: dict[str, Any] = {}
|
||||
if op.name is not None:
|
||||
changes["name"] = op.name
|
||||
if op.entity_type is not None:
|
||||
changes["entity_type"] = op.entity_type
|
||||
await apply_entity_update(
|
||||
self._session, self._audit, entities[op.id], changes, now=now, description=IMPORT_DESCRIPTION
|
||||
)
|
||||
|
||||
async def _record(self, name: str, content: bytes, report: ImportReport, now: datetime.datetime) -> None:
|
||||
"""След боевого импорта: событие аудита с отчётом и исходный файл в S3 с записью в `s3_object`."""
|
||||
event = await self._audit.record(
|
||||
action=AuditAction.DIRECTORY_IMPORT,
|
||||
section=AuditSection.DIRECTORIES,
|
||||
object_type=OBJECT_TYPE,
|
||||
description=name,
|
||||
after_value=report.to_dict(),
|
||||
)
|
||||
if self._storage is None:
|
||||
logger.warning("S3 не подключено: файл импорта не сохранён. audit_event_id={event_id}", event_id=event.id)
|
||||
return
|
||||
key = f"{S3_PREFIX}/{event.id}/{name}"
|
||||
# Строка реестра — до загрузки: конфликт ключа или ошибка БД не оставят в бакете объект без записи
|
||||
self._session.add(
|
||||
S3Object(
|
||||
id=uuid.uuid4(),
|
||||
bucket_role=S3BucketRole.uploads,
|
||||
bucket=self._storage.bucket,
|
||||
key=key,
|
||||
content_type=CONTENT_TYPES.get(PurePosixPath(name).suffix.lstrip(".").lower()),
|
||||
size_bytes=len(content),
|
||||
sha256=hashlib.sha256(content).hexdigest(),
|
||||
created_by=f"user:{self._audit.user_id}",
|
||||
created_at=now,
|
||||
)
|
||||
)
|
||||
await self._session.flush()
|
||||
await self._storage.put_object(key, content)
|
||||
558
api/tests/test_api_directory_import.py
Normal file
558
api/tests/test_api_directory_import.py
Normal file
@ -0,0 +1,558 @@
|
||||
"""API импорта справочников из выгрузки ЕЦП (spec directory-import, audit-trail; `05-domain-rules.md` §13.3).
|
||||
|
||||
Тесты боевого импорта идут поверх фикстуры `app` (транзакция теста с SAVEPOINT). `dry_run` проверяется на отдельной
|
||||
временной БД без SAVEPOINT-обёртки (`committed_app`): сервис делает явный `session.rollback()`, а внутри
|
||||
SAVEPOINT-обёртки `db_session` такой откат снимает корневую точку сохранения сессии теста вместе с seed — поведение
|
||||
отличалось бы от прода (там откатывается только транзакция запроса). Поэтому `dry_run` гоняется так же, как в проде:
|
||||
сессия запроса `async with session.begin()`, данные seed зафиксированы.
|
||||
"""
|
||||
|
||||
import re
|
||||
import uuid
|
||||
from collections.abc import AsyncIterator, Awaitable, Callable
|
||||
from contextlib import AsyncExitStack
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from fastapi import FastAPI
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||||
|
||||
from scripts.seed import run_seed
|
||||
from src.api.router import build_api_router
|
||||
from src.core.auth import build_verifier
|
||||
from src.core.config import get_settings
|
||||
from src.core.exception_handlers import register_exception_handlers
|
||||
from src.db.db import get_async_session
|
||||
from src.models.access import AppUser
|
||||
from src.models.audit_ex import AuditEvent
|
||||
from src.models.directories import (
|
||||
CompanyGroup,
|
||||
CompanyGroupHistory,
|
||||
LegalEntity,
|
||||
LegalEntityHistory,
|
||||
RegionalOffice,
|
||||
RelatedGroup,
|
||||
RelatedGroupMember,
|
||||
)
|
||||
from src.models.enums import AuditAction, AuditSection, EntityType, LeHistoryEvent, RelatedGroupKind, S3BucketRole
|
||||
from src.models.files import S3Object
|
||||
from tests.conftest import AuthClientFactory
|
||||
from tests.factories import make_company_group, make_legal_entity
|
||||
from tests.pg_helpers import alembic_async, create_scratch_db, drop_scratch_db
|
||||
|
||||
HTTP_OK = 200
|
||||
HTTP_FORBIDDEN = 403
|
||||
HTTP_TOO_LARGE = 413
|
||||
HTTP_UNPROCESSABLE = 422
|
||||
IMPORT = "/directories/import"
|
||||
ADMIN = "admin@armka.local"
|
||||
IMPORT_MAX_BYTES = 20 * 1024 * 1024
|
||||
XLSX_TYPE = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
||||
|
||||
RB_FILE = "ecp-export-russkiy-biskvit.xlsx"
|
||||
RB_CRM_ID = "1-AZFL6BQ"
|
||||
RB_GROUP = "Русский Бисквит"
|
||||
RB_OFFICE = "Санкт-Петербургский РФ"
|
||||
RB_FL_ROW = 12
|
||||
RB_ROWS = 21
|
||||
RB_ENTITIES = 20 # 18 ЮЛ + 2 ИП (R-15)
|
||||
RB_LEGAL_ENTITIES = 18
|
||||
RB_GSK_MEMBERS = ["ПЕКАРНЯ СЧАСТЛИВОЧНАЯ, ООО", "РУССКИЙ БИСКВИТ, АО"]
|
||||
RB_COUNTERS = {
|
||||
"rows": RB_ROWS,
|
||||
"groups_created": 1,
|
||||
"groups_updated": 0,
|
||||
"groups_skipped": 0,
|
||||
"entities_created": RB_ENTITIES,
|
||||
"entities_updated": 0,
|
||||
"entities_skipped": 0,
|
||||
"memberships_created": 2,
|
||||
"offices_created": 1,
|
||||
"errors": 1,
|
||||
}
|
||||
COUNTER_KEYS = tuple(RB_COUNTERS)
|
||||
FINAL_CODES = {
|
||||
"UNSUPPORTED_TYPE",
|
||||
"INVALID_INN",
|
||||
"DUPLICATE_IN_FILE",
|
||||
"LE_IN_OTHER_GROUP",
|
||||
"GK_NAME_CONFLICT",
|
||||
"LE_CREATED",
|
||||
"LE_UPDATED",
|
||||
"LE_EXISTS",
|
||||
}
|
||||
DIRECTORY_TABLES = (
|
||||
RegionalOffice,
|
||||
CompanyGroup,
|
||||
CompanyGroupHistory,
|
||||
LegalEntity,
|
||||
LegalEntityHistory,
|
||||
RelatedGroup,
|
||||
RelatedGroupMember,
|
||||
)
|
||||
TRACE_TABLES = (AuditEvent, S3Object)
|
||||
|
||||
HEADER = ["ИНН", "iD ГК", "Базовый РФ ГК в CRM", "Группа компаний", "Организация", "ГСК", "ГСЗ", "Тип"]
|
||||
KRASNODAR = "Краснодарский региональный филиал"
|
||||
|
||||
|
||||
class FakeStorage:
|
||||
"""Заглушка `S3Storage`: тот же интерфейс `bucket`/`put_object(obj_name, file_data)`, объекты — в словаре."""
|
||||
|
||||
def __init__(self, bucket: str = "arm-ka") -> None:
|
||||
self.bucket = bucket
|
||||
self.objects: dict[str, bytes] = {}
|
||||
|
||||
async def put_object(self, obj_name: str, file_data: bytes) -> None:
|
||||
self.objects[obj_name] = file_data
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def storage(app: FastAPI) -> FakeStorage:
|
||||
fake = FakeStorage()
|
||||
app.state.storage = fake
|
||||
return fake
|
||||
|
||||
|
||||
def _csv(*rows: list[str], header: list[str] = HEADER) -> bytes:
|
||||
return ("\n".join(";".join(cells) for cells in [header, *rows]) + "\n").encode("utf-8")
|
||||
|
||||
|
||||
def _row( # noqa: PLR0913 - ячейки строки выгрузки, необязательные — keyword-only
|
||||
inn: str, crm_id: str, group: str, entity: str, *, office: str = KRASNODAR, gsk: str = ""
|
||||
) -> list[str]:
|
||||
return [inn, crm_id, office, group, entity, gsk, "", "ЮЛ"]
|
||||
|
||||
|
||||
async def _post( # noqa: PLR0913 - поля multipart запроса импорта, необязательные — keyword-only
|
||||
client: httpx.AsyncClient,
|
||||
filename: str,
|
||||
content: bytes,
|
||||
*,
|
||||
dry_run: bool | None,
|
||||
on_existing: str | None = "skip",
|
||||
content_type: str = "text/csv",
|
||||
) -> httpx.Response:
|
||||
data: dict[str, str] = {}
|
||||
if dry_run is not None:
|
||||
data["dry_run"] = "true" if dry_run else "false"
|
||||
if on_existing is not None:
|
||||
data["on_existing"] = on_existing
|
||||
return await client.post(IMPORT, files={"file": (filename, content, content_type)}, data=data)
|
||||
|
||||
|
||||
async def _report(client: httpx.AsyncClient, filename: str, content: bytes, **kwargs: Any) -> dict[str, Any]:
|
||||
response = await _post(client, filename, content, **kwargs)
|
||||
assert response.status_code == HTTP_OK, response.text
|
||||
report: dict[str, Any] = response.json()
|
||||
return report
|
||||
|
||||
|
||||
async def _counts(
|
||||
session: AsyncSession, models: tuple[type[Any], ...] = DIRECTORY_TABLES + TRACE_TABLES
|
||||
) -> dict[str, int]:
|
||||
return {
|
||||
model.__tablename__: (await session.execute(select(func.count()).select_from(model))).scalar_one()
|
||||
for model in models
|
||||
}
|
||||
|
||||
|
||||
def _finals(report: dict[str, Any]) -> dict[int, str]:
|
||||
"""Итоговый код каждой строки (ровно один на строку)."""
|
||||
finals: dict[int, str] = {}
|
||||
for line in report["lines"]:
|
||||
if line["code"] in FINAL_CODES:
|
||||
assert line["row"] not in finals, line
|
||||
finals[line["row"]] = line["code"]
|
||||
return finals
|
||||
|
||||
|
||||
def _codes(report: dict[str, Any]) -> list[str]:
|
||||
return [line["code"] for line in report["lines"]]
|
||||
|
||||
|
||||
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 _import_events(session: AsyncSession) -> list[AuditEvent]:
|
||||
stmt = (
|
||||
select(AuditEvent)
|
||||
.where(AuditEvent.action == AuditAction.DIRECTORY_IMPORT)
|
||||
.order_by(AuditEvent.occurred_at, AuditEvent.id)
|
||||
)
|
||||
return list((await session.execute(stmt)).scalars().all())
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- боевой импорт
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.fixtures
|
||||
async def test_russkiy_biskvit_export_creates_directory(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage, fixtures_dir: Path
|
||||
) -> None:
|
||||
"""Срез РБ на базе после seed: 1 ГК, 1 РФ, 20 ЮЛ/ИП, 1 ГСК с двумя членами; аудит и файл в S3."""
|
||||
client = await auth_client(ADMIN)
|
||||
content = (fixtures_dir / "directories" / RB_FILE).read_bytes()
|
||||
before = await _counts(seeded)
|
||||
|
||||
report = await _report(client, RB_FILE, content, dry_run=False, content_type=XLSX_TYPE)
|
||||
|
||||
assert {key: report[key] for key in COUNTER_KEYS} == RB_COUNTERS
|
||||
assert _finals(report) == {n: "UNSUPPORTED_TYPE" if n == RB_FL_ROW else "LE_CREATED" for n in range(2, 23)}
|
||||
assert "DUPLICATE_IN_FILE" not in _codes(report)
|
||||
|
||||
after = await _counts(seeded)
|
||||
delta = {table: after[table] - before[table] for table in after}
|
||||
assert delta == {
|
||||
"regional_office": 1,
|
||||
"company_group": 1,
|
||||
"company_group_history": 0,
|
||||
"legal_entity": RB_ENTITIES,
|
||||
"legal_entity_history": 0,
|
||||
"related_group": 1,
|
||||
"related_group_member": 2,
|
||||
"audit_event": 1,
|
||||
"s3_object": 1,
|
||||
}
|
||||
|
||||
group = (await seeded.execute(select(CompanyGroup).where(CompanyGroup.crm_id == RB_CRM_ID))).scalar_one()
|
||||
office = (await seeded.execute(select(RegionalOffice).where(RegionalOffice.name == RB_OFFICE))).scalar_one()
|
||||
assert group.name == RB_GROUP
|
||||
assert re.fullmatch(r"GK-\d{6}", group.code)
|
||||
assert group.base_regional_office_id == office.id
|
||||
assert (office.is_head_office, office.is_active) == (False, True)
|
||||
entities = list(
|
||||
(await seeded.execute(select(LegalEntity).where(LegalEntity.company_group_id == group.id))).scalars().all()
|
||||
)
|
||||
assert len(entities) == RB_ENTITIES
|
||||
assert [e.entity_type for e in entities].count(EntityType.ЮЛ) == RB_LEGAL_ENTITIES
|
||||
assert {e.regional_office_id for e in entities} == {office.id}
|
||||
admin_id = await _user_id(seeded, ADMIN)
|
||||
assert {e.updated_by for e in entities} == {admin_id}
|
||||
|
||||
related = (await seeded.execute(select(RelatedGroup).where(RelatedGroup.company_group_id == group.id))).scalar_one()
|
||||
assert (related.kind, related.name) == (RelatedGroupKind.gsk, RB_GROUP)
|
||||
assert re.fullmatch(r"GSK-\d{6}", related.code)
|
||||
members = (
|
||||
await seeded.execute(
|
||||
select(RelatedGroupMember, LegalEntity)
|
||||
.join(LegalEntity, LegalEntity.id == RelatedGroupMember.legal_entity_id)
|
||||
.where(RelatedGroupMember.related_group_id == related.id)
|
||||
)
|
||||
).all()
|
||||
assert sorted(entity.name for _, entity in members) == RB_GSK_MEMBERS
|
||||
assert all(m.kind is RelatedGroupKind.gsk and m.is_current and m.created_by == admin_id for m, _ in members)
|
||||
|
||||
(event,) = await _import_events(seeded)
|
||||
assert (event.section, event.object_type, event.description, event.user_id) == (
|
||||
AuditSection.DIRECTORIES,
|
||||
"directory_import",
|
||||
RB_FILE,
|
||||
admin_id,
|
||||
)
|
||||
assert event.after_value is not None
|
||||
assert event.after_value["rows"] == RB_ROWS
|
||||
assert event.after_value == report
|
||||
|
||||
key = f"uploads/imports/{event.id}/{RB_FILE}"
|
||||
s3_object = (await seeded.execute(select(S3Object).where(S3Object.key == key))).scalar_one()
|
||||
assert (s3_object.bucket_role, s3_object.bucket, s3_object.size_bytes, s3_object.created_by) == (
|
||||
S3BucketRole.uploads,
|
||||
"arm-ka",
|
||||
len(content),
|
||||
f"user:{admin_id}",
|
||||
)
|
||||
assert s3_object.content_type == XLSX_TYPE
|
||||
assert storage.objects == {key: content}
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.fixtures
|
||||
async def test_repeated_import_with_skip_is_idempotent(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage, fixtures_dir: Path
|
||||
) -> None:
|
||||
"""Повтор того же файла с `skip`: все применённые строки — `LE_EXISTS`, ГК — `GK_EXISTS`, справочник не меняется."""
|
||||
client = await auth_client(ADMIN)
|
||||
content = (fixtures_dir / "directories" / RB_FILE).read_bytes()
|
||||
await _report(client, RB_FILE, content, dry_run=False)
|
||||
before = await _counts(seeded, DIRECTORY_TABLES)
|
||||
|
||||
report = await _report(client, RB_FILE, content, dry_run=False)
|
||||
|
||||
assert await _counts(seeded, DIRECTORY_TABLES) == before
|
||||
assert _finals(report) == {n: "UNSUPPORTED_TYPE" if n == RB_FL_ROW else "LE_EXISTS" for n in range(2, 23)}
|
||||
assert _codes(report).count("GK_EXISTS") == 1
|
||||
assert {key: report[key] for key in COUNTER_KEYS} == {
|
||||
**dict.fromkeys(COUNTER_KEYS, 0),
|
||||
"rows": RB_ROWS,
|
||||
"groups_skipped": 1,
|
||||
"entities_skipped": RB_ENTITIES,
|
||||
"errors": 1,
|
||||
}
|
||||
assert len(await _import_events(seeded)) == 2 # noqa: PLR2004
|
||||
assert len(storage.objects) == 2 # noqa: PLR2004 - у каждого импорта свой ключ (id события аудита)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_update_strategy_updates_names_with_history(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
|
||||
) -> None:
|
||||
"""`skip` не трогает изменённое имя (`LE_EXISTS`); `update` — имя обновлено, снимки истории, `LE_UPDATED`."""
|
||||
client = await auth_client(ADMIN)
|
||||
await _report(client, "first.csv", _csv(_row("7701000001", "1-IMP01", "ГК Импорт", "ООО Альфа")), dry_run=False)
|
||||
changed = _csv(_row("7701000001", "1-IMP01", "ГК Импорт Новая", "ООО Альфа Плюс"))
|
||||
|
||||
skipped = await _report(client, "changed.csv", changed, dry_run=False, on_existing="skip")
|
||||
entity = (await seeded.execute(select(LegalEntity).where(LegalEntity.inn == "7701000001"))).scalar_one()
|
||||
assert _finals(skipped) == {2: "LE_EXISTS"}
|
||||
assert "GK_EXISTS" in _codes(skipped)
|
||||
assert entity.name == "ООО Альфа"
|
||||
|
||||
updated = await _report(client, "changed.csv", changed, dry_run=False, on_existing="update")
|
||||
|
||||
assert _finals(updated) == {2: "LE_UPDATED"}
|
||||
assert "GK_UPDATED" in _codes(updated)
|
||||
assert (updated["entities_updated"], updated["groups_updated"]) == (1, 1)
|
||||
await seeded.refresh(entity)
|
||||
group = (await seeded.execute(select(CompanyGroup).where(CompanyGroup.crm_id == "1-IMP01"))).scalar_one()
|
||||
assert (entity.name, group.name) == ("ООО Альфа Плюс", "ГК Импорт Новая")
|
||||
(snapshot,) = (
|
||||
(await seeded.execute(select(LegalEntityHistory).where(LegalEntityHistory.legal_entity_id == entity.id)))
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
assert (snapshot.event_type, snapshot.name) == (LeHistoryEvent.UPDATE, "ООО Альфа")
|
||||
(group_snapshot,) = (
|
||||
(await seeded.execute(select(CompanyGroupHistory).where(CompanyGroupHistory.company_group_id == group.id)))
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
assert group_snapshot.name == "ГК Импорт"
|
||||
changes = (
|
||||
(
|
||||
await seeded.execute(
|
||||
select(AuditEvent).where(
|
||||
AuditEvent.action.in_([AuditAction.LEGAL_ENTITY_CHANGE, AuditAction.GROUP_CHANGE]),
|
||||
AuditEvent.object_id.in_([entity.id, group.id]),
|
||||
)
|
||||
)
|
||||
)
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
assert {(e.action, e.description, e.changed_field) for e in changes} == {
|
||||
(AuditAction.LEGAL_ENTITY_CHANGE, "импорт ЕЦП", "name"),
|
||||
(AuditAction.GROUP_CHANGE, "импорт ЕЦП", "name"),
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_entity_in_other_group_is_not_moved(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
|
||||
) -> None:
|
||||
"""ИНН уже в другой ГК → `LE_IN_OTHER_GROUP`, ЮЛ не переведено, остальные строки применяются."""
|
||||
other = await make_company_group(seeded, name="ГК Чужая", crm_id="1-OTHER")
|
||||
entity = await make_legal_entity(seeded, name="ООО Чужое", inn="7702000002", group=other)
|
||||
client = await auth_client(ADMIN)
|
||||
content = _csv(
|
||||
_row("7702000002", "1-NEW01", "ГК Новая", "ООО Чужое"),
|
||||
_row("7702000003", "1-NEW01", "ГК Новая", "ООО Своё"),
|
||||
)
|
||||
|
||||
report = await _report(client, "other.csv", content, dry_run=False)
|
||||
|
||||
assert _finals(report) == {2: "LE_IN_OTHER_GROUP", 3: "LE_CREATED"}
|
||||
await seeded.refresh(entity)
|
||||
await seeded.refresh(other)
|
||||
assert (entity.company_group_id, entity.name) == (other.id, "ООО Чужое")
|
||||
assert other.name == "ГК Чужая"
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_import_without_storage_writes_audit_only(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||||
"""dev без S3 (`app.state.storage is None`): импорт применяется, аудит пишется, `s3_object` не создаётся."""
|
||||
client = await auth_client(ADMIN)
|
||||
before = await _counts(seeded, TRACE_TABLES)
|
||||
|
||||
report = await _report(
|
||||
client, "nos3.csv", _csv(_row("7703000001", "1-NOS3", "ГК Без S3", "ООО Бета")), dry_run=False
|
||||
)
|
||||
|
||||
assert report["entities_created"] == 1
|
||||
assert await _counts(seeded, TRACE_TABLES) == {"audit_event": before["audit_event"] + 1, "s3_object": 0}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- ошибки файла и права
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_missing_column_is_422(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
|
||||
) -> None:
|
||||
client = await auth_client(ADMIN)
|
||||
before = await _counts(seeded)
|
||||
content = _csv(_row("7704000001", "1-MISS", "ГК", "ООО")[:-1], header=HEADER[:-1])
|
||||
|
||||
response = await _post(client, "no-type.csv", content, dry_run=False)
|
||||
|
||||
assert response.status_code == HTTP_UNPROCESSABLE, response.text
|
||||
detail = response.json()["detail"]
|
||||
assert (detail["code"], detail["missing"]) == ("MISSING_COLUMNS", ["Тип"])
|
||||
assert await _counts(seeded) == before
|
||||
assert storage.objects == {}
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_cp1251_csv_is_422_not_500(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||||
"""Review Focus 3: csv из Excel в cp1251 → 422 `UNSUPPORTED_FILE` «ожидается UTF-8»."""
|
||||
client = await auth_client(ADMIN)
|
||||
content = _csv(_row("7705000001", "1-CP", "ГК Кодировка", "ООО Кодировка")).decode("utf-8").encode("cp1251")
|
||||
|
||||
response = await _post(client, "cp1251.csv", content, dry_run=False)
|
||||
|
||||
assert response.status_code == HTTP_UNPROCESSABLE, response.text
|
||||
detail = response.json()["detail"]
|
||||
assert detail["code"] == "UNSUPPORTED_FILE"
|
||||
assert "UTF-8" in detail["message"]
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_file_over_limit_is_413(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||||
assert get_settings().IMPORT_MAX_BYTES == IMPORT_MAX_BYTES
|
||||
client = await auth_client(ADMIN)
|
||||
before = await _counts(seeded)
|
||||
|
||||
response = await _post(client, "big.csv", b"x" * (IMPORT_MAX_BYTES + 1), dry_run=False)
|
||||
|
||||
assert response.status_code == HTTP_TOO_LARGE, response.text
|
||||
assert await _counts(seeded) == before
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_import_requires_directory_admin(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||||
client = await auth_client("ka_go@armka.local")
|
||||
before = await _counts(seeded)
|
||||
|
||||
response = await _post(client, "x.csv", _csv(_row("7706000001", "1-DENY", "ГК", "ООО")), dry_run=False)
|
||||
|
||||
assert response.status_code == HTTP_FORBIDDEN
|
||||
assert "DIRECTORY_ADMIN" in response.json()["detail"]
|
||||
assert await _counts(seeded) == before
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- dry_run (как в проде, без SAVEPOINT)
|
||||
|
||||
|
||||
ClientFactory = Callable[[str], Awaitable[httpx.AsyncClient]]
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def committed_sessionmaker() -> AsyncIterator[async_sessionmaker[AsyncSession]]:
|
||||
"""Отдельная временная БД: `alembic upgrade head` и seed (dev) ЗАФИКСИРОВАНЫ; удаляется после теста."""
|
||||
name = f"armcore_import_{uuid.uuid4().hex[:12]}"
|
||||
url = (await create_scratch_db(name)).render_as_string(hide_password=False)
|
||||
engine = create_async_engine(url, connect_args={"server_settings": {"search_path": "dkb"}})
|
||||
try:
|
||||
await alembic_async(url, "upgrade", "head")
|
||||
maker = async_sessionmaker(engine, expire_on_commit=False)
|
||||
async with maker() as session, session.begin():
|
||||
await run_seed(session, dev=True)
|
||||
yield maker
|
||||
finally:
|
||||
await engine.dispose()
|
||||
await drop_scratch_db(name)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def committed_client(
|
||||
committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch
|
||||
) -> AsyncIterator[ClientFactory]:
|
||||
"""Клиенты приложения, сессия запроса которого устроена как в проде (`src.db.db.get_async_session`)."""
|
||||
monkeypatch.setattr("src.core.auth.logger", SimpleNamespace(warning=lambda *args, **kwargs: None))
|
||||
settings = get_settings().model_copy(update={"DEBUG": True, "AUTH_DEV_ISSUER": True})
|
||||
application = FastAPI()
|
||||
register_exception_handlers(application)
|
||||
application.include_router(build_api_router(settings))
|
||||
|
||||
async def production_like_session() -> AsyncIterator[AsyncSession]:
|
||||
async with committed_sessionmaker() as session, session.begin():
|
||||
yield session
|
||||
|
||||
application.dependency_overrides[get_async_session] = production_like_session
|
||||
application.state.token_verifier = build_verifier(settings)
|
||||
application.state.storage = FakeStorage()
|
||||
|
||||
async with AsyncExitStack() as stack:
|
||||
|
||||
async def factory(email: str) -> httpx.AsyncClient:
|
||||
transport = httpx.ASGITransport(app=application)
|
||||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as anonymous:
|
||||
response = await anonymous.post("/api/v1/dev/token", json={"email": email})
|
||||
assert response.status_code == HTTP_OK, response.text
|
||||
token = response.json()["access_token"]
|
||||
client = httpx.AsyncClient(
|
||||
transport=transport, base_url="http://test/api/v1", headers={"Authorization": f"Bearer {token}"}
|
||||
)
|
||||
return await stack.enter_async_context(client)
|
||||
|
||||
yield factory
|
||||
assert isinstance(application.state.storage, FakeStorage)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_dry_run_leaves_no_trace_and_repeats_identically(
|
||||
committed_client: ClientFactory, committed_sessionmaker: async_sessionmaker[AsyncSession]
|
||||
) -> None:
|
||||
"""`dry_run` (по умолчанию): отчёт как у боевого, ни строк справочника, ни аудита, ни `s3_object`; повтор `dry_run`
|
||||
даёт тот же отчёт (Review Focus 4: коды из последовательностей расходуются, но не видны и не ломают прогон)."""
|
||||
client = await committed_client(ADMIN)
|
||||
content = _csv(
|
||||
_row("7707000001", "1-DRY01", "ГК Проба", "ООО Проба", office="Новый РФ", gsk="входит в ГСК"),
|
||||
_row("7707000002", "1-DRY01", "ГК Проба", "ООО Проба-2", office="Новый РФ", gsk="входит в ГСК"),
|
||||
_row("7707000001", "1-DRY01", "ГК Проба", "ООО Проба (дубль)", office="Новый РФ"),
|
||||
)
|
||||
|
||||
async def counts() -> dict[str, int]:
|
||||
async with committed_sessionmaker() as session:
|
||||
return await _counts(session)
|
||||
|
||||
before = await counts()
|
||||
|
||||
first = await _report(client, "dry.csv", content, dry_run=None, on_existing=None) # значения по умолчанию
|
||||
assert await counts() == before
|
||||
second = await _report(client, "dry.csv", content, dry_run=True)
|
||||
assert await counts() == before
|
||||
assert second == first
|
||||
assert {key: first[key] for key in COUNTER_KEYS} == {
|
||||
**dict.fromkeys(COUNTER_KEYS, 0),
|
||||
"rows": 3,
|
||||
"groups_created": 1,
|
||||
"entities_created": 2,
|
||||
"memberships_created": 2,
|
||||
"offices_created": 1,
|
||||
"errors": 1,
|
||||
}
|
||||
assert _finals(first) == {2: "LE_CREATED", 3: "LE_CREATED", 4: "DUPLICATE_IN_FILE"}
|
||||
|
||||
real = await _report(client, "dry.csv", content, dry_run=False)
|
||||
|
||||
assert real == first # отчёт dry_run идентичен боевому (§13.3)
|
||||
after = await counts()
|
||||
assert {table: after[table] - before[table] for table in after} == {
|
||||
"regional_office": 1,
|
||||
"company_group": 1,
|
||||
"company_group_history": 0,
|
||||
"legal_entity": 2,
|
||||
"legal_entity_history": 0,
|
||||
"related_group": 1,
|
||||
"related_group_member": 2,
|
||||
"audit_event": 1,
|
||||
"s3_object": 1,
|
||||
}
|
||||
@ -153,7 +153,7 @@ async def run(self, *, filename: str, content: bytes, dry_run: bool, on_existing
|
||||
- Эндпоинт `POST /api/v1/directories/import` (`require("DIRECTORY_ADMIN")`): `file: UploadFile = File(...)`, `dry_run: bool = Form(True)`, `on_existing: OnExisting = Form(OnExisting.skip)`; `storage = request.app.state.storage`.
|
||||
- `load_snapshot` строит `EntitySnap.gsk_group_id/gsz_group_id` из текущих членств (`is_current`), `GroupSnap.gsk_id/gsz_id` — первая по `code` группа каждого вида у ГК.
|
||||
|
||||
- [ ] **Step 1: Тесты** (`tests/test_api_directory_import.py`, integration; клиент `admin@armka.local`; `app.state.storage` — заглушка `FakeStorage(bucket="arm-ka")` с `put_object(key, data)` в словарь):
|
||||
- [x] **Step 1: Тесты** (`tests/test_api_directory_import.py`, integration; клиент `admin@armka.local`; `app.state.storage` — заглушка `FakeStorage(bucket="arm-ka")` с `put_object(key, data)` в словарь):
|
||||
- срез РБ `dry_run=false, skip` на seed-базе → 200; счётчики как в Task 9; в БД 1 ГК с `crm_id="1-AZFL6BQ"`, `regional_office` «Санкт-Петербургский РФ», 20 ЮЛ/ИП, 1 ГСК (`name` = имя ГК, 2 члена с `kind=gsk`), `audit_event(DIRECTORY_IMPORT)` с `after_value["rows"] == 21`, `S3Object` с ключом `uploads/imports/<event.id>/ecp-export-russkiy-biskvit.xlsx`, байты в `FakeStorage`;
|
||||
- повтор `skip` → все `LE_EXISTS`/`GK_EXISTS`, число строк не изменилось;
|
||||
- csv с изменённым именем ЮЛ + `update` → имя обновлено, `legal_entity_history(UPDATE)`, код `LE_UPDATED`;
|
||||
@ -161,8 +161,8 @@ async def run(self, *, filename: str, content: bytes, dry_run: bool, on_existing
|
||||
- `dry_run=true` → 200, счётчики как боевые, число строк во всех таблицах и `audit_event`/`s3_object` не изменилось; второй `dry_run` — тот же отчёт (Review Focus 4);
|
||||
- csv без «Тип» → 422 `MISSING_COLUMNS` с `missing`; csv в cp1251 → 422 `UNSUPPORTED_FILE` (Review Focus 3); файл 20 МБ + 1 байт → 413;
|
||||
- `ka_go` (нет `DIRECTORY_ADMIN`) → 403.
|
||||
- [ ] **Step 2–4:** падает → реализация → `pytest -m integration tests/test_api_directory_import.py -v`, `make check` → PASS.
|
||||
- [ ] **Step 5: Commit** `feat(directories): импорт выгрузки ЕЦП — сервис, эндпоинт, аудит, файл в S3`.
|
||||
- [x] **Step 2–4:** падает → реализация → `pytest -m integration tests/test_api_directory_import.py -v`, `make check` → PASS.
|
||||
- [x] **Step 5: Commit** `feat(directories): импорт выгрузки ЕЦП — сервис, эндпоинт, аудит, файл в S3`.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@ -52,7 +52,7 @@ test-integration`). Пути — от корня репозитория. Доп
|
||||
Unit-тесты без БД: синтетический csv, покрывающий каждый код §13.2 ровно один раз; xlsx-фикстура
|
||||
`../fixtures/directories/ecp-export-russkiy-biskvit.xlsx` → ожидаемый план (1 ГК, 1 РФ, 20 ЮЛ/ИП, 1 ГСК с 2 членами,
|
||||
коды ФЛ/дубля по фактическому порядку строк).
|
||||
- [ ] 3.2 **Сервис и эндпоинт.** `services/directory_import.py` (снимок справочника одним запросом на таблицу, применение плана,
|
||||
- [x] 3.2 **Сервис и эндпоинт.** `services/directory_import.py` (снимок справочника одним запросом на таблицу, применение плана,
|
||||
`dry_run` → rollback, `audit_event(DIRECTORY_IMPORT, after_value = отчёт)`, файл в S3 `uploads/imports/{audit_event_id}/{name}`
|
||||
+ `s3_object`), `POST /api/v1/directories/import` (multipart, `IMPORT_MAX_BYTES` = 20 МБ → 413, `DIRECTORY_ADMIN`),
|
||||
`schemas/directory_import.py`. Интеграционные тесты: сценарии спеки `directory-import` (срез РБ, идемпотентность, `update`,
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user