72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
"""Фабрики тестовых данных для БД-тестов: минимальные цепочки FK поверх сессии (`flush`, без `commit`)."""
|
||
|
||
import datetime
|
||
import decimal
|
||
import uuid
|
||
from typing import Any
|
||
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from src.models.directories import LegalEntity
|
||
from src.models.enums import (
|
||
DocumentStatus,
|
||
DocumentType,
|
||
FileStatus,
|
||
FileType,
|
||
RecognitionPipeline,
|
||
RecognitionStatus,
|
||
S3BucketRole,
|
||
)
|
||
from src.models.files import Document, RecognitionJob, S3Object, StoredFile
|
||
from src.models.osv import OsvEntry
|
||
|
||
|
||
async def make_osv_entry(session: AsyncSession, **overrides: Any) -> OsvEntry:
|
||
"""Строка ОСВ с минимальной цепочкой FK (ЮЛ, s3_object, stored_file, recognition_job, document); `flush`.
|
||
|
||
ИНН ЮЛ случайный, поэтому фабрику можно вызывать несколько раз в одном тесте.
|
||
|
||
`overrides` перекрывают поля `OsvEntry` (например, `closing_debit=Decimal("1.00")`). Данные остаются в
|
||
транзакции сессии: фикстура `db_session` откатывает их после теста.
|
||
"""
|
||
now = datetime.datetime.now(datetime.UTC)
|
||
entity = LegalEntity(inn=str(uuid.uuid4().int)[:10], name="ООО Тест", valid_from=now)
|
||
s3 = S3Object(bucket_role=S3BucketRole.uploads, bucket="uploads", key=f"{uuid.uuid4()}/osv.xlsx", created_at=now)
|
||
session.add_all([entity, s3])
|
||
await session.flush()
|
||
file = StoredFile(
|
||
legal_entity_id=entity.id, file_type=FileType.XLSX, s3_object_id=s3.id, status=FileStatus.UPLOADED
|
||
)
|
||
session.add(file)
|
||
await session.flush()
|
||
job = RecognitionJob(
|
||
file_id=file.id,
|
||
pipeline=RecognitionPipeline.tabular,
|
||
request_subject="recognition.tabular.request",
|
||
schema_version="1",
|
||
status=RecognitionStatus.queued,
|
||
)
|
||
session.add(job)
|
||
await session.flush()
|
||
document = Document(
|
||
stored_file_id=file.id,
|
||
legal_entity_id=entity.id,
|
||
document_type=DocumentType.OSV,
|
||
osv_account_code="62",
|
||
status=DocumentStatus.RECOGNIZED,
|
||
)
|
||
session.add(document)
|
||
await session.flush()
|
||
fields: dict[str, Any] = {
|
||
"recognition_job_id": job.id,
|
||
"document_id": document.id,
|
||
"owner_legal_entity_id": entity.id,
|
||
"row_number": 1,
|
||
"closing_debit": decimal.Decimal("0.00"),
|
||
}
|
||
fields.update(overrides)
|
||
entry = OsvEntry(**fields)
|
||
session.add(entry)
|
||
await session.flush()
|
||
return entry
|