291 lines
14 KiB
Python
291 lines
14 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.common.exceptions import ConflictError
|
|
from src.domain.policy import VisibilityScope
|
|
from src.models.audit_ex import AuditEvent
|
|
from src.models.consolidation import ConsolidationMember, ConsolidationVersion
|
|
from src.models.enums import ConsolidationStatus, FileStatus, RecognitionPipeline, RecognitionStatus
|
|
from src.models.files import RecognitionJob
|
|
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
|
|
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"] == [{"file_id": str(bad.id), "code": "FILE_NOT_RETRYABLE"}]
|
|
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 sorted(len(result.skipped) for result in results) == [0, 1]
|
|
|
|
|
|
@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 == []
|