diff --git a/tests/audio/test_audio_job_consumer.py b/tests/audio/test_audio_job_consumer.py index e4cad94..3d1ac35 100644 --- a/tests/audio/test_audio_job_consumer.py +++ b/tests/audio/test_audio_job_consumer.py @@ -78,6 +78,10 @@ async def test_processes_day_job_successfully( job_id = uuid4() day_id = uuid4() mock_get_job.return_value = {"job_id": str(job_id), "status": AudioJobStatus.PENDING.value} + mock_update_status.side_effect = [ + {"job_id": str(job_id), "status": AudioJobStatus.PROCESSING.value}, + {"job_id": str(job_id), "status": AudioJobStatus.COMPLETED.value}, + ] mock_generate.return_value = { "audio_url": "https://example.com/a.wav", "audio_duration_ms": 1200, @@ -120,6 +124,10 @@ async def test_marks_failed_on_generation_error( ): job_id = uuid4() mock_get_job.return_value = {"job_id": str(job_id), "status": AudioJobStatus.PENDING.value} + mock_update_status.side_effect = [ + {"job_id": str(job_id), "status": AudioJobStatus.PROCESSING.value}, + {"job_id": str(job_id), "status": AudioJobStatus.FAILED.value}, + ] mock_generate.side_effect = HTTPException(status_code=404, detail={"message": "Sub task not found"}) message = { @@ -168,6 +176,81 @@ async def test_skips_completed_job( mock_update_status.assert_not_awaited() mock_delete.assert_called_once_with("abc") + @pytest.mark.asyncio + @patch("worker_api.audio.services.audio_job_consumer.delete_audio_job_message") + @patch("worker_api.audio.services.audio_job_consumer.generate_plan_audio_service", new_callable=AsyncMock) + @patch("worker_api.audio.services.audio_job_consumer.update_audio_job_status", new_callable=AsyncMock) + @patch("worker_api.audio.services.audio_job_consumer.get_audio_job_status", new_callable=AsyncMock) + async def test_skips_when_claim_conflicts( + self, + mock_get_job, + mock_update_status, + mock_generate, + mock_delete, + ): + job_id = uuid4() + mock_get_job.return_value = {"job_id": str(job_id), "status": AudioJobStatus.PENDING.value} + mock_update_status.side_effect = HTTPException( + status_code=409, + detail={"error": "Bad request", "message": "Audio job is already being processed"}, + ) + + message = { + "ReceiptHandle": "abc", + "Body": json.dumps( + { + "job_id": str(job_id), + "day_id": str(uuid4()), + "language": "bo", + "type": "TEXT_READING", + } + ), + } + + await process_audio_job_message(message) + + mock_generate.assert_not_called() + mock_update_status.assert_awaited_once() + assert mock_update_status.await_args.kwargs["status"] == AudioJobStatus.PROCESSING + mock_delete.assert_called_once_with("abc") + + @pytest.mark.asyncio + @patch("worker_api.audio.services.audio_job_consumer.delete_audio_job_message") + @patch("worker_api.audio.services.audio_job_consumer.generate_plan_audio_service", new_callable=AsyncMock) + @patch("worker_api.audio.services.audio_job_consumer.update_audio_job_status", new_callable=AsyncMock) + @patch("worker_api.audio.services.audio_job_consumer.get_audio_job_status", new_callable=AsyncMock) + async def test_skips_when_claim_returns_terminal( + self, + mock_get_job, + mock_update_status, + mock_generate, + mock_delete, + ): + job_id = uuid4() + mock_get_job.return_value = {"job_id": str(job_id), "status": AudioJobStatus.PENDING.value} + mock_update_status.return_value = { + "job_id": str(job_id), + "status": AudioJobStatus.COMPLETED.value, + } + + message = { + "ReceiptHandle": "abc", + "Body": json.dumps( + { + "job_id": str(job_id), + "day_id": str(uuid4()), + "language": "bo", + "type": "TEXT_READING", + } + ), + } + + await process_audio_job_message(message) + + mock_generate.assert_not_called() + mock_update_status.assert_awaited_once() + mock_delete.assert_called_once_with("abc") + @pytest.mark.asyncio @patch("worker_api.audio.services.audio_job_consumer.delete_audio_job_message") @patch("worker_api.audio.services.audio_job_consumer.parse_audio_job_message_body", return_value=None) @@ -210,6 +293,10 @@ async def test_fails_when_language_missing( ): job_id = uuid4() mock_get_job.return_value = {"job_id": str(job_id), "status": AudioJobStatus.PENDING.value} + mock_update_status.side_effect = [ + {"job_id": str(job_id), "status": AudioJobStatus.PROCESSING.value}, + {"job_id": str(job_id), "status": AudioJobStatus.FAILED.value}, + ] message = { "ReceiptHandle": "abc", "Body": json.dumps( @@ -243,6 +330,10 @@ async def test_fails_when_target_ids_missing( ): job_id = uuid4() mock_get_job.return_value = {"job_id": str(job_id), "status": AudioJobStatus.PENDING.value} + mock_update_status.side_effect = [ + {"job_id": str(job_id), "status": AudioJobStatus.PROCESSING.value}, + {"job_id": str(job_id), "status": AudioJobStatus.FAILED.value}, + ] message = { "ReceiptHandle": "abc", "Body": json.dumps({"job_id": str(job_id), "language": "en"}), diff --git a/worker_api/audio/services/audio_job_consumer.py b/worker_api/audio/services/audio_job_consumer.py index 20423f9..ed1909d 100644 --- a/worker_api/audio/services/audio_job_consumer.py +++ b/worker_api/audio/services/audio_job_consumer.py @@ -100,7 +100,23 @@ async def process_audio_job_message(message: Dict[str, Any]) -> None: delete_audio_job_message(receipt_handle) return - await update_audio_job_status(job_id=job_id, status=AudioJobStatus.PROCESSING) + try: + claimed = await update_audio_job_status(job_id=job_id, status=AudioJobStatus.PROCESSING) + except HTTPException as exc: + # Another worker already claimed this job (duplicate SQS delivery). + if exc.status_code == 409: + logger.info("Skipping already claimed audio job %s", job_id) + if receipt_handle: + delete_audio_job_message(receipt_handle) + return + raise + + claimed_status = str(claimed.get("status") or "") + if claimed_status != AudioJobStatus.PROCESSING.value: + logger.info("Skipping audio job %s after claim (%s)", job_id, claimed_status) + if receipt_handle: + delete_audio_job_message(receipt_handle) + return try: day_id = _parse_uuid(body.get("day_id"))