Source Google Ads: Improve unit and integration tests (#12651)

* #12650 source Googel ads: tests

* #12650 source google ads: add changelog item

* #12650 source google ads: add comments to tests

* auto-bump connector version

Co-authored-by: Octavia Squidington III <[email protected]>

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -311,7 +311,7 @@
 - name: Google Ads
   sourceDefinitionId: 253487c0-2246-43ba-a21f-5116b20a2c50
   dockerRepository: airbyte/source-google-ads
-  dockerImageTag: 0.1.36
+  dockerImageTag: 0.1.37
   documentationUrl: https://docs.airbyte.io/integrations/sources/google-ads
   icon: google-adwords.svg
   sourceType: api

airbyte-config/init/src/main/resources/seed/source_specs.yaml

@@ -2688,7 +2688,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-google-ads:0.1.36"
+- dockerImage: "airbyte/source-google-ads:0.1.37"
   spec:
     documentationUrl: "https://docs.airbyte.com/integrations/sources/google-ads"
     connectionSpecification:

airbyte-integrations/connectors/source-google-ads/Dockerfile

@@ -6,12 +6,12 @@ RUN apt-get update && apt-get install -y bash && rm -rf /var/lib/apt/lists/*
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 
 WORKDIR /airbyte/integration_code
-COPY source_google_ads ./source_google_ads
-COPY main.py ./
 COPY setup.py ./
 RUN pip install .
+COPY source_google_ads ./source_google_ads
+COPY main.py ./
 
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.1.36
+LABEL io.airbyte.version=0.1.37
 LABEL io.airbyte.name=airbyte/source-google-ads

airbyte-integrations/connectors/source-google-ads/acceptance-test-config.yml

@@ -28,7 +28,9 @@ tests:
       configured_catalog_path: "integration_tests/configured_catalog_protobuf_msg.json"
       # expect_records:
       #   path: "integration_tests/expected_records_msg.txt"
-  # TODO incremental test is disabled because records output from the report streams can be up to 14 days older than the input state
+  # These tests are disabled because of the issues https://github.com/airbytehq/airbyte/issues/12665
+  # and https://github.com/airbytehq/airbyte/issues/12467. Instead, custom integration tests are implemented.
+  # As soon as the above issues are resolved, standard SATs can be enabled and custom tests removed.
   #  incremental:
   #    - config_path: "secrets/config.json"
   #      configured_catalog_path: "integration_tests/configured_catalog.json"

airbyte-integrations/connectors/source-google-ads/integration_tests/abnormal_state.json

@@ -1,5 +1,5 @@
 {
   "ad_group_ad_report": {
-    "segments.date": "2021-06-07"
+    "segments.date": "2221-06-07"
   }
 }

airbyte-integrations/connectors/source-google-ads/integration_tests/test_incremental.py

@@ -3,93 +3,88 @@
 #
 
 import pendulum
+import pytest
 from airbyte_cdk.logger import AirbyteLogger
 from airbyte_cdk.models import ConfiguredAirbyteCatalog, Type
 from source_google_ads.source import SourceGoogleAds
 
-SAMPLE_CATALOG = {
-    "streams": [
-        {
-            "stream": {
-                "name": "ad_group_ad_report",
-                "json_schema": {
-                    "type": "object",
-                    "title": "Ad Group Ad Report",
-                    "description": "An ad group ad.",
-                    "properties": {
-                        "accent_color": {
-                            "description": "AccentColor",
-                            "type": ["null", "string"],
-                            "field": "ad_group_ad.ad.legacy_responsive_display_ad.accent_color",
-                        },
-                        "account_currency_code": {
-                            "description": "AccountCurrencyCode",
-                            "type": ["null", "string"],
-                            "field": "customer.currency_code",
-                        },
-                        "account_descriptive_name": {
-                            "description": "AccountDescriptiveName",
-                            "type": ["null", "string"],
-                            "field": "customer.descriptive_name",
-                        },
-                        "segments.date": {"description": "Date", "type": ["null", "string"], "field": "segments.date"},
-                    },
+
+@pytest.fixture
+def configured_catalog():
+    return {
+        "streams": [
+            {
+                "stream": {
+                    "name": "ad_group_ad_report",
+                    "json_schema": {},
+                    "supported_sync_modes": ["full_refresh", "incremental"],
+                    "source_defined_cursor": True,
+                    "default_cursor_field": ["segments.date"],
                 },
-                "supported_sync_modes": ["full_refresh", "incremental"],
-                "source_defined_cursor": True,
-                "default_cursor_field": ["segments.date"],
-            },
-            "sync_mode": "incremental",
-            "destination_sync_mode": "overwrite",
-            "cursor_field": ["segments.date"],
-        }
-    ]
-}
+                "sync_mode": "incremental",
+                "destination_sync_mode": "overwrite",
+                "cursor_field": ["segments.date"],
+            }
+        ]
+    }
+
+
+GAP_DAYS = 14
+
 
+def test_incremental_sync(config, configured_catalog):
+    today = pendulum.now().date()
+    start_date = today.subtract(months=1)
+    config["start_date"] = start_date.to_date_string()
 
-def test_incremental_sync(config):
     google_ads_client = SourceGoogleAds()
-    state = "2021-05-24"
-    records = google_ads_client.read(
-        AirbyteLogger(), config, ConfiguredAirbyteCatalog.parse_obj(SAMPLE_CATALOG), {"ad_group_ad_report": {"segments.date": state}}
+    records = list(google_ads_client.read(AirbyteLogger(), config, ConfiguredAirbyteCatalog.parse_obj(configured_catalog)))
+    latest_state = None
+    for record in records[::-1]:
+        if record and record.type == Type.STATE:
+            latest_state = record.state.data["ad_group_ad_report"][config["customer_id"]]["segments.date"]
+            break
+
+    for message in records:
+        if not message or message.type != Type.RECORD:
+            continue
+        cursor_value = message.record.data["segments.date"]
+        assert cursor_value <= latest_state
+        assert cursor_value >= start_date.subtract(days=GAP_DAYS).to_date_string()
+
+    #  next sync
+    records = list(
+        google_ads_client.read(
+            AirbyteLogger(),
+            config,
+            ConfiguredAirbyteCatalog.parse_obj(configured_catalog),
+            {"ad_group_ad_report": {"segments.date": latest_state}},
+        )
     )
-    current_state = pendulum.parse(state).subtract(days=14).to_date_string()
 
     for record in records:
-        if record and record.type == Type.STATE:
-            print(record)
-            temp_state = record.state.data["ad_group_ad_report"]
-            current_state = (
-                temp_state[config["customer_id"]]["segments.date"] if temp_state.get(config["customer_id"]) else temp_state["segments.date"]
-            )
-        if record and record.type == Type.RECORD:
-            assert record.record.data["segments.date"] >= current_state
+        if record.type == Type.RECORD:
+            assert record.record.data["segments.date"] >= pendulum.parse(latest_state).subtract(days=GAP_DAYS).to_date_string()
+        if record.type == Type.STATE:
+            assert record.state.data["ad_group_ad_report"][config["customer_id"]]["segments.date"] >= latest_state
+
 
-    # Next sync
-    state = "2021-06-04"
+def test_abnormally_large_state(config, configured_catalog):
+    google_ads_client = SourceGoogleAds()
     records = google_ads_client.read(
-        AirbyteLogger(), config, ConfiguredAirbyteCatalog.parse_obj(SAMPLE_CATALOG), {"ad_group_ad_report": {"segments.date": state}}
+        AirbyteLogger(),
+        config,
+        ConfiguredAirbyteCatalog.parse_obj(configured_catalog),
+        {"ad_group_ad_report": {"segments.date": "2222-06-04"}},
     )
-    current_state = pendulum.parse(state).subtract(days=14).to_date_string()
 
+    no_data_records = True
+    state_records = False
     for record in records:
         if record and record.type == Type.STATE:
-            current_state = record.state.data["ad_group_ad_report"][config["customer_id"]]["segments.date"]
+            state_records = True
         if record and record.type == Type.RECORD:
-            assert record.record.data["segments.date"] >= current_state
-
-    # # Abnormal state
-    # This part of the test is broken need to understand what is causing this.
-    # state = "2029-06-04"
-    # records = google_ads_client.read(
-    #     AirbyteLogger(), config, ConfiguredAirbyteCatalog.parse_obj(SAMPLE_CATALOG), {"ad_group_ad_report": {"segments.date": state}}
-    # )
-
-    # no_records = True
-    # for record in records:
-    #     if record and record.type == Type.STATE:
-    #         assert record.state.data["ad_group_ad_report"]["segments.date"] == state
-    #     if record and record.type == Type.RECORD:
-    #         no_records = False
+            no_data_records = False
 
-    # assert no_records
+    assert no_data_records
+    assert state_records

airbyte-integrations/connectors/source-google-ads/setup.py

@@ -7,7 +7,7 @@
 
 MAIN_REQUIREMENTS = ["airbyte-cdk~=0.1", "google-ads==14.1.0", "pendulum"]
 
-TEST_REQUIREMENTS = ["pytest~=6.1", "pytest-mock", "freezegun"]
+TEST_REQUIREMENTS = ["pytest~=6.1", "pytest-mock", "freezegun", "requests-mock"]
 
 setup(
     name="source_google_ads",

airbyte-integrations/connectors/source-google-ads/source_google_ads/custom_query_stream.py

@@ -34,7 +34,7 @@ def get_query(self, stream_slice: Mapping[str, Any] = None) -> str:
         return self.insert_segments_date_expr(self.user_defined_query, start_date, end_date)
 
     # IncrementalGoogleAdsStream uses get_json_schema a lot while parsing
-    # responses, caching plaing crucial role for performance here.
+    # responses, caching playing crucial role for performance here.
     @lru_cache()
     def get_json_schema(self) -> Dict[str, Any]:
         """

airbyte-integrations/connectors/source-google-ads/source_google_ads/streams.py

@@ -48,7 +48,7 @@ def chunk_date_range(
     days_of_data_storage: int = None,
     range_days: int = None,
     time_zone=None,
-) -> Iterable[Mapping[str, any]]:
+) -> Iterable[Optional[Mapping[str, any]]]:
     """
     Passing optional parameter end_date for testing
     Returns a list of the beginning and ending timestamps of each `range_days` between the start date and now.
@@ -64,7 +64,7 @@ def chunk_date_range(
 
     # As in to return some state when state in abnormal
     if start_date > end_date:
-        start_date = end_date
+        return [None]
 
     # applying conversion window
     start_date = start_date.subtract(days=conversion_window)
@@ -99,7 +99,9 @@ def stream_slices(self, stream_state: Mapping[str, Any] = None, **kwargs) -> Ite
             self._customer_id = customer_id
             yield {}
 
-    def read_records(self, sync_mode, stream_slice: Mapping[str, Any] = None, **kwargs) -> Iterable[Mapping[str, Any]]:
+    def read_records(self, sync_mode, stream_slice: Optional[Mapping[str, Any]] = None, **kwargs) -> Iterable[Mapping[str, Any]]:
+        if not stream_slice:
+            return []
         account_responses = self.google_ads_client.send_request(self.get_query(stream_slice), customer_id=self._customer_id)
         for response in account_responses:
             yield from self.parse_response(response)

airbyte-integrations/connectors/source-google-ads/unit_tests/conftest.py

@@ -11,3 +11,11 @@
 def config_fixture():
     with open("secrets/config.json", "r") as config_file:
         return json.load(config_file)
+
+
+@pytest.fixture(autouse=True)
+def mock_oauth_call(requests_mock):
+    yield requests_mock.post(
+        "https://accounts.google.com/o/oauth2/token",
+        json={"access_token": "access_token", "refresh_token": "refresh_token", "expires_in": 0},
+    )