Source twilio: implement rolling windows

[Lex-MUTTDATA]

What

Describe what the change is solving
It helps to add screenshots if it affects the frontend.

-Adding lookback setting to have rolling window that allows updating based on the date_updated field
-Add page size setting from UI
-Update to IncrementalMixin (deprecated get_update_state)

How

Describe the solution

carried out the changes on the twilio connector to integrate the functionality of being able to set a lookback (rolling window) in order to be able to bring information with windows of, for example, 24 hours back and be able to update. And the option to set the page size from the UI

Recommended reading order

1. x.java
2. y.python

🚨 User Impact 🚨

Are there any breaking changes? What is the end result perceived by the user? If yes, please merge this PR with the 🚨🚨 emoji so changelog authors can further highlight this if needed.

Pre-merge Checklist

Expand the relevant checklist and delete the others.

New Connector

Community member or Airbyter

• Community member? Grant edit access to maintainers (instructions)
• Secrets in the connector's spec are annotated with airbyte_secret
• Unit & integration tests added and passing. Community members, please provide proof of success locally e.g: screenshot or copy-paste unit, integration, and acceptance test output. To run acceptance tests for a Python connector, follow instructions in the README. For java connectors run ./gradlew :airbyte-integrations:connectors:<name>:integrationTest.
• Code reviews completed
• Documentation updated
  • Connector's README.md
  • Connector's bootstrap.md. See description and examples
  • docs/SUMMARY.md
  • docs/integrations/<source or destination>/<name>.md including changelog. See changelog example
  • docs/integrations/README.md
  • airbyte-integrations/builds.md
• PR name follows PR naming conventions

Airbyter

If this is a community PR, the Airbyte engineer reviewing this PR is responsible for the below items.

• Create a non-forked branch based on this PR and test the below items on it
• Build is successful
• If new credentials are required for use in CI, add them to GSM. Instructions.
• /test connector=connectors/<name> command is passing
• New Connector version released on Dockerhub by running the /publish command described here
• After the connector is published, connector added to connector index as described here
• Seed specs have been re-generated by building the platform and committing the changes to the seed spec files, as described here

Updating a connector

Community member or Airbyter

• Grant edit access to maintainers (instructions)
• Secrets in the connector's spec are annotated with airbyte_secret
• Unit & integration tests added and passing. Community members, please provide proof of success locally e.g: screenshot or copy-paste unit, integration, and acceptance test output. To run acceptance tests for a Python connector, follow instructions in the README. For java connectors run ./gradlew :airbyte-integrations:connectors:<name>:integrationTest.
• Code reviews completed
• Documentation updated
  • Connector's README.md
  • Connector's bootstrap.md. See description and examples
  • Changelog updated in docs/integrations/<source or destination>/<name>.md including changelog. See changelog example
• PR name follows PR naming conventions

Airbyter

If this is a community PR, the Airbyte engineer reviewing this PR is responsible for the below items.

• Create a non-forked branch based on this PR and test the below items on it
• Build is successful
• If new credentials are required for use in CI, add them to GSM. Instructions.
• /test connector=connectors/<name> command is passing
• New Connector version released on Dockerhub and connector version bumped by running the /publish command described here

Connector Generator

• Issue acceptance criteria met
• PR name follows PR naming conventions
• If adding a new generator, add it to the list of scaffold modules being tested
• The generator test modules (all connectors with -scaffold in their name) have been updated with the latest scaffold by running ./gradlew :airbyte-integrations:connector-templates:generator:testScaffoldTemplates then checking in your changes
• Documentation which references the generator is updated as needed

Tests

Unit

Put your unit tests output here.

Integration

Put your integration tests output here.

Acceptance

Put your acceptance tests output here.

[CLAassistant]

All committers have signed the CLA.

[alafanechere]

Thank you @alejandroRosano-muttdata for this contribution! Could you please sign the CLA?
I'm converting this to draft as it's WIP according to your PR title.
I also read that you are making a change to get_updated_state. This method is now deprecated in favor of the usage of IncrementalMixin, do you mind trying to use IncrementalMixin in this PR.
More about IncrementalMixin here

[Lex-MUTTDATA]

Thank you @alejandroRosano-muttdata for this contribution! Could you please sign the CLA? I'm converting this to draft as it's WIP according to your PR title. I also read that you are making a change to get_updated_state. This method is now deprecated in favor of the usage of IncrementalMixin, do you mind trying to use IncrementalMixin in this PR. More about IncrementalMixin here

HI @alafanechere.
I have researched IncrementalMixin and found that this functionality is available from version v0.36.0-alpha and the version used by the environment where I am testing the twilio custom connector is running v0.35.14-alpha, I understand that For it to work, I should upgrade the version of airbyte in my environment, is that right?

[alafanechere]

I should upgrade the version of airbyte in my environment, is that right?

I don't think so. This change was introduced in v0.35.31. But I don't think this means it will not be backward compatible with your current environment as the latest CDK version will be bundled in the connector container and it's agnostic from the Airbyte version you are using. The only obligation is make you PR branch from our master to make sure it's using our latest CDK.

[gvillafanetapia]

@alafanechere ready for review 😁

[alafanechere]

/test connector=connectors/source-twilio

🕑 connectors/source-twilio https://github.com/airbytehq/airbyte/actions/runs/2334185265
❌ connectors/source-twilio https://github.com/airbytehq/airbyte/actions/runs/2334185265
🐛 https://gradle.com/s/figop7y347yly

[alafanechere]

@gvillafanetapia could you please make sure the code is properly linted and formatted by running
./gradlew format and the acceptance tests with ./gradlew :airbyte-integrations:connectors:source-twilio:integrationTest.
The test I launched above because of an error running the flake check: you have a bunch of unused imports.

	 ./source_twilio/streams.py:7:1: F401 'asyncio.log.logger' imported but unused
	 ./source_twilio/streams.py:10:1: F401 'typing.List' imported but unused
	 ./source_twilio/streams.py:19:1: F401 'typing_extensions.Self' imported but unused

[alafanechere]

/test connector=connectors/source-twilio

🕑 connectors/source-twilio https://github.com/airbytehq/airbyte/actions/runs/2352593880
❌ connectors/source-twilio https://github.com/airbytehq/airbyte/actions/runs/2352593880
🐛 https://gradle.com/s/jaj2czzoogepm

[ext-gvillafane]

@alafanechere

having issues with gradle picking up the wrong version of python

$ python -V
Python 3.9.11

$ ./gradlew format
Building all of Airbyte.
/home/bollo/urepos/tools/airbyte-muttdata-fork/airbyte-integrations/connectors
> Task :airbyte-cdk:python:checkPython FAILED
> Task :octavia-cli:checkPython FAILED
> Task :airbyte-commons:compileJava FAILED

FAILURE: Build completed with 3 failures.

1: Task failed with an exception.
-----------
* What went wrong:
Execution failed for task ':airbyte-cdk:python:checkPython'.
> Python (/usr) verion 3.8.10 does not match minimal required version: 3.9

* Try:
> Run with --stacktrace option to get the stack trace.
> Run with --info or --debug option to get more log output.
> Run with --scan to get full insights.
==============================================================================

2: Task failed with an exception.
-----------
* What went wrong:
Execution failed for task ':octavia-cli:checkPython'.
> Python (/usr) verion 3.8.10 does not match minimal required version: 3.9

* Try:
> Run with --stacktrace option to get the stack trace.
> Run with --info or --debug option to get more log output.
> Run with --scan to get full insights.
==============================================================================

3: Task failed with an exception.
-----------
* What went wrong:
Execution failed for task ':airbyte-commons:compileJava'.
> error: invalid source release: 17

* Try:
> Run with --stacktrace option to get the stack trace.
> Run with --info or --debug option to get more log output.
> Run with --scan to get full insights.
==============================================================================

* Get more help at https://help.gradle.org

Deprecated Gradle features were used in this build, making it incompatible with Gradle 8.0.

You can use '--warning-mode all' to show the individual deprecation warnings and determine if they come from your own scripts or plugins.

See https://docs.gradle.org/7.4/userguide/command_line_interface.html#sec:command_line_warnings

BUILD FAILED in 14s
23 actionable tasks: 4 executed, 19 up-to-date

[alafanechere]

@alafanechere

having issues with gradle picking up the wrong version of python

$ python -V
Python 3.9.11

$ ./gradlew format
Building all of Airbyte.
/home/bollo/urepos/tools/airbyte-muttdata-fork/airbyte-integrations/connectors
> Task :airbyte-cdk:python:checkPython FAILED
> Task :octavia-cli:checkPython FAILED
> Task :airbyte-commons:compileJava FAILED

FAILURE: Build completed with 3 failures.

1: Task failed with an exception.
-----------
* What went wrong:
Execution failed for task ':airbyte-cdk:python:checkPython'.
> Python (/usr) verion 3.8.10 does not match minimal required version: 3.9

* Try:
> Run with --stacktrace option to get the stack trace.
> Run with --info or --debug option to get more log output.
> Run with --scan to get full insights.
==============================================================================

2: Task failed with an exception.
-----------
* What went wrong:
Execution failed for task ':octavia-cli:checkPython'.
> Python (/usr) verion 3.8.10 does not match minimal required version: 3.9

* Try:
> Run with --stacktrace option to get the stack trace.
> Run with --info or --debug option to get more log output.
> Run with --scan to get full insights.
==============================================================================

3: Task failed with an exception.
-----------
* What went wrong:
Execution failed for task ':airbyte-commons:compileJava'.
> error: invalid source release: 17

* Try:
> Run with --stacktrace option to get the stack trace.
> Run with --info or --debug option to get more log output.
> Run with --scan to get full insights.
==============================================================================

* Get more help at https://help.gradle.org

Deprecated Gradle features were used in this build, making it incompatible with Gradle 8.0.

You can use '--warning-mode all' to show the individual deprecation warnings and determine if they come from your own scripts or plugins.

See https://docs.gradle.org/7.4/userguide/command_line_interface.html#sec:command_line_warnings

BUILD FAILED in 14s
23 actionable tasks: 4 executed, 19 up-to-date

Did you already work on the Airbyte repo in the past? I think you might need to delete all the old virtualenv that might have been created before we bumped the python version to 3.9. I personally used this command, but be careful this is destructive 😄 : find . -name ".venv" -type d -exec rm -rf "{}" \;

[alafanechere]

Hey @alejandroRosano-muttdata did you try to recreate the virtualenv?
Could you please resolve the conflicts in the branch?
FYI I'll be off until next Monday, so bear with me if there is a bit of a delay reviewing your upcoming changes.

[alafanechere]

/test connector=connectors/source-twilio

🕑 connectors/source-twilio https://github.com/airbytehq/airbyte/actions/runs/2466715352
❌ connectors/source-twilio https://github.com/airbytehq/airbyte/actions/runs/2466715352
🐛 https://gradle.com/s/pz6imqavsbcww

Build Failed

Test summary info:

=========================== short test summary info ============================
FAILED test_core.py::TestConnection::test_check[inputs0] - docker.errors.Cont...
FAILED test_core.py::TestConnection::test_check[inputs1] - docker.errors.Cont...
FAILED test_core.py::TestDiscovery::test_discover[inputs0] - docker.errors.Co...
ERROR test_core.py::TestSpec::test_config_match_spec[inputs0] - docker.errors...
ERROR test_core.py::TestSpec::test_match_expected[inputs0] - json.decoder.JSO...
ERROR test_core.py::TestSpec::test_docker_env[inputs0] - docker.errors.Contai...
ERROR test_core.py::TestSpec::test_oneof_usage[inputs0] - docker.errors.Conta...
ERROR test_core.py::TestSpec::test_defined_refs_exist_in_json_spec_file[inputs0]
ERROR test_core.py::TestSpec::test_oauth_flow_parameters[inputs0] - docker.er...
ERROR test_core.py::TestDiscovery::test_defined_cursors_exist_in_schema[inputs0]
ERROR test_core.py::TestDiscovery::test_defined_refs_exist_in_schema[inputs0]
ERROR test_core.py::TestDiscovery::test_defined_keyword_exist_in_schema[inputs0-allOf]
ERROR test_core.py::TestDiscovery::test_defined_keyword_exist_in_schema[inputs0-not]
ERROR test_core.py::TestDiscovery::test_primary_keys_exist_in_schema[inputs0]
ERROR test_core.py::TestBasicRead::test_read[inputs0] - docker.errors.Contain...
ERROR test_full_refresh.py::TestFullRefresh::test_sequential_reads[inputs0]
ERROR test_incremental.py::TestIncremental::test_two_sequential_reads[inputs0]
ERROR test_incremental.py::TestIncremental::test_read_sequential_slices[inputs0]
ERROR test_incremental.py::TestIncremental::test_state_with_abnormally_large_values[inputs0]
==================== 3 failed, 5 passed, 16 errors in 9.94s ====================

[misteryeo]

A bit confused by this description.

Could we adjust to this be clearer and/or link to some documentation that elaborates possibly with clearer examples?

[alafanechere]

Hey @alejandroRosano-muttdata I tried to run the acceptance tests locally but they are failing due to multiple typos in your code.
I added comments for you to spot these.
Unfortunately, even after fixing these typo locally the tests are failing.
Could you please make the fixes required to make tests pass after running:
./gradlew :airbyte-integrations:connectors:source-twilio:integrationTest

My current tests output:

> Task :airbyte-integrations:connectors:source-twilio:sourceAcceptanceTest
============================= test session starts ==============================
platform linux -- Python 3.9.11, pytest-6.2.5, py-1.11.0, pluggy-1.0.0
rootdir: /test_input
plugins: timeout-1.4.2, sugar-0.9.4
collected 24 items

test_core.py ..................F.                                        [ 83%]
test_full_refresh.py .                                                   [ 87%]
test_incremental.py FF.                                                  [100%]


=================================== FAILURES ===================================
_______________________ TestBasicRead.test_read[inputs0] _______________________

self = <source_acceptance_test.tests.test_core.TestBasicRead object at 0xffff83567970>
connector_config = SecretDict(******)
configured_catalog = ConfiguredAirbyteCatalog(streams=[ConfiguredAirbyteStream(stream=AirbyteStream(name='accounts', json_schema={'properti...l_refresh'>, cursor_field=None, destination_sync_mode=<DestinationSyncMode.overwrite: 'overwrite'>, primary_key=None)])
inputs = BasicReadTestConfig(config_path='secrets/config.json', configured_catalog_path='integration_tests/no_empty_streams_cat...rds=None, validate_schema=True, validate_data_points=False, expect_trace_message_on_failure=True, timeout_seconds=None)
expected_records = []
docker_runner = <source_acceptance_test.utils.connector_runner.ConnectorRunner object at 0xffff835677f0>
detailed_logger = <Logger detailed_logger /test_input/acceptance_tests_logs/test_core.py__TestBasicRead__test_read[inputs0].txt (DEBUG)>

    def test_read(
        self,
        connector_config,
        configured_catalog,
        inputs: BasicReadTestConfig,
        expected_records: List[AirbyteRecordMessage],
        docker_runner: ConnectorRunner,
        detailed_logger,
    ):
>       output = docker_runner.call_read(connector_config, configured_catalog)

/usr/local/lib/python3.9/site-packages/source_acceptance_test/tests/test_core.py:390:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/usr/local/lib/python3.9/site-packages/source_acceptance_test/utils/connector_runner.py:83: in call_read
    output = list(self.run(cmd=cmd, config=config, catalog=catalog, **kwargs))
/usr/local/lib/python3.9/site-packages/source_acceptance_test/utils/connector_runner.py:106: in run
    for line in self.read(container, command=cmd, with_ext=raise_container_error):
/usr/local/lib/python3.9/site-packages/source_acceptance_test/utils/connector_runner.py:119: in read
    for chunk in container.logs(stdout=True, stderr=True, stream=True, follow=True):
/usr/local/lib/python3.9/site-packages/docker/types/daemon.py:32: in __next__
    return next(self._stream)
/usr/local/lib/python3.9/site-packages/docker/api/client.py:386: in _multiplexed_response_stream_helper
    header = response.raw.read(STREAM_HEADER_SIZE_BYTES)
/usr/local/lib/python3.9/site-packages/urllib3/response.py:522: in read
    data = self._fp.read(amt) if not fp_closed else b""
/usr/local/lib/python3.9/http/client.py:463: in read
    n = self.readinto(b)
/usr/local/lib/python3.9/http/client.py:497: in readinto
    return self._readinto_chunked(b)
/usr/local/lib/python3.9/http/client.py:592: in _readinto_chunked
    chunk_left = self._get_chunk_left()
/usr/local/lib/python3.9/http/client.py:560: in _get_chunk_left
    chunk_left = self._read_next_chunk_size()
/usr/local/lib/python3.9/http/client.py:520: in _read_next_chunk_size
    line = self.fp.readline(_MAXLINE + 1)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <socket.SocketIO object at 0xffff8371d340>
b = <memory at 0xffff839eaa00>

    def readinto(self, b):
        """Read up to len(b) bytes into the writable buffer *b* and return
        the number of bytes read.  If the socket is non-blocking and no bytes
        are available, None is returned.

        If *b* is non-empty, a 0 return value indicates that the connection
        was shutdown at the other end.
        """
        self._checkClosed()
        self._checkReadable()
        if self._timeout_occurred:
            raise OSError("cannot read from timed out object")
        while True:
            try:
>               return self._sock.recv_into(b)
E               Failed: Timeout >300.0s

/usr/local/lib/python3.9/socket.py:704: Failed
______________ TestIncremental.test_two_sequential_reads[inputs0] ______________

self = <source_acceptance_test.tests.test_incremental.TestIncremental object at 0xffff825b2760>
inputs = IncrementalConfig(config_path='secrets/config.json', configured_catalog_path='integration_tests/no_empty_streams_no_us...gration_tests/abnormal_state.json', timeout_seconds=None, threshold_days=0, skip_comprehensive_incremental_tests=False)
connector_config = SecretDict(******)
configured_catalog_for_incremental = ConfiguredAirbyteCatalog(streams=[ConfiguredAirbyteStream(stream=AirbyteStream(name='calls', json_schema={'properties'...tal'>, cursor_field=['date_updated'], destination_sync_mode=<DestinationSyncMode.append: 'append'>, primary_key=None)])
cursor_paths = {'alerts': ['date_updated'], 'calls': ['end_time'], 'conferences': ['date_updated'], 'message_media': ['date_created'], ...}
docker_runner = <source_acceptance_test.utils.connector_runner.ConnectorRunner object at 0xffff825b2250>

    def test_two_sequential_reads(
        self,
        inputs: IncrementalConfig,
        connector_config: SecretDict,
        configured_catalog_for_incremental: ConfiguredAirbyteCatalog,
        cursor_paths: dict[str, list[str]],
        docker_runner: ConnectorRunner,
    ):
        threshold_days = getattr(inputs, "threshold_days") or 0
        stream_mapping = {stream.stream.name: stream for stream in configured_catalog_for_incremental.streams}

        output = docker_runner.call_read(connector_config, configured_catalog_for_incremental)
        records_1 = filter_output(output, type_=Type.RECORD)
        states_1 = filter_output(output, type_=Type.STATE)

        assert states_1, "Should produce at least one state"
        assert records_1, "Should produce at least one record"

        latest_state = states_1[-1].state.data
        for record_value, state_value, stream_name in records_with_state(records_1, latest_state, stream_mapping, cursor_paths):
>           assert (
                record_value <= state_value
            ), f"First incremental sync should produce records younger or equal to cursor value from the state. Stream: {stream_name}"
E           AssertionError: First incremental sync should produce records younger or equal to cursor value from the state. Stream: calls
E           assert DateTime(2022, 6, 2, 12, 54, 5, tzinfo=Timezone('UTC')) <= DateTime(2022, 3, 13, 23, 56, 37, tzinfo=Timezone('+00:00'))

/usr/local/lib/python3.9/site-packages/source_acceptance_test/tests/test_incremental.py:133: AssertionError
_____________ TestIncremental.test_read_sequential_slices[inputs0] _____________

self = <source_acceptance_test.tests.test_incremental.TestIncremental object at 0xffff82bbe1f0>
inputs = IncrementalConfig(config_path='secrets/config.json', configured_catalog_path='integration_tests/no_empty_streams_no_us...gration_tests/abnormal_state.json', timeout_seconds=None, threshold_days=0, skip_comprehensive_incremental_tests=False)
connector_config = SecretDict(******)
configured_catalog_for_incremental = ConfiguredAirbyteCatalog(streams=[ConfiguredAirbyteStream(stream=AirbyteStream(name='calls', json_schema={'properties'...tal'>, cursor_field=['date_updated'], destination_sync_mode=<DestinationSyncMode.append: 'append'>, primary_key=None)])
cursor_paths = {'alerts': ['date_updated'], 'calls': ['end_time'], 'conferences': ['date_updated'], 'message_media': ['date_created'], ...}
docker_runner = <source_acceptance_test.utils.connector_runner.ConnectorRunner object at 0xffff82bbea00>

    def test_read_sequential_slices(
        self, inputs: IncrementalConfig, connector_config, configured_catalog_for_incremental, cursor_paths, docker_runner: ConnectorRunner
    ):
        """
        Incremental test that makes calls the read method without a state checkpoint. Then we partition the results by stream and
        slice checkpoints resulting in batches of messages that look like:
        <state message>
        <record message>
        ...
        <record message>

        Using these batches, we then make additional read method calls using the state message and verify the correctness of the
        messages in the response.
        """
        if inputs.skip_comprehensive_incremental_tests:
            pytest.skip("Skipping new incremental test based on acceptance-test-config.yml")
            return

        threshold_days = getattr(inputs, "threshold_days") or 0
        stream_mapping = {stream.stream.name: stream for stream in configured_catalog_for_incremental.streams}

        output = docker_runner.call_read(connector_config, configured_catalog_for_incremental)
        records_1 = filter_output(output, type_=Type.RECORD)
        states_1 = filter_output(output, type_=Type.STATE)

        assert states_1, "Should produce at least one state"
        assert records_1, "Should produce at least one record"

        latest_state = states_1[-1].state.data
        for record_value, state_value, stream_name in records_with_state(records_1, latest_state, stream_mapping, cursor_paths):
>           assert (
                record_value <= state_value
            ), f"First incremental sync should produce records younger or equal to cursor value from the state. Stream: {stream_name}"
E           AssertionError: First incremental sync should produce records younger or equal to cursor value from the state. Stream: calls
E           assert DateTime(2022, 6, 2, 12, 54, 5, tzinfo=Timezone('UTC')) <= DateTime(2022, 3, 13, 23, 56, 37, tzinfo=Timezone('+00:00'))

/usr/local/lib/python3.9/site-packages/source_acceptance_test/tests/test_incremental.py:175: AssertionError
=========================== short test summary info ============================
FAILED test_core.py::TestBasicRead::test_read[inputs0] - Failed: Timeout >300.0s
FAILED test_incremental.py::TestIncremental::test_two_sequential_reads[inputs0]
FAILED test_incremental.py::TestIncremental::test_read_sequential_slices[inputs0]
=================== 3 failed, 21 passed in 593.61s (0:09:53) ===================

[Lex-MUTTDATA]

Hey @alejandroRosano-muttdata I tried to run the acceptance tests locally but they are failing due to multiple typos in your code. I added comments for you to spot these. Unfortunately, even after fixing these typo locally the tests are failing. Could you please make the fixes required to make tests pass after running: ./gradlew :airbyte-integrations:connectors:source-twilio:integrationTest

My current tests output:

> Task :airbyte-integrations:connectors:source-twilio:sourceAcceptanceTest
============================= test session starts ==============================
platform linux -- Python 3.9.11, pytest-6.2.5, py-1.11.0, pluggy-1.0.0
rootdir: /test_input
plugins: timeout-1.4.2, sugar-0.9.4
collected 24 items

test_core.py ..................F.                                        [ 83%]
test_full_refresh.py .                                                   [ 87%]
test_incremental.py FF.                                                  [100%]


=================================== FAILURES ===================================
_______________________ TestBasicRead.test_read[inputs0] _______________________

self = <source_acceptance_test.tests.test_core.TestBasicRead object at 0xffff83567970>
connector_config = SecretDict(******)
configured_catalog = ConfiguredAirbyteCatalog(streams=[ConfiguredAirbyteStream(stream=AirbyteStream(name='accounts', json_schema={'properti...l_refresh'>, cursor_field=None, destination_sync_mode=<DestinationSyncMode.overwrite: 'overwrite'>, primary_key=None)])
inputs = BasicReadTestConfig(config_path='secrets/config.json', configured_catalog_path='integration_tests/no_empty_streams_cat...rds=None, validate_schema=True, validate_data_points=False, expect_trace_message_on_failure=True, timeout_seconds=None)
expected_records = []
docker_runner = <source_acceptance_test.utils.connector_runner.ConnectorRunner object at 0xffff835677f0>
detailed_logger = <Logger detailed_logger /test_input/acceptance_tests_logs/test_core.py__TestBasicRead__test_read[inputs0].txt (DEBUG)>

    def test_read(
        self,
        connector_config,
        configured_catalog,
        inputs: BasicReadTestConfig,
        expected_records: List[AirbyteRecordMessage],
        docker_runner: ConnectorRunner,
        detailed_logger,
    ):
>       output = docker_runner.call_read(connector_config, configured_catalog)

/usr/local/lib/python3.9/site-packages/source_acceptance_test/tests/test_core.py:390:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/usr/local/lib/python3.9/site-packages/source_acceptance_test/utils/connector_runner.py:83: in call_read
    output = list(self.run(cmd=cmd, config=config, catalog=catalog, **kwargs))
/usr/local/lib/python3.9/site-packages/source_acceptance_test/utils/connector_runner.py:106: in run
    for line in self.read(container, command=cmd, with_ext=raise_container_error):
/usr/local/lib/python3.9/site-packages/source_acceptance_test/utils/connector_runner.py:119: in read
    for chunk in container.logs(stdout=True, stderr=True, stream=True, follow=True):
/usr/local/lib/python3.9/site-packages/docker/types/daemon.py:32: in __next__
    return next(self._stream)
/usr/local/lib/python3.9/site-packages/docker/api/client.py:386: in _multiplexed_response_stream_helper
    header = response.raw.read(STREAM_HEADER_SIZE_BYTES)
/usr/local/lib/python3.9/site-packages/urllib3/response.py:522: in read
    data = self._fp.read(amt) if not fp_closed else b""
/usr/local/lib/python3.9/http/client.py:463: in read
    n = self.readinto(b)
/usr/local/lib/python3.9/http/client.py:497: in readinto
    return self._readinto_chunked(b)
/usr/local/lib/python3.9/http/client.py:592: in _readinto_chunked
    chunk_left = self._get_chunk_left()
/usr/local/lib/python3.9/http/client.py:560: in _get_chunk_left
    chunk_left = self._read_next_chunk_size()
/usr/local/lib/python3.9/http/client.py:520: in _read_next_chunk_size
    line = self.fp.readline(_MAXLINE + 1)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <socket.SocketIO object at 0xffff8371d340>
b = <memory at 0xffff839eaa00>

    def readinto(self, b):
        """Read up to len(b) bytes into the writable buffer *b* and return
        the number of bytes read.  If the socket is non-blocking and no bytes
        are available, None is returned.

        If *b* is non-empty, a 0 return value indicates that the connection
        was shutdown at the other end.
        """
        self._checkClosed()
        self._checkReadable()
        if self._timeout_occurred:
            raise OSError("cannot read from timed out object")
        while True:
            try:
>               return self._sock.recv_into(b)
E               Failed: Timeout >300.0s

/usr/local/lib/python3.9/socket.py:704: Failed
______________ TestIncremental.test_two_sequential_reads[inputs0] ______________

self = <source_acceptance_test.tests.test_incremental.TestIncremental object at 0xffff825b2760>
inputs = IncrementalConfig(config_path='secrets/config.json', configured_catalog_path='integration_tests/no_empty_streams_no_us...gration_tests/abnormal_state.json', timeout_seconds=None, threshold_days=0, skip_comprehensive_incremental_tests=False)
connector_config = SecretDict(******)
configured_catalog_for_incremental = ConfiguredAirbyteCatalog(streams=[ConfiguredAirbyteStream(stream=AirbyteStream(name='calls', json_schema={'properties'...tal'>, cursor_field=['date_updated'], destination_sync_mode=<DestinationSyncMode.append: 'append'>, primary_key=None)])
cursor_paths = {'alerts': ['date_updated'], 'calls': ['end_time'], 'conferences': ['date_updated'], 'message_media': ['date_created'], ...}
docker_runner = <source_acceptance_test.utils.connector_runner.ConnectorRunner object at 0xffff825b2250>

    def test_two_sequential_reads(
        self,
        inputs: IncrementalConfig,
        connector_config: SecretDict,
        configured_catalog_for_incremental: ConfiguredAirbyteCatalog,
        cursor_paths: dict[str, list[str]],
        docker_runner: ConnectorRunner,
    ):
        threshold_days = getattr(inputs, "threshold_days") or 0
        stream_mapping = {stream.stream.name: stream for stream in configured_catalog_for_incremental.streams}

        output = docker_runner.call_read(connector_config, configured_catalog_for_incremental)
        records_1 = filter_output(output, type_=Type.RECORD)
        states_1 = filter_output(output, type_=Type.STATE)

        assert states_1, "Should produce at least one state"
        assert records_1, "Should produce at least one record"

        latest_state = states_1[-1].state.data
        for record_value, state_value, stream_name in records_with_state(records_1, latest_state, stream_mapping, cursor_paths):
>           assert (
                record_value <= state_value
            ), f"First incremental sync should produce records younger or equal to cursor value from the state. Stream: {stream_name}"
E           AssertionError: First incremental sync should produce records younger or equal to cursor value from the state. Stream: calls
E           assert DateTime(2022, 6, 2, 12, 54, 5, tzinfo=Timezone('UTC')) <= DateTime(2022, 3, 13, 23, 56, 37, tzinfo=Timezone('+00:00'))

/usr/local/lib/python3.9/site-packages/source_acceptance_test/tests/test_incremental.py:133: AssertionError
_____________ TestIncremental.test_read_sequential_slices[inputs0] _____________

self = <source_acceptance_test.tests.test_incremental.TestIncremental object at 0xffff82bbe1f0>
inputs = IncrementalConfig(config_path='secrets/config.json', configured_catalog_path='integration_tests/no_empty_streams_no_us...gration_tests/abnormal_state.json', timeout_seconds=None, threshold_days=0, skip_comprehensive_incremental_tests=False)
connector_config = SecretDict(******)
configured_catalog_for_incremental = ConfiguredAirbyteCatalog(streams=[ConfiguredAirbyteStream(stream=AirbyteStream(name='calls', json_schema={'properties'...tal'>, cursor_field=['date_updated'], destination_sync_mode=<DestinationSyncMode.append: 'append'>, primary_key=None)])
cursor_paths = {'alerts': ['date_updated'], 'calls': ['end_time'], 'conferences': ['date_updated'], 'message_media': ['date_created'], ...}
docker_runner = <source_acceptance_test.utils.connector_runner.ConnectorRunner object at 0xffff82bbea00>

    def test_read_sequential_slices(
        self, inputs: IncrementalConfig, connector_config, configured_catalog_for_incremental, cursor_paths, docker_runner: ConnectorRunner
    ):
        """
        Incremental test that makes calls the read method without a state checkpoint. Then we partition the results by stream and
        slice checkpoints resulting in batches of messages that look like:
        <state message>
        <record message>
        ...
        <record message>

        Using these batches, we then make additional read method calls using the state message and verify the correctness of the
        messages in the response.
        """
        if inputs.skip_comprehensive_incremental_tests:
            pytest.skip("Skipping new incremental test based on acceptance-test-config.yml")
            return

        threshold_days = getattr(inputs, "threshold_days") or 0
        stream_mapping = {stream.stream.name: stream for stream in configured_catalog_for_incremental.streams}

        output = docker_runner.call_read(connector_config, configured_catalog_for_incremental)
        records_1 = filter_output(output, type_=Type.RECORD)
        states_1 = filter_output(output, type_=Type.STATE)

        assert states_1, "Should produce at least one state"
        assert records_1, "Should produce at least one record"

        latest_state = states_1[-1].state.data
        for record_value, state_value, stream_name in records_with_state(records_1, latest_state, stream_mapping, cursor_paths):
>           assert (
                record_value <= state_value
            ), f"First incremental sync should produce records younger or equal to cursor value from the state. Stream: {stream_name}"
E           AssertionError: First incremental sync should produce records younger or equal to cursor value from the state. Stream: calls
E           assert DateTime(2022, 6, 2, 12, 54, 5, tzinfo=Timezone('UTC')) <= DateTime(2022, 3, 13, 23, 56, 37, tzinfo=Timezone('+00:00'))

/usr/local/lib/python3.9/site-packages/source_acceptance_test/tests/test_incremental.py:175: AssertionError
=========================== short test summary info ============================
FAILED test_core.py::TestBasicRead::test_read[inputs0] - Failed: Timeout >300.0s
FAILED test_incremental.py::TestIncremental::test_two_sequential_reads[inputs0]
FAILED test_incremental.py::TestIncremental::test_read_sequential_slices[inputs0]
=================== 3 failed, 21 passed in 593.61s (0:09:53) ===================

I will be work in resolve this today.

[Lex-MUTTDATA]

Hi @alafanechere ,

In my last commit when i run :
./gradlew :airbyte-integrations:connectors:source-twilio:integrationTest
I get :

latest_state = states_1[-1].state.data
for record_value, state_value, stream_name in records_with_state(records_1, latest_state, stream_mapping, cursor_paths):
assert (
  record_value <= state_value
), f"First incremental sync should produce records younger or equal to cursor value from the state. Stream: {stream_name}"
E           AssertionError: First incremental sync should produce records younger or equal to cursor value from the state. Stream: calls
E           assert DateTime(2022, 6, 13, 13, 44, 25, tzinfo=Timezone('UTC')) <= DateTime(2022, 6, 13, 13, 30, 0, tzinfo=Timezone('+00:00'))

I really don't know where to go, it seems or I understand that it is caused by the lookback_window implementation.
have you ever heard something like that?. I would greatly appreciate your help. Thanks a lot

[alafanechere]

@alejandroRosano-muttdata I'm going to try to debug it locally. I'll keep you updated.

[alafanechere]

@alejandroRosano-muttdata I spent a good amount of time figuring the root cause of the test failure.
The tests did not pass on master too so I had to fix some cursor definition related issue before considering lookback window.
I opened a PR based on your branch here #13896 . I'm going to merge this if the tests passes in the CI

[alafanechere]

Closing in favour of #13896