diff --git a/airbyte/sources/base.py b/airbyte/sources/base.py index dbd90a1ec..07096eec6 100644 --- a/airbyte/sources/base.py +++ b/airbyte/sources/base.py @@ -455,14 +455,14 @@ def _get_sync_mode(stream: AirbyteStream) -> SyncMode: # Use getattr to handle mocks or streams without supported_sync_modes attribute supported_modes = getattr(stream, "supported_sync_modes", None) - if force_full_refresh: - # When force_full_refresh is True, prefer full_refresh if supported - if supported_modes and SyncMode.full_refresh in supported_modes: - return SyncMode.full_refresh - # Fall back to incremental if full_refresh is not supported - return SyncMode.incremental - - # Default behavior: preserve previous semantics (always incremental) + if ( + supported_modes + and SyncMode.full_refresh in supported_modes + and (force_full_refresh or SyncMode.incremental not in supported_modes) + ): + return SyncMode.full_refresh + + # Prefer incremental unless full_refresh is required or explicitly requested. return SyncMode.incremental return ConfiguredAirbyteCatalog( diff --git a/tests/unit_tests/sources/test_source_sync_modes.py b/tests/unit_tests/sources/test_source_sync_modes.py new file mode 100644 index 000000000..00deb2832 --- /dev/null +++ b/tests/unit_tests/sources/test_source_sync_modes.py @@ -0,0 +1,65 @@ +# Copyright (c) 2023 Airbyte, Inc., all rights reserved. +"""Catalog sync modes must respect the stream's advertised capabilities.""" + +from unittest.mock import Mock, patch + +import pytest +from airbyte.sources.base import Source +from airbyte_protocol.models import AirbyteCatalog, AirbyteStream, SyncMode + + +@pytest.mark.parametrize( + "supported_modes,force_full_refresh,expected_mode", + [ + pytest.param( + [SyncMode.full_refresh], False, SyncMode.full_refresh, id="full-only" + ), + pytest.param( + [SyncMode.incremental], False, SyncMode.incremental, id="incremental-only" + ), + pytest.param( + [SyncMode.full_refresh, SyncMode.incremental], + False, + SyncMode.incremental, + id="prefer-incremental", + ), + pytest.param( + [SyncMode.full_refresh], True, SyncMode.full_refresh, id="force-full-only" + ), + pytest.param( + [SyncMode.incremental], + True, + SyncMode.incremental, + id="force-incremental-only", + ), + pytest.param( + [SyncMode.full_refresh, SyncMode.incremental], + True, + SyncMode.full_refresh, + id="force-prefer-full", + ), + ], +) +def test_configured_catalog_respects_supported_sync_modes( + supported_modes: list[SyncMode], + force_full_refresh: bool, + expected_mode: SyncMode, +) -> None: + """A full-refresh-only stream must never be configured as incremental.""" + discovered_catalog = AirbyteCatalog( + streams=[ + AirbyteStream( + name="records", + json_schema={"type": "object", "properties": {}}, + supported_sync_modes=supported_modes, + ) + ] + ) + source = Source(executor=Mock(), name="source-test") + + with patch.object(Source, "_discover", return_value=discovered_catalog): + configured_catalog = source.get_configured_catalog( + force_full_refresh=force_full_refresh + ) + + assert configured_catalog.streams[0].sync_mode == expected_mode