""" Tests for API ingress retry handling These tests verify: 1. process_api_request reclaims a previously failed request so an Amazon retry (same order, new trackingId) is reprocessed instead of swallowed 2. True duplicates are still skipped without calling 1Rx 3. call_1rx_api retries with backoff on 5xx/429/connection errors and does not retry on other 4xx responses """ import asyncio import sys import unittest from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch sys.path.insert(0, str(Path(__file__).resolve().parents[2])) import app.src.file_processor as file_processor_mod from app.src.file_processor import FileProcessor class MockConfig: """Mock configuration for testing""" def __init__(self): self.rx_api_key = "test-api-key" self.rx_api_base_url = "http://localhost:10000/api/1rx/v1" self.rx_api_retry_attempts = 3 self.rx_api_retry_delay_seconds = 0.01 self.rx_api_timeout_seconds = 1.0 def make_amazon_request(tracking_id="trk-1"): """Minimal valid Amazon API ingress request""" return { "messageIdentifiers": { "primaryReferenceId": "404-1234567#a1b2c3d4", "secondaryReferenceId": "cons-123", "trackingId": tracking_id, "isTestData": False, }, "data": { "requestMetadata": { "messageId": "msg-1", "messageSentTime": 1699968883231, }, }, } def make_file_processor(config, request_store): with patch.object(file_processor_mod, "SFTPHandler"), patch.object( file_processor_mod, "PGPHandler" ): fp = FileProcessor(config, request_store) return fp class TestProcessApiRequestReclaim(unittest.TestCase): """Test cases for reclaim-on-retry behavior in process_api_request""" def setUp(self): self.config = MockConfig() self.request_store = MagicMock() self.fp = make_file_processor(self.config, self.request_store) self.fp.call_1rx_api = AsyncMock(return_value={"data": {"orderId": "1"}}) def test_new_request_calls_1rx(self): self.request_store.store_request.return_value = True result = asyncio.run(self.fp.process_api_request(make_amazon_request())) self.assertTrue(result["success"]) self.fp.call_1rx_api.assert_awaited_once() self.request_store.reclaim_failed_request.assert_not_called() self.request_store.update_request_status.assert_called_with( "cons-123", "sent_to_1rx" ) def test_true_duplicate_skips_1rx(self): self.request_store.store_request.return_value = False self.request_store.reclaim_failed_request.return_value = False result = asyncio.run(self.fp.process_api_request(make_amazon_request("trk-2"))) self.assertTrue(result["success"]) self.assertIn("Duplicate", result["message"]) self.fp.call_1rx_api.assert_not_awaited() def test_failed_request_is_reclaimed_and_retried(self): self.request_store.store_request.return_value = False self.request_store.reclaim_failed_request.return_value = True result = asyncio.run(self.fp.process_api_request(make_amazon_request("trk-2"))) self.assertTrue(result["success"]) self.fp.call_1rx_api.assert_awaited_once() self.request_store.update_request_status.assert_called_with( "cons-123", "sent_to_1rx" ) def test_reclaimed_retry_can_fail_again(self): self.request_store.store_request.return_value = False self.request_store.reclaim_failed_request.return_value = True self.fp.call_1rx_api = AsyncMock(return_value=None) result = asyncio.run(self.fp.process_api_request(make_amazon_request("trk-3"))) self.assertFalse(result["success"]) self.request_store.update_request_status.assert_called_with( "cons-123", "failed_1rx" ) def test_duplicate_response_echoes_message_identifiers(self): self.request_store.store_request.return_value = False self.request_store.reclaim_failed_request.return_value = False request = make_amazon_request("trk-4") result = asyncio.run(self.fp.process_api_request(request)) self.assertEqual( result["message_identifiers"], request["messageIdentifiers"] ) class FakeResponse: """Fake aiohttp response usable as an async context manager""" def __init__(self, status, json_body=None, text_body=""): self.status = status self._json = json_body self._text = text_body async def json(self): return self._json async def text(self): return self._text async def __aenter__(self): return self async def __aexit__(self, *args): return False class FailingPost: """Fake post context manager that raises on entry""" def __init__(self, exc): self.exc = exc async def __aenter__(self): raise self.exc async def __aexit__(self, *args): return False class FakeSession: """Fake aiohttp session returning queued post outcomes""" def __init__(self, outcomes): self.outcomes = list(outcomes) self.calls = 0 def post(self, *args, **kwargs): self.calls += 1 outcome = self.outcomes.pop(0) if isinstance(outcome, Exception): return FailingPost(outcome) return outcome async def __aenter__(self): return self async def __aexit__(self, *args): return False class TestCall1rxApiRetry(unittest.TestCase): """Test cases for retry behavior in call_1rx_api""" def setUp(self): self.config = MockConfig() self.fp = make_file_processor(self.config, MagicMock()) def call(self, outcomes): session = FakeSession(outcomes) with patch.object( file_processor_mod.aiohttp, "ClientSession", return_value=session ): result = asyncio.run(self.fp.call_1rx_api({"payload": 1})) return result, session def test_retries_on_5xx_then_succeeds(self): result, session = self.call( [ FakeResponse(500, text_body="server error"), FakeResponse(202, json_body={"data": {"orderId": "1"}}), ] ) self.assertEqual(result, {"data": {"orderId": "1"}}) self.assertEqual(session.calls, 2) def test_retries_on_429_then_succeeds(self): result, session = self.call( [ FakeResponse(429, text_body="throttled"), FakeResponse(202, json_body={"data": {}}), ] ) self.assertEqual(result, {"data": {}}) self.assertEqual(session.calls, 2) def test_retries_on_connection_error_then_succeeds(self): result, session = self.call( [ file_processor_mod.aiohttp.ClientConnectionError("refused"), FakeResponse(202, json_body={"data": {}}), ] ) self.assertEqual(result, {"data": {}}) self.assertEqual(session.calls, 2) def test_retries_on_timeout_then_succeeds(self): result, session = self.call( [ asyncio.TimeoutError(), FakeResponse(202, json_body={"data": {}}), ] ) self.assertEqual(result, {"data": {}}) self.assertEqual(session.calls, 2) def test_exhausts_attempts_and_returns_none(self): result, session = self.call( [ FakeResponse(500, text_body="err"), FakeResponse(503, text_body="err"), FakeResponse(500, text_body="err"), ] ) self.assertIsNone(result) self.assertEqual(session.calls, 3) def test_does_not_retry_on_4xx(self): result, session = self.call([FakeResponse(400, text_body="bad payload")]) self.assertIsNone(result) self.assertEqual(session.calls, 1) if __name__ == "__main__": unittest.main()