2d69a9e867
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
259 lines
11 KiB
Python
259 lines
11 KiB
Python
"""Фабрики справочников и фикстуры `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, async_sessionmaker
|
|
|
|
from src.core.storage import ObjectNotFoundError
|
|
from src.models.access import UserRole
|
|
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,
|
|
)
|
|
|
|
HTTP_OK = 200
|
|
INN_LENGTH = 10
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_directory_factories_build_consistent_chain(db_session: AsyncSession) -> None:
|
|
office = await make_regional_office(db_session)
|
|
head = await make_regional_office(db_session, name="ГО Тест", is_head_office=True)
|
|
group = await make_company_group(db_session, office=office, crm_id="CRM-1")
|
|
entity = await make_legal_entity(db_session, group=group)
|
|
other = await make_legal_entity(db_session, name="ИП Тест", entity_type=EntityType.ИП, office=head)
|
|
gsk = await make_related_group(db_session, group=group)
|
|
gsz = await make_related_group(db_session, group=group, kind=RelatedGroupKind.gsz, name="ГСЗ Тест")
|
|
member = await make_membership(db_session, related_group=gsk, entity=entity)
|
|
# v5.14: одно ЮЛ одновременно в ГСК и ГСЗ
|
|
member_gsz = await make_membership(db_session, related_group=gsz, entity=entity)
|
|
|
|
assert head.is_head_office is True
|
|
assert office.is_head_office is False
|
|
assert group.code.startswith("GK-T")
|
|
assert group.base_regional_office_id == office.id
|
|
assert group.valid_from is not None
|
|
assert entity.regional_office_id == office.id # по умолчанию — филиал группы
|
|
assert entity.entity_type == EntityType.ЮЛ
|
|
assert len(entity.inn) == INN_LENGTH
|
|
assert entity.inn != other.inn
|
|
assert other.regional_office_id == head.id
|
|
assert other.company_group_id is None
|
|
assert gsk.code.startswith("GSK-T")
|
|
assert gsz.code.startswith("GSZ-T")
|
|
assert member.kind == RelatedGroupKind.gsk
|
|
assert member.is_current is True
|
|
assert member.valid_to is None
|
|
assert member_gsz.kind == RelatedGroupKind.gsz
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_membership_not_current_is_closed(db_session: AsyncSession) -> None:
|
|
group = await make_company_group(db_session)
|
|
entity = await make_legal_entity(db_session, group=group)
|
|
related = await make_related_group(db_session, group=group)
|
|
closed = await make_membership(db_session, related_group=related, entity=entity, is_current=False)
|
|
again = await make_membership(db_session, related_group=related, entity=entity) # закрытая версия не мешает
|
|
assert closed.is_current is False
|
|
assert closed.valid_to is not None
|
|
assert again.is_current is True
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_second_current_membership_in_same_kind_is_rejected(db_session: AsyncSession) -> None:
|
|
"""Фабрики не обходят ограничения БД: ЮЛ в двух ГСК одновременно — нарушение partial UNIQUE (v5.14)."""
|
|
group = await make_company_group(db_session)
|
|
entity = await make_legal_entity(db_session, group=group)
|
|
first = await make_related_group(db_session, group=group)
|
|
second = await make_related_group(db_session, group=group, name="ГСК-2")
|
|
await make_membership(db_session, related_group=first, entity=entity)
|
|
with pytest.raises(IntegrityError):
|
|
async with db_session.begin_nested():
|
|
await make_membership(db_session, related_group=second, entity=entity)
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_make_user_assigns_seeded_roles(seeded: AsyncSession) -> None:
|
|
office = await make_regional_office(seeded)
|
|
user = await make_user(seeded, email="factory@armka.local", office=office, roles=["KA_RF", "EXPERT"])
|
|
plain = await make_user(seeded, email="plain@armka.local", office=office, is_active=False)
|
|
count = len((await seeded.execute(select(UserRole).where(UserRole.user_id == user.id))).scalars().all())
|
|
assert count == 2 # noqa: PLR2004
|
|
assert user.regional_office_id == office.id
|
|
assert plain.is_active is False
|
|
assert not (await seeded.execute(select(UserRole).where(UserRole.user_id == plain.id))).scalars().all()
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_auth_client_gets_me_for_dev_user(auth_client: AuthClientFactory) -> None:
|
|
client = await auth_client("ka_rf@armka.local")
|
|
response = await client.get("/me")
|
|
assert response.status_code == HTTP_OK, response.text
|
|
assert response.json()["email"] == "ka_rf@armka.local"
|
|
assert response.json()["roles"] == ["KA_RF"]
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_auth_client_sees_data_created_by_factories(
|
|
db_session: AsyncSession, auth_client: AuthClientFactory
|
|
) -> None:
|
|
"""Пользователь, созданный фабрикой в той же сессии, получает токен и проходит `/me` (одна `db_session`)."""
|
|
office = await make_regional_office(db_session)
|
|
await make_user(db_session, email="made@armka.local", office=office, roles=["KA_RF"])
|
|
client = await auth_client("made@armka.local")
|
|
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()
|