139 lines
5.9 KiB
Python
139 lines
5.9 KiB
Python
#
|
|
# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
|
|
#
|
|
|
|
import json
|
|
import time
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
from typing import Mapping
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from airbyte_cdk import AirbyteLogger
|
|
from airbyte_cdk.models import SyncMode
|
|
from requests.exceptions import HTTPError
|
|
from source_intercom.source import Companies, ConversationParts, SourceIntercom, VersionApiAuthenticator
|
|
|
|
LOGGER = AirbyteLogger()
|
|
# from unittest.mock import Mock
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def stream_attributes() -> Mapping[str, str]:
|
|
filename = HERE.parent / "secrets/config.json"
|
|
with open(filename) as json_file:
|
|
return json.load(json_file)
|
|
|
|
|
|
@pytest.mark.skip(reason="need to refresh this test, it is very slow")
|
|
@pytest.mark.parametrize(
|
|
"version,not_supported_streams,custom_companies_data_field",
|
|
(
|
|
(1.0, ["company_segments", "company_attributes", "contact_attributes"], "companies"),
|
|
(1.1, ["company_segments", "company_attributes", "contact_attributes"], "companies"),
|
|
(1.2, ["company_segments", "company_attributes", "contact_attributes"], "companies"),
|
|
(1.3, ["company_segments", "company_attributes", "contact_attributes"], "companies"),
|
|
(1.4, ["company_segments"], "companies"),
|
|
(2.0, [], "data"),
|
|
(2.1, [], "data"),
|
|
(2.2, [], "data"),
|
|
(2.3, [], "data"),
|
|
),
|
|
)
|
|
def test_supported_versions(stream_attributes, version, not_supported_streams, custom_companies_data_field):
|
|
class CustomVersionApiAuthenticator(VersionApiAuthenticator):
|
|
relevant_supported_version = str(version)
|
|
|
|
authenticator = CustomVersionApiAuthenticator(token=stream_attributes["access_token"])
|
|
for stream in SourceIntercom().streams(deepcopy(stream_attributes)):
|
|
stream._authenticator = authenticator
|
|
if stream.name == "companies":
|
|
stream.data_fields = [custom_companies_data_field]
|
|
elif hasattr(stream, "parent_stream_class") and stream.parent_stream_class == Companies:
|
|
stream.parent_stream_class.data_fields = [custom_companies_data_field]
|
|
|
|
if stream.name in not_supported_streams:
|
|
LOGGER.info(f"version {version} shouldn't be supported the stream '{stream.name}'")
|
|
with pytest.raises(HTTPError) as err:
|
|
for slice in stream.stream_slices(sync_mode=SyncMode.full_refresh):
|
|
next(stream.read_records(sync_mode=None, stream_slice=slice), None)
|
|
break
|
|
# example of response errors:
|
|
# {"type": "error.list", "request_id": "000hjqhpf95ef3b8f8v0",
|
|
# "errors": [{"code": "intercom_version_invalid", "message": "The requested version could not be found"}]}
|
|
assert len(err.value.response.json()["errors"]) > 0
|
|
err_data = err.value.response.json()["errors"][0]
|
|
LOGGER.info(f"version {version} doesn't support the stream '{stream.name}', error: {err_data}")
|
|
else:
|
|
LOGGER.info(f"version {version} should be supported the stream '{stream.name}'")
|
|
for slice in stream.stream_slices(sync_mode=SyncMode.full_refresh):
|
|
records = stream.read_records(sync_mode=None, stream_slice=slice)
|
|
if stream.name == "companies":
|
|
# need to read all records for scroll resetting
|
|
list(records)
|
|
else:
|
|
next(records, None)
|
|
|
|
|
|
def test_companies_scroll(stream_attributes):
|
|
authenticator = VersionApiAuthenticator(token=stream_attributes["access_token"])
|
|
stream1 = Companies(authenticator=authenticator)
|
|
stream2 = Companies(authenticator=authenticator)
|
|
stream3 = Companies(authenticator=authenticator)
|
|
|
|
# read the first stream and stop
|
|
for slice in stream1.stream_slices(sync_mode=SyncMode.full_refresh):
|
|
next(stream1.read_records(sync_mode=SyncMode.full_refresh, stream_slice=slice), None)
|
|
break
|
|
|
|
start_time = time.time()
|
|
# read all records
|
|
records = []
|
|
for slice in stream2.stream_slices(sync_mode=SyncMode.full_refresh):
|
|
records += list(stream2.read_records(sync_mode=SyncMode.full_refresh, stream_slice=slice))
|
|
assert len(records) == 3
|
|
assert (time.time() - start_time) > 60.0
|
|
|
|
start_time = time.time()
|
|
# read all records
|
|
records = []
|
|
for slice in stream3.stream_slices(sync_mode=SyncMode.full_refresh):
|
|
records += list(stream3.read_records(sync_mode=SyncMode.full_refresh, stream_slice=slice))
|
|
assert len(records) == 3
|
|
assert (time.time() - start_time) < 5.0
|
|
|
|
|
|
@patch("source_intercom.source.Companies.can_use_scroll", lambda *args: False)
|
|
def test_switch_to_standard_endpoint(stream_attributes):
|
|
authenticator = VersionApiAuthenticator(token=stream_attributes["access_token"])
|
|
stream1 = Companies(authenticator=authenticator)
|
|
stream2 = Companies(authenticator=authenticator)
|
|
stream3 = ConversationParts(authenticator=authenticator)
|
|
|
|
# read the first stream and stop
|
|
for slice in stream1.stream_slices(sync_mode=SyncMode.full_refresh):
|
|
next(stream1.read_records(sync_mode=SyncMode.full_refresh, stream_slice=slice), None)
|
|
break
|
|
|
|
start_time = time.time()
|
|
# read all records
|
|
records = []
|
|
assert stream2._endpoint_type == Companies.EndpointType.scroll
|
|
for slice in stream2.stream_slices(sync_mode=SyncMode.full_refresh):
|
|
records += list(stream2.read_records(sync_mode=SyncMode, stream_slice=slice))
|
|
assert stream2._endpoint_type == Companies.EndpointType.standard
|
|
assert stream2._total_count == 3
|
|
assert len(records) == 3
|
|
assert (time.time() - start_time) < 5.0
|
|
|
|
start_time = time.time()
|
|
# read all children records
|
|
records = []
|
|
for slice in stream3.stream_slices(sync_mode=SyncMode.full_refresh):
|
|
records += list(stream3.read_records(sync_mode=SyncMode, stream_slice=slice))
|
|
assert len(records) == 12
|
|
assert (time.time() - start_time) < 5.0
|