feat(file-based): Migrate unstructured document parser to MarkItDown#1062
Draft
Aaron ("AJ") Steers (aaronsteers) wants to merge 3 commits into
Draft
feat(file-based): Migrate unstructured document parser to MarkItDown#1062Aaron ("AJ") Steers (aaronsteers) wants to merge 3 commits into
Aaron ("AJ") Steers (aaronsteers) wants to merge 3 commits into
GitHub Actions / PyTest Results (Full)
failed
Jul 1, 2026 in 0s
104 fail, 12 skipped, 3 997 pass in 20635d 21h 27m 42s
Annotations
Check warning on line 0 in unit_tests.test_entrypoint
github-actions / PyTest Results (Full)
test_run_check_with_exception (unit_tests.test_entrypoint) failed
build/test-results/pytest-results.xml [took 1m 43s]
Raw output
pydantic_core._pydantic_core.ValidationError: 1 validation error for RunnablePassthrough[Any]
name
Field required [type=missing, input_value={'func': None, 'afunc': None, 'input_type': None}, input_type=dict]
For further information visit https://errors.pydantic.dev/2.14/v/missing
args = ()
kwargs = {'config_mock': {'username': 'fake'}, 'entrypoint': <airbyte_cdk.entrypoint.AirbyteEntrypoint object at 0x7fa63b942300>, 'mocker': <pytest_mock.plugin.MockerFixture object at 0x7fa63b8a9fd0>, 'spec_mock': <MagicMock id='140351940862336'>}
@functools.wraps(func)
def wrapper(*args: "P.args", **kwargs: "P.kwargs") -> T:
> with self as time_factory:
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/freezegun/api.py:950:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/freezegun/api.py:738: in __enter__
return self.start()
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/freezegun/api.py:827: in start
module_attrs = _get_cached_module_attributes(module)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/freezegun/api.py:144: in _get_cached_module_attributes
_setup_module_cache(module)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/freezegun/api.py:123: in _setup_module_cache
all_module_attributes = _get_module_attributes(module)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/freezegun/api.py:112: in _get_module_attributes
attribute_value = getattr(module, attribute_name)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/langchain_core/runnables/__init__.py:130: in __getattr__
result = import_attr(attr_name, module_name, __spec__.parent)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/langchain_core/_import_utils.py:36: in import_attr
module = import_module(f".{module_name}", package=package)
/usr/lib/python3.12/importlib/__init__.py:90: in import_module
return _bootstrap._gcd_import(name[level:], package, level)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/langchain_core/runnables/passthrough.py:349: in <module>
_graph_passthrough = RunnablePassthrough[Any]()
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/langchain_core/runnables/passthrough.py:178: in __init__
super().__init__(func=func, afunc=afunc, input_type=input_type, **kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = RunnablePassthrough[Any](), args = ()
kwargs = {'afunc': None, 'func': None, 'input_type': None}
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""""" # noqa: D419 # Intentional blank docstring
> super().__init__(*args, **kwargs)
E pydantic_core._pydantic_core.ValidationError: 1 validation error for RunnablePassthrough[Any]
E name
E Field required [type=missing, input_value={'func': None, 'afunc': None, 'input_type': None}, input_type=dict]
E For further information visit https://errors.pydantic.dev/2.14/v/missing
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/langchain_core/load/serializable.py:118: ValidationError
Check warning on line 0 in unit_tests.connector_builder.test_connector_builder_handler
github-actions / PyTest Results (Full)
test_full_resolve_manifest (unit_tests.connector_builder.test_connector_builder_handler) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
AttributeError: 'NoneType' object has no attribute 'data'
valid_resolve_manifest_config_file = PosixPath('/tmp/pytest-of-runner/pytest-0/test_full_resolve_manifest0/config.json')
def test_full_resolve_manifest(valid_resolve_manifest_config_file):
config = copy.deepcopy(RESOLVE_DYNAMIC_STREAM_MANIFEST_CONFIG)
command = config["__command"]
source = ConcurrentDeclarativeSource(
catalog=None, config=config, state=None, source_config=DYNAMIC_STREAM_MANIFEST
)
limits = TestLimits(max_streams=2)
with HttpMocker() as http_mocker:
http_mocker.get(
HttpRequest(url="https://api.test.com/parents"),
HttpResponse(body=json.dumps([{"id": 1}, {"id": 2}])),
)
parent_ids = [1, 2]
for parent_id in parent_ids:
http_mocker.get(
HttpRequest(url=f"https://api.test.com/parent/{parent_id}/items"),
HttpResponse(
body=json.dumps(
[
{"id": 1, "name": "item_1"},
{"id": 2, "name": "item_2"},
]
)
),
)
resolved_manifest = handle_connector_builder_request(
source, command, config, create_configured_catalog("dummy_stream"), _A_STATE, limits
)
expected_resolved_manifest = {
"version": "0.30.3",
"definitions": {
"retriever": {
"paginator": {
"type": "DefaultPaginator",
"page_size": 2,
"page_size_option": {
"inject_into": "request_parameter",
"field_name": "page_size",
},
"page_token_option": {"inject_into": "path", "type": "RequestPath"},
"pagination_strategy": {
"type": "CursorPagination",
"cursor_value": "{{ response._metadata.next }}",
"page_size": 2,
},
},
"partition_router": {
"type": "ListPartitionRouter",
"values": ["0", "1", "2", "3", "4", "5", "6", "7"],
"cursor_field": "item_id",
},
"requester": {
"path": "/v3/marketing/lists",
"authenticator": {
"type": "BearerAuthenticator",
"api_token": "{{ config.apikey }}",
},
"request_parameters": {"a_param": "10"},
},
"record_selector": {"extractor": {"field_path": ["result"]}},
}
},
"streams": [
{
"type": "DeclarativeStream",
"retriever": {
"paginator": {
"type": "DefaultPaginator",
"page_size": 2,
"page_size_option": {
"inject_into": "request_parameter",
"field_name": "page_size",
"type": "RequestOption",
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
"$parameters": {
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
},
},
"page_token_option": {
"inject_into": "path",
"type": "RequestPath",
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
"$parameters": {
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
},
},
"pagination_strategy": {
"type": "CursorPagination",
"cursor_value": "{{ response._metadata.next }}",
"page_size": 2,
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
"$parameters": {
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
},
},
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
"$parameters": {
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
},
},
"partition_router": {
"type": "ListPartitionRouter",
"values": ["0", "1", "2", "3", "4", "5", "6", "7"],
"cursor_field": "item_id",
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
"$parameters": {
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
},
},
"requester": {
"path": "/v3/marketing/lists",
"authenticator": {
"type": "BearerAuthenticator",
"api_token": "{{ config.apikey }}",
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
"$parameters": {
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
},
},
"request_parameters": {"a_param": "10"},
"type": "HttpRequester",
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
"$parameters": {
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
},
},
"record_selector": {
"extractor": {
"field_path": ["result"],
"type": "DpathExtractor",
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
"$parameters": {
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
},
},
"type": "RecordSelector",
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
"$parameters": {
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
},
},
"type": "SimpleRetriever",
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
"$parameters": {
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
},
},
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
"$parameters": {
"name": "stream_with_custom_requester",
"primary_key": "id",
"url_base": "https://api.sendgrid.com",
},
"dynamic_stream_name": None,
},
{
"type": "DeclarativeStream",
"name": "parent_1_item_1",
"primary_key": [],
"schema_loader": {
"type": "InlineSchemaLoader",
"schema": {
"$schema": "http://json-schema.org/schema#",
"properties": {"ABC": {"type": "number"}, "AED": {"type": "number"}},
"type": "object",
},
},
"retriever": {
"type": "SimpleRetriever",
"requester": {
"type": "HttpRequester",
"url_base": "https://api.test.com",
"path": "1/1",
"http_method": "GET",
"authenticator": {
"type": "ApiKeyAuthenticator",
"header": "apikey",
"api_token": "{{ config['api_key'] }}",
},
},
"record_selector": {
"type": "RecordSelector",
"extractor": {"type": "DpathExtractor", "field_path": []},
},
"paginator": {"type": "NoPagination"},
},
"dynamic_stream_name": "TestDynamicStream",
},
{
"type": "DeclarativeStream",
"name": "parent_1_item_2",
"primary_key": [],
"schema_loader": {
"type": "InlineSchemaLoader",
"schema": {
"$schema": "http://json-schema.org/schema#",
"properties": {"ABC": {"type": "number"}, "AED": {"type": "number"}},
"type": "object",
},
},
"retriever": {
"type": "SimpleRetriever",
"requester": {
"type": "HttpRequester",
"url_base": "https://api.test.com",
"path": "1/2",
"http_method": "GET",
"authenticator": {
"type": "ApiKeyAuthenticator",
"header": "apikey",
"api_token": "{{ config['api_key'] }}",
},
},
"record_selector": {
"type": "RecordSelector",
"extractor": {"type": "DpathExtractor", "field_path": []},
},
"paginator": {"type": "NoPagination"},
},
"dynamic_stream_name": "TestDynamicStream",
},
],
"check": {"type": "CheckStream", "stream_names": ["lists"]},
"spec": {
"connection_specification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"type": "object",
"required": [],
"properties": {},
"additionalProperties": True,
},
"type": "Spec",
},
"dynamic_streams": [
{
"type": "DynamicDeclarativeStream",
"name": "TestDynamicStream",
"stream_template": {
"type": "DeclarativeStream",
"name": "",
"primary_key": [],
"schema_loader": {
"type": "InlineSchemaLoader",
"schema": {
"$schema": "http://json-schema.org/schema#",
"properties": {"ABC": {"type": "number"}, "AED": {"type": "number"}},
"type": "object",
},
},
"retriever": {
"type": "SimpleRetriever",
"requester": {
"type": "HttpRequester",
"url_base": "https://api.test.com",
"path": "",
"http_method": "GET",
"authenticator": {
"type": "ApiKeyAuthenticator",
"header": "apikey",
"api_token": "{{ config['api_key'] }}",
},
},
"record_selector": {
"type": "RecordSelector",
"extractor": {"type": "DpathExtractor", "field_path": []},
},
"paginator": {"type": "NoPagination"},
},
},
"components_resolver": {
"type": "HttpComponentsResolver",
"retriever": {
"type": "SimpleRetriever",
"requester": {
"type": "HttpRequester",
"url_base": "https://api.test.com",
"path": "parent/{{ stream_partition.parent_id }}/items",
"http_method": "GET",
"authenticator": {
"type": "ApiKeyAuthenticator",
"header": "apikey",
"api_token": "{{ config['api_key'] }}",
},
"use_cache": True,
},
"record_selector": {
"type": "RecordSelector",
"extractor": {"type": "DpathExtractor", "field_path": []},
},
"paginator": {"type": "NoPagination"},
"partition_router": {
"type": "SubstreamPartitionRouter",
"parent_stream_configs": [
{
"type": "ParentStreamConfig",
"parent_key": "id",
"partition_field": "parent_id",
"stream": {
"type": "DeclarativeStream",
"name": "parent",
"retriever": {
"type": "SimpleRetriever",
"requester": {
"type": "HttpRequester",
"url_base": "https://api.test.com",
"path": "/parents",
"http_method": "GET",
"authenticator": {
"type": "ApiKeyAuthenticator",
"header": "apikey",
"api_token": "{{ config['api_key'] }}",
},
},
"record_selector": {
"type": "RecordSelector",
"extractor": {
"type": "DpathExtractor",
"field_path": [],
},
},
},
"schema_loader": {
"type": "InlineSchemaLoader",
"schema": {
"$schema": "http://json-schema.org/schema#",
"properties": {"id": {"type": "integer"}},
"type": "object",
},
},
},
}
],
},
},
"components_mapping": [
{
"type": "ComponentMappingDefinition",
"field_path": ["name"],
"value": "parent_{{stream_slice['parent_id']}}_{{components_values['name']}}",
},
{
"type": "ComponentMappingDefinition",
"field_path": ["retriever", "requester", "path"],
"value": "{{ stream_slice['parent_id'] }}/{{ components_values['id'] }}",
},
],
},
}
],
"type": "DeclarativeSource",
}
> assert resolved_manifest.record.data["manifest"] == expected_resolved_manifest
E AttributeError: 'NoneType' object has no attribute 'data'
unit_tests/connector_builder/test_connector_builder_handler.py:1805: AttributeError
Check warning on line 0 in unit_tests.legacy.sources.declarative.test_manifest_declarative_source
github-actions / PyTest Results (Full)
test_slice_checkpoint[test_with_pagination_and_partition_router-manifest0-pages0-2] (unit_tests.legacy.sources.declarative.test_manifest_declarative_source) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
AssertionError: assert 1 == 2
+ where 1 = len([AirbyteStateMessage(type=<AirbyteStateType.STREAM: 'STREAM'>, stream=AirbyteStreamState(stream_descriptor=StreamDescriptor(name='Rates', namespace=None), stream_state=AirbyteStateBlob()), global_=None, data=None, sourceStats=None, destinationStats=None)])
test_name = 'test_with_pagination_and_partition_router'
manifest = {'check': {'stream_names': ['Rates'], 'type': 'CheckStream'}, 'spec': {'connection_specification': {'$schema': 'http:/...T', 'path': '/exchangerates_data/latest', 'request_body_json': {}, ...}, ...}, ...}], 'type': 'DeclarativeSource', ...}
pages = (<Response [200]>, <Response [200]>, <Response [200]>)
expected_states_qty = 2
@pytest.mark.parametrize(
"test_name, manifest, pages, expected_states_qty",
[
(
"test_with_pagination_and_partition_router",
{
"version": "0.34.2",
"type": "DeclarativeSource",
"check": {"type": "CheckStream", "stream_names": ["Rates"]},
"streams": [
{
"type": "DeclarativeStream",
"name": "Rates",
"primary_key": [],
"schema_loader": {
"type": "InlineSchemaLoader",
"schema": {
"$schema": "http://json-schema.org/schema#",
"properties": {
"ABC": {"type": "number"},
"AED": {"type": "number"},
"partition": {"type": "number"},
},
"type": "object",
},
},
"retriever": {
"type": "SimpleRetriever",
"requester": {
"type": "HttpRequester",
"url_base": "https://api.apilayer.com",
"path": "/exchangerates_data/latest",
"http_method": "GET",
"request_parameters": {},
"request_headers": {},
"request_body_json": {},
"authenticator": {
"type": "ApiKeyAuthenticator",
"header": "apikey",
"api_token": "{{ config['api_key'] }}",
},
},
"partition_router": {
"type": "ListPartitionRouter",
"values": ["0", "1"],
"cursor_field": "partition",
},
"record_selector": {
"type": "RecordSelector",
"extractor": {"type": "DpathExtractor", "field_path": ["rates"]},
},
"paginator": {
"type": "DefaultPaginator",
"page_size": 2,
"page_size_option": {
"inject_into": "request_parameter",
"field_name": "page_size",
},
"page_token_option": {"inject_into": "path", "type": "RequestPath"},
"pagination_strategy": {
"type": "CursorPagination",
"cursor_value": "{{ response._metadata.next }}",
"page_size": 2,
},
},
},
"incremental_sync": {
"type": "DatetimeBasedCursor",
"cursor_datetime_formats": ["%Y-%m-%dT%H:%M:%S.%fZ"],
"datetime_format": "%Y-%m-%dT%H:%M:%S.%fZ",
"cursor_field": "updated_at",
"start_datetime": {
"datetime": "{{ config.get('start_date', '2020-10-16T00:00:00.000Z') }}"
},
},
}
],
"spec": {
"connection_specification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"type": "object",
"required": ["api_key"],
"properties": {
"api_key": {
"type": "string",
"title": "API Key",
"airbyte_secret": True,
},
"start_date": {
"title": "Start Date",
"description": "UTC date and time in the format YYYY-MM-DDTHH:MM:SS.000Z. During incremental sync, any data generated before this date will not be replicated. If left blank, the start date will be set to 2 years before the present date.",
"pattern": "^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$",
"pattern_descriptor": "YYYY-MM-DDTHH:MM:SS.000Z",
"examples": ["2020-11-16T00:00:00.000Z"],
"type": "string",
"format": "date-time",
},
},
"additionalProperties": True,
},
"documentation_url": "https://example.org",
"type": "Spec",
},
},
(
_create_page(
{
"rates": [
{"ABC": 0, "partition": 0, "updated_at": "2020-11-16T00:00:00.000Z"},
{"AED": 1, "partition": 0, "updated_at": "2020-11-16T00:00:00.000Z"},
],
"_metadata": {"next": "next"},
}
),
_create_page(
{
"rates": [
{"USD": 3, "partition": 0, "updated_at": "2020-11-16T00:00:00.000Z"}
],
"_metadata": {},
}
),
_create_page(
{
"rates": [
{"ABC": 2, "partition": 1, "updated_at": "2020-11-16T00:00:00.000Z"}
],
"_metadata": {},
}
),
),
2,
),
],
)
def test_slice_checkpoint(test_name, manifest, pages, expected_states_qty):
_stream_name = "Rates"
with patch.object(SimpleRetriever, "_fetch_next_page", side_effect=pages):
states = [message.state for message in _run_read(manifest, _stream_name) if message.state]
> assert len(states) == expected_states_qty
E AssertionError: assert 1 == 2
E + where 1 = len([AirbyteStateMessage(type=<AirbyteStateType.STREAM: 'STREAM'>, stream=AirbyteStreamState(stream_descriptor=StreamDescriptor(name='Rates', namespace=None), stream_state=AirbyteStateBlob()), global_=None, data=None, sourceStats=None, destinationStats=None)])
unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py:1969: AssertionError
github-actions / PyTest Results (Full)
test_given_multiple_cursor_datetime_format_then_slice_using_first_format (unit_tests.legacy.sources.declarative.incremental.test_datetime_based_cursor) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
AssertionError: assert [{'start_time...'1970-01-01'}] == [{'end_time':...'2021-01-01'}]
At index 0 diff: {'start_time': '1970-01-01', 'end_time': '1970-01-01'} != {'start_time': '2021-01-01', 'end_time': '2023-01-10'}
Full diff:
- [{'end_time': '2023-01-10', 'start_time': '2021-01-01'}]
+ [{'start_time': '1970-01-01', 'end_time': '1970-01-01'}]
def test_given_multiple_cursor_datetime_format_then_slice_using_first_format():
cursor = DatetimeBasedCursor(
start_datetime=MinMaxDatetime("2021-01-01", parameters={}),
end_datetime=MinMaxDatetime("2023-01-10", parameters={}),
cursor_field=InterpolatedString(cursor_field, parameters={}),
datetime_format="%Y-%m-%d",
cursor_datetime_formats=["%Y-%m-%dT%H:%M:%S", "%Y-%m-%d"],
config=config,
parameters={},
)
stream_slices = cursor.stream_slices()
> assert stream_slices == [{"start_time": "2021-01-01", "end_time": "2023-01-10"}]
E AssertionError: assert [{'start_time...'1970-01-01'}] == [{'end_time':...'2021-01-01'}]
E At index 0 diff: {'start_time': '1970-01-01', 'end_time': '1970-01-01'} != {'start_time': '2021-01-01', 'end_time': '2023-01-10'}
E Full diff:
E - [{'end_time': '2023-01-10', 'start_time': '2021-01-01'}]
E + [{'start_time': '1970-01-01', 'end_time': '1970-01-01'}]
unit_tests/legacy/sources/declarative/incremental/test_datetime_based_cursor.py:1109: AssertionError
github-actions / PyTest Results (Full)
test_no_cursor_granularity_and_no_step_then_only_return_one_slice (unit_tests.legacy.sources.declarative.incremental.test_datetime_based_cursor) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
AssertionError: assert [{'start_time...'1970-01-01'}] == [{'end_time':...'2021-01-01'}]
At index 0 diff: {'start_time': '1970-01-01', 'end_time': '1970-01-01'} != {'start_time': '2021-01-01', 'end_time': '2023-01-01'}
Full diff:
- [{'end_time': '2023-01-01', 'start_time': '2021-01-01'}]
+ [{'start_time': '1970-01-01', 'end_time': '1970-01-01'}]
def test_no_cursor_granularity_and_no_step_then_only_return_one_slice():
cursor = DatetimeBasedCursor(
start_datetime=MinMaxDatetime("2021-01-01", parameters={}),
end_datetime=MinMaxDatetime("2023-01-01", parameters={}),
cursor_field=InterpolatedString(cursor_field, parameters={}),
datetime_format="%Y-%m-%d",
config=config,
parameters={},
)
stream_slices = cursor.stream_slices()
> assert stream_slices == [{"start_time": "2021-01-01", "end_time": "2023-01-01"}]
E AssertionError: assert [{'start_time...'1970-01-01'}] == [{'end_time':...'2021-01-01'}]
E At index 0 diff: {'start_time': '1970-01-01', 'end_time': '1970-01-01'} != {'start_time': '2021-01-01', 'end_time': '2023-01-01'}
E Full diff:
E - [{'end_time': '2023-01-01', 'start_time': '2021-01-01'}]
E + [{'start_time': '1970-01-01', 'end_time': '1970-01-01'}]
unit_tests/legacy/sources/declarative/incremental/test_datetime_based_cursor.py:1122: AssertionError
github-actions / PyTest Results (Full)
test_given_no_state_and_start_before_cursor_value_when_should_be_synced_then_return_true (unit_tests.legacy.sources.declarative.incremental.test_datetime_based_cursor) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
AssertionError: assert False
+ where False = <bound method DatetimeBasedCursor.should_be_synced of DatetimeBasedCursor(start_datetime=MinMaxDatetime(datetime=InterpolatedString(string='2021-01-01', default='2021-01-01'), datetime_format='%Y-%m-%d', min_datetime=None, max_datetime=None), cursor_field=InterpolatedString(string='created', default='created'), datetime_format='%Y-%m-%d', config={'start_date': '2021-01-01T00:00:00.000000+0000', 'start_date_ymd': '2021-01-01'}, end_datetime=None, step=None, cursor_granularity=None, start_time_option=None, end_time_option=None, partition_field_start=None, partition_field_end=None, lookback_window=None, message_repository=None, is_compare_strictly=False, cursor_datetime_formats=['%Y-%m-%d'])>({'created': '2022-01-01'})
+ where <bound method DatetimeBasedCursor.should_be_synced of DatetimeBasedCursor(start_datetime=MinMaxDatetime(datetime=InterpolatedString(string='2021-01-01', default='2021-01-01'), datetime_format='%Y-%m-%d', min_datetime=None, max_datetime=None), cursor_field=InterpolatedString(string='created', default='created'), datetime_format='%Y-%m-%d', config={'start_date': '2021-01-01T00:00:00.000000+0000', 'start_date_ymd': '2021-01-01'}, end_datetime=None, step=None, cursor_granularity=None, start_time_option=None, end_time_option=None, partition_field_start=None, partition_field_end=None, lookback_window=None, message_repository=None, is_compare_strictly=False, cursor_datetime_formats=['%Y-%m-%d'])> = DatetimeBasedCursor(start_datetime=MinMaxDatetime(datetime=InterpolatedString(string='2021-01-01', default='2021-01-01'), datetime_format='%Y-%m-%d', min_datetime=None, max_datetime=None), cursor_field=InterpolatedString(string='created', default='created'), datetime_format='%Y-%m-%d', config={'start_date': '2021-01-01T00:00:00.000000+0000', 'start_date_ymd': '2021-01-01'}, end_datetime=None, step=None, cursor_granularity=None, start_time_option=None, end_time_option=None, partition_field_start=None, partition_field_end=None, lookback_window=None, message_repository=None, is_compare_strictly=False, cursor_datetime_formats=['%Y-%m-%d']).should_be_synced
+ and {'created': '2022-01-01'} = Record({'created': '2022-01-01'}, {})
def test_given_no_state_and_start_before_cursor_value_when_should_be_synced_then_return_true():
cursor = DatetimeBasedCursor(
start_datetime=MinMaxDatetime("2021-01-01", parameters={}),
cursor_field=InterpolatedString(cursor_field, parameters={}),
datetime_format="%Y-%m-%d",
config=config,
parameters={},
)
> assert cursor.should_be_synced(Record({cursor_field: "2022-01-01"}, ANY_SLICE))
E AssertionError: assert False
E + where False = <bound method DatetimeBasedCursor.should_be_synced of DatetimeBasedCursor(start_datetime=MinMaxDatetime(datetime=InterpolatedString(string='2021-01-01', default='2021-01-01'), datetime_format='%Y-%m-%d', min_datetime=None, max_datetime=None), cursor_field=InterpolatedString(string='created', default='created'), datetime_format='%Y-%m-%d', config={'start_date': '2021-01-01T00:00:00.000000+0000', 'start_date_ymd': '2021-01-01'}, end_datetime=None, step=None, cursor_granularity=None, start_time_option=None, end_time_option=None, partition_field_start=None, partition_field_end=None, lookback_window=None, message_repository=None, is_compare_strictly=False, cursor_datetime_formats=['%Y-%m-%d'])>({'created': '2022-01-01'})
E + where <bound method DatetimeBasedCursor.should_be_synced of DatetimeBasedCursor(start_datetime=MinMaxDatetime(datetime=InterpolatedString(string='2021-01-01', default='2021-01-01'), datetime_format='%Y-%m-%d', min_datetime=None, max_datetime=None), cursor_field=InterpolatedString(string='created', default='created'), datetime_format='%Y-%m-%d', config={'start_date': '2021-01-01T00:00:00.000000+0000', 'start_date_ymd': '2021-01-01'}, end_datetime=None, step=None, cursor_granularity=None, start_time_option=None, end_time_option=None, partition_field_start=None, partition_field_end=None, lookback_window=None, message_repository=None, is_compare_strictly=False, cursor_datetime_formats=['%Y-%m-%d'])> = DatetimeBasedCursor(start_datetime=MinMaxDatetime(datetime=InterpolatedString(string='2021-01-01', default='2021-01-01'), datetime_format='%Y-%m-%d', min_datetime=None, max_datetime=None), cursor_field=InterpolatedString(string='created', default='created'), datetime_format='%Y-%m-%d', config={'start_date': '2021-01-01T00:00:00.000000+0000', 'start_date_ymd': '2021-01-01'}, end_datetime=None, step=None, cursor_granularity=None, start_time_option=None, end_time_option=None, partition_field_start=None, partition_field_end=None, lookback_window=None, message_repository=None, is_compare_strictly=False, cursor_datetime_formats=['%Y-%m-%d']).should_be_synced
E + and {'created': '2022-01-01'} = Record({'created': '2022-01-01'}, {})
unit_tests/legacy/sources/declarative/incremental/test_datetime_based_cursor.py:1147: AssertionError
Check warning on line 0 in unit_tests.manifest_server.helpers.test_auth.TestVerifyJwtToken
github-actions / PyTest Results (Full)
test_valid_token_passes (unit_tests.manifest_server.helpers.test_auth.TestVerifyJwtToken) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
fastapi.exceptions.HTTPException: 401: Invalid token
credentials = HTTPAuthorizationCredentials(scheme='Bearer', credentials='eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE3ODI5NDQ5NjMsImlhdCI6MTc4Mjk0MTM2Mywic3ViIjoidGVzdC11c2VyIn0.CGQYMf0N8EvQp5OEtxV6xQ-a8Vs--vgWdwV8hFy97kI')
def verify_jwt_token(
credentials: Optional[HTTPAuthorizationCredentials] = Security(security),
) -> None:
"""
Verify JWT token if AB_JWT_SIGNATURE_SECRET is set, otherwise allow through.
Args:
credentials: Bearer token credentials from request header
Raises:
HTTPException: If token is invalid or missing when secret is configured
"""
jwt_secret = os.getenv("AB_JWT_SIGNATURE_SECRET")
# If no secret is configured, allow all requests through
if not jwt_secret:
return
if not credentials:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Bearer token required",
headers={"WWW-Authenticate": "Bearer"},
)
try:
> jwt.decode(credentials.credentials, jwt_secret, algorithms=["HS256"])
airbyte_cdk/manifest_server/helpers/auth.py:37:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/jwt/api_jwt.py:368: in decode
decoded = self.decode_complete(
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/jwt/api_jwt.py:275: in decode_complete
self._validate_claims(
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/jwt/api_jwt.py:402: in _validate_claims
self._validate_iat(payload, now, leeway)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <jwt.api_jwt.PyJWT object at 0x7fa65f967b00>
payload = {'exp': 1782944963, 'iat': 1782941363, 'sub': 'test-user'}
now = 0.001, leeway = 0
def _validate_iat(
self,
payload: dict[str, Any],
now: float,
leeway: float,
) -> None:
try:
iat = int(payload["iat"])
except ValueError:
raise InvalidIssuedAtError(
"Issued At claim (iat) must be an integer."
) from None
if iat > (now + leeway):
> raise ImmatureSignatureError("The token is not yet valid (iat)")
E jwt.exceptions.ImmatureSignatureError: The token is not yet valid (iat)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/jwt/api_jwt.py:481: ImmatureSignatureError
During handling of the above exception, another exception occurred:
self = <unit_tests.manifest_server.helpers.test_auth.TestVerifyJwtToken object at 0x7fa6400931a0>
def test_valid_token_passes(self):
"""Test that valid JWT tokens pass verification."""
secret = "test-secret-key"
payload = {
"exp": datetime.now(timezone.utc) + timedelta(hours=1),
"iat": datetime.now(timezone.utc),
"sub": "test-user",
}
valid_token = jwt.encode(payload, secret, algorithm="HS256")
with patch.dict(os.environ, {"AB_JWT_SIGNATURE_SECRET": secret}):
valid_credentials = HTTPAuthorizationCredentials(
scheme="Bearer", credentials=valid_token
)
# Should not raise any exception
> verify_jwt_token(valid_credentials)
unit_tests/manifest_server/helpers/test_auth.py:76:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
credentials = HTTPAuthorizationCredentials(scheme='Bearer', credentials='eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE3ODI5NDQ5NjMsImlhdCI6MTc4Mjk0MTM2Mywic3ViIjoidGVzdC11c2VyIn0.CGQYMf0N8EvQp5OEtxV6xQ-a8Vs--vgWdwV8hFy97kI')
def verify_jwt_token(
credentials: Optional[HTTPAuthorizationCredentials] = Security(security),
) -> None:
"""
Verify JWT token if AB_JWT_SIGNATURE_SECRET is set, otherwise allow through.
Args:
credentials: Bearer token credentials from request header
Raises:
HTTPException: If token is invalid or missing when secret is configured
"""
jwt_secret = os.getenv("AB_JWT_SIGNATURE_SECRET")
# If no secret is configured, allow all requests through
if not jwt_secret:
return
if not credentials:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Bearer token required",
headers={"WWW-Authenticate": "Bearer"},
)
try:
jwt.decode(credentials.credentials, jwt_secret, algorithms=["HS256"])
except jwt.InvalidTokenError:
> raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token",
headers={"WWW-Authenticate": "Bearer"},
)
E fastapi.exceptions.HTTPException: 401: Invalid token
airbyte_cdk/manifest_server/helpers/auth.py:39: HTTPException
Check warning on line 0 in unit_tests.sources.declarative.test_concurrent_declarative_source
github-actions / PyTest Results (Full)
test_check (unit_tests.sources.declarative.test_concurrent_declarative_source) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
assert <Status.FAILED: 'FAILED'> == <Status.SUCCEEDED: 'SUCCEEDED'>
+ where <Status.FAILED: 'FAILED'> = AirbyteConnectionStatus(status=<Status.FAILED: 'FAILED'>, message="'Stream party_members is not available: Cannot attempt to connect to stream party_members - no stream slices were found'").status
+ and <Status.SUCCEEDED: 'SUCCEEDED'> = Status.SUCCEEDED
def test_check():
"""
Verifies that the ConcurrentDeclarativeSource check command is run against synchronous streams
"""
with HttpMocker() as http_mocker:
http_mocker.get(
HttpRequest(
"https://persona.metaverse.com/party_members?start=2024-07-01&end=2024-07-15"
),
HttpResponse(
json.dumps(
{
"id": "amamiya",
"first_name": "ren",
"last_name": "amamiya",
"updated_at": "2024-07-10",
}
)
),
)
http_mocker.get(
HttpRequest("https://persona.metaverse.com/palaces"),
HttpResponse(json.dumps({"id": "palace_1"})),
)
http_mocker.get(
HttpRequest(
"https://persona.metaverse.com/locations?m=active&i=1&g=country&start=2024-07-01&end=2024-07-31"
),
HttpResponse(json.dumps({"id": "location_1"})),
)
source = ConcurrentDeclarativeSource(
source_config=_MANIFEST, config=_CONFIG, catalog=None, state=None
)
connection_status = source.check(logger=source.logger, config=_CONFIG)
> assert connection_status.status == Status.SUCCEEDED
E assert <Status.FAILED: 'FAILED'> == <Status.SUCCEEDED: 'SUCCEEDED'>
E + where <Status.FAILED: 'FAILED'> = AirbyteConnectionStatus(status=<Status.FAILED: 'FAILED'>, message="'Stream party_members is not available: Cannot attempt to connect to stream party_members - no stream slices were found'").status
E + and <Status.SUCCEEDED: 'SUCCEEDED'> = Status.SUCCEEDED
unit_tests/sources/declarative/test_concurrent_declarative_source.py:819: AssertionError
Check warning on line 0 in unit_tests.sources.declarative.test_concurrent_declarative_source
github-actions / PyTest Results (Full)
test_read_with_concurrent_and_synchronous_streams_with_concurrent_state (unit_tests.sources.declarative.test_concurrent_declarative_source) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
airbyte_cdk.utils.traced_exception.AirbyteTracedException: During the sync, the following streams did not sync successfully: party_members_skills: NameError("name 'RequestsCookieJar' is not defined")
party_members: NameError("name 'RequestsCookieJar' is not defined")
party_members: NameError("name 'RequestsCookieJar' is not defined")
party_members: NameError("name 'RequestsCookieJar' is not defined")
palaces: NameError("name 'RequestsCookieJar' is not defined")
@freezegun.freeze_time(_NOW)
@patch(
"airbyte_cdk.sources.streams.concurrent.state_converters.abstract_stream_state_converter.AbstractStreamStateConverter.__init__",
mocked_init,
)
def test_read_with_concurrent_and_synchronous_streams_with_concurrent_state():
"""
Verifies that a ConcurrentDeclarativeSource processes concurrent streams correctly using the incoming
concurrent state format
"""
state = [
AirbyteStateMessage(
type=AirbyteStateType.STREAM,
stream=AirbyteStreamState(
stream_descriptor=StreamDescriptor(name="locations", namespace=None),
stream_state=AirbyteStateBlob(
state_type="date-range",
slices=[{"start": "2024-07-01", "end": "2024-07-31"}],
),
),
),
AirbyteStateMessage(
type=AirbyteStateType.STREAM,
stream=AirbyteStreamState(
stream_descriptor=StreamDescriptor(name="party_members", namespace=None),
stream_state=AirbyteStateBlob(
state_type="date-range",
slices=[
{"start": "2024-07-16", "end": "2024-07-30"},
{"start": "2024-07-31", "end": "2024-08-14"},
{"start": "2024-08-30", "end": "2024-09-09"},
],
),
),
),
]
party_members_slices_and_responses = _NO_STATE_PARTY_MEMBERS_SLICES_AND_RESPONSES + [
(
{"start": "2024-09-04", "end": "2024-09-10"}, # considering lookback window
HttpResponse(
json.dumps(
[
{
"id": "yoshizawa",
"first_name": "sumire",
"last_name": "yoshizawa",
"updated_at": "2024-09-10",
}
]
)
),
)
]
location_slices = [
{"start": "2024-07-26", "end": "2024-08-25"},
{"start": "2024-08-26", "end": "2024-09-10"},
]
source = ConcurrentDeclarativeSource(
source_config=_MANIFEST, config=_CONFIG, catalog=_CATALOG, state=state
)
with HttpMocker() as http_mocker:
_mock_party_members_requests(http_mocker, party_members_slices_and_responses)
_mock_locations_requests(http_mocker, location_slices)
http_mocker.get(HttpRequest("https://persona.metaverse.com/palaces"), _PALACES_RESPONSE)
_mock_party_members_skills_requests(http_mocker)
> messages = list(
source.read(logger=source.logger, config=_CONFIG, catalog=_CATALOG, state=state)
)
unit_tests/sources/declarative/test_concurrent_declarative_source.py:1080:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:388: in read
yield from self._concurrent_source.read(selected_concurrent_streams)
airbyte_cdk/sources/concurrent_source/concurrent_source.py:131: in read
yield from self._consume_from_queue(
airbyte_cdk/sources/concurrent_source/concurrent_source.py:159: in _consume_from_queue
if queue.empty() and concurrent_stream_processor.is_done():
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <airbyte_cdk.sources.concurrent_source.concurrent_read_processor.ConcurrentReadProcessor object at 0x7fa6330ccaa0>
def is_done(self) -> bool:
"""
This method is called to check if the sync is done.
The sync is done when:
1. There are no more streams generating partitions
2. There are no more streams to read from
3. All partitions for all streams are closed
"""
is_done = all(
[
self._is_stream_done(stream_name)
for stream_name in self._stream_name_to_instance.keys()
]
)
if is_done and self._stream_instances_to_start_partition_generation:
stuck_stream_names = [
s.name for s in self._stream_instances_to_start_partition_generation
]
raise AirbyteTracedException(
message="Partition generation queue is not empty after all streams completed.",
internal_message=f"Streams {stuck_stream_names} remained in the partition generation queue after all streams were marked done.",
failure_type=FailureType.system_error,
)
if is_done and self._active_groups:
raise AirbyteTracedException(
message="Active stream groups are not empty after all streams completed.",
internal_message=f"Groups {dict(self._active_groups)} still active after all streams were marked done.",
failure_type=FailureType.system_error,
)
if is_done and self._exceptions_per_stream_name:
error_message = generate_failed_streams_error_message(self._exceptions_per_stream_name)
self._logger.info(error_message)
# We still raise at least one exception when a stream raises an exception because the platform currently relies
# on a non-zero exit code to determine if a sync attempt has failed. We also raise the exception as a config_error
# type because this combined error isn't actionable, but rather the previously emitted individual errors.
> raise AirbyteTracedException(
message=error_message,
internal_message="Concurrent read failure",
failure_type=FailureType.config_error,
)
E airbyte_cdk.utils.traced_exception.AirbyteTracedException: During the sync, the following streams did not sync successfully: party_members_skills: NameError("name 'RequestsCookieJar' is not defined")
E party_members: NameError("name 'RequestsCookieJar' is not defined")
E party_members: NameError("name 'RequestsCookieJar' is not defined")
E party_members: NameError("name 'RequestsCookieJar' is not defined")
E palaces: NameError("name 'RequestsCookieJar' is not defined")
airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py:416: AirbyteTracedException
Check warning on line 0 in unit_tests.sources.declarative.test_concurrent_declarative_source
github-actions / PyTest Results (Full)
test_read_with_concurrent_and_synchronous_streams_with_sequential_state (unit_tests.sources.declarative.test_concurrent_declarative_source) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
airbyte_cdk.utils.traced_exception.AirbyteTracedException: During the sync, the following streams did not sync successfully: party_members: NameError("name 'RequestsCookieJar' is not defined")
party_members: NameError("name 'RequestsCookieJar' is not defined")
palaces: NameError("name 'RequestsCookieJar' is not defined")
party_members_skills: NameError("name 'RequestsCookieJar' is not defined")
@freezegun.freeze_time(_NOW)
@patch(
"airbyte_cdk.sources.streams.concurrent.state_converters.abstract_stream_state_converter.AbstractStreamStateConverter.__init__",
mocked_init,
)
def test_read_with_concurrent_and_synchronous_streams_with_sequential_state():
"""
Verifies that a ConcurrentDeclarativeSource processes concurrent streams correctly using the incoming
legacy state format
"""
state = [
AirbyteStateMessage(
type=AirbyteStateType.STREAM,
stream=AirbyteStreamState(
stream_descriptor=StreamDescriptor(name="locations", namespace=None),
stream_state=AirbyteStateBlob(updated_at="2024-08-06"),
),
),
AirbyteStateMessage(
type=AirbyteStateType.STREAM,
stream=AirbyteStreamState(
stream_descriptor=StreamDescriptor(name="party_members", namespace=None),
stream_state=AirbyteStateBlob(updated_at="2024-08-21"),
),
),
]
source = ConcurrentDeclarativeSource(
source_config=_MANIFEST, config=_CONFIG, catalog=_CATALOG, state=state
)
party_members_slices_and_responses = _NO_STATE_PARTY_MEMBERS_SLICES_AND_RESPONSES + [
(
{"start": "2024-08-16", "end": "2024-08-30"},
HttpResponse(
json.dumps(
[
{
"id": "nijima",
"first_name": "makoto",
"last_name": "nijima",
"updated_at": "2024-08-10",
}
]
)
),
), # considering lookback window
(
{"start": "2024-08-31", "end": "2024-09-10"},
HttpResponse(
json.dumps(
[
{
"id": "yoshizawa",
"first_name": "sumire",
"last_name": "yoshizawa",
"updated_at": "2024-09-10",
}
]
)
),
),
]
location_slices = [
{"start": "2024-08-01", "end": "2024-08-31"},
{"start": "2024-09-01", "end": "2024-09-10"},
]
with HttpMocker() as http_mocker:
_mock_party_members_requests(http_mocker, party_members_slices_and_responses)
_mock_locations_requests(http_mocker, location_slices)
http_mocker.get(HttpRequest("https://persona.metaverse.com/palaces"), _PALACES_RESPONSE)
_mock_party_members_skills_requests(http_mocker)
> messages = list(
source.read(logger=source.logger, config=_CONFIG, catalog=_CATALOG, state=state)
)
unit_tests/sources/declarative/test_concurrent_declarative_source.py:1209:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:388: in read
yield from self._concurrent_source.read(selected_concurrent_streams)
airbyte_cdk/sources/concurrent_source/concurrent_source.py:131: in read
yield from self._consume_from_queue(
airbyte_cdk/sources/concurrent_source/concurrent_source.py:159: in _consume_from_queue
if queue.empty() and concurrent_stream_processor.is_done():
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <airbyte_cdk.sources.concurrent_source.concurrent_read_processor.ConcurrentReadProcessor object at 0x7fa63306d310>
def is_done(self) -> bool:
"""
This method is called to check if the sync is done.
The sync is done when:
1. There are no more streams generating partitions
2. There are no more streams to read from
3. All partitions for all streams are closed
"""
is_done = all(
[
self._is_stream_done(stream_name)
for stream_name in self._stream_name_to_instance.keys()
]
)
if is_done and self._stream_instances_to_start_partition_generation:
stuck_stream_names = [
s.name for s in self._stream_instances_to_start_partition_generation
]
raise AirbyteTracedException(
message="Partition generation queue is not empty after all streams completed.",
internal_message=f"Streams {stuck_stream_names} remained in the partition generation queue after all streams were marked done.",
failure_type=FailureType.system_error,
)
if is_done and self._active_groups:
raise AirbyteTracedException(
message="Active stream groups are not empty after all streams completed.",
internal_message=f"Groups {dict(self._active_groups)} still active after all streams were marked done.",
failure_type=FailureType.system_error,
)
if is_done and self._exceptions_per_stream_name:
error_message = generate_failed_streams_error_message(self._exceptions_per_stream_name)
self._logger.info(error_message)
# We still raise at least one exception when a stream raises an exception because the platform currently relies
# on a non-zero exit code to determine if a sync attempt has failed. We also raise the exception as a config_error
# type because this combined error isn't actionable, but rather the previously emitted individual errors.
> raise AirbyteTracedException(
message=error_message,
internal_message="Concurrent read failure",
failure_type=FailureType.config_error,
)
E airbyte_cdk.utils.traced_exception.AirbyteTracedException: During the sync, the following streams did not sync successfully: party_members: NameError("name 'RequestsCookieJar' is not defined")
E party_members: NameError("name 'RequestsCookieJar' is not defined")
E palaces: NameError("name 'RequestsCookieJar' is not defined")
E party_members_skills: NameError("name 'RequestsCookieJar' is not defined")
airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py:416: AirbyteTracedException
Check warning on line 0 in unit_tests.sources.declarative.test_concurrent_declarative_source
github-actions / PyTest Results (Full)
test_read_concurrent_skip_streams_not_in_catalog (unit_tests.sources.declarative.test_concurrent_declarative_source) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
airbyte_cdk.utils.traced_exception.AirbyteTracedException: During the sync, the following streams did not sync successfully: palaces: NameError("name 'RequestsCookieJar' is not defined")
@freezegun.freeze_time(_NOW)
@patch(
"airbyte_cdk.sources.streams.concurrent.state_converters.abstract_stream_state_converter.AbstractStreamStateConverter.__init__",
mocked_init,
)
def test_read_concurrent_skip_streams_not_in_catalog():
"""
Verifies that the ConcurrentDeclarativeSource only syncs streams that are specified in the incoming ConfiguredCatalog
"""
with HttpMocker() as http_mocker:
catalog = ConfiguredAirbyteCatalog(
streams=[
ConfiguredAirbyteStream(
stream=AirbyteStream(
name="palaces", json_schema={}, supported_sync_modes=[SyncMode.full_refresh]
),
sync_mode=SyncMode.full_refresh,
destination_sync_mode=DestinationSyncMode.append,
),
ConfiguredAirbyteStream(
stream=AirbyteStream(
name="locations",
json_schema={},
supported_sync_modes=[SyncMode.incremental],
),
sync_mode=SyncMode.incremental,
destination_sync_mode=DestinationSyncMode.append,
),
]
)
source = ConcurrentDeclarativeSource(
source_config=_MANIFEST, config=_CONFIG, catalog=catalog, state=None
)
# locations requests
location_slices = [
{"start": "2024-07-01", "end": "2024-07-31"},
{"start": "2024-08-01", "end": "2024-08-31"},
{"start": "2024-09-01", "end": "2024-09-10"},
]
locations_query_params = list(
map(lambda _slice: _slice | {"m": "active", "i": "1", "g": "country"}, location_slices)
)
_mock_requests(
http_mocker,
"https://persona.metaverse.com/locations",
locations_query_params,
[_LOCATIONS_RESPONSE] * len(location_slices),
)
# palaces requests
http_mocker.get(HttpRequest("https://persona.metaverse.com/palaces"), _PALACES_RESPONSE)
> messages = list(
source.read(logger=source.logger, config=_CONFIG, catalog=catalog, state=[])
)
unit_tests/sources/declarative/test_concurrent_declarative_source.py:1528:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:388: in read
yield from self._concurrent_source.read(selected_concurrent_streams)
airbyte_cdk/sources/concurrent_source/concurrent_source.py:131: in read
yield from self._consume_from_queue(
airbyte_cdk/sources/concurrent_source/concurrent_source.py:159: in _consume_from_queue
if queue.empty() and concurrent_stream_processor.is_done():
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <airbyte_cdk.sources.concurrent_source.concurrent_read_processor.ConcurrentReadProcessor object at 0x7fa633165cd0>
def is_done(self) -> bool:
"""
This method is called to check if the sync is done.
The sync is done when:
1. There are no more streams generating partitions
2. There are no more streams to read from
3. All partitions for all streams are closed
"""
is_done = all(
[
self._is_stream_done(stream_name)
for stream_name in self._stream_name_to_instance.keys()
]
)
if is_done and self._stream_instances_to_start_partition_generation:
stuck_stream_names = [
s.name for s in self._stream_instances_to_start_partition_generation
]
raise AirbyteTracedException(
message="Partition generation queue is not empty after all streams completed.",
internal_message=f"Streams {stuck_stream_names} remained in the partition generation queue after all streams were marked done.",
failure_type=FailureType.system_error,
)
if is_done and self._active_groups:
raise AirbyteTracedException(
message="Active stream groups are not empty after all streams completed.",
internal_message=f"Groups {dict(self._active_groups)} still active after all streams were marked done.",
failure_type=FailureType.system_error,
)
if is_done and self._exceptions_per_stream_name:
error_message = generate_failed_streams_error_message(self._exceptions_per_stream_name)
self._logger.info(error_message)
# We still raise at least one exception when a stream raises an exception because the platform currently relies
# on a non-zero exit code to determine if a sync attempt has failed. We also raise the exception as a config_error
# type because this combined error isn't actionable, but rather the previously emitted individual errors.
> raise AirbyteTracedException(
message=error_message,
internal_message="Concurrent read failure",
failure_type=FailureType.config_error,
)
E airbyte_cdk.utils.traced_exception.AirbyteTracedException: During the sync, the following streams did not sync successfully: palaces: NameError("name 'RequestsCookieJar' is not defined")
airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py:416: AirbyteTracedException
Check warning on line 0 in unit_tests.sources.declarative.test_concurrent_declarative_source
github-actions / PyTest Results (Full)
test_slice_checkpoint[test_with_pagination_and_partition_router-manifest0-pages0-2] (unit_tests.sources.declarative.test_concurrent_declarative_source) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
AssertionError: assert 1 == 2
+ where 1 = len([AirbyteStateMessage(type=<AirbyteStateType.STREAM: 'STREAM'>, stream=AirbyteStreamState(stream_descriptor=StreamDescriptor(name='Rates', namespace=None), stream_state=AirbyteStateBlob()), global_=None, data=None, sourceStats=None, destinationStats=None)])
test_name = 'test_with_pagination_and_partition_router'
manifest = {'check': {'stream_names': ['Rates'], 'type': 'CheckStream'}, 'spec': {'connection_specification': {'$schema': 'http:/...T', 'path': '/exchangerates_data/latest', 'request_body_json': {}, ...}, ...}, ...}], 'type': 'DeclarativeSource', ...}
pages = (<Response [200]>, <Response [200]>, <Response [200]>)
expected_states_qty = 2
@pytest.mark.parametrize(
"test_name, manifest, pages, expected_states_qty",
[
(
"test_with_pagination_and_partition_router",
{
"version": "0.34.2",
"type": "DeclarativeSource",
"check": {"type": "CheckStream", "stream_names": ["Rates"]},
"streams": [
{
"type": "DeclarativeStream",
"name": "Rates",
"primary_key": [],
"schema_loader": {
"type": "InlineSchemaLoader",
"schema": {
"$schema": "http://json-schema.org/schema#",
"properties": {
"ABC": {"type": "number"},
"AED": {"type": "number"},
"partition": {"type": "number"},
},
"type": "object",
},
},
"retriever": {
"type": "SimpleRetriever",
"requester": {
"type": "HttpRequester",
"url_base": "https://api.apilayer.com",
"path": "/exchangerates_data/latest",
"http_method": "GET",
"request_parameters": {},
"request_headers": {},
"request_body_json": {},
"authenticator": {
"type": "ApiKeyAuthenticator",
"header": "apikey",
"api_token": "{{ config['api_key'] }}",
},
},
"partition_router": {
"type": "ListPartitionRouter",
"values": ["0", "1"],
"cursor_field": "partition",
},
"record_selector": {
"type": "RecordSelector",
"extractor": {"type": "DpathExtractor", "field_path": ["rates"]},
},
"paginator": {
"type": "DefaultPaginator",
"page_size": 2,
"page_size_option": {
"inject_into": "request_parameter",
"field_name": "page_size",
},
"page_token_option": {"inject_into": "path", "type": "RequestPath"},
"pagination_strategy": {
"type": "CursorPagination",
"cursor_value": "{{ response._metadata.next }}",
"page_size": 2,
},
},
},
"incremental_sync": {
"type": "DatetimeBasedCursor",
"cursor_datetime_formats": ["%Y-%m-%dT%H:%M:%S.%fZ"],
"datetime_format": "%Y-%m-%dT%H:%M:%S.%fZ",
"cursor_field": "updated_at",
"start_datetime": {
"datetime": "{{ config.get('start_date', '2020-10-16T00:00:00.000Z') }}"
},
},
}
],
"spec": {
"connection_specification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"type": "object",
"required": ["api_key"],
"properties": {
"api_key": {
"type": "string",
"title": "API Key",
"airbyte_secret": True,
},
"start_date": {
"title": "Start Date",
"description": "UTC date and time in the format YYYY-MM-DDTHH:MM:SS.000Z. During incremental sync, any data generated before this date will not be replicated. If left blank, the start date will be set to 2 years before the present date.",
"pattern": "^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$",
"pattern_descriptor": "YYYY-MM-DDTHH:MM:SS.000Z",
"examples": ["2020-11-16T00:00:00.000Z"],
"type": "string",
"format": "date-time",
},
},
"additionalProperties": True,
},
"documentation_url": "https://example.org",
"type": "Spec",
},
},
(
_create_page(
{
"rates": [
{"ABC": 0, "partition": 0, "updated_at": "2020-11-16T00:00:00.000Z"},
{"AED": 1, "partition": 0, "updated_at": "2020-11-16T00:00:00.000Z"},
],
"_metadata": {"next": "next"},
}
),
_create_page(
{
"rates": [
{"USD": 3, "partition": 0, "updated_at": "2020-11-16T00:00:00.000Z"}
],
"_metadata": {},
}
),
_create_page(
{
"rates": [
{"ABC": 2, "partition": 1, "updated_at": "2020-11-16T00:00:00.000Z"}
],
"_metadata": {},
}
),
),
2,
),
],
)
def test_slice_checkpoint(test_name, manifest, pages, expected_states_qty):
_stream_name = "Rates"
with patch.object(SimpleRetriever, "_fetch_next_page", side_effect=pages):
states = [message.state for message in _run_read(manifest, _stream_name) if message.state]
> assert len(states) == expected_states_qty
E AssertionError: assert 1 == 2
E + where 1 = len([AirbyteStateMessage(type=<AirbyteStateType.STREAM: 'STREAM'>, stream=AirbyteStreamState(stream_descriptor=StreamDescriptor(name='Rates', namespace=None), stream_state=AirbyteStateBlob()), global_=None, data=None, sourceStats=None, destinationStats=None)])
unit_tests/sources/declarative/test_concurrent_declarative_source.py:4254: AssertionError
Check warning on line 0 in unit_tests.sources.declarative.test_state_delegating_stream
github-actions / PyTest Results (Full)
test_parent_state_delegating_stream_retention_falls_back_to_full_refresh (unit_tests.sources.declarative.test_state_delegating_stream) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
airbyte_cdk.utils.traced_exception.AirbyteTracedException: During the sync, the following streams did not sync successfully: ChildStream: NameError("name 'RequestsCookieJar' is not defined")
@freezegun.freeze_time("2024-07-15")
def test_parent_state_delegating_stream_retention_falls_back_to_full_refresh():
"""When parent StateDelegatingStream has old cursor in child state, retention triggers full refresh for parent."""
manifest = _create_parent_child_manifest_with_retention_period("P7D")
with HttpMocker() as http_mocker:
http_mocker.get(
HttpRequest(url="https://api.test.com/parents"),
HttpResponse(
body=json.dumps([{"id": 1, "name": "parent_1", "updated_at": "2024-07-14"}])
),
)
http_mocker.get(
HttpRequest(url="https://api.test.com/children/1"),
HttpResponse(
body=json.dumps([{"id": 10, "name": "child_1", "updated_at": "2024-07-14"}])
),
)
state = [
AirbyteStateMessage(
type=AirbyteStateType.STREAM,
stream=AirbyteStreamState(
stream_descriptor=StreamDescriptor(name="ChildStream", namespace=None),
stream_state=AirbyteStateBlob(
use_global_cursor=False,
state={"updated_at": "2024-07-14"},
states=[],
parent_state={"ParentStream": {"updated_at": "2024-06-01"}},
lookback_window=0,
),
),
)
]
source = ConcurrentDeclarativeSource(
source_config=manifest, config=_CONFIG, catalog=None, state=state
)
configured_catalog = create_configured_catalog(source, _CONFIG)
> records = get_records(source, _CONFIG, configured_catalog, state)
unit_tests/sources/declarative/test_state_delegating_stream.py:853:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
unit_tests/sources/declarative/test_state_delegating_stream.py:185: in get_records
return [
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:388: in read
yield from self._concurrent_source.read(selected_concurrent_streams)
airbyte_cdk/sources/concurrent_source/concurrent_source.py:131: in read
yield from self._consume_from_queue(
airbyte_cdk/sources/concurrent_source/concurrent_source.py:159: in _consume_from_queue
if queue.empty() and concurrent_stream_processor.is_done():
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <airbyte_cdk.sources.concurrent_source.concurrent_read_processor.ConcurrentReadProcessor object at 0x7fa630cba3c0>
def is_done(self) -> bool:
"""
This method is called to check if the sync is done.
The sync is done when:
1. There are no more streams generating partitions
2. There are no more streams to read from
3. All partitions for all streams are closed
"""
is_done = all(
[
self._is_stream_done(stream_name)
for stream_name in self._stream_name_to_instance.keys()
]
)
if is_done and self._stream_instances_to_start_partition_generation:
stuck_stream_names = [
s.name for s in self._stream_instances_to_start_partition_generation
]
raise AirbyteTracedException(
message="Partition generation queue is not empty after all streams completed.",
internal_message=f"Streams {stuck_stream_names} remained in the partition generation queue after all streams were marked done.",
failure_type=FailureType.system_error,
)
if is_done and self._active_groups:
raise AirbyteTracedException(
message="Active stream groups are not empty after all streams completed.",
internal_message=f"Groups {dict(self._active_groups)} still active after all streams were marked done.",
failure_type=FailureType.system_error,
)
if is_done and self._exceptions_per_stream_name:
error_message = generate_failed_streams_error_message(self._exceptions_per_stream_name)
self._logger.info(error_message)
# We still raise at least one exception when a stream raises an exception because the platform currently relies
# on a non-zero exit code to determine if a sync attempt has failed. We also raise the exception as a config_error
# type because this combined error isn't actionable, but rather the previously emitted individual errors.
> raise AirbyteTracedException(
message=error_message,
internal_message="Concurrent read failure",
failure_type=FailureType.config_error,
)
E airbyte_cdk.utils.traced_exception.AirbyteTracedException: During the sync, the following streams did not sync successfully: ChildStream: NameError("name 'RequestsCookieJar' is not defined")
airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py:416: AirbyteTracedException
Check warning on line 0 in unit_tests.sources.declarative.test_state_delegating_stream
github-actions / PyTest Results (Full)
test_unconfigured_parent_stream_does_not_emit_state_on_retention_fallback (unit_tests.sources.declarative.test_state_delegating_stream) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
airbyte_cdk.utils.traced_exception.AirbyteTracedException: During the sync, the following streams did not sync successfully: ChildStream: NameError("name 'RequestsCookieJar' is not defined")
@freezegun.freeze_time("2024-07-15")
def test_unconfigured_parent_stream_does_not_emit_state_on_retention_fallback():
"""When a parent StateDelegatingStream has stale cursor state but is NOT in the
configured catalog (only the child is selected), no state message should be emitted
for the parent. Previously this would emit a state message for the parent stream,
causing the destination to crash with 'Stream not found'."""
manifest = _create_parent_child_manifest_with_retention_period("P7D")
with HttpMocker() as http_mocker:
http_mocker.get(
HttpRequest(url="https://api.test.com/parents"),
HttpResponse(
body=json.dumps([{"id": 1, "name": "parent_1", "updated_at": "2024-07-14"}])
),
)
http_mocker.get(
HttpRequest(url="https://api.test.com/children/1"),
HttpResponse(
body=json.dumps([{"id": 10, "name": "child_1", "updated_at": "2024-07-14"}])
),
)
# ParentStream has stale state (older than 7 days) but ParentStream is NOT
# in the configured catalog — only ChildStream is selected.
state = [
AirbyteStateMessage(
type=AirbyteStateType.STREAM,
stream=AirbyteStreamState(
stream_descriptor=StreamDescriptor(name="ParentStream", namespace=None),
stream_state=AirbyteStateBlob(updated_at="2024-06-01"),
),
),
AirbyteStateMessage(
type=AirbyteStateType.STREAM,
stream=AirbyteStreamState(
stream_descriptor=StreamDescriptor(name="ChildStream", namespace=None),
stream_state=AirbyteStateBlob(
use_global_cursor=False,
state={"updated_at": "2024-07-14"},
states=[],
parent_state={"ParentStream": {"updated_at": "2024-06-01"}},
lookback_window=0,
),
),
),
]
source = ConcurrentDeclarativeSource(
source_config=manifest, config=_CONFIG, catalog=None, state=state
)
configured_catalog = create_configured_catalog(source, _CONFIG)
> all_messages = list(
source.read(logger=MagicMock(), config=_CONFIG, catalog=configured_catalog, state=state)
)
unit_tests/sources/declarative/test_state_delegating_stream.py:908:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:388: in read
yield from self._concurrent_source.read(selected_concurrent_streams)
airbyte_cdk/sources/concurrent_source/concurrent_source.py:131: in read
yield from self._consume_from_queue(
airbyte_cdk/sources/concurrent_source/concurrent_source.py:159: in _consume_from_queue
if queue.empty() and concurrent_stream_processor.is_done():
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <airbyte_cdk.sources.concurrent_source.concurrent_read_processor.ConcurrentReadProcessor object at 0x7fa630ad2c30>
def is_done(self) -> bool:
"""
This method is called to check if the sync is done.
The sync is done when:
1. There are no more streams generating partitions
2. There are no more streams to read from
3. All partitions for all streams are closed
"""
is_done = all(
[
self._is_stream_done(stream_name)
for stream_name in self._stream_name_to_instance.keys()
]
)
if is_done and self._stream_instances_to_start_partition_generation:
stuck_stream_names = [
s.name for s in self._stream_instances_to_start_partition_generation
]
raise AirbyteTracedException(
message="Partition generation queue is not empty after all streams completed.",
internal_message=f"Streams {stuck_stream_names} remained in the partition generation queue after all streams were marked done.",
failure_type=FailureType.system_error,
)
if is_done and self._active_groups:
raise AirbyteTracedException(
message="Active stream groups are not empty after all streams completed.",
internal_message=f"Groups {dict(self._active_groups)} still active after all streams were marked done.",
failure_type=FailureType.system_error,
)
if is_done and self._exceptions_per_stream_name:
error_message = generate_failed_streams_error_message(self._exceptions_per_stream_name)
self._logger.info(error_message)
# We still raise at least one exception when a stream raises an exception because the platform currently relies
# on a non-zero exit code to determine if a sync attempt has failed. We also raise the exception as a config_error
# type because this combined error isn't actionable, but rather the previously emitted individual errors.
> raise AirbyteTracedException(
message=error_message,
internal_message="Concurrent read failure",
failure_type=FailureType.config_error,
)
E airbyte_cdk.utils.traced_exception.AirbyteTracedException: During the sync, the following streams did not sync successfully: ChildStream: NameError("name 'RequestsCookieJar' is not defined")
airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py:416: AirbyteTracedException
Check warning on line 0 in unit_tests.sources.declarative.async_job.test_job
github-actions / PyTest Results (Full)
test_given_timer_is_out_when_status_then_return_timed_out (unit_tests.sources.declarative.async_job.test_job) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
AssertionError: assert <AsyncJobStatus.RUNNING: ('RUNNING', False)> == <AsyncJobStatus.TIMED_OUT: ('TIMED_OUT', True)>
+ where <AsyncJobStatus.RUNNING: ('RUNNING', False)> = <bound method AsyncJob.status of AsyncJob(api_job_id=an api job id, job_parameters={}, status=AsyncJobStatus.RUNNING)>()
+ where <bound method AsyncJob.status of AsyncJob(api_job_id=an api job id, job_parameters={}, status=AsyncJobStatus.RUNNING)> = AsyncJob(api_job_id=an api job id, job_parameters={}, status=AsyncJobStatus.RUNNING).status
+ and <AsyncJobStatus.TIMED_OUT: ('TIMED_OUT', True)> = AsyncJobStatus.TIMED_OUT
def test_given_timer_is_out_when_status_then_return_timed_out() -> None:
job = AsyncJob(_AN_API_JOB_ID, _ANY_STREAM_SLICE, _IMMEDIATELY_TIMED_OUT)
time.sleep(0.001)
> assert job.status() == AsyncJobStatus.TIMED_OUT
E AssertionError: assert <AsyncJobStatus.RUNNING: ('RUNNING', False)> == <AsyncJobStatus.TIMED_OUT: ('TIMED_OUT', True)>
E + where <AsyncJobStatus.RUNNING: ('RUNNING', False)> = <bound method AsyncJob.status of AsyncJob(api_job_id=an api job id, job_parameters={}, status=AsyncJobStatus.RUNNING)>()
E + where <bound method AsyncJob.status of AsyncJob(api_job_id=an api job id, job_parameters={}, status=AsyncJobStatus.RUNNING)> = AsyncJob(api_job_id=an api job id, job_parameters={}, status=AsyncJobStatus.RUNNING).status
E + and <AsyncJobStatus.TIMED_OUT: ('TIMED_OUT', True)> = AsyncJobStatus.TIMED_OUT
unit_tests/sources/declarative/async_job/test_job.py:27: AssertionError
Check warning on line 0 in unit_tests.sources.declarative.async_job.test_job
github-actions / PyTest Results (Full)
test_retry_after[retry_after_in_past] (unit_tests.sources.declarative.async_job.test_job) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
assert False == True
+ where False = <bound method AsyncJob.ready_to_retry of AsyncJob(api_job_id=an api job id, job_parameters={}, status=AsyncJobStatus.RUNNING)>()
+ where <bound method AsyncJob.ready_to_retry of AsyncJob(api_job_id=an api job id, job_parameters={}, status=AsyncJobStatus.RUNNING)> = AsyncJob(api_job_id=an api job id, job_parameters={}, status=AsyncJobStatus.RUNNING).ready_to_retry
retry_after_offset = datetime.timedelta(days=-1, seconds=86399)
expected_deferred = True, expected_ready = True
@pytest.mark.parametrize(
"retry_after_offset,expected_deferred,expected_ready",
[
pytest.param(None, False, True, id="no_retry_after_set"),
pytest.param(timedelta(hours=1), True, False, id="retry_after_in_future"),
pytest.param(-timedelta(seconds=1), True, True, id="retry_after_in_past"),
],
)
def test_retry_after(
retry_after_offset: Optional[timedelta], expected_deferred: bool, expected_ready: bool
) -> None:
job = AsyncJob(_AN_API_JOB_ID, _ANY_STREAM_SLICE, _A_VERY_BIG_TIMEOUT)
if retry_after_offset is not None:
job.set_retry_after(datetime.now(tz=timezone.utc) + retry_after_offset)
assert job.retry_deferred() == expected_deferred
> assert job.ready_to_retry() == expected_ready
E assert False == True
E + where False = <bound method AsyncJob.ready_to_retry of AsyncJob(api_job_id=an api job id, job_parameters={}, status=AsyncJobStatus.RUNNING)>()
E + where <bound method AsyncJob.ready_to_retry of AsyncJob(api_job_id=an api job id, job_parameters={}, status=AsyncJobStatus.RUNNING)> = AsyncJob(api_job_id=an api job id, job_parameters={}, status=AsyncJobStatus.RUNNING).ready_to_retry
unit_tests/sources/declarative/async_job/test_job.py:45: AssertionError
github-actions / PyTest Results (Full)
test_given_failed_retry_wait_time_when_job_fails_then_defers_retry (unit_tests.sources.declarative.async_job.test_job_orchestrator.AsyncJobOrchestratorTest) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
AssertionError: Expected 'start' to have been called once. Called 0 times.
self = <Mock name='mock.start' id='140351748529216'>
def assert_called_once(self):
"""assert that the mock was called only once.
"""
if not self.call_count == 1:
msg = ("Expected '%s' to have been called once. Called %s times.%s"
% (self._mock_name or 'mock',
self.call_count,
self._calls_repr()))
> raise AssertionError(msg)
E AssertionError: Expected 'start' to have been called once. Called 0 times.
/usr/lib/python3.12/unittest/mock.py:923: AssertionError
During handling of the above exception, another exception occurred:
self = <unit_tests.sources.declarative.async_job.test_job_orchestrator.AsyncJobOrchestratorTest testMethod=test_given_failed_retry_wait_time_when_job_fails_then_defers_retry>
mock_sleep = <MagicMock name='sleep' id='140351748536368'>
@mock.patch(sleep_mock_target)
def test_given_failed_retry_wait_time_when_job_fails_then_defers_retry(
self, mock_sleep: MagicMock
) -> None:
"""When failed_retry_wait_time_in_seconds is set and a job fails, the retry should
be deferred: first call sets retry_after timestamp and skips, subsequent calls skip
until cooldown elapses, then the job is replaced."""
job_tracker = JobTracker(_NO_JOB_LIMIT)
job = self._an_async_job("deferred-job", _A_STREAM_SLICE)
job_tracker._jobs.add("deferred-job")
partition = AsyncPartition([job], _A_STREAM_SLICE)
orchestrator = AsyncJobOrchestrator(
self._job_repository,
[],
job_tracker,
self._message_repository,
failed_retry_wait_time_in_seconds=1800,
)
job.update_status(AsyncJobStatus.FAILED)
# First call: should set retry_after and NOT replace
orchestrator._replace_failed_jobs(partition)
assert job.retry_deferred()
assert not job.ready_to_retry()
self._job_repository.start.assert_not_called()
# Second call while cooldown hasn't elapsed: should still skip
orchestrator._replace_failed_jobs(partition)
self._job_repository.start.assert_not_called()
# Simulate cooldown elapsed by setting retry_after to the past
job.set_retry_after(datetime.now(tz=timezone.utc) - timedelta(seconds=1))
replacement_job = self._an_async_job("replacement-job", _A_STREAM_SLICE)
self._job_repository.start.return_value = replacement_job
# Third call after cooldown: should replace the job
orchestrator._replace_failed_jobs(partition)
> self._job_repository.start.assert_called_once()
E AssertionError: Expected 'start' to have been called once. Called 0 times.
unit_tests/sources/declarative/async_job/test_job_orchestrator.py:458: AssertionError
github-actions / PyTest Results (Full)
test_given_real_failed_then_cooldown_elapses_then_start_returns_creation_failure_then_no_rearm (unit_tests.sources.declarative.async_job.test_job_orchestrator.AsyncJobOrchestratorTest) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
AssertionError: Expected 'start' to have been called once. Called 0 times.
self = <Mock name='mock.start' id='140351750889376'>
def assert_called_once(self):
"""assert that the mock was called only once.
"""
if not self.call_count == 1:
msg = ("Expected '%s' to have been called once. Called %s times.%s"
% (self._mock_name or 'mock',
self.call_count,
self._calls_repr()))
> raise AssertionError(msg)
E AssertionError: Expected 'start' to have been called once. Called 0 times.
/usr/lib/python3.12/unittest/mock.py:923: AssertionError
During handling of the above exception, another exception occurred:
self = <unit_tests.sources.declarative.async_job.test_job_orchestrator.AsyncJobOrchestratorTest testMethod=test_given_real_failed_then_cooldown_elapses_then_start_returns_creation_failure_then_no_rearm>
mock_sleep = <MagicMock name='sleep' id='140351748948288'>
@mock.patch(sleep_mock_target)
def test_given_real_failed_then_cooldown_elapses_then_start_returns_creation_failure_then_no_rearm(
self, mock_sleep: MagicMock
) -> None:
"""Regression test for the original bug: real FAILED -> arm cooldown -> cooldown
elapses -> _start_job returns a creation-failure job (API still rejects) -> the
replacement must NOT re-arm cooldown; it should be replaced immediately on the
next tick."""
job_tracker = JobTracker(_NO_JOB_LIMIT)
real_job = self._an_async_job("real-job", _A_STREAM_SLICE)
job_tracker._jobs.add("real-job")
partition = AsyncPartition([real_job], _A_STREAM_SLICE)
orchestrator = AsyncJobOrchestrator(
self._job_repository,
[],
job_tracker,
self._message_repository,
failed_retry_wait_time_in_seconds=1800,
)
real_job.update_status(AsyncJobStatus.FAILED)
# Tick 1: arms cooldown on the real FAILED job
orchestrator._replace_failed_jobs(partition)
assert real_job.retry_deferred()
self._job_repository.start.assert_not_called()
# Simulate cooldown elapsed
real_job.set_retry_after(datetime.now(tz=timezone.utc) - timedelta(seconds=1))
# _start_job returns a creation-failure job (API still rejects with 429)
creation_failure_replacement = AsyncJob(
"creation-failure-replacement", _A_STREAM_SLICE, is_creation_failure=True
)
creation_failure_replacement.update_status(AsyncJobStatus.FAILED)
self._job_repository.start.return_value = creation_failure_replacement
job_tracker._jobs.add("creation-failure-replacement")
# Tick 2: cooldown elapsed, replaces with creation-failure job
orchestrator._replace_failed_jobs(partition)
> self._job_repository.start.assert_called_once()
E AssertionError: Expected 'start' to have been called once. Called 0 times.
unit_tests/sources/declarative/async_job/test_job_orchestrator.py:577: AssertionError
Check warning on line 0 in unit_tests.sources.declarative.auth.test_jwt.TestJwtAuthenticator
github-actions / PyTest Results (Full)
test_get_jwt_payload[test_iss-test_sub-test_aud-additional_jwt_payload0-expected0] (unit_tests.sources.declarative.auth.test_jwt.TestJwtAuthenticator) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
AssertionError: assert {'aud': 'test...est_iss', ...} == {'aud': 'test...est_iss', ...}
Omitting 4 identical items, use -vv to show
Differing items:
{'iat': 1640995200} != {'iat': 1782941539}
{'exp': 1640996200} != {'exp': 1782942539}
{'nbf': 1640995200} != {'nbf': 1782941539}
Full diff:
{
'aud': 'test_aud',
- 'exp': 1782942539,
- 'iat': 1782941539,
+ 'exp': 1640996200,
+ 'iat': 1640995200,
'iss': 'test_iss',
- 'nbf': 1782941539,
+ 'nbf': 1640995200,
'sub': 'test_sub',
'test': 'test',
}
self = <unit_tests.sources.declarative.auth.test_jwt.TestJwtAuthenticator object at 0x7fa64029ec30>
iss = 'test_iss', sub = 'test_sub', aud = 'test_aud'
additional_jwt_payload = {'test': 'test'}
expected = {'aud': 'test_aud', 'exp': 1782942539, 'iat': 1782941539, 'iss': 'test_iss', ...}
@pytest.mark.parametrize(
"iss, sub, aud, additional_jwt_payload, expected",
[
(
"test_iss",
"test_sub",
"test_aud",
{"test": "test"},
{"iss": "test_iss", "sub": "test_sub", "aud": "test_aud", "test": "test"},
),
(None, None, None, None, {}),
],
)
def test_get_jwt_payload(self, iss, sub, aud, additional_jwt_payload, expected):
authenticator = JwtAuthenticator(
config={},
parameters={},
algorithm="ALGORITHM",
secret_key="test_key",
token_duration=1000,
iss=iss,
sub=sub,
aud=aud,
additional_jwt_payload=additional_jwt_payload,
)
with freezegun.freeze_time("2022-01-01 00:00:00"):
expected["iat"] = int(datetime.now().timestamp())
expected["exp"] = expected["iat"] + 1000
expected["nbf"] = expected["iat"]
> assert authenticator._get_jwt_payload() == expected
E AssertionError: assert {'aud': 'test...est_iss', ...} == {'aud': 'test...est_iss', ...}
E Omitting 4 identical items, use -vv to show
E Differing items:
E {'iat': 1640995200} != {'iat': 1782941539}
E {'exp': 1640996200} != {'exp': 1782942539}
E {'nbf': 1640995200} != {'nbf': 1782941539}
E Full diff:
E {
E 'aud': 'test_aud',
E - 'exp': 1782942539,
E - 'iat': 1782941539,
E + 'exp': 1640996200,
E + 'iat': 1640995200,
E 'iss': 'test_iss',
E - 'nbf': 1782941539,
E + 'nbf': 1640995200,
E 'sub': 'test_sub',
E 'test': 'test',
E }
unit_tests/sources/declarative/auth/test_jwt.py:104: AssertionError
Check warning on line 0 in unit_tests.sources.declarative.auth.test_jwt.TestJwtAuthenticator
github-actions / PyTest Results (Full)
test_get_jwt_payload[None-None-None-None-expected1] (unit_tests.sources.declarative.auth.test_jwt.TestJwtAuthenticator) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
AssertionError: assert {'exp': 16409...': 1640995200} == {'exp': 17829...': 1782941539}
Differing items:
{'iat': 1640995200} != {'iat': 1782941539}
{'exp': 1640996200} != {'exp': 1782942539}
{'nbf': 1640995200} != {'nbf': 1782941539}
Full diff:
- {'exp': 1782942539, 'iat': 1782941539, 'nbf': 1782941539}
+ {'exp': 1640996200, 'iat': 1640995200, 'nbf': 1640995200}
self = <unit_tests.sources.declarative.auth.test_jwt.TestJwtAuthenticator object at 0x7fa64029d9a0>
iss = None, sub = None, aud = None, additional_jwt_payload = None
expected = {'exp': 1782942539, 'iat': 1782941539, 'nbf': 1782941539}
@pytest.mark.parametrize(
"iss, sub, aud, additional_jwt_payload, expected",
[
(
"test_iss",
"test_sub",
"test_aud",
{"test": "test"},
{"iss": "test_iss", "sub": "test_sub", "aud": "test_aud", "test": "test"},
),
(None, None, None, None, {}),
],
)
def test_get_jwt_payload(self, iss, sub, aud, additional_jwt_payload, expected):
authenticator = JwtAuthenticator(
config={},
parameters={},
algorithm="ALGORITHM",
secret_key="test_key",
token_duration=1000,
iss=iss,
sub=sub,
aud=aud,
additional_jwt_payload=additional_jwt_payload,
)
with freezegun.freeze_time("2022-01-01 00:00:00"):
expected["iat"] = int(datetime.now().timestamp())
expected["exp"] = expected["iat"] + 1000
expected["nbf"] = expected["iat"]
> assert authenticator._get_jwt_payload() == expected
E AssertionError: assert {'exp': 16409...': 1640995200} == {'exp': 17829...': 1782941539}
E Differing items:
E {'iat': 1640995200} != {'iat': 1782941539}
E {'exp': 1640996200} != {'exp': 1782942539}
E {'nbf': 1640995200} != {'nbf': 1782941539}
E Full diff:
E - {'exp': 1782942539, 'iat': 1782941539, 'nbf': 1782941539}
E + {'exp': 1640996200, 'iat': 1640995200, 'nbf': 1640995200}
unit_tests/sources/declarative/auth/test_jwt.py:104: AssertionError
Check warning on line 0 in unit_tests.sources.declarative.checks.test_check_dynamic_stream
github-actions / PyTest Results (Full)
test_check_dynamic_stream[test_stream_unavailable_unhandled_error] (unit_tests.sources.declarative.checks.test_check_dynamic_stream) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
NameError: name 'RequestsCookieJar' is not defined
response_code = 404, available_expectation = <Status.FAILED: 'FAILED'>
use_check_availability = True
expected_messages = ['Not found. The requested resource was not found on the server.']
@pytest.mark.parametrize(
"response_code, available_expectation, use_check_availability, expected_messages",
[
pytest.param(
404,
Status.FAILED,
True,
["Not found. The requested resource was not found on the server."],
id="test_stream_unavailable_unhandled_error",
),
pytest.param(
403,
Status.FAILED,
True,
["Forbidden. You don't have permission to access this resource."],
id="test_stream_unavailable_handled_error",
),
pytest.param(200, Status.SUCCEEDED, True, [], id="test_stream_available"),
pytest.param(200, Status.SUCCEEDED, False, [], id="test_stream_available"),
pytest.param(
401,
Status.FAILED,
True,
["Unauthorized. Please ensure you are authenticated correctly."],
id="test_stream_unauthorized_error",
),
],
)
def test_check_dynamic_stream(
response_code, available_expectation, use_check_availability, expected_messages
):
manifest = deepcopy(_MANIFEST)
with HttpMocker() as http_mocker:
items_request = HttpRequest(url="https://api.test.com/items")
items_response = HttpResponse(
body=json.dumps([{"id": 1, "name": "item_1"}, {"id": 2, "name": "item_2"}])
)
http_mocker.get(items_request, items_response)
item_request = HttpRequest(url="https://api.test.com/items/1")
item_response = HttpResponse(body=json.dumps(expected_messages), status_code=response_code)
item_request_count = 1
http_mocker.get(item_request, item_response)
if not use_check_availability:
manifest["check"]["use_check_availability"] = False
item_request_count = 0 # stream only created and data request not called
source = ConcurrentDeclarativeSource(
source_config=manifest,
config=_CONFIG,
catalog=None,
state=None,
)
> connection_status = source.check(logger, _CONFIG)
unit_tests/sources/declarative/checks/test_check_dynamic_stream.py:164:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:586: in check
check_succeeded, error = connection_checker.check_connection(self, logger, self._config)
airbyte_cdk/sources/declarative/checks/check_dynamic_stream.py:41: in check_connection
streams: List[Union[Stream, AbstractStream]] = source.streams(config=config) # type: ignore # this is a migration step and we expect the declarative CDK to migrate off of ConnectionChecker
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:413: in streams
stream_configs = self._stream_configs(self._source_config) + self.dynamic_streams
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:593: in dynamic_streams
return self._dynamic_stream_configs(
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:668: in _dynamic_stream_configs
for dynamic_stream in components_resolver.resolve_components(
airbyte_cdk/sources/declarative/resolvers/http_components_resolver.py:95: in resolve_components
for components_values in self.retriever.read_records(
airbyte_cdk/sources/declarative/retrievers/simple_retriever.py:460: in read_records
yield from self._read_pages(record_generator, _slice)
airbyte_cdk/sources/declarative/retrievers/simple_retriever.py:398: in _read_pages
response = self._fetch_next_page(stream_slice, next_page_token)
airbyte_cdk/sources/declarative/retrievers/simple_retriever.py:315: in _fetch_next_page
return self.requester.send_request(
airbyte_cdk/sources/declarative/requesters/http_requester.py:458: in send_request
request, response = self._http_client.send_request(
airbyte_cdk/sources/streams/http/http_client.py:611: in send_request
response: requests.Response = self._send_with_retry(
airbyte_cdk/sources/streams/http/http_client.py:292: in _send_with_retry
response = backoff_handler(rate_limit_backoff_handler(user_backoff_handler))(
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/backoff/_sync.py:105: in retry
ret = target(*args, **kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/backoff/_sync.py:105: in retry
ret = target(*args, **kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/backoff/_sync.py:105: in retry
ret = target(*args, **kwargs)
airbyte_cdk/sources/streams/http/http_client.py:348: in _send
response = self._session.send(request, **request_kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/session.py:230: in send
response = self._send_and_cache(request, actions, cached_response, **kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/session.py:258: in _send_and_cache
self.cache.save_response(response, actions.cache_key, actions.expires)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/base.py:100: in save_response
self.responses[cache_key] = cached_response
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/sqlite.py:327: in __setitem__
self._write(key, value)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/sqlite.py:347: in _write
value = self.serialize(value)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/sqlite.py:397: in serialize
value = super().serialize(value)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/base.py:343: in serialize
return self.serializer.dumps(value) if self.serializer else value
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/serializers/pipeline.py:56: in dumps
value = step(value)
/usr/lib/python3.12/functools.py:946: in _method
return method.__get__(obj, cls)(*args, **kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/serializers/cattrs.py:74: in _
response_dict = self.converter.unstructure(value)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/cattrs/converters.py:324: in unstructure
return self._unstructure_func.dispatch(
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/cattrs/dispatch.py:133: in dispatch_without_caching
res = self._function_dispatch.dispatch(typ)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/cattrs/dispatch.py:76: in dispatch
return handler(typ)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/cattrs/converters.py:1266: in gen_unstructure_attrs_fromdict
resolve_types(origin or cl)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/attr/_funcs.py:487: in resolve_types
hints = typing.get_type_hints(cls, **kwargs)
/usr/lib/python3.12/typing.py:2244: in get_type_hints
value = _eval_type(value, base_globals, base_locals)
/usr/lib/python3.12/typing.py:414: in _eval_type
return t._evaluate(globalns, localns, recursive_guard)
/usr/lib/python3.12/typing.py:924: in _evaluate
eval(self.__forward_code__, globalns, localns),
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E NameError: name 'RequestsCookieJar' is not defined
<string>:1: NameError
Check warning on line 0 in unit_tests.sources.declarative.checks.test_check_dynamic_stream
github-actions / PyTest Results (Full)
test_check_dynamic_stream[test_stream_unavailable_handled_error] (unit_tests.sources.declarative.checks.test_check_dynamic_stream) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
NameError: name 'RequestsCookieJar' is not defined
response_code = 403, available_expectation = <Status.FAILED: 'FAILED'>
use_check_availability = True
expected_messages = ["Forbidden. You don't have permission to access this resource."]
@pytest.mark.parametrize(
"response_code, available_expectation, use_check_availability, expected_messages",
[
pytest.param(
404,
Status.FAILED,
True,
["Not found. The requested resource was not found on the server."],
id="test_stream_unavailable_unhandled_error",
),
pytest.param(
403,
Status.FAILED,
True,
["Forbidden. You don't have permission to access this resource."],
id="test_stream_unavailable_handled_error",
),
pytest.param(200, Status.SUCCEEDED, True, [], id="test_stream_available"),
pytest.param(200, Status.SUCCEEDED, False, [], id="test_stream_available"),
pytest.param(
401,
Status.FAILED,
True,
["Unauthorized. Please ensure you are authenticated correctly."],
id="test_stream_unauthorized_error",
),
],
)
def test_check_dynamic_stream(
response_code, available_expectation, use_check_availability, expected_messages
):
manifest = deepcopy(_MANIFEST)
with HttpMocker() as http_mocker:
items_request = HttpRequest(url="https://api.test.com/items")
items_response = HttpResponse(
body=json.dumps([{"id": 1, "name": "item_1"}, {"id": 2, "name": "item_2"}])
)
http_mocker.get(items_request, items_response)
item_request = HttpRequest(url="https://api.test.com/items/1")
item_response = HttpResponse(body=json.dumps(expected_messages), status_code=response_code)
item_request_count = 1
http_mocker.get(item_request, item_response)
if not use_check_availability:
manifest["check"]["use_check_availability"] = False
item_request_count = 0 # stream only created and data request not called
source = ConcurrentDeclarativeSource(
source_config=manifest,
config=_CONFIG,
catalog=None,
state=None,
)
> connection_status = source.check(logger, _CONFIG)
unit_tests/sources/declarative/checks/test_check_dynamic_stream.py:164:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:586: in check
check_succeeded, error = connection_checker.check_connection(self, logger, self._config)
airbyte_cdk/sources/declarative/checks/check_dynamic_stream.py:41: in check_connection
streams: List[Union[Stream, AbstractStream]] = source.streams(config=config) # type: ignore # this is a migration step and we expect the declarative CDK to migrate off of ConnectionChecker
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:413: in streams
stream_configs = self._stream_configs(self._source_config) + self.dynamic_streams
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:593: in dynamic_streams
return self._dynamic_stream_configs(
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:668: in _dynamic_stream_configs
for dynamic_stream in components_resolver.resolve_components(
airbyte_cdk/sources/declarative/resolvers/http_components_resolver.py:95: in resolve_components
for components_values in self.retriever.read_records(
airbyte_cdk/sources/declarative/retrievers/simple_retriever.py:460: in read_records
yield from self._read_pages(record_generator, _slice)
airbyte_cdk/sources/declarative/retrievers/simple_retriever.py:398: in _read_pages
response = self._fetch_next_page(stream_slice, next_page_token)
airbyte_cdk/sources/declarative/retrievers/simple_retriever.py:315: in _fetch_next_page
return self.requester.send_request(
airbyte_cdk/sources/declarative/requesters/http_requester.py:458: in send_request
request, response = self._http_client.send_request(
airbyte_cdk/sources/streams/http/http_client.py:611: in send_request
response: requests.Response = self._send_with_retry(
airbyte_cdk/sources/streams/http/http_client.py:292: in _send_with_retry
response = backoff_handler(rate_limit_backoff_handler(user_backoff_handler))(
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/backoff/_sync.py:105: in retry
ret = target(*args, **kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/backoff/_sync.py:105: in retry
ret = target(*args, **kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/backoff/_sync.py:105: in retry
ret = target(*args, **kwargs)
airbyte_cdk/sources/streams/http/http_client.py:348: in _send
response = self._session.send(request, **request_kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/session.py:230: in send
response = self._send_and_cache(request, actions, cached_response, **kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/session.py:258: in _send_and_cache
self.cache.save_response(response, actions.cache_key, actions.expires)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/base.py:100: in save_response
self.responses[cache_key] = cached_response
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/sqlite.py:327: in __setitem__
self._write(key, value)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/sqlite.py:347: in _write
value = self.serialize(value)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/sqlite.py:397: in serialize
value = super().serialize(value)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/base.py:343: in serialize
return self.serializer.dumps(value) if self.serializer else value
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/serializers/pipeline.py:56: in dumps
value = step(value)
/usr/lib/python3.12/functools.py:946: in _method
return method.__get__(obj, cls)(*args, **kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/serializers/cattrs.py:74: in _
response_dict = self.converter.unstructure(value)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/cattrs/converters.py:324: in unstructure
return self._unstructure_func.dispatch(
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/cattrs/dispatch.py:133: in dispatch_without_caching
res = self._function_dispatch.dispatch(typ)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/cattrs/dispatch.py:76: in dispatch
return handler(typ)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/cattrs/converters.py:1266: in gen_unstructure_attrs_fromdict
resolve_types(origin or cl)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/attr/_funcs.py:487: in resolve_types
hints = typing.get_type_hints(cls, **kwargs)
/usr/lib/python3.12/typing.py:2244: in get_type_hints
value = _eval_type(value, base_globals, base_locals)
/usr/lib/python3.12/typing.py:414: in _eval_type
return t._evaluate(globalns, localns, recursive_guard)
/usr/lib/python3.12/typing.py:924: in _evaluate
eval(self.__forward_code__, globalns, localns),
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E NameError: name 'RequestsCookieJar' is not defined
<string>:1: NameError
Check warning on line 0 in unit_tests.sources.declarative.checks.test_check_dynamic_stream
github-actions / PyTest Results (Full)
test_check_dynamic_stream[test_stream_available0] (unit_tests.sources.declarative.checks.test_check_dynamic_stream) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
NameError: name 'RequestsCookieJar' is not defined
response_code = 200, available_expectation = <Status.SUCCEEDED: 'SUCCEEDED'>
use_check_availability = True, expected_messages = []
@pytest.mark.parametrize(
"response_code, available_expectation, use_check_availability, expected_messages",
[
pytest.param(
404,
Status.FAILED,
True,
["Not found. The requested resource was not found on the server."],
id="test_stream_unavailable_unhandled_error",
),
pytest.param(
403,
Status.FAILED,
True,
["Forbidden. You don't have permission to access this resource."],
id="test_stream_unavailable_handled_error",
),
pytest.param(200, Status.SUCCEEDED, True, [], id="test_stream_available"),
pytest.param(200, Status.SUCCEEDED, False, [], id="test_stream_available"),
pytest.param(
401,
Status.FAILED,
True,
["Unauthorized. Please ensure you are authenticated correctly."],
id="test_stream_unauthorized_error",
),
],
)
def test_check_dynamic_stream(
response_code, available_expectation, use_check_availability, expected_messages
):
manifest = deepcopy(_MANIFEST)
with HttpMocker() as http_mocker:
items_request = HttpRequest(url="https://api.test.com/items")
items_response = HttpResponse(
body=json.dumps([{"id": 1, "name": "item_1"}, {"id": 2, "name": "item_2"}])
)
http_mocker.get(items_request, items_response)
item_request = HttpRequest(url="https://api.test.com/items/1")
item_response = HttpResponse(body=json.dumps(expected_messages), status_code=response_code)
item_request_count = 1
http_mocker.get(item_request, item_response)
if not use_check_availability:
manifest["check"]["use_check_availability"] = False
item_request_count = 0 # stream only created and data request not called
source = ConcurrentDeclarativeSource(
source_config=manifest,
config=_CONFIG,
catalog=None,
state=None,
)
> connection_status = source.check(logger, _CONFIG)
unit_tests/sources/declarative/checks/test_check_dynamic_stream.py:164:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:586: in check
check_succeeded, error = connection_checker.check_connection(self, logger, self._config)
airbyte_cdk/sources/declarative/checks/check_dynamic_stream.py:41: in check_connection
streams: List[Union[Stream, AbstractStream]] = source.streams(config=config) # type: ignore # this is a migration step and we expect the declarative CDK to migrate off of ConnectionChecker
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:413: in streams
stream_configs = self._stream_configs(self._source_config) + self.dynamic_streams
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:593: in dynamic_streams
return self._dynamic_stream_configs(
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:668: in _dynamic_stream_configs
for dynamic_stream in components_resolver.resolve_components(
airbyte_cdk/sources/declarative/resolvers/http_components_resolver.py:95: in resolve_components
for components_values in self.retriever.read_records(
airbyte_cdk/sources/declarative/retrievers/simple_retriever.py:460: in read_records
yield from self._read_pages(record_generator, _slice)
airbyte_cdk/sources/declarative/retrievers/simple_retriever.py:398: in _read_pages
response = self._fetch_next_page(stream_slice, next_page_token)
airbyte_cdk/sources/declarative/retrievers/simple_retriever.py:315: in _fetch_next_page
return self.requester.send_request(
airbyte_cdk/sources/declarative/requesters/http_requester.py:458: in send_request
request, response = self._http_client.send_request(
airbyte_cdk/sources/streams/http/http_client.py:611: in send_request
response: requests.Response = self._send_with_retry(
airbyte_cdk/sources/streams/http/http_client.py:292: in _send_with_retry
response = backoff_handler(rate_limit_backoff_handler(user_backoff_handler))(
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/backoff/_sync.py:105: in retry
ret = target(*args, **kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/backoff/_sync.py:105: in retry
ret = target(*args, **kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/backoff/_sync.py:105: in retry
ret = target(*args, **kwargs)
airbyte_cdk/sources/streams/http/http_client.py:348: in _send
response = self._session.send(request, **request_kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/session.py:230: in send
response = self._send_and_cache(request, actions, cached_response, **kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/session.py:258: in _send_and_cache
self.cache.save_response(response, actions.cache_key, actions.expires)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/base.py:100: in save_response
self.responses[cache_key] = cached_response
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/sqlite.py:327: in __setitem__
self._write(key, value)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/sqlite.py:347: in _write
value = self.serialize(value)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/sqlite.py:397: in serialize
value = super().serialize(value)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/base.py:343: in serialize
return self.serializer.dumps(value) if self.serializer else value
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/serializers/pipeline.py:56: in dumps
value = step(value)
/usr/lib/python3.12/functools.py:946: in _method
return method.__get__(obj, cls)(*args, **kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/serializers/cattrs.py:74: in _
response_dict = self.converter.unstructure(value)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/cattrs/converters.py:324: in unstructure
return self._unstructure_func.dispatch(
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/cattrs/dispatch.py:133: in dispatch_without_caching
res = self._function_dispatch.dispatch(typ)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/cattrs/dispatch.py:76: in dispatch
return handler(typ)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/cattrs/converters.py:1266: in gen_unstructure_attrs_fromdict
resolve_types(origin or cl)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/attr/_funcs.py:487: in resolve_types
hints = typing.get_type_hints(cls, **kwargs)
/usr/lib/python3.12/typing.py:2244: in get_type_hints
value = _eval_type(value, base_globals, base_locals)
/usr/lib/python3.12/typing.py:414: in _eval_type
return t._evaluate(globalns, localns, recursive_guard)
/usr/lib/python3.12/typing.py:924: in _evaluate
eval(self.__forward_code__, globalns, localns),
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E NameError: name 'RequestsCookieJar' is not defined
<string>:1: NameError
Check warning on line 0 in unit_tests.sources.declarative.checks.test_check_dynamic_stream
github-actions / PyTest Results (Full)
test_check_dynamic_stream[test_stream_available1] (unit_tests.sources.declarative.checks.test_check_dynamic_stream) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
NameError: name 'RequestsCookieJar' is not defined
response_code = 200, available_expectation = <Status.SUCCEEDED: 'SUCCEEDED'>
use_check_availability = False, expected_messages = []
@pytest.mark.parametrize(
"response_code, available_expectation, use_check_availability, expected_messages",
[
pytest.param(
404,
Status.FAILED,
True,
["Not found. The requested resource was not found on the server."],
id="test_stream_unavailable_unhandled_error",
),
pytest.param(
403,
Status.FAILED,
True,
["Forbidden. You don't have permission to access this resource."],
id="test_stream_unavailable_handled_error",
),
pytest.param(200, Status.SUCCEEDED, True, [], id="test_stream_available"),
pytest.param(200, Status.SUCCEEDED, False, [], id="test_stream_available"),
pytest.param(
401,
Status.FAILED,
True,
["Unauthorized. Please ensure you are authenticated correctly."],
id="test_stream_unauthorized_error",
),
],
)
def test_check_dynamic_stream(
response_code, available_expectation, use_check_availability, expected_messages
):
manifest = deepcopy(_MANIFEST)
with HttpMocker() as http_mocker:
items_request = HttpRequest(url="https://api.test.com/items")
items_response = HttpResponse(
body=json.dumps([{"id": 1, "name": "item_1"}, {"id": 2, "name": "item_2"}])
)
http_mocker.get(items_request, items_response)
item_request = HttpRequest(url="https://api.test.com/items/1")
item_response = HttpResponse(body=json.dumps(expected_messages), status_code=response_code)
item_request_count = 1
http_mocker.get(item_request, item_response)
if not use_check_availability:
manifest["check"]["use_check_availability"] = False
item_request_count = 0 # stream only created and data request not called
source = ConcurrentDeclarativeSource(
source_config=manifest,
config=_CONFIG,
catalog=None,
state=None,
)
> connection_status = source.check(logger, _CONFIG)
unit_tests/sources/declarative/checks/test_check_dynamic_stream.py:164:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:586: in check
check_succeeded, error = connection_checker.check_connection(self, logger, self._config)
airbyte_cdk/sources/declarative/checks/check_dynamic_stream.py:41: in check_connection
streams: List[Union[Stream, AbstractStream]] = source.streams(config=config) # type: ignore # this is a migration step and we expect the declarative CDK to migrate off of ConnectionChecker
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:413: in streams
stream_configs = self._stream_configs(self._source_config) + self.dynamic_streams
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:593: in dynamic_streams
return self._dynamic_stream_configs(
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:668: in _dynamic_stream_configs
for dynamic_stream in components_resolver.resolve_components(
airbyte_cdk/sources/declarative/resolvers/http_components_resolver.py:95: in resolve_components
for components_values in self.retriever.read_records(
airbyte_cdk/sources/declarative/retrievers/simple_retriever.py:460: in read_records
yield from self._read_pages(record_generator, _slice)
airbyte_cdk/sources/declarative/retrievers/simple_retriever.py:398: in _read_pages
response = self._fetch_next_page(stream_slice, next_page_token)
airbyte_cdk/sources/declarative/retrievers/simple_retriever.py:315: in _fetch_next_page
return self.requester.send_request(
airbyte_cdk/sources/declarative/requesters/http_requester.py:458: in send_request
request, response = self._http_client.send_request(
airbyte_cdk/sources/streams/http/http_client.py:611: in send_request
response: requests.Response = self._send_with_retry(
airbyte_cdk/sources/streams/http/http_client.py:292: in _send_with_retry
response = backoff_handler(rate_limit_backoff_handler(user_backoff_handler))(
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/backoff/_sync.py:105: in retry
ret = target(*args, **kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/backoff/_sync.py:105: in retry
ret = target(*args, **kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/backoff/_sync.py:105: in retry
ret = target(*args, **kwargs)
airbyte_cdk/sources/streams/http/http_client.py:348: in _send
response = self._session.send(request, **request_kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/session.py:230: in send
response = self._send_and_cache(request, actions, cached_response, **kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/session.py:258: in _send_and_cache
self.cache.save_response(response, actions.cache_key, actions.expires)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/base.py:100: in save_response
self.responses[cache_key] = cached_response
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/sqlite.py:327: in __setitem__
self._write(key, value)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/sqlite.py:347: in _write
value = self.serialize(value)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/sqlite.py:397: in serialize
value = super().serialize(value)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/base.py:343: in serialize
return self.serializer.dumps(value) if self.serializer else value
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/serializers/pipeline.py:56: in dumps
value = step(value)
/usr/lib/python3.12/functools.py:946: in _method
return method.__get__(obj, cls)(*args, **kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/serializers/cattrs.py:74: in _
response_dict = self.converter.unstructure(value)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/cattrs/converters.py:324: in unstructure
return self._unstructure_func.dispatch(
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/cattrs/dispatch.py:133: in dispatch_without_caching
res = self._function_dispatch.dispatch(typ)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/cattrs/dispatch.py:76: in dispatch
return handler(typ)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/cattrs/converters.py:1266: in gen_unstructure_attrs_fromdict
resolve_types(origin or cl)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/attr/_funcs.py:487: in resolve_types
hints = typing.get_type_hints(cls, **kwargs)
/usr/lib/python3.12/typing.py:2244: in get_type_hints
value = _eval_type(value, base_globals, base_locals)
/usr/lib/python3.12/typing.py:414: in _eval_type
return t._evaluate(globalns, localns, recursive_guard)
/usr/lib/python3.12/typing.py:924: in _evaluate
eval(self.__forward_code__, globalns, localns),
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E NameError: name 'RequestsCookieJar' is not defined
<string>:1: NameError
Check warning on line 0 in unit_tests.sources.declarative.checks.test_check_dynamic_stream
github-actions / PyTest Results (Full)
test_check_dynamic_stream[test_stream_unauthorized_error] (unit_tests.sources.declarative.checks.test_check_dynamic_stream) failed
build/test-results/pytest-results.xml [took 0s]
Raw output
NameError: name 'RequestsCookieJar' is not defined
response_code = 401, available_expectation = <Status.FAILED: 'FAILED'>
use_check_availability = True
expected_messages = ['Unauthorized. Please ensure you are authenticated correctly.']
@pytest.mark.parametrize(
"response_code, available_expectation, use_check_availability, expected_messages",
[
pytest.param(
404,
Status.FAILED,
True,
["Not found. The requested resource was not found on the server."],
id="test_stream_unavailable_unhandled_error",
),
pytest.param(
403,
Status.FAILED,
True,
["Forbidden. You don't have permission to access this resource."],
id="test_stream_unavailable_handled_error",
),
pytest.param(200, Status.SUCCEEDED, True, [], id="test_stream_available"),
pytest.param(200, Status.SUCCEEDED, False, [], id="test_stream_available"),
pytest.param(
401,
Status.FAILED,
True,
["Unauthorized. Please ensure you are authenticated correctly."],
id="test_stream_unauthorized_error",
),
],
)
def test_check_dynamic_stream(
response_code, available_expectation, use_check_availability, expected_messages
):
manifest = deepcopy(_MANIFEST)
with HttpMocker() as http_mocker:
items_request = HttpRequest(url="https://api.test.com/items")
items_response = HttpResponse(
body=json.dumps([{"id": 1, "name": "item_1"}, {"id": 2, "name": "item_2"}])
)
http_mocker.get(items_request, items_response)
item_request = HttpRequest(url="https://api.test.com/items/1")
item_response = HttpResponse(body=json.dumps(expected_messages), status_code=response_code)
item_request_count = 1
http_mocker.get(item_request, item_response)
if not use_check_availability:
manifest["check"]["use_check_availability"] = False
item_request_count = 0 # stream only created and data request not called
source = ConcurrentDeclarativeSource(
source_config=manifest,
config=_CONFIG,
catalog=None,
state=None,
)
> connection_status = source.check(logger, _CONFIG)
unit_tests/sources/declarative/checks/test_check_dynamic_stream.py:164:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:586: in check
check_succeeded, error = connection_checker.check_connection(self, logger, self._config)
airbyte_cdk/sources/declarative/checks/check_dynamic_stream.py:41: in check_connection
streams: List[Union[Stream, AbstractStream]] = source.streams(config=config) # type: ignore # this is a migration step and we expect the declarative CDK to migrate off of ConnectionChecker
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:413: in streams
stream_configs = self._stream_configs(self._source_config) + self.dynamic_streams
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:593: in dynamic_streams
return self._dynamic_stream_configs(
airbyte_cdk/sources/declarative/concurrent_declarative_source.py:668: in _dynamic_stream_configs
for dynamic_stream in components_resolver.resolve_components(
airbyte_cdk/sources/declarative/resolvers/http_components_resolver.py:95: in resolve_components
for components_values in self.retriever.read_records(
airbyte_cdk/sources/declarative/retrievers/simple_retriever.py:460: in read_records
yield from self._read_pages(record_generator, _slice)
airbyte_cdk/sources/declarative/retrievers/simple_retriever.py:398: in _read_pages
response = self._fetch_next_page(stream_slice, next_page_token)
airbyte_cdk/sources/declarative/retrievers/simple_retriever.py:315: in _fetch_next_page
return self.requester.send_request(
airbyte_cdk/sources/declarative/requesters/http_requester.py:458: in send_request
request, response = self._http_client.send_request(
airbyte_cdk/sources/streams/http/http_client.py:611: in send_request
response: requests.Response = self._send_with_retry(
airbyte_cdk/sources/streams/http/http_client.py:292: in _send_with_retry
response = backoff_handler(rate_limit_backoff_handler(user_backoff_handler))(
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/backoff/_sync.py:105: in retry
ret = target(*args, **kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/backoff/_sync.py:105: in retry
ret = target(*args, **kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/backoff/_sync.py:105: in retry
ret = target(*args, **kwargs)
airbyte_cdk/sources/streams/http/http_client.py:348: in _send
response = self._session.send(request, **request_kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/session.py:230: in send
response = self._send_and_cache(request, actions, cached_response, **kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/session.py:258: in _send_and_cache
self.cache.save_response(response, actions.cache_key, actions.expires)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/base.py:100: in save_response
self.responses[cache_key] = cached_response
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/sqlite.py:327: in __setitem__
self._write(key, value)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/sqlite.py:347: in _write
value = self.serialize(value)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/sqlite.py:397: in serialize
value = super().serialize(value)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/backends/base.py:343: in serialize
return self.serializer.dumps(value) if self.serializer else value
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/serializers/pipeline.py:56: in dumps
value = step(value)
/usr/lib/python3.12/functools.py:946: in _method
return method.__get__(obj, cls)(*args, **kwargs)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/requests_cache/serializers/cattrs.py:74: in _
response_dict = self.converter.unstructure(value)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/cattrs/converters.py:324: in unstructure
return self._unstructure_func.dispatch(
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/cattrs/dispatch.py:133: in dispatch_without_caching
res = self._function_dispatch.dispatch(typ)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/cattrs/dispatch.py:76: in dispatch
return handler(typ)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/cattrs/converters.py:1266: in gen_unstructure_attrs_fromdict
resolve_types(origin or cl)
../../../.cache/pypoetry/virtualenvs/airbyte-cdk-LAWzOSa7-py3.11/lib/python3.12/site-packages/attr/_funcs.py:487: in resolve_types
hints = typing.get_type_hints(cls, **kwargs)
/usr/lib/python3.12/typing.py:2244: in get_type_hints
value = _eval_type(value, base_globals, base_locals)
/usr/lib/python3.12/typing.py:414: in _eval_type
return t._evaluate(globalns, localns, recursive_guard)
/usr/lib/python3.12/typing.py:924: in _evaluate
eval(self.__forward_code__, globalns, localns),
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E NameError: name 'RequestsCookieJar' is not defined
<string>:1: NameError
Loading