diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index eadd756458..66e518084f 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -2884,6 +2884,7 @@ definitions: - RETRY - IGNORE - RESET_PAGINATION + - RESET_OPAQUE_TOKEN - RATE_LIMITED - REFRESH_TOKEN_THEN_RETRY - REDUCE_PAGE_SIZE @@ -2894,6 +2895,7 @@ definitions: - RETRY - IGNORE - RESET_PAGINATION + - RESET_OPAQUE_TOKEN - RATE_LIMITED - REFRESH_TOKEN_THEN_RETRY - REDUCE_PAGE_SIZE diff --git a/airbyte_cdk/sources/declarative/incremental/opaque_token_cursor.py b/airbyte_cdk/sources/declarative/incremental/opaque_token_cursor.py index cd75943f7e..3bd1c56f8c 100644 --- a/airbyte_cdk/sources/declarative/incremental/opaque_token_cursor.py +++ b/airbyte_cdk/sources/declarative/incremental/opaque_token_cursor.py @@ -197,6 +197,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 change token. " + "Check the RESET_OPAQUE_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 01ed2f3bd8..a7d79fc62d 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_OPAQUE_TOKEN = "RESET_OPAQUE_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_OPAQUE_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 23902e1484..7d8d04eb7a 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -4182,6 +4182,9 @@ def _get_url(req: Requester) -> str: page_observer=( cursor.on_page_completed if isinstance(cursor, OpaqueTokenCursor) else None ), + token_reset_handler=( + cursor.restart_without_token if isinstance(cursor, OpaqueTokenCursor) 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..06560593af 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_OPAQUE_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 d7b63e85fc..6adac54946 100644 --- a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py +++ b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py @@ -57,6 +57,10 @@ from airbyte_cdk.sources.declarative.stream_slicers.stream_slicer import StreamSlicer from airbyte_cdk.sources.source import ExperimentalClassWarning from airbyte_cdk.sources.streams.core import StreamData +from airbyte_cdk.sources.streams.http.opaque_token_reset_exception import ( + OpaqueTokenResetNotSupportedException, + OpaqueTokenResetRequiredException, +) from airbyte_cdk.sources.streams.http.page_size_reduction_exception import ( PageSizeReductionNotSupportedException, PageSizeReductionRequiredException, @@ -154,6 +158,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) @@ -703,6 +708,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 OpaqueTokenResetRequiredException as exception: + if self.token_reset_handler is None: + raise OpaqueTokenResetNotSupportedException(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 change 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..129e086d49 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_OPAQUE_TOKEN = "RESET_OPAQUE_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..97cf6e125d 100644 --- a/airbyte_cdk/sources/streams/http/http_client.py +++ b/airbyte_cdk/sources/streams/http/http_client.py @@ -43,6 +43,9 @@ RequestBodyException, UserDefinedBackoffException, ) +from airbyte_cdk.sources.streams.http.opaque_token_reset_exception import ( + OpaqueTokenResetRequiredException, +) from airbyte_cdk.sources.streams.http.page_size_reduction_exception import ( PageSizeReductionRequiredException, ) @@ -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_OPAQUE_TOKEN: + raise OpaqueTokenResetRequiredException(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/opaque_token_reset_exception.py b/airbyte_cdk/sources/streams/http/opaque_token_reset_exception.py new file mode 100644 index 0000000000..b03e36d4be --- /dev/null +++ b/airbyte_cdk/sources/streams/http/opaque_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 OpaqueTokenResetRequiredException(Exception): + def __init__(self, error_message: Optional[str] = None) -> None: + self.error_message = error_message + super().__init__(error_message) + + +class OpaqueTokenResetNotSupportedException(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 an OpaqueTokenCursor. " + "Remove the RESET_OPAQUE_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_opaque_token_cursor_integration.py b/unit_tests/sources/declarative/incremental/test_opaque_token_cursor_integration.py index 652d45f851..17939dbf33 100644 --- a/unit_tests/sources/declarative/incremental/test_opaque_token_cursor_integration.py +++ b/unit_tests/sources/declarative/incremental/test_opaque_token_cursor_integration.py @@ -13,6 +13,7 @@ ConfiguredAirbyteCatalog, ConfiguredAirbyteStream, DestinationSyncMode, + FailureType, SyncMode, Type, ) @@ -215,6 +216,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_OPAQUE_TOKEN", + "predicate": "{{ response.get('success') == false }}", + "error_message": "Ashby rejected the stored change 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( @@ -311,6 +349,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": "opaque_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 change 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": "opaque_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": "opaque_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 change token. " + "Check the RESET_OPAQUE_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": "opaque_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": "opaque_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": "opaque_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_OPAQUE_TOKEN" in str(exc_info.value) + + def test_t_builder_page_limit_and_max_records_do_not_commit_tokens(): with HttpMocker() as http_mocker: http_mocker.post( 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..4bfbdfc5fa 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_opaque_token(): + reset_handler = MagicMock() + reset_resolution = ErrorResolution( + response_action=ResponseAction.RESET_OPAQUE_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..abc7a03596 100644 --- a/unit_tests/sources/streams/http/test_http_client.py +++ b/unit_tests/sources/streams/http/test_http_client.py @@ -35,6 +35,9 @@ UserDefinedBackoffException, ) from airbyte_cdk.sources.streams.http.http_client import MessageRepresentationAirbyteTracedErrors +from airbyte_cdk.sources.streams.http.opaque_token_reset_exception import ( + OpaqueTokenResetRequiredException, +) from airbyte_cdk.sources.streams.http.page_size_reduction_exception import ( PageSizeReductionRequiredException, ) @@ -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_opaque_token_action_raises_exception_with_error_message(): + http_client = test_http_client() + error_message = "sync_token_expired" + with pytest.raises(OpaqueTokenResetRequiredException) as exc_info: + http_client._handle_error_resolution( + response=None, + exc=None, + request=requests.PreparedRequest(), + error_resolution=ErrorResolution( + response_action=ResponseAction.RESET_OPAQUE_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)