422 lines
21 KiB
Python
422 lines
21 KiB
Python
"""Регрессии AURORA-1442: новая попытка после сбоя обмена, без повторной загрузки."""
|
||
|
||
import asyncio
|
||
from datetime import timedelta
|
||
from http import HTTPStatus
|
||
|
||
import pytest
|
||
from sqlalchemy import delete, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from src.api.dependencies.audit import RequestAudit
|
||
from src.api.dependencies.directories import get_app_settings
|
||
from src.common.exceptions import ConflictError
|
||
from src.contract import CONTRACT_VERSION
|
||
from src.domain.policy import VisibilityScope
|
||
from src.models.audit_ex import AuditEvent
|
||
from src.models.consolidation import Consolidation, ConsolidationMember, ConsolidationVersion
|
||
from src.models.enums import ConsolidationStatus, FileStatus, RecognitionPipeline, RecognitionStatus
|
||
from src.models.files import RecognitionJob, StoredFile
|
||
from src.services.audit import AuditService
|
||
from src.services.file_retry import FileRetryService
|
||
from src.services.recognition_exchange import handle_message, publish_pending
|
||
from tests.conftest import AuthClientFactory
|
||
from tests.factories import make_document, make_stored_file, make_version
|
||
from tests.test_api_consolidation_members import _user, _version
|
||
from tests.test_api_files_read import KA_RF, _rostov_entity
|
||
from tests.test_recognition_exchange import FakeNats
|
||
from tests.test_recognition_inbox import envelope, fake_msg
|
||
|
||
pytestmark = pytest.mark.integration
|
||
|
||
|
||
@pytest.mark.parametrize("state,code", [(RecognitionStatus.timeout, "TIMEOUT"), (RecognitionStatus.dlq, "NO_ACK")])
|
||
async def test_retry_preserves_old_job_and_records_audit(
|
||
seeded: AsyncSession, auth_client: AuthClientFactory, state: RecognitionStatus, code: str
|
||
) -> None:
|
||
entity = await _rostov_entity(seeded)
|
||
file, old = await make_stored_file(seeded, entity=entity, status=FileStatus.ERROR, job_status=state)
|
||
old.max_attempts = 3
|
||
old.error_code = code
|
||
old.metrics = {"republish_count": 3}
|
||
source = file.s3_object_id
|
||
await seeded.flush()
|
||
client = await auth_client(KA_RF)
|
||
|
||
response = await client.post(f"/files/{file.id}/retry")
|
||
assert response.status_code == HTTPStatus.ACCEPTED, response.text
|
||
await seeded.refresh(file)
|
||
await seeded.refresh(old)
|
||
new = await seeded.get(RecognitionJob, file.current_recognition_job_id)
|
||
assert new is not None and new.id != old.id
|
||
assert (new.status, new.attempt, new.max_attempts) == (RecognitionStatus.queued, 2, 3)
|
||
assert new.error_code is None and new.metrics is None and new.result_s3_object_id is None
|
||
assert file.s3_object_id == source and file.status == FileStatus.UPLOADED
|
||
assert old.status == state and old.error_code == code
|
||
event = (
|
||
await seeded.execute(
|
||
select(AuditEvent).where(AuditEvent.object_id == file.id, AuditEvent.action == "FILE_REPROCESS")
|
||
)
|
||
).scalar_one()
|
||
assert event.before_value == {"job_id": str(old.id), "attempt": 1, "status": state.value, "error_code": code}
|
||
assert event.after_value == {"job_id": str(new.id), "attempt": 2}
|
||
again = await client.post(f"/files/{file.id}/retry")
|
||
assert again.status_code == HTTPStatus.CONFLICT
|
||
|
||
|
||
@pytest.mark.parametrize("reason", ["limit", "ocr", "service_error", "document"])
|
||
async def test_retry_rejects_ineligible_file(seeded: AsyncSession, auth_client: AuthClientFactory, reason: str) -> None:
|
||
entity = await _rostov_entity(seeded)
|
||
file, job = await make_stored_file(
|
||
seeded, entity=entity, status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
|
||
)
|
||
job.error_code = "TIMEOUT"
|
||
job.max_attempts = 3
|
||
if reason == "limit":
|
||
job.attempt = 3
|
||
elif reason == "ocr":
|
||
job.pipeline = RecognitionPipeline.image
|
||
elif reason == "service_error":
|
||
job.status = RecognitionStatus.error
|
||
else:
|
||
await make_document(seeded, file=file, job=job)
|
||
await seeded.flush()
|
||
client = await auth_client(KA_RF)
|
||
response = await client.post(f"/files/{file.id}/retry")
|
||
assert response.status_code == HTTPStatus.CONFLICT, response.text
|
||
expected = "MAX_ATTEMPTS_REACHED" if reason == "limit" else "FILE_NOT_RETRYABLE"
|
||
assert response.json()["detail"]["code"] == expected
|
||
await seeded.refresh(file)
|
||
assert file.current_recognition_job_id == job.id
|
||
|
||
|
||
async def test_batch_filters_perimeter_and_skips_ineligible(
|
||
seeded: AsyncSession, auth_client: AuthClientFactory
|
||
) -> None:
|
||
|
||
version, inside, outside, _ = await _version(seeded, members=1, spare=1, locked_by=None)
|
||
good, job = await make_stored_file(
|
||
seeded, entity=inside[0], status=FileStatus.ERROR, job_status=RecognitionStatus.dlq
|
||
)
|
||
job.error_code, job.max_attempts = "NO_ACK", 3
|
||
good.consolidation_id = version.consolidation_id
|
||
bad, _ = await make_stored_file(seeded, entity=inside[0])
|
||
bad.consolidation_id = version.consolidation_id
|
||
excluded, _ = await make_stored_file(seeded, entity=outside[0])
|
||
excluded.consolidation_id = version.consolidation_id
|
||
desk, _ = await make_stored_file(seeded, entity=inside[0])
|
||
await seeded.flush()
|
||
client = await auth_client(KA_RF)
|
||
response = await client.post(f"/consolidations/versions/{version.id}/files/retry")
|
||
assert response.status_code == HTTPStatus.ACCEPTED, response.text
|
||
body = response.json()
|
||
assert [row["file_id"] for row in body["accepted"]] == [str(good.id)]
|
||
assert body["skipped"] == []
|
||
event = (
|
||
await seeded.execute(
|
||
select(AuditEvent).where(AuditEvent.object_id == good.id, AuditEvent.action == "FILE_REPROCESS")
|
||
)
|
||
).scalar_one()
|
||
assert event.after_value["consolidation_id"] == str(version.consolidation_id)
|
||
assert str(desk.id) not in str(body)
|
||
|
||
|
||
async def test_batch_rejects_verified_and_returns_empty_for_empty_perimeter(
|
||
seeded: AsyncSession, auth_client: AuthClientFactory
|
||
) -> None:
|
||
|
||
version, _, _, _ = await _version(seeded, members=1, locked_by=None)
|
||
client = await auth_client(KA_RF)
|
||
url = f"/consolidations/versions/{version.id}/files/retry"
|
||
response = await client.post(url)
|
||
assert response.status_code == HTTPStatus.ACCEPTED
|
||
assert response.json() == {"accepted": [], "skipped": []}
|
||
version.status = ConsolidationStatus.VERIFIED
|
||
await seeded.flush()
|
||
response = await client.post(url)
|
||
assert response.status_code == HTTPStatus.CONFLICT
|
||
assert response.json()["detail"]["code"] == "VERSION_NOT_EDITABLE"
|
||
|
||
|
||
async def test_concurrent_retries_create_only_one_attempt(committed_sessionmaker) -> None:
|
||
|
||
async with committed_sessionmaker() as session, session.begin():
|
||
entity = await _rostov_entity(session)
|
||
file, job = await make_stored_file(
|
||
session, entity=entity, status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
|
||
)
|
||
job.error_code, job.max_attempts = "TIMEOUT", 3
|
||
user = await _user(session, KA_RF)
|
||
file_id, user_id = file.id, user.id
|
||
|
||
async def retry():
|
||
async with committed_sessionmaker() as session:
|
||
try:
|
||
async with session.begin():
|
||
audit = RequestAudit(AuditService(session), user_id, None, None)
|
||
return await FileRetryService(session, audit, VisibilityScope(None)).retry(file_id)
|
||
except ConflictError as error:
|
||
return error.code
|
||
|
||
results = await asyncio.wait_for(asyncio.gather(retry(), retry()), timeout=10)
|
||
assert sum(isinstance(result, str) for result in results) == 1
|
||
assert "FILE_NOT_RETRYABLE" in results
|
||
async with committed_sessionmaker() as session:
|
||
jobs = list((await session.scalars(select(RecognitionJob).where(RecognitionJob.file_id == file_id))).all())
|
||
assert sorted(job.attempt for job in jobs) == [1, 2]
|
||
|
||
|
||
async def test_retry_outbox_and_late_old_result(committed_sessionmaker) -> None:
|
||
|
||
async with committed_sessionmaker() as session, session.begin():
|
||
entity = await _rostov_entity(session)
|
||
file, old = await make_stored_file(
|
||
session, entity=entity, status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
|
||
)
|
||
old.error_code, old.max_attempts = "TIMEOUT", 3
|
||
user = await _user(session, KA_RF)
|
||
file_id, old_id = file.id, old.id
|
||
audit = RequestAudit(AuditService(session), user.id, None, None)
|
||
card = await FileRetryService(session, audit, VisibilityScope(None)).retry(file.id)
|
||
assert card.job is not None
|
||
new_id = card.job.id
|
||
|
||
for state in ("processing", "done"):
|
||
await handle_message(fake_msg(envelope(old_id, file_id, state), None), session_factory=committed_sessionmaker)
|
||
offline = FakeNats(connected=False)
|
||
await publish_pending(committed_sessionmaker, offline, ack_timeout=timedelta(seconds=30), max_republish=3)
|
||
async with committed_sessionmaker() as session:
|
||
new = await session.get(RecognitionJob, new_id)
|
||
assert new is not None and new.status is RecognitionStatus.queued
|
||
assert new.reply_deadline is None and new.result_s3_object_id is None
|
||
online = FakeNats()
|
||
await publish_pending(committed_sessionmaker, online, ack_timeout=timedelta(seconds=30), max_republish=3)
|
||
async with committed_sessionmaker() as session:
|
||
new = await session.get(RecognitionJob, new_id)
|
||
assert new is not None and new.status is RecognitionStatus.published
|
||
assert new.reply_deadline is not None and new.request_msg_id == f"{new_id}:2"
|
||
old = await session.get(RecognitionJob, old_id)
|
||
assert old is not None and old.status is RecognitionStatus.timeout
|
||
|
||
|
||
async def test_retry_rolls_back_job_pointer_and_audit(committed_sessionmaker) -> None:
|
||
async with committed_sessionmaker() as session, session.begin():
|
||
entity = await _rostov_entity(session)
|
||
file, job = await make_stored_file(
|
||
session, entity=entity, status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
|
||
)
|
||
job.error_code, job.max_attempts = "TIMEOUT", 3
|
||
user = await _user(session, KA_RF)
|
||
file_id, job_id, user_id = file.id, job.id, user.id
|
||
with pytest.raises(RuntimeError, match="rollback"):
|
||
async with committed_sessionmaker() as session, session.begin():
|
||
audit = RequestAudit(AuditService(session), user_id, None, None)
|
||
await FileRetryService(session, audit, VisibilityScope(None)).retry(file_id)
|
||
raise RuntimeError("rollback")
|
||
async with committed_sessionmaker() as session:
|
||
jobs = list((await session.scalars(select(RecognitionJob).where(RecognitionJob.file_id == file_id))).all())
|
||
assert [job.id for job in jobs] == [job_id]
|
||
events = list((await session.scalars(select(AuditEvent).where(AuditEvent.object_id == file_id))).all())
|
||
assert events == []
|
||
|
||
|
||
async def test_concurrent_batch_retry(committed_sessionmaker) -> None:
|
||
async with committed_sessionmaker() as session, session.begin():
|
||
version, members, _, _ = await _version(session, members=1, locked_by=None)
|
||
file, job = await make_stored_file(
|
||
session, entity=members[0], status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
|
||
)
|
||
file.consolidation_id = version.consolidation_id
|
||
job.error_code, job.max_attempts = "TIMEOUT", 3
|
||
user = await _user(session, KA_RF)
|
||
version_id, user_id = version.id, user.id
|
||
|
||
async def retry():
|
||
async with committed_sessionmaker() as session, session.begin():
|
||
audit = RequestAudit(AuditService(session), user_id, None, None)
|
||
return await FileRetryService(session, audit, VisibilityScope(None)).retry_version(version_id)
|
||
|
||
results = await asyncio.wait_for(asyncio.gather(retry(), retry()), timeout=10)
|
||
assert sorted(len(result.accepted) for result in results) == [0, 1]
|
||
assert all(result.skipped == [] for result in results)
|
||
|
||
|
||
@pytest.mark.parametrize("change", ["status", "perimeter"])
|
||
async def test_batch_rechecks_version_after_waiting_for_job(
|
||
committed_sessionmaker, monkeypatch: pytest.MonkeyPatch, change: str
|
||
) -> None:
|
||
|
||
async with committed_sessionmaker() as session, session.begin():
|
||
version, members, _, _ = await _version(session, members=1, locked_by=None)
|
||
file, job = await make_stored_file(
|
||
session, entity=members[0], status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
|
||
)
|
||
file.consolidation_id = version.consolidation_id
|
||
job.error_code, job.max_attempts = "TIMEOUT", 3
|
||
user = await _user(session, KA_RF)
|
||
version_id, job_id, user_id = version.id, job.id, user.id
|
||
reached = asyncio.Event()
|
||
original = FileRetryService._lock_files
|
||
|
||
async def signal(self, files):
|
||
reached.set()
|
||
return await original(self, files)
|
||
|
||
monkeypatch.setattr(FileRetryService, "_lock_files", signal)
|
||
|
||
async def retry():
|
||
async with committed_sessionmaker() as session:
|
||
try:
|
||
async with session.begin():
|
||
audit = RequestAudit(AuditService(session), user_id, None, None)
|
||
return await FileRetryService(session, audit, VisibilityScope(None)).retry_version(version_id)
|
||
except ConflictError as error:
|
||
return error.code
|
||
|
||
async with committed_sessionmaker() as holder, holder.begin():
|
||
await holder.execute(select(RecognitionJob).where(RecognitionJob.id == job_id).with_for_update())
|
||
pending = asyncio.create_task(retry())
|
||
await asyncio.wait_for(reached.wait(), timeout=10)
|
||
version = await holder.get(ConsolidationVersion, version_id)
|
||
assert version is not None
|
||
if change == "status":
|
||
version.status = ConsolidationStatus.VERIFIED
|
||
else:
|
||
await holder.execute(
|
||
delete(ConsolidationMember).where(ConsolidationMember.consolidation_version_id == version_id)
|
||
)
|
||
result = await asyncio.wait_for(pending, timeout=10)
|
||
if change == "status":
|
||
assert result == "VERSION_NOT_EDITABLE"
|
||
else:
|
||
assert result.accepted == [] and result.skipped == []
|
||
|
||
|
||
async def test_batch_mixed_package_skips_ocr_and_exhausted_files(
|
||
seeded: AsyncSession, auth_client: AuthClientFactory
|
||
) -> None:
|
||
"""Смешанный пакет: подходящий файл принят, OCR в ERROR и исчерпанный лимит попыток — в skipped с кодами."""
|
||
version, inside, _, _ = await _version(seeded, members=1, locked_by=None)
|
||
good, good_job = await make_stored_file(
|
||
seeded, entity=inside[0], status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
|
||
)
|
||
good_job.error_code, good_job.max_attempts = "TIMEOUT", 3
|
||
ocr, ocr_job = await make_stored_file(
|
||
seeded, entity=inside[0], status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
|
||
)
|
||
ocr_job.pipeline, ocr_job.error_code, ocr_job.max_attempts = RecognitionPipeline.image, "TIMEOUT", 3
|
||
spent, spent_job = await make_stored_file(
|
||
seeded, entity=inside[0], status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
|
||
)
|
||
spent_job.error_code, spent_job.max_attempts, spent_job.attempt = "TIMEOUT", 3, 3
|
||
for item in (good, ocr, spent):
|
||
item.consolidation_id = version.consolidation_id
|
||
await seeded.flush()
|
||
client = await auth_client(KA_RF)
|
||
response = await client.post(f"/consolidations/versions/{version.id}/files/retry")
|
||
assert response.status_code == HTTPStatus.ACCEPTED, response.text
|
||
body = response.json()
|
||
assert [row["file_id"] for row in body["accepted"]] == [str(good.id)]
|
||
skipped = {row["file_id"]: row["code"] for row in body["skipped"]}
|
||
assert skipped == {str(ocr.id): "FILE_NOT_RETRYABLE", str(spent.id): "MAX_ATTEMPTS_REACHED"}
|
||
for skipped_file, skipped_job in ((ocr, ocr_job), (spent, spent_job)):
|
||
await seeded.refresh(skipped_file)
|
||
assert skipped_file.current_recognition_job_id == skipped_job.id and skipped_file.status == FileStatus.ERROR
|
||
events = await seeded.scalars(
|
||
select(AuditEvent).where(AuditEvent.object_id == skipped_file.id, AuditEvent.action == "FILE_REPROCESS")
|
||
)
|
||
assert list(events.all()) == []
|
||
|
||
|
||
async def test_batch_no_ack_at_attempt_limit_is_max_attempts_reached(
|
||
seeded: AsyncSession, auth_client: AuthClientFactory
|
||
) -> None:
|
||
"""NO_ACK (dlq) на пределе попыток в пакете даёт MAX_ATTEMPTS_REACHED, новое задание не создаётся."""
|
||
version, inside, _, _ = await _version(seeded, members=1, locked_by=None)
|
||
file, job = await make_stored_file(
|
||
seeded, entity=inside[0], status=FileStatus.ERROR, job_status=RecognitionStatus.dlq
|
||
)
|
||
job.error_code, job.max_attempts, job.attempt = "NO_ACK", 3, 3
|
||
file.consolidation_id = version.consolidation_id
|
||
await seeded.flush()
|
||
client = await auth_client(KA_RF)
|
||
response = await client.post(f"/consolidations/versions/{version.id}/files/retry")
|
||
assert response.status_code == HTTPStatus.ACCEPTED, response.text
|
||
assert response.json() == {"accepted": [], "skipped": [{"file_id": str(file.id), "code": "MAX_ATTEMPTS_REACHED"}]}
|
||
jobs = await seeded.scalars(select(RecognitionJob).where(RecognitionJob.file_id == file.id))
|
||
assert [item.id for item in jobs.all()] == [job.id]
|
||
|
||
|
||
async def test_retry_uses_current_contract_and_subject(
|
||
seeded: AsyncSession, auth_client: AuthClientFactory, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
entity = await _rostov_entity(seeded)
|
||
file, old = await make_stored_file(
|
||
seeded, entity=entity, status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
|
||
)
|
||
old.error_code, old.max_attempts = "TIMEOUT", 3
|
||
old.schema_version, old.request_subject = "0.9", "old.tabular.subject"
|
||
monkeypatch.setattr(get_app_settings().OPENBAO.NATS, "NATS_TABULAR_REQUEST_SUBJECT", "current.tabular.subject")
|
||
await seeded.flush()
|
||
client = await auth_client(KA_RF)
|
||
response = await client.post(f"/files/{file.id}/retry")
|
||
assert response.status_code == HTTPStatus.ACCEPTED, response.text
|
||
await seeded.refresh(file)
|
||
new = await seeded.get(RecognitionJob, file.current_recognition_job_id)
|
||
assert new is not None
|
||
assert new.schema_version == CONTRACT_VERSION and new.request_subject == "current.tabular.subject"
|
||
await seeded.refresh(old)
|
||
assert old.schema_version == "0.9" and old.request_subject == "old.tabular.subject"
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"normal_status", [FileStatus.UPLOADED, FileStatus.PROCESSING, FileStatus.PROCESSED, FileStatus.SUPERSEDED]
|
||
)
|
||
async def test_batch_does_not_lock_normal_files(committed_sessionmaker, normal_status: FileStatus) -> None:
|
||
async with committed_sessionmaker() as session, session.begin():
|
||
version, members, _, _ = await _version(session, members=1, locked_by=None)
|
||
normal, job = await make_stored_file(
|
||
session, entity=members[0], status=normal_status, job_status=RecognitionStatus.processing
|
||
)
|
||
normal.consolidation_id = version.consolidation_id
|
||
user = await _user(session, KA_RF)
|
||
version_id, file_id, job_id, user_id = version.id, normal.id, job.id, user.id
|
||
|
||
async def retry():
|
||
async with committed_sessionmaker() as session, session.begin():
|
||
audit = RequestAudit(AuditService(session), user_id, None, None)
|
||
return await FileRetryService(session, audit, VisibilityScope(None)).retry_version(version_id)
|
||
|
||
async with committed_sessionmaker() as holder, holder.begin():
|
||
await holder.execute(select(RecognitionJob).where(RecognitionJob.id == job_id).with_for_update())
|
||
await holder.execute(select(StoredFile).where(StoredFile.id == file_id).with_for_update())
|
||
result = await asyncio.wait_for(retry(), timeout=5)
|
||
assert result.accepted == [] and result.skipped == []
|
||
|
||
|
||
@pytest.mark.parametrize("closed_status", [ConsolidationStatus.VERIFIED, ConsolidationStatus.CANCELLED])
|
||
async def test_single_retry_does_not_lock_closed_versions(committed_sessionmaker, closed_status) -> None:
|
||
async with committed_sessionmaker() as session, session.begin():
|
||
working, members, _, _ = await _version(session, members=1, locked_by=None)
|
||
consolidation = await session.get(Consolidation, working.consolidation_id)
|
||
user = await _user(session, KA_RF)
|
||
closed = await make_version(
|
||
session, consolidation=consolidation, version_no=2, members=members, initiator=user, status=closed_status
|
||
)
|
||
file, job = await make_stored_file(
|
||
session, entity=members[0], status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
|
||
)
|
||
file.consolidation_id = working.consolidation_id
|
||
job.error_code, job.max_attempts = "TIMEOUT", 3
|
||
file_id, user_id, closed_id = file.id, user.id, closed.id
|
||
|
||
async def retry():
|
||
async with committed_sessionmaker() as session, session.begin():
|
||
audit = RequestAudit(AuditService(session), user_id, None, None)
|
||
return await FileRetryService(session, audit, VisibilityScope(None)).retry(file_id)
|
||
|
||
async with committed_sessionmaker() as holder, holder.begin():
|
||
await holder.execute(select(ConsolidationVersion).where(ConsolidationVersion.id == closed_id).with_for_update())
|
||
card = await asyncio.wait_for(retry(), timeout=5)
|
||
assert card.status == FileStatus.UPLOADED
|