From 2d69a9e8679cd2c6c915dd87788d94cd28d7585b Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Mon, 5 Oct 2026 21:35:41 +0400 Subject: [PATCH] =?UTF-8?q?chore:=20S3=20get=5Fbytes,=20=D0=BF=D1=80=D0=BE?= =?UTF-8?q?=D1=82=D0=BE=D0=BA=D0=BE=D0=BB=20ObjectStorage,=20=D1=84=D0=B0?= =?UTF-8?q?=D0=B1=D1=80=D0=B8=D0=BA=D0=B8=20=D1=84=D0=B0=D0=B9=D0=BB=D0=BE?= =?UTF-8?q?=D0=B2/=D0=B4=D0=BE=D0=BA=D1=83=D0=BC=D0=B5=D0=BD=D1=82=D0=BE?= =?UTF-8?q?=D0=B2,=20=D0=BD=D0=B0=D1=81=D1=82=D1=80=D0=BE=D0=B9=D0=BA?= =?UTF-8?q?=D0=B8=20=D0=B7=D0=B0=D0=B3=D1=80=D1=83=D0=B7=D0=BA=D0=B8=20?= =?UTF-8?q?=D0=B8=20=D0=BC=D0=B0=D0=BF=D0=BF=D0=B8=D0=BD=D0=B3=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2 --- api/.env.example | 6 + api/.env.test | 5 + api/pytest.ini | 1 + api/src/core/clients.py | 19 +++ api/src/core/config.py | 7 + api/src/core/storage.py | 23 +++ api/src/services/directory_import.py | 11 +- api/tests/conftest.py | 9 +- api/tests/factories.py | 131 ++++++++++++++- api/tests/test_clients.py | 85 ++++++++++ api/tests/test_config.py | 9 +- api/tests/test_factories.py | 152 +++++++++++++++++- .../plans/2026-10-05-slice-2c-osv-ingest.md | 6 +- openspec/changes/slice-2c-osv-ingest/tasks.md | 2 +- 14 files changed, 446 insertions(+), 20 deletions(-) create mode 100644 api/src/core/storage.py create mode 100644 api/tests/test_clients.py diff --git a/api/.env.example b/api/.env.example index 2a5ed79..cae2f29 100644 --- a/api/.env.example +++ b/api/.env.example @@ -57,3 +57,9 @@ OUTBOX_MAX_REPUBLISH=3 # Импорт справочников из выгрузки ЕЦП CRM: предельный размер файла в байтах (20 МБ), больше — ответ 413 IMPORT_MAX_BYTES=20971520 + +# Загрузка первичных файлов (ББ/ОФР/ОСВ): предельный размер файла в байтах (50 МБ), больше — ответ 413 +UPLOAD_MAX_BYTES=52428800 +# Маппинг результата распознавания: период опроса готовых заданий (секунды) и число попыток до отметки ошибки +MAPPING_POLL_SECONDS=5 +MAPPING_MAX_ATTEMPTS=3 diff --git a/api/.env.test b/api/.env.test index 8f8be01..979d167 100644 --- a/api/.env.test +++ b/api/.env.test @@ -50,3 +50,8 @@ OUTBOX_MAX_REPUBLISH=3 # Импорт справочников: предельный размер файла (20 МБ), как в .env.example IMPORT_MAX_BYTES=20971520 + +# Загрузка первичных файлов и маппинг результатов: значения как в .env.example +UPLOAD_MAX_BYTES=52428800 +MAPPING_POLL_SECONDS=5 +MAPPING_MAX_ATTEMPTS=3 diff --git a/api/pytest.ini b/api/pytest.ini index 12c269f..12d42ea 100644 --- a/api/pytest.ini +++ b/api/pytest.ini @@ -24,3 +24,4 @@ filterwarnings = markers = integration: тесты, которым нужны PostgreSQL/NATS (docker compose); запускаются отдельно от юнит-тестов fixtures: тесты на реальных данных из ../fixtures (внутренний доступ) + e2e: сквозные тесты Среза 2c на реальных данных (NATS + tabular-parser + MinIO + PostgreSQL); запускаются отдельно diff --git a/api/src/core/clients.py b/api/src/core/clients.py index ceb87e5..36dd199 100644 --- a/api/src/core/clients.py +++ b/api/src/core/clients.py @@ -5,6 +5,11 @@ from io import BytesIO from miniopy_async import Minio +from miniopy_async.error import S3Error + +from src.core.storage import ObjectNotFoundError + +NO_SUCH_KEY = "NoSuchKey" class S3Storage: @@ -26,3 +31,17 @@ class S3Storage: return response except: raise + + async def get_bytes(self, obj_name: str) -> bytes: + """Содержимое объекта целиком; response закрывается и отпускается; `NoSuchKey` → `ObjectNotFoundError`.""" + try: + response = await self.client.get_object(self.bucket, obj_name) + except S3Error as error: + if error.code == NO_SUCH_KEY: + raise ObjectNotFoundError(obj_name) from error + raise + try: + return await response.read() + finally: + response.close() + await response.release() diff --git a/api/src/core/config.py b/api/src/core/config.py index 69d174d..5f15f84 100644 --- a/api/src/core/config.py +++ b/api/src/core/config.py @@ -87,6 +87,13 @@ class Settings(BaseSettings): # Импорт справочников из выгрузки ЕЦП (§13.3): предельный размер файла, больше — 413 IMPORT_MAX_BYTES: PositiveInt = 20 * 1024 * 1024 + # Загрузка первичных файлов (Срез 2c): предельный размер файла, больше — 413 + UPLOAD_MAX_BYTES: PositiveInt = 50 * 1024 * 1024 + # Маппинг результата распознавания в `document`/`osv_entry`: период опроса заданий в статусе done/needs_check без + # документов и число попыток (0 и отрицательные значения отвергаются при старте) + MAPPING_POLL_SECONDS: PositiveInt = 5 + MAPPING_MAX_ATTEMPTS: PositiveInt = 3 + @field_validator("CORS_ORIGINS", mode="before") @classmethod def _split_origins(cls, value: Any) -> Any: diff --git a/api/src/core/storage.py b/api/src/core/storage.py new file mode 100644 index 0000000..5ef1544 --- /dev/null +++ b/api/src/core/storage.py @@ -0,0 +1,23 @@ +"""Контракт хранилища объектов (S3), которым пользуются сервисы ядра. + +Реализация — `src.core.clients.S3Storage`, заглушка тестов — `tests.conftest.FakeStorage`. Протокол вынесен из +`services/directory_import.py`: тот же интерфейс нужен загрузке файлов и маппингу результатов (Срез 2c, D-12). +""" + +from typing import Protocol + + +class ObjectNotFoundError(Exception): + """Объекта с таким ключом нет в бакете (S3 `NoSuchKey`).""" + + +class ObjectStorage(Protocol): + """Что ядру нужно от S3: бакет, запись и чтение объекта целиком.""" + + bucket: str + + async def put_object(self, obj_name: str, file_data: bytes) -> None: ... + + async def get_bytes(self, obj_name: str) -> bytes: + """Содержимое объекта; отсутствие объекта — `ObjectNotFoundError`.""" + ... diff --git a/api/src/services/directory_import.py b/api/src/services/directory_import.py index cf1c6b8..2cc47b2 100644 --- a/api/src/services/directory_import.py +++ b/api/src/services/directory_import.py @@ -19,7 +19,7 @@ import unicodedata import uuid from collections import defaultdict from pathlib import PurePosixPath -from typing import Any, Protocol +from typing import Any from fastapi import HTTPException, status from loguru import logger @@ -30,6 +30,7 @@ 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.core.storage import ObjectStorage from src.domain.directory_import import ( AddMember, CreateEntity, @@ -74,14 +75,6 @@ MAX_FILENAME_LENGTH = 200 # символов: ключ S3 `uploads/imports/{uui MAX_SUFFIX_LENGTH = 16 -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: diff --git a/api/tests/conftest.py b/api/tests/conftest.py index 7687153..dade4ff 100644 --- a/api/tests/conftest.py +++ b/api/tests/conftest.py @@ -38,6 +38,7 @@ 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.core.storage import ObjectNotFoundError from src.db.db import get_async_session, session_transaction from src.services.session_audit import SessionAudit from tests.fixtures_helpers import fixtures_skip_reason, resolve_fixtures_dir @@ -205,7 +206,7 @@ async def auth_client(app: FastAPI, seeded: AsyncSession) -> AsyncIterator[AuthC class FakeStorage: - """Заглушка `S3Storage`: тот же интерфейс `bucket`/`put_object(obj_name, file_data)`, объекты — в словаре.""" + """Заглушка `S3Storage` (протокол `ObjectStorage`): `bucket`/`put_object`/`get_bytes`, объекты — в словаре.""" def __init__(self, bucket: str = "arm-ka") -> None: self.bucket = bucket @@ -214,6 +215,12 @@ class FakeStorage: async def put_object(self, obj_name: str, file_data: bytes) -> None: self.objects[obj_name] = file_data + async def get_bytes(self, obj_name: str) -> bytes: + try: + return self.objects[obj_name] + except KeyError: + raise ObjectNotFoundError(obj_name) from None + @pytest_asyncio.fixture async def committed_sessionmaker() -> AsyncIterator[async_sessionmaker[AsyncSession]]: diff --git a/api/tests/factories.py b/api/tests/factories.py index c98b579..5af4174 100644 --- a/api/tests/factories.py +++ b/api/tests/factories.py @@ -4,6 +4,9 @@ `make_user` — справочники Среза 1 (`flush`, без `commit`): откатываются фикстурой `db_session`. Идентификаторы по умолчанию случайные (ИНН, коды, имена филиалов), поэтому фабрики можно вызывать несколько раз в одном тесте. - `make_osv_entry` — минимальная цепочка FK поверх сессии (`flush`, без `commit`): откатывается фикстурой `db_session`. +- `make_stored_file`, `make_document`, `make_result_object` — цепочка файла/задания/документа и объект результата в + `FakeStorage` для тестов загрузки и маппинга (Срез 2c; `flush`, без `commit`). `make_entries_from_result` создаёт + Task 9 (R-2). - `CommittedJobs` — задания распознавания, ЗАФИКСИРОВАННЫЕ в БД (`commit`): для фоновых процессов обмена, которые работают в собственных транзакциях и не видят чужую незафиксированную транзакцию. Всё созданное удаляется `cleanup()` в порядке FK, включая строки `s3_object` файлов результата, зарегистрированные inbox'ом. @@ -11,12 +14,13 @@ import datetime import decimal +import json import secrets import uuid from collections.abc import Sequence from typing import Any -from sqlalchemy import delete, func, select +from sqlalchemy import delete, func, select, update from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from src.models.access import AppUser, Role, UserRole @@ -33,6 +37,7 @@ from src.models.enums import ( EntityType, FileStatus, FileType, + PeriodQuarter, RecognitionPipeline, RecognitionStatus, RelatedGroupKind, @@ -40,8 +45,11 @@ from src.models.enums import ( ) from src.models.files import Document, RecognitionJob, S3Object, StoredFile from src.models.osv import OsvEntry +from tests.conftest import FakeStorage DEFAULT_FILE_TYPES = {RecognitionPipeline.tabular: FileType.XLSX, RecognitionPipeline.image: FileType.PDF} +TABULAR_FILE_TYPES = frozenset({FileType.XLS, FileType.XLSX, FileType.XLSB}) +FACTORY_BUCKET = "arm-ka" def _now() -> datetime.datetime: @@ -159,6 +167,116 @@ async def make_user( # noqa: PLR0913 - сигнатура фабрики фик return user +async def make_stored_file( # noqa: PLR0913 - сигнатура фабрики фиксирована планом + session: AsyncSession, + *, + entity: LegalEntity, + status: FileStatus = FileStatus.UPLOADED, + file_type: FileType = FileType.XLS, + original_name: str = "osv.xls", + content_hash: str | None = None, + job_status: RecognitionStatus = RecognitionStatus.queued, + s3_key: str | None = None, + uploaded_by: uuid.UUID | None = None, +) -> tuple[StoredFile, RecognitionJob]: + """`s3_object` (uploads) + `stored_file` + `recognition_job` (attempt 1), `current_recognition_job_id` = задание. + + Конвейер задания выводится из типа файла (XLS/XLSX/XLSB — tabular, иначе image). Данные остаются в транзакции + сессии (`flush`). + """ + now = _now() + s3 = S3Object( + bucket_role=S3BucketRole.uploads, + bucket=FACTORY_BUCKET, + key=s3_key or f"uploads/{uuid.uuid4()}/{original_name}", + created_at=now, + ) + session.add(s3) + await session.flush() + file = StoredFile( + legal_entity_id=entity.id, + file_type=file_type, + original_name=original_name, + s3_object_id=s3.id, + content_hash=content_hash, + status=status, + uploaded_by=uploaded_by, + uploaded_at=now, + ) + session.add(file) + await session.flush() + pipeline = RecognitionPipeline.tabular if file_type in TABULAR_FILE_TYPES else RecognitionPipeline.image + job = RecognitionJob( + file_id=file.id, + pipeline=pipeline, + request_subject=f"recognition.{pipeline.value}.request", + schema_version="1.0", + status=job_status, + ) + session.add(job) + await session.flush() + file.current_recognition_job_id = job.id + await session.flush() + return file, job + + +async def make_document( # noqa: PLR0913 - сигнатура фабрики фиксирована планом + session: AsyncSession, + *, + file: StoredFile, + job: RecognitionJob, + account: str = "60", + year: int = 2025, + quarter: PeriodQuarter = PeriodQuarter.Q1, + status: DocumentStatus = DocumentStatus.RECOGNIZED, + is_cumulative: bool = True, +) -> Document: + """ОСВ-документ по счёту `account` файла `file` (`flush`). + + `job` — задание, породившее документ (у `document` FK на задание нет; параметр держит вызовы единообразными с + `make_stored_file`). + """ + assert job.file_id == file.id # noqa: S101 - защита от перепутанных аргументов в тестах + now = _now() + document = Document( + stored_file_id=file.id, + legal_entity_id=file.legal_entity_id, + document_type=DocumentType.OSV, + osv_account_code=account, + period_year=year, + period_quarter=quarter, + is_cumulative=is_cumulative, + status=status, + created_at=now, + updated_at=now, + ) + session.add(document) + await session.flush() + return document + + +async def make_result_object( + session: AsyncSession, *, job: RecognitionJob, storage: FakeStorage, result: dict[str, Any] +) -> S3Object: + """JSON результата — в `storage` под `results/{file_id}/{job_id}/result.json` + `s3_object` + ссылка в задании.""" + key = f"results/{job.file_id}/{job.id}/result.json" + payload = json.dumps(result, ensure_ascii=False).encode("utf-8") + await storage.put_object(key, payload) + s3 = S3Object( + bucket_role=S3BucketRole.results, + bucket=storage.bucket, + key=key, + content_type="application/json", + size_bytes=len(payload), + created_at=_now(), + ) + session.add(s3) + await session.flush() + job.result_s3_object_id = s3.id + await session.flush() + return s3 + + async def make_osv_entry(session: AsyncSession, **overrides: Any) -> OsvEntry: """Строка ОСВ с минимальной цепочкой FK (ЮЛ, s3_object, stored_file, recognition_job, document); `flush`. @@ -320,9 +438,17 @@ class CommittedJobs: self._extra_s3_ids.append(s3_object_id) async def cleanup(self) -> None: - """Удаляет всё созданное в порядке FK (файлы результата, на которые ссылаются задания, — после заданий).""" + """Удаляет всё созданное в порядке FK: строки ОСВ и документы файлов → задания → файлы → объекты S3 и ЮЛ.""" async with self._session_factory() as session, session.begin(): result_s3_ids: list[uuid.UUID] = [] + if self._file_ids: + # задания ссылаются на файл, файл — на текущее задание (use_alter): сначала разрываем обратную ссылку + await session.execute( + update(StoredFile).where(StoredFile.id.in_(self._file_ids)).values(current_recognition_job_id=None) + ) + documents = select(Document.id).where(Document.stored_file_id.in_(self._file_ids)) + await session.execute(delete(OsvEntry).where(OsvEntry.document_id.in_(documents))) + await session.execute(delete(Document).where(Document.stored_file_id.in_(self._file_ids))) if self.job_ids: rows = await session.execute( select(RecognitionJob.result_s3_object_id).where( @@ -330,6 +456,7 @@ class CommittedJobs: ) ) result_s3_ids = [row[0] for row in rows] + await session.execute(delete(OsvEntry).where(OsvEntry.recognition_job_id.in_(self.job_ids))) await session.execute(delete(RecognitionJob).where(RecognitionJob.id.in_(self.job_ids))) if self._file_ids: await session.execute(delete(StoredFile).where(StoredFile.id.in_(self._file_ids))) diff --git a/api/tests/test_clients.py b/api/tests/test_clients.py new file mode 100644 index 0000000..4f7c6eb --- /dev/null +++ b/api/tests/test_clients.py @@ -0,0 +1,85 @@ +"""`S3Storage.get_bytes` на заглушке Minio (без сети): +чтение, закрытие response, `NoSuchKey` → `ObjectNotFoundError`.""" + +from typing import Any +from unittest.mock import MagicMock + +import pytest +from miniopy_async.error import S3Error + +from src.core.clients import S3Storage +from src.core.storage import ObjectNotFoundError + + +class StubResponse: + """Ответ `aiohttp` с флагами закрытия.""" + + def __init__(self, data: bytes, *, fail_read: bool = False) -> None: + self._data = data + self._fail_read = fail_read + self.closed = False + self.released = False + + async def read(self) -> bytes: + if self._fail_read: + raise ConnectionError("обрыв чтения") + return self._data + + def close(self) -> None: + self.closed = True + + async def release(self) -> None: + self.released = True + + +class StubMinio: + def __init__(self, response: StubResponse | None = None, error: Exception | None = None) -> None: + self._response = response + self._error = error + self.calls: list[tuple[str, str]] = [] + + async def get_object(self, bucket: str, obj_name: str) -> StubResponse: + self.calls.append((bucket, obj_name)) + if self._error is not None: + raise self._error + assert self._response is not None # noqa: S101 + return self._response + + +def _s3_error(code: str) -> S3Error: + return S3Error(code, "сообщение", None, None, None, MagicMock()) + + +def _storage(minio: StubMinio) -> S3Storage: + client: Any = minio + return S3Storage(client, "arm-ka") + + +async def test_get_bytes_returns_content_and_closes_response() -> None: + response = StubResponse(b"payload") + minio = StubMinio(response) + + assert await _storage(minio).get_bytes("uploads/a/b.xls") == b"payload" + assert minio.calls == [("arm-ka", "uploads/a/b.xls")] + assert response.closed is True + assert response.released is True + + +async def test_get_bytes_closes_response_when_read_fails() -> None: + response = StubResponse(b"", fail_read=True) + + with pytest.raises(ConnectionError): + await _storage(StubMinio(response)).get_bytes("k") + assert response.closed is True + assert response.released is True + + +async def test_get_bytes_no_such_key_is_object_not_found() -> None: + with pytest.raises(ObjectNotFoundError): + await _storage(StubMinio(error=_s3_error("NoSuchKey"))).get_bytes("missing") + + +async def test_get_bytes_other_s3_error_is_not_swallowed() -> None: + with pytest.raises(S3Error) as caught: + await _storage(StubMinio(error=_s3_error("AccessDenied"))).get_bytes("k") + assert caught.value.code == "AccessDenied" diff --git a/api/tests/test_config.py b/api/tests/test_config.py index 23cc200..c570c8a 100644 --- a/api/tests/test_config.py +++ b/api/tests/test_config.py @@ -15,7 +15,10 @@ API_DIR = Path(__file__).resolve().parent.parent def clean_env(monkeypatch: pytest.MonkeyPatch) -> pytest.MonkeyPatch: """Убирает из окружения процесса переменные, которые перебили бы значения из env-файла.""" for key in list(os.environ): - if key.startswith(("OPENBAO__", "AUTH_", "OUTBOX_", "ACK_", "SERVER__")) or key in {"NATS_ENABLED", "DEBUG"}: + if key.startswith(("OPENBAO__", "AUTH_", "OUTBOX_", "ACK_", "SERVER__", "UPLOAD_", "MAPPING_")) or key in { + "NATS_ENABLED", + "DEBUG", + }: monkeypatch.delenv(key) return monkeypatch @@ -54,6 +57,7 @@ def test_defaults(clean_env: pytest.MonkeyPatch) -> None: assert s.AUTH_EMAIL_CLAIM == "email" assert s.NATS_ENABLED is False assert (s.OUTBOX_POLL_SECONDS, s.ACK_TIMEOUT_SECONDS, s.OUTBOX_MAX_REPUBLISH) == (5, 30, 3) + assert (s.UPLOAD_MAX_BYTES, s.MAPPING_POLL_SECONDS, s.MAPPING_MAX_ATTEMPTS) == (50 * 1024 * 1024, 5, 3) assert s.OPENBAO.NATS.NATS_SEED == "" assert s.OPENBAO.NATS.NATS_JWT == "" assert s.OPENBAO.MINIO.MINIO_SECURE is True @@ -90,6 +94,9 @@ REQUIRED_ENV = { ("ACK_TIMEOUT_SECONDS", "0"), ("ACK_TIMEOUT_SECONDS", "-5"), ("OUTBOX_MAX_REPUBLISH", "-1"), + ("UPLOAD_MAX_BYTES", "0"), + ("MAPPING_POLL_SECONDS", "0"), + ("MAPPING_MAX_ATTEMPTS", "0"), ], ) def test_exchange_settings_reject_invalid(clean_env: pytest.MonkeyPatch, key: str, value: str) -> None: diff --git a/api/tests/test_factories.py b/api/tests/test_factories.py index c04d5f5..6b753b6 100644 --- a/api/tests/test_factories.py +++ b/api/tests/test_factories.py @@ -1,19 +1,40 @@ """Фабрики справочников и фикстуры `app` / `auth_client` / `seeded` (задача 1.3): проверка до API-тестов.""" +import json +from decimal import Decimal + import pytest from sqlalchemy import select from sqlalchemy.exc import IntegrityError -from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker +from src.core.storage import ObjectNotFoundError from src.models.access import UserRole -from src.models.enums import EntityType, RelatedGroupKind -from tests.conftest import AuthClientFactory +from src.models.enums import ( + DocumentStatus, + DocumentType, + EntityType, + FileStatus, + FileType, + PeriodQuarter, + RecognitionPipeline, + RecognitionStatus, + RelatedGroupKind, + S3BucketRole, +) +from src.models.files import Document, RecognitionJob, S3Object, StoredFile +from src.models.osv import OsvEntry +from tests.conftest import AuthClientFactory, FakeStorage from tests.factories import ( + CommittedJobs, make_company_group, + make_document, make_legal_entity, make_membership, make_regional_office, make_related_group, + make_result_object, + make_stored_file, make_user, ) @@ -110,3 +131,128 @@ async def test_auth_client_sees_data_created_by_factories( response = await client.get("/me") assert response.status_code == HTTP_OK, response.text assert response.json()["regional_office"]["id"] == str(office.id) + + +@pytest.mark.integration +async def test_make_stored_file_builds_file_job_chain(db_session: AsyncSession) -> None: + entity = await make_legal_entity(db_session) + file, job = await make_stored_file(db_session, entity=entity) + + assert file.legal_entity_id == entity.id + assert file.status == FileStatus.UPLOADED + assert file.file_type == FileType.XLS + assert file.original_name == "osv.xls" + assert file.content_hash is None + assert file.current_recognition_job_id == job.id + assert job.file_id == file.id + assert job.status == RecognitionStatus.queued + assert job.pipeline == RecognitionPipeline.tabular + s3 = await db_session.get(S3Object, file.s3_object_id) + assert s3 is not None + assert s3.bucket_role == S3BucketRole.uploads + assert s3.key.endswith("/osv.xls") + + +@pytest.mark.integration +async def test_make_stored_file_overrides_and_image_pipeline(db_session: AsyncSession) -> None: + entity = await make_legal_entity(db_session) + office = await make_regional_office(db_session) + user = await make_user(db_session, email="uploader@armka.local", office=office) + file, job = await make_stored_file( + db_session, + entity=entity, + status=FileStatus.PROCESSING, + file_type=FileType.PDF, + original_name="bb.pdf", + content_hash="a" * 64, + job_status=RecognitionStatus.done, + s3_key="uploads/custom/bb.pdf", + uploaded_by=user.id, + ) + + assert (file.status, file.file_type, file.content_hash, file.uploaded_by) == ( + FileStatus.PROCESSING, + FileType.PDF, + "a" * 64, + user.id, + ) + assert job.pipeline == RecognitionPipeline.image + assert job.status == RecognitionStatus.done + s3 = await db_session.get(S3Object, file.s3_object_id) + assert s3 is not None + assert s3.key == "uploads/custom/bb.pdf" + + +@pytest.mark.integration +async def test_make_document_defaults_and_uniqueness(db_session: AsyncSession) -> None: + entity = await make_legal_entity(db_session) + file, job = await make_stored_file(db_session, entity=entity) + document = await make_document(db_session, file=file, job=job) + other = await make_document(db_session, file=file, job=job, account="62", status=DocumentStatus.NEEDS_CHECK) + + assert document.stored_file_id == file.id + assert document.legal_entity_id == entity.id + assert document.document_type == DocumentType.OSV + assert (document.osv_account_code, document.period_year, document.period_quarter) == ("60", 2025, PeriodQuarter.Q1) + assert document.status == DocumentStatus.RECOGNIZED + assert document.is_cumulative is True + assert other.status == DocumentStatus.NEEDS_CHECK + assert other.id != document.id + + +@pytest.mark.integration +async def test_make_result_object_stores_json_and_links_job(db_session: AsyncSession) -> None: + entity = await make_legal_entity(db_session) + file, job = await make_stored_file(db_session, entity=entity) + storage = FakeStorage() + result = {"schema_version": "1.0", "значение": "1.50"} + + s3 = await make_result_object(db_session, job=job, storage=storage, result=result) + + key = f"results/{file.id}/{job.id}/result.json" + assert s3.key == key + assert s3.bucket == storage.bucket + assert s3.bucket_role == S3BucketRole.results + assert job.result_s3_object_id == s3.id + assert json.loads(await storage.get_bytes(key)) == result + + +async def test_fake_storage_get_bytes_missing_object() -> None: + storage = FakeStorage() + await storage.put_object("a", b"x") + assert await storage.get_bytes("a") == b"x" + with pytest.raises(ObjectNotFoundError): + await storage.get_bytes("b") + + +@pytest.mark.integration +async def test_committed_jobs_cleanup_removes_documents_and_entries( + committed_sessionmaker: async_sessionmaker[AsyncSession], +) -> None: + jobs = CommittedJobs(committed_sessionmaker) + job_id = await jobs.make_queued_job() + try: + async with committed_sessionmaker() as session, session.begin(): + job = await session.get(RecognitionJob, job_id) + assert job is not None + file = await session.get(StoredFile, job.file_id) + assert file is not None + file.current_recognition_job_id = job.id + document = await make_document(session, file=file, job=job) + session.add( + OsvEntry( + recognition_job_id=job.id, + document_id=document.id, + owner_legal_entity_id=file.legal_entity_id, + row_number=1, + closing_debit=Decimal("1.00"), + ) + ) + file_id = file.id + finally: + await jobs.cleanup() + + async with committed_sessionmaker() as session: + assert await session.get(StoredFile, file_id) is None + assert not (await session.execute(select(Document).where(Document.stored_file_id == file_id))).all() + assert not (await session.execute(select(OsvEntry).where(OsvEntry.recognition_job_id == job_id))).all() diff --git a/docs/superpowers/plans/2026-10-05-slice-2c-osv-ingest.md b/docs/superpowers/plans/2026-10-05-slice-2c-osv-ingest.md index e5f04b2..6241eed 100644 --- a/docs/superpowers/plans/2026-10-05-slice-2c-osv-ingest.md +++ b/docs/superpowers/plans/2026-10-05-slice-2c-osv-ingest.md @@ -82,9 +82,9 @@ async def make_result_object(session, *, job: RecognitionJob, storage: FakeStora # CommittedJobs.cleanup(): перед stored_file удалять osv_entry, document (по file_id) ``` -- [ ] **Step 1:** тесты: `get_bytes` возвращает байты и закрывает response (заглушка с флагами), NoSuchKey → `ObjectNotFoundError`; фабрики создают цепочку и `cleanup` удаляет документы. -- [ ] **Step 2–4:** падает → реализация → `make check`, `pytest -m integration tests/test_factories.py -q`. -- [ ] **Step 5: Commit** `chore: S3 get_bytes, протокол ObjectStorage, фабрики файлов/документов, настройки загрузки и маппинга`. +- [x] **Step 1:** тесты: `get_bytes` возвращает байты и закрывает response (заглушка с флагами), NoSuchKey → `ObjectNotFoundError`; фабрики создают цепочку и `cleanup` удаляет документы. +- [x] **Step 2–4:** падает → реализация → `make check`, `pytest -m integration tests/test_factories.py -q`. +- [x] **Step 5: Commit** `chore: S3 get_bytes, протокол ObjectStorage, фабрики файлов/документов, настройки загрузки и маппинга`. --- diff --git a/openspec/changes/slice-2c-osv-ingest/tasks.md b/openspec/changes/slice-2c-osv-ingest/tasks.md index 04f1e39..d63b557 100644 --- a/openspec/changes/slice-2c-osv-ingest/tasks.md +++ b/openspec/changes/slice-2c-osv-ingest/tasks.md @@ -11,7 +11,7 @@ `dbml_parity.DBML_FILE_NAME`, `HEAD` в `test_migrations`, `test_enums`, `test_check_dbml_script` (фиктивная 5.16); нормализация email в нижний регистр в seed (`_apply_users`) и в `users_admin.create_user/update_user`; тесты seed/admin на регистр. Проверка: `make dbml-check` пуст; `make migration-cycle`. -- [ ] 1.2 **S3 и фабрики.** `S3Storage.get_bytes(key) -> bytes` (+ закрытие response), `ObjectStorage`-протокол с `get_bytes`; +- [x] 1.2 **S3 и фабрики.** `S3Storage.get_bytes(key) -> bytes` (+ закрытие response), `ObjectStorage`-протокол с `get_bytes`; `FakeStorage.get_bytes`; `tests/factories.py`: `make_stored_file(session, *, entity, status, content_hash, job_status)`, `make_document`, `make_entries_from_result(session, file, result_json)`; `CommittedJobs.cleanup()` удаляет `osv_entry`, `document`. Настройки `UPLOAD_MAX_BYTES` (50 МиБ), `MAPPING_POLL_SECONDS` (5), `MAPPING_MAX_ATTEMPTS` (3) + `.env.example`/`.env.test`.