From c9c3b690764b8d3b41af470c30669cb16be345ce Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sun, 4 Oct 2026 23:15:26 +0400 Subject: [PATCH] =?UTF-8?q?feat(directories):=20=D0=B8=D0=BC=D0=BF=D0=BE?= =?UTF-8?q?=D1=80=D1=82=20=D0=B2=D1=8B=D0=B3=D1=80=D1=83=D0=B7=D0=BA=D0=B8?= =?UTF-8?q?=20=D0=95=D0=A6=D0=9F=20=E2=80=94=20=D1=81=D0=B5=D1=80=D0=B2?= =?UTF-8?q?=D0=B8=D1=81,=20=D1=8D=D0=BD=D0=B4=D0=BF=D0=BE=D0=B8=D0=BD?= =?UTF-8?q?=D1=82,=20=D0=B0=D1=83=D0=B4=D0=B8=D1=82,=20=D1=84=D0=B0=D0=B9?= =?UTF-8?q?=D0=BB=20=D0=B2=20S3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2 --- api/.env.example | 3 + api/.env.test | 3 + api/src/api/dependencies/directories.py | 25 +- api/src/api/endpoints/directories/__init__.py | 3 +- api/src/api/endpoints/directories/import_.py | 45 ++ api/src/core/config.py | 3 + api/src/repositories/directories.py | 73 ++- api/src/schemas/directory_import.py | 30 + api/src/services/directory_import.py | 265 +++++++++ api/tests/test_api_directory_import.py | 558 ++++++++++++++++++ .../2026-10-04-slice-1-directories-phase2.md | 6 +- openspec/changes/slice-1-directories/tasks.md | 2 +- 12 files changed, 1009 insertions(+), 7 deletions(-) create mode 100644 api/src/api/endpoints/directories/import_.py create mode 100644 api/src/schemas/directory_import.py create mode 100644 api/src/services/directory_import.py create mode 100644 api/tests/test_api_directory_import.py diff --git a/api/.env.example b/api/.env.example index 97311dc..5255f9c 100644 --- a/api/.env.example +++ b/api/.env.example @@ -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 diff --git a/api/.env.test b/api/.env.test index 8801f1d..59fcd77 100644 --- a/api/.env.test +++ b/api/.env.test @@ -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 diff --git a/api/src/api/dependencies/directories.py b/api/src/api/dependencies/directories.py index ca4390e..e4cd836 100644 --- a/api/src/api/dependencies/directories.py +++ b/api/src/api/dependencies/directories.py @@ -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)] diff --git a/api/src/api/endpoints/directories/__init__.py b/api/src/api/endpoints/directories/__init__.py index 979ca2f..9454158 100644 --- a/api/src/api/endpoints/directories/__init__.py +++ b/api/src/api/endpoints/directories/__init__.py @@ -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) diff --git a/api/src/api/endpoints/directories/import_.py b/api/src/api/endpoints/directories/import_.py new file mode 100644 index 0000000..3a45b94 --- /dev/null +++ b/api/src/api/endpoints/directories/import_.py @@ -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()) diff --git a/api/src/core/config.py b/api/src/core/config.py index 47f093d..91fdeb6 100644 --- a/api/src/core/config.py +++ b/api/src/core/config.py @@ -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"), diff --git a/api/src/repositories/directories.py b/api/src/repositories/directories.py index 5f883c4..0eea753 100644 --- a/api/src/repositories/directories.py +++ b/api/src/repositories/directories.py @@ -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) diff --git a/api/src/schemas/directory_import.py b/api/src/schemas/directory_import.py new file mode 100644 index 0000000..7cafcd5 --- /dev/null +++ b/api/src/schemas/directory_import.py @@ -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] diff --git a/api/src/services/directory_import.py b/api/src/services/directory_import.py new file mode 100644 index 0000000..bbe5245 --- /dev/null +++ b/api/src/services/directory_import.py @@ -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) diff --git a/api/tests/test_api_directory_import.py b/api/tests/test_api_directory_import.py new file mode 100644 index 0000000..a0d01a3 --- /dev/null +++ b/api/tests/test_api_directory_import.py @@ -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, + } diff --git a/docs/superpowers/plans/2026-10-04-slice-1-directories-phase2.md b/docs/superpowers/plans/2026-10-04-slice-1-directories-phase2.md index 1d3df8e..d44778a 100644 --- a/docs/superpowers/plans/2026-10-04-slice-1-directories-phase2.md +++ b/docs/superpowers/plans/2026-10-04-slice-1-directories-phase2.md @@ -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//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`. --- diff --git a/openspec/changes/slice-1-directories/tasks.md b/openspec/changes/slice-1-directories/tasks.md index 171265d..6fdfe28 100644 --- a/openspec/changes/slice-1-directories/tasks.md +++ b/openspec/changes/slice-1-directories/tasks.md @@ -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`,