diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index 0e8a6c7273..e7495a085c 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -2886,6 +2886,7 @@ definitions: - RETRY - IGNORE - RESET_PAGINATION + - RESET_SYNC_TOKEN - RATE_LIMITED - REFRESH_TOKEN_THEN_RETRY - REDUCE_PAGE_SIZE @@ -2896,6 +2897,7 @@ definitions: - RETRY - IGNORE - RESET_PAGINATION + - RESET_SYNC_TOKEN - RATE_LIMITED - REFRESH_TOKEN_THEN_RETRY - REDUCE_PAGE_SIZE diff --git a/airbyte_cdk/sources/declarative/incremental/sync_token_cursor.py b/airbyte_cdk/sources/declarative/incremental/sync_token_cursor.py index 83051a45e0..abdd672282 100644 --- a/airbyte_cdk/sources/declarative/incremental/sync_token_cursor.py +++ b/airbyte_cdk/sources/declarative/incremental/sync_token_cursor.py @@ -218,6 +218,30 @@ def on_page_completed(self, page: PageCompleted) -> None: .replace("+00:00", "Z"), } + def restart_without_token( + self, stream_slice: StreamSlice, reason: Optional[str] = None + ) -> StreamSlice: + if "token" not in stream_slice.cursor_slice: + message = ( + f"Stream {self._stream_name}: the API rejected a read without a sync token. " + "Check the RESET_SYNC_TOKEN response filter." + ) + raise AirbyteTracedException( + internal_message=message, + message=message, + failure_type=FailureType.system_error, + ) + + partition_key = self._serializer.to_partition_key(stream_slice.partition or {}) + with self._lock: + self._pending.pop(partition_key, None) + + return StreamSlice( + partition=stream_slice.partition, + cursor_slice={}, + extra_fields=stream_slice.extra_fields, + ) + def _coerce_token(self, value: Any, path: list[str]) -> Optional[str]: if value is None or value == "": return None diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index b3944869ae..a05773921a 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -614,6 +614,7 @@ class Action(Enum): RETRY = "RETRY" IGNORE = "IGNORE" RESET_PAGINATION = "RESET_PAGINATION" + RESET_SYNC_TOKEN = "RESET_SYNC_TOKEN" RATE_LIMITED = "RATE_LIMITED" REFRESH_TOKEN_THEN_RETRY = "REFRESH_TOKEN_THEN_RETRY" REDUCE_PAGE_SIZE = "REDUCE_PAGE_SIZE" @@ -637,6 +638,7 @@ class HttpResponseFilter(BaseModel): "RETRY", "IGNORE", "RESET_PAGINATION", + "RESET_SYNC_TOKEN", "RATE_LIMITED", "REFRESH_TOKEN_THEN_RETRY", "REDUCE_PAGE_SIZE", diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index 3696778c7d..6d0f568a7c 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -4207,6 +4207,9 @@ def _get_url(req: Requester) -> str: page_observer=( cursor.on_page_completed if isinstance(cursor, SyncTokenCursor) else None ), + token_reset_handler=( + cursor.restart_without_token if isinstance(cursor, SyncTokenCursor) else None + ), parameters=model.parameters or {}, ) diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/composite_error_handler.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/composite_error_handler.py index 2ef35a06ff..0d416adaaf 100644 --- a/airbyte_cdk/sources/declarative/requesters/error_handlers/composite_error_handler.py +++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/composite_error_handler.py @@ -71,6 +71,7 @@ def interpret_response( ResponseAction.RETRY, ResponseAction.IGNORE, ResponseAction.RESET_PAGINATION, + ResponseAction.RESET_SYNC_TOKEN, ResponseAction.REDUCE_PAGE_SIZE, ResponseAction.SPLIT_REQUEST_WINDOW, ]: diff --git a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py index 48593377b9..841cd21f8e 100644 --- a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py +++ b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py @@ -68,6 +68,10 @@ RequestWindowSplitNotSupportedException, RequestWindowSplitRequiredException, ) +from airbyte_cdk.sources.streams.http.sync_token_reset_exception import ( + SyncTokenResetNotSupportedException, + SyncTokenResetRequiredException, +) from airbyte_cdk.sources.types import Config, Record, StreamSlice from airbyte_cdk.utils.mapping_helpers import combine_mappings from airbyte_cdk.utils.traced_exception import AirbyteTracedException @@ -158,6 +162,7 @@ class SimpleRetriever(Retriever): Callable[[StreamSlice, Optional[datetime.timedelta]], Optional[List[StreamSlice]]] ] = None page_observer: Optional[Callable[["PageCompleted"], None]] = None + token_reset_handler: Optional[Callable[[StreamSlice, Optional[str]], StreamSlice]] = None def __post_init__(self, parameters: Mapping[str, Any]) -> None: self._paginator = self.paginator or NoPagination(parameters=parameters) @@ -707,6 +712,19 @@ def _read_records_or_split_request_window( yield from self._read_records_or_split_request_window( records_schema, child, original_slice, depth + 1 ) + except SyncTokenResetRequiredException as exception: + if self.token_reset_handler is None: + raise SyncTokenResetNotSupportedException(stream_name=self.name) from exception + + reset_slice = self.token_reset_handler(stream_slice, exception.error_message) + reason = exception.error_message or "the API rejected the stored sync token" + LOGGER.warning( + f"Stream {self.name}: {reason}; reading this stream in full. " + f"{emitted_count} records read earlier in this attempt may be emitted again." + ) + yield from self._read_records_or_split_request_window( + records_schema, reset_slice, original_slice, depth + 1 + ) def _with_request_window_failure_message(self, message: str) -> str: """ diff --git a/airbyte_cdk/sources/streams/http/error_handlers/response_models.py b/airbyte_cdk/sources/streams/http/error_handlers/response_models.py index 4cda91ffb8..6b53582494 100644 --- a/airbyte_cdk/sources/streams/http/error_handlers/response_models.py +++ b/airbyte_cdk/sources/streams/http/error_handlers/response_models.py @@ -17,6 +17,7 @@ class ResponseAction(Enum): FAIL = "FAIL" IGNORE = "IGNORE" RESET_PAGINATION = "RESET_PAGINATION" + RESET_SYNC_TOKEN = "RESET_SYNC_TOKEN" RATE_LIMITED = "RATE_LIMITED" REFRESH_TOKEN_THEN_RETRY = "REFRESH_TOKEN_THEN_RETRY" REDUCE_PAGE_SIZE = "REDUCE_PAGE_SIZE" diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py index 8da499aa32..aae206966a 100644 --- a/airbyte_cdk/sources/streams/http/http_client.py +++ b/airbyte_cdk/sources/streams/http/http_client.py @@ -65,6 +65,9 @@ ResponseAwareAuthenticator, TokenRotatingAuthenticator, ) +from airbyte_cdk.sources.streams.http.sync_token_reset_exception import ( + SyncTokenResetRequiredException, +) from airbyte_cdk.sources.streams.http.tcp_keepalive import TcpKeepaliveHTTPAdapter from airbyte_cdk.sources.utils.types import JsonType from airbyte_cdk.utils.airbyte_secrets_utils import filter_secrets @@ -620,6 +623,9 @@ def _handle_error_resolution( if error_resolution.response_action == ResponseAction.RESET_PAGINATION: raise PaginationResetRequiredException() + if error_resolution.response_action == ResponseAction.RESET_SYNC_TOKEN: + raise SyncTokenResetRequiredException(error_resolution.error_message) + if error_resolution.response_action == ResponseAction.REDUCE_PAGE_SIZE: raise PageSizeReductionRequiredException( stream_name=self._name, error_message=error_resolution.error_message diff --git a/airbyte_cdk/sources/streams/http/sync_token_reset_exception.py b/airbyte_cdk/sources/streams/http/sync_token_reset_exception.py new file mode 100644 index 0000000000..b8ec4cfd64 --- /dev/null +++ b/airbyte_cdk/sources/streams/http/sync_token_reset_exception.py @@ -0,0 +1,28 @@ +# +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +# + +from typing import Optional + +from airbyte_cdk.models import FailureType +from airbyte_cdk.utils.traced_exception import AirbyteTracedException + + +class SyncTokenResetRequiredException(Exception): + def __init__(self, error_message: Optional[str] = None) -> None: + self.error_message = error_message + super().__init__(error_message) + + +class SyncTokenResetNotSupportedException(AirbyteTracedException): + def __init__(self, stream_name: Optional[str] = None) -> None: + stream = f"Stream {stream_name}" if stream_name else "The stream" + message = ( + f"{stream} resolves to a token reset but does not use a SyncTokenCursor. " + "Remove the RESET_SYNC_TOKEN action from its error handler." + ) + super().__init__( + internal_message=message, + message=message, + failure_type=FailureType.config_error, + ) diff --git a/unit_tests/sources/declarative/incremental/test_sync_token_cursor_integration.py b/unit_tests/sources/declarative/incremental/test_sync_token_cursor_integration.py index a0879f815b..0a9e468d1e 100644 --- a/unit_tests/sources/declarative/incremental/test_sync_token_cursor_integration.py +++ b/unit_tests/sources/declarative/incremental/test_sync_token_cursor_integration.py @@ -15,6 +15,7 @@ ConfiguredAirbyteCatalog, ConfiguredAirbyteStream, DestinationSyncMode, + FailureType, SyncMode, Type, ) @@ -217,6 +218,43 @@ def _incoming_state(stream_state): return [state_manager.create_state_message("events", None).state] +def _reset_error_handler(*, reduce_page_size=False, pagination_reset=False): + response_filters = [] + if reduce_page_size: + response_filters.append( + { + "type": "HttpResponseFilter", + "action": "REDUCE_PAGE_SIZE", + "http_codes": [413], + "error_message": "page too large", + } + ) + response_filters.append( + { + "type": "HttpResponseFilter", + "action": "RESET_SYNC_TOKEN", + "predicate": "{{ response.get('success') == false }}", + "error_message": "Ashby rejected the stored sync token (sync_token_expired)", + } + ) + if pagination_reset: + response_filters.append( + { + "type": "HttpResponseFilter", + "action": "RESET_PAGINATION", + "http_codes": [400], + "failure_type": "system_error", + "error_message": "The page cursor expired", + } + ) + return { + "type": "DefaultErrorHandler", + "max_retries": 0, + "backoff_strategies": [], + "response_filters": response_filters, + } + + def test_t_boot_commits_after_three_pages_and_records(): with HttpMocker() as http_mocker: http_mocker.post( @@ -317,6 +355,283 @@ def test_t_fail_preserves_committed_token_and_emits_incomplete_status(): ) +def test_t_reset_restarts_without_token_and_commits_terminal_token(caplog): + state = _incoming_state({"state_type": "sync_token", "token": "T1"}) + manifest = _manifest(error_handler=_reset_error_handler()) + with HttpMocker() as http_mocker: + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1"}), + HttpResponse(json.dumps({"items": [{"id": "old-1"}], "nextCursor": "old-p2"})), + ) + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1", "nextCursor": "old-p2"}), + HttpResponse(json.dumps({"items": [{"id": "old-2"}], "nextCursor": "old-p3"})), + ) + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1", "nextCursor": "old-p3"}), + HttpResponse(json.dumps({"success": False, "error": "sync_token_expired"})), + ) + http_mocker.post( + HttpRequest(_URL, body=None), + HttpResponse(json.dumps({"items": [{"id": "new-1"}], "nextCursor": "new-p2"})), + ) + http_mocker.post( + HttpRequest(_URL, body={"nextCursor": "new-p2"}), + HttpResponse(json.dumps({"items": [{"id": "new-2"}], "syncToken": "T2"})), + ) + source, catalog = _source_and_catalog(manifest, state) + messages = _read(source, catalog, state) + + assert [message.record.data["id"] for message in messages if message.type == Type.RECORD] == [ + "old-1", + "old-2", + "new-1", + "new-2", + ] + assert len(_state(messages)) == 1 + assert _state(messages)[0].stream.stream_state.__dict__["token"] == "T2" + assert ( + "Ashby rejected the stored sync token (sync_token_expired); reading this stream in full." + in caplog.text + ) + assert "2 records read earlier in this attempt may be emitted again." in caplog.text + + +def test_t_reset_context_is_fresh_after_page_size_reduction(caplog): + state = _incoming_state({"state_type": "sync_token", "token": "T1"}) + manifest = _manifest(error_handler=_reset_error_handler(reduce_page_size=True)) + stream = manifest["streams"][0] + stream["retriever"]["paginator"]["page_size_option"] = { + "type": "RequestOption", + "field_name": "pageSize", + "inject_into": "body_json", + } + stream["retriever"]["paginator"]["pagination_strategy"]["page_size"] = 8 + stream["retriever"]["page_size_reduction"] = { + "type": "PageSizeReduction", + "minimum_page_size": 1, + "backoff_seconds": 0, + } + stream["retriever"]["record_selector"]["record_filter"] = { + "type": "RecordFilter", + "condition": "{{ record['expected_token'] == stream_interval.get('token', 'none') }}", + } + stream["transformations"] = [ + { + "type": "AddFields", + "fields": [ + { + "path": ["slice_token"], + "value": "{{ stream_interval.get('token', 'none') }}", + } + ], + } + ] + stream["schema_loader"]["schema"]["properties"].update( + {"expected_token": {"type": "string"}, "slice_token": {"type": "string"}} + ) + + with HttpMocker() as http_mocker: + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1", "pageSize": 8}), + HttpResponse("{}", status_code=413), + ) + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1", "pageSize": 4}), + HttpResponse( + json.dumps( + { + "items": [{"id": "old", "expected_token": "T1"}], + "nextCursor": "old-p2", + } + ) + ), + ) + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1", "nextCursor": "old-p2", "pageSize": 4}), + HttpResponse(json.dumps({"success": False, "error": "sync_token_expired"})), + ) + http_mocker.post( + HttpRequest(_URL, body={"pageSize": 8}), + HttpResponse( + json.dumps( + { + "items": [{"id": "new", "expected_token": "none"}], + "nextCursor": "new-p2", + } + ) + ), + ) + http_mocker.post( + HttpRequest(_URL, body={"nextCursor": "new-p2", "pageSize": 8}), + HttpResponse(json.dumps({"items": [], "syncToken": "T2"})), + ) + source, catalog = _source_and_catalog(manifest, state) + messages = _read(source, catalog, state) + + records = [message.record.data for message in messages if message.type == Type.RECORD] + assert records == [ + {"id": "old", "expected_token": "T1", "slice_token": "T1"}, + {"id": "new", "expected_token": "none", "slice_token": "none"}, + ] + assert _state(messages)[0].stream.stream_state.__dict__["token"] == "T2" + assert "sync_token_expired" in caplog.text + + +def test_t_reset_twice_errors_and_preserves_committed_state(): + state = _incoming_state({"state_type": "sync_token", "token": "T1"}) + manifest = _manifest(error_handler=_reset_error_handler()) + with HttpMocker() as http_mocker: + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1"}), + HttpResponse(json.dumps({"items": [{"id": "old"}], "nextCursor": "old-p2"})), + ) + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1", "nextCursor": "old-p2"}), + HttpResponse(json.dumps({"success": False, "error": "sync_token_expired"})), + ) + http_mocker.post( + HttpRequest(_URL, body=None), + HttpResponse(json.dumps({"items": [{"id": "new"}], "nextCursor": "new-p2"})), + ) + http_mocker.post( + HttpRequest(_URL, body={"nextCursor": "new-p2"}), + HttpResponse(json.dumps({"success": False, "error": "sync_token_expired"})), + ) + source, catalog = _source_and_catalog(manifest, state) + messages = _read_until_error(source, catalog, state) + + assert any(message.type == Type.RECORD for message in messages) + assert _state(messages)[-1].stream.stream_state.__dict__["token"] == "T1" + assert any( + message.type == Type.TRACE + and message.trace.error + and message.trace.error.failure_type == FailureType.system_error + and message.trace.error.message + == "Stream events: the API rejected a read without a sync token. " + "Check the RESET_SYNC_TOKEN response filter." + for message in messages + ) + + +def test_t_rpag_reset_replays_resumed_page_one_with_the_same_token(): + state = _incoming_state({"state_type": "sync_token", "token": "T1"}) + manifest = _manifest(error_handler=_reset_error_handler(pagination_reset=True)) + with HttpMocker() as http_mocker: + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1"}), + [ + HttpResponse(json.dumps({"items": [{"id": "page-1"}], "nextCursor": "old-p2"})), + HttpResponse( + json.dumps({"items": [{"id": "replayed-page-1"}], "nextCursor": "new-p2"}) + ), + ], + ) + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1", "nextCursor": "old-p2"}), + HttpResponse("{}", status_code=400), + ) + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1", "nextCursor": "new-p2"}), + HttpResponse(json.dumps({"items": [{"id": "replayed-page-2"}], "syncToken": "T2"})), + ) + source, catalog = _source_and_catalog(manifest, state) + messages = _read(source, catalog, state) + + assert [message.record.data["id"] for message in messages if message.type == Type.RECORD] == [ + "page-1", + "replayed-page-1", + "replayed-page-2", + ] + assert _state(messages)[-1].stream.stream_state.__dict__["token"] == "T2" + + +def test_t_rpag_reset_after_token_reset_replays_tokenless_page_one(): + state = _incoming_state({"state_type": "sync_token", "token": "T1"}) + manifest = _manifest(error_handler=_reset_error_handler(pagination_reset=True)) + with HttpMocker() as http_mocker: + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1"}), + HttpResponse(json.dumps({"items": [{"id": "old-page-1"}], "nextCursor": "old-p2"})), + ) + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1", "nextCursor": "old-p2"}), + HttpResponse(json.dumps({"success": False, "error": "sync_token_expired"})), + ) + http_mocker.post( + HttpRequest(_URL, body=None), + [ + HttpResponse(json.dumps({"items": [{"id": "page-1"}], "nextCursor": "p2"})), + HttpResponse( + json.dumps({"items": [{"id": "replayed-page-1"}], "nextCursor": "new-p2"}) + ), + ], + ) + http_mocker.post( + HttpRequest(_URL, body={"nextCursor": "p2"}), + HttpResponse("{}", status_code=400), + ) + http_mocker.post( + HttpRequest(_URL, body={"nextCursor": "new-p2"}), + HttpResponse(json.dumps({"items": [{"id": "replayed-page-2"}], "syncToken": "T2"})), + ) + source, catalog = _source_and_catalog(manifest, state) + messages = _read(source, catalog, state) + + assert [message.record.data["id"] for message in messages if message.type == Type.RECORD] == [ + "old-page-1", + "page-1", + "replayed-page-1", + "replayed-page-2", + ] + assert _state(messages)[-1].stream.stream_state.__dict__["token"] == "T2" + + +def test_t_rpag_second_pagination_reset_is_system_error(): + state = _incoming_state({"state_type": "sync_token", "token": "T1"}) + manifest = _manifest(error_handler=_reset_error_handler(pagination_reset=True)) + with HttpMocker() as http_mocker: + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1"}), + [ + HttpResponse(json.dumps({"items": [{"id": "page-1"}], "nextCursor": "p2"})), + HttpResponse( + json.dumps({"items": [{"id": "replayed-page-1"}], "nextCursor": "p2"}) + ), + ], + ) + http_mocker.post( + HttpRequest(_URL, body={"syncToken": "T1", "nextCursor": "p2"}), + [HttpResponse("{}", status_code=400), HttpResponse("{}", status_code=400)], + ) + source, catalog = _source_and_catalog(manifest, state) + messages = _read_until_error(source, catalog, state) + + assert any( + message.type == Type.TRACE + and message.trace.error + and message.trace.error.failure_type == FailureType.system_error + for message in messages + ) + assert _state(messages)[-1].stream.stream_state.__dict__["token"] == "T1" + + +def test_reset_action_on_non_token_stream_is_not_supported(): + manifest = _manifest(error_handler=_reset_error_handler()) + del manifest["streams"][0]["incremental_sync"] + with HttpMocker() as http_mocker: + http_mocker.post( + HttpRequest(_URL, body=None), + HttpResponse(json.dumps({"success": False, "error": "unexpected_reset"})), + ) + source, catalog = _source_and_catalog(manifest, sync_mode=SyncMode.full_refresh) + with pytest.raises(AirbyteTracedException) as exc_info: + _read(source, catalog) + + assert exc_info.value.failure_type.value == "config_error" + assert "RESET_SYNC_TOKEN" in str(exc_info.value) + + def test_t_resume_replays_existing_partition_and_bootstraps_missing_partition(): state = _incoming_state( { diff --git a/unit_tests/sources/declarative/requesters/error_handlers/test_composite_error_handler.py b/unit_tests/sources/declarative/requesters/error_handlers/test_composite_error_handler.py index 78d3b7ee8a..6181cd1b72 100644 --- a/unit_tests/sources/declarative/requesters/error_handlers/test_composite_error_handler.py +++ b/unit_tests/sources/declarative/requesters/error_handlers/test_composite_error_handler.py @@ -132,6 +132,24 @@ def test_given_unmatched_response_or_exception_then_return_default_error_resolut assert error_resolution.failure_type == FailureType.system_error +def test_composite_error_handler_preserves_reset_sync_token(): + reset_handler = MagicMock() + reset_resolution = ErrorResolution( + response_action=ResponseAction.RESET_SYNC_TOKEN, + error_message="sync_token_expired", + ) + reset_handler.interpret_response.return_value = reset_resolution + success_handler = MagicMock() + success_handler.interpret_response.return_value = ErrorResolution( + response_action=ResponseAction.SUCCESS + ) + composite_error_handler = CompositeErrorHandler( + error_handlers=[reset_handler, success_handler], parameters={} + ) + + assert composite_error_handler.interpret_response(MagicMock()) == reset_resolution + + def test_composite_error_handler_no_handlers(): try: CompositeErrorHandler(error_handlers=[], parameters={}) diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py index 31d4ff9a1c..0b8bb28d1f 100644 --- a/unit_tests/sources/streams/http/test_http_client.py +++ b/unit_tests/sources/streams/http/test_http_client.py @@ -45,6 +45,9 @@ Oauth2Authenticator, TokenAuthenticator, ) +from airbyte_cdk.sources.streams.http.sync_token_reset_exception import ( + SyncTokenResetRequiredException, +) from airbyte_cdk.sources.streams.http.tcp_keepalive import TcpKeepaliveHTTPAdapter from airbyte_cdk.utils.constants import ENV_REQUEST_CACHE_PATH from airbyte_cdk.utils.datetime_helpers import ab_datetime_now @@ -244,6 +247,23 @@ def test_send_raises_airbyte_traced_exception_with_fail_response_action(): assert http_client._session.send.call_count == 1 +def test_reset_sync_token_action_raises_exception_with_error_message(): + http_client = test_http_client() + error_message = "sync_token_expired" + with pytest.raises(SyncTokenResetRequiredException) as exc_info: + http_client._handle_error_resolution( + response=None, + exc=None, + request=requests.PreparedRequest(), + error_resolution=ErrorResolution( + response_action=ResponseAction.RESET_SYNC_TOKEN, + error_message=error_message, + ), + ) + + assert exc_info.value.error_message == error_message + + def test_send_ignores_with_ignore_reponse_action_and_returns_response(): mocked_session = MagicMock(spec=requests.Session) mocked_response = MagicMock(spec=requests.Response)