Source Google Analytics Data API: fix errors; improve acceptance tests; test coverage 90%

Author: roman-yermilov-gl

airbyte-integrations/connectors/source-google-analytics-data-api/Dockerfile

@@ -28,5 +28,5 @@ COPY source_google_analytics_data_api ./source_google_analytics_data_api
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.0.3
+LABEL io.airbyte.version=0.0.4
 LABEL io.airbyte.name=airbyte/source-google-analytics-data-api

airbyte-integrations/connectors/source-google-analytics-data-api/acceptance-test-config.yml

@@ -1,31 +1,41 @@
 # See [Source Acceptance Tests](https://docs.airbyte.com/connector-development/testing-connectors/source-acceptance-tests-reference)
 # for more information about how to configure these tests
 connector_image: airbyte/source-google-analytics-data-api:dev
-tests:
+acceptance_tests:
   spec:
+    tests:
     - spec_path: "source_google_analytics_data_api/spec.json"
   connection:
-    - config_path: "secrets/config.json"
-      status: "succeed"
-    - config_path: "integration_tests/invalid_config.json"
-      status: "failed"
+    tests:
+      - config_path: "secrets/config.json"
+        status: "succeed"
+      - config_path: "integration_tests/invalid_config.json"
+        status: "failed"
   discovery:
-    - config_path: "secrets/config.json"
-      backward_compatibility_tests_config:
-        disable_for_version: "0.0.2"
+    tests:
+      - config_path: "secrets/config.json"
+        backward_compatibility_tests_config:
+          disable_for_version: "0.0.2"
   basic_read:
-    - config_path: "secrets/config.json"
-      configured_catalog_path: "integration_tests/configured_catalog.json"
-      empty_streams: []
+    tests:
+      - config_path: "secrets/config.json"
+        empty_streams: []
+  incremental:
+    tests:
+      - config_path: "secrets/config.json"
+        configured_catalog_path: "integration_tests/configured_catalog.json"
+        future_state:
+          future_state_path: "integration_tests/abnormal_state.json"
   full_refresh:
-    - config_path: "secrets/config.json"
-      configured_catalog_path: "integration_tests/configured_catalog.json"
-      ignored_fields:
-        "daily_active_users": ["uuid"]
-        "weekly_active_users": ["uuid"]
-        "four_weekly_active_users": ["uuid"]
-        "devices": ["uuid"]
-        "locations": ["uuid"]
-        "pages": ["uuid"]
-        "traffic_sources": ["uuid"]
-        "website_overview": ["uuid"]
+    tests:
+      - config_path: "secrets/config.json"
+        configured_catalog_path: "integration_tests/configured_catalog.json"
+        ignored_fields:
+          "daily_active_users": ["uuid"]
+          "weekly_active_users": ["uuid"]
+          "four_weekly_active_users": ["uuid"]
+          "devices": ["uuid"]
+          "locations": ["uuid"]
+          "pages": ["uuid"]
+          "traffic_sources": ["uuid"]
+          "website_overview": ["uuid"]

airbyte-integrations/connectors/source-google-analytics-data-api/integration_tests/abnormal_state.json

@@ -0,0 +1,10 @@
+{
+  "daily_active_users": {"date": "20501201"},
+  "weekly_active_users": {"date": "20501201"},
+  "four_weekly_active_users": {"date": "20501201"},
+  "devices": {"date": "20501201"},
+  "locations": {"date": "20501201"},
+  "pages": {"date": "20501201"},
+  "traffic_sources": {"date": "20501201"},
+  "website_overview": {"date": "20501201"}
+}

airbyte-integrations/connectors/source-google-analytics-data-api/integration_tests/sample_state.json

@@ -0,0 +1,10 @@
+{
+  "daily_active_users": {"date": "20221201"},
+  "weekly_active_users": {"date": "20221201"},
+  "four_weekly_active_users": {"date": "20221201"},
+  "devices": {"date": "20221201"},
+  "locations": {"date": "20221201"},
+  "pages": {"date": "20221201"},
+  "traffic_sources": {"date": "20221201"},
+  "website_overview": {"date": "20221201"}
+}

airbyte-integrations/connectors/source-google-analytics-data-api/source_google_analytics_data_api/source.py

@@ -222,44 +222,56 @@ def parse_response(
         rows = []
 
         for row in r.get("rows", []):
-            rows.append(
-                collections.ChainMap(
-                    *[
-                        self.add_primary_key(),
-                        self.add_property_id(self.config["property_id"]),
-                        self.add_dimensions(dimensions, row),
-                        self.add_metrics(metrics, metrics_type_map, row),
-                    ]
-                )
+            chain_row = collections.ChainMap(
+                *[
+                    self.add_primary_key(),
+                    self.add_property_id(self.config["property_id"]),
+                    self.add_dimensions(dimensions, row),
+                    self.add_metrics(metrics, metrics_type_map, row),
+                ]
             )
+            rows.append(dict(chain_row))
         r["records"] = rows
 
         yield r
 
 
 class IncrementalGoogleAnalyticsDataApiStream(GoogleAnalyticsDataApiBaseStream, IncrementalMixin, ABC):
-    _date_format = "%Y-%m-%d"
+    _date_format: str = "%Y-%m-%d"
 
     def __init__(self, *args, **kwargs):
         super(IncrementalGoogleAnalyticsDataApiStream, self).__init__(*args, **kwargs)
-        self._cursor_value = None
+        self._cursor_value: str = ""
 
 
 class GoogleAnalyticsDataApiGenericStream(IncrementalGoogleAnalyticsDataApiStream):
-    _default_window_in_days = 1
-    _record_date_format = "%Y%m%d"
+    _default_window_in_days: int = 1
+    _record_date_format:     str = "%Y%m%d"
 
     @property
     def cursor_field(self) -> Union[str, List[str]]:
         return "date"
 
     @property
     def state(self) -> MutableMapping[str, Any]:
-        return {self.cursor_field: self._cursor_value or utils.string_to_date(self.config["date_ranges_start_date"], self._date_format)}
+        if self._cursor_value:
+            return {self.cursor_field: self._cursor_value}
+        return {
+            self.cursor_field: utils.date_to_string(self._date_parse_probe(self.config["date_ranges_start_date"]), self._record_date_format)
+        }
 
     @state.setter
-    def state(self, value):
-        self._cursor_value = utils.string_to_date(value[self.cursor_field], self._date_format) + datetime.timedelta(days=1)
+    def state(self, value: dict):
+        self._cursor_value = utils.date_to_string(
+            self._date_parse_probe(value[self.cursor_field]) + datetime.timedelta(days=1),
+            self._record_date_format
+        )
+
+    def _date_parse_probe(self, date_string: str) -> datetime.date:
+        try:
+            return utils.string_to_date(date_string, self._record_date_format)
+        except ValueError:
+            return utils.string_to_date(date_string, self._date_format)
 
     def request_body_json(
         self,
@@ -285,19 +297,18 @@ def read_records(
         records = super().read_records(sync_mode=sync_mode, cursor_field=cursor_field, stream_slice=stream_slice, stream_state=stream_state)
         for record in records:
             for row in record["records"]:
-                next_cursor_value = utils.string_to_date(row[self.cursor_field], self._record_date_format)
-                self._cursor_value = max(self._cursor_value, next_cursor_value) if self._cursor_value else next_cursor_value
+                self._cursor_value: str = max(self._cursor_value, row[self.cursor_field]) if self._cursor_value else row[self.cursor_field]
                 yield row
 
     def stream_slices(
         self, *, sync_mode: SyncMode, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None
     ) -> Iterable[Optional[Mapping[str, Any]]]:
         dates = []
 
-        today: datetime.date = datetime.date.today()
-        start_date: datetime.date = self.state[self.cursor_field]
+        today:      datetime.date = datetime.date.today()
+        start_date: datetime.date = utils.string_to_date(self.state[self.cursor_field], self._record_date_format)
 
-        timedelta: int = self.config["window_in_days"] or self._default_window_in_days
+        timedelta: int = self.config.get("window_in_days", self._default_window_in_days)
 
         while start_date <= today:
             end_date: datetime.date = start_date + datetime.timedelta(days=timedelta)

airbyte-integrations/connectors/source-google-analytics-data-api/unit_tests/test_authenticator.py

@@ -0,0 +1,56 @@
+import datetime
+
+from source_google_analytics_data_api import utils
+from source_google_analytics_data_api.authenticator import GoogleServiceKeyAuthenticator
+
+
+private_key = """
+-----BEGIN RSA PRIVATE KEY-----
+MIIBPAIBAAJBAIy64eoS8VCwNnu6+kcyvRc7w/Cw20fnZWcpftLIl33ZWdXl/Q+W
+sEQkm2RRWO2R9CGC2bJRZYEbiAabuG4T1LkCAwEAAQJBAIbr5Qv1fUZOqu2VJb58
+9qz/r6ti49jcEGwHbH/JsPQFpXWTyKdonibIblLDB4AbZdR25zEM2k2G42OErWjJ
+0wECIQD21o4uOMJtAe7GLvg3AeQrb4t9sWlCPuNNxnmKfn8PfQIhAJH0F7vSyD7D
+UwRcMht0PU65zWUZArBaI7BlpmFhgNbtAiEA1rIZ6vQtkDjtIW37MYUwm+Milgo4
+vokKljx6vM5339UCIEb1Owyvn3cUEypNgHbkfmHl5zu9exct26gI42j4tGDJAiEA
+3uFuRTUY3W5s8hxyHcATsGhWfGa4VCSjlKE+sSchksc=
+-----END RSA PRIVATE KEY-----
+"""
+
+
+def test_authenticator(mocker):
+    requests = mocker.MagicMock()
+    requests.request.return_value.json.side_effect = [
+        {
+            "expires_in": GoogleServiceKeyAuthenticator._default_token_lifetime_secs,
+            "access_token": "ga-access-token-1"
+        },
+        {
+            "expires_in": GoogleServiceKeyAuthenticator._default_token_lifetime_secs,
+            "access_token": "ga-access-token-2"
+        }
+    ]
+
+    mocker.patch("source_google_analytics_data_api.authenticator.requests", requests)
+
+    authenticator = GoogleServiceKeyAuthenticator(credentials={
+        "client_email": "[email protected]",
+        "private_key": private_key,
+        "client_id": "c-airbyte-001"
+    })
+
+    request_object = mocker.MagicMock()
+    request_object.headers = {}
+
+    authenticator(request_object)
+    assert requests.request.call_count == 1
+    assert request_object.headers["Authorization"] == f"Bearer ga-access-token-1"
+
+    authenticator(request_object)
+    assert requests.request.call_count == 1
+    assert request_object.headers["Authorization"] == f"Bearer ga-access-token-1"
+
+    authenticator._token["expires_at"] = utils.datetime_to_secs(datetime.datetime.utcnow()) - GoogleServiceKeyAuthenticator._default_token_lifetime_secs
+
+    authenticator(request_object)
+    assert requests.request.call_count == 2
+    assert request_object.headers["Authorization"] == f"Bearer ga-access-token-2"