🐛Source Amazon Seller Partner: add integration tests

airbyte-integrations/connectors/source-amazon-seller-partner/source_amazon_seller_partner/schemas/GET_LEDGER_SUMMARY_VIEW_DATA.json

@@ -4,7 +4,7 @@
   "type": "object",
   "$schema": "http://json-schema.org/draft-07/schema#",
   "properties": {
-    "Date": { "type": ["null", "string"], "format": "date-time" },
+    "Date": { "type": ["null", "string"], "format": "date" },
     "FNSKU": { "type": ["null", "string"] },
     "ASIN": { "type": ["null", "string"] },
     "MSKU": { "type": ["null", "string"] },

airbyte-integrations/connectors/source-amazon-seller-partner/source_amazon_seller_partner/schemas/GET_SALES_AND_TRAFFIC_REPORT.json

@@ -14,6 +14,9 @@
     "childAsin": {
       "type": ["null", "string"]
     },
+    "sku": {
+      "type": ["null", "string"]
+    },
     "salesByAsin": {
       "type": "object",
       "properties": {

airbyte-integrations/connectors/source-amazon-seller-partner/source_amazon_seller_partner/streams.py

@@ -872,12 +872,15 @@ def __init__(self, *args, **kwargs):
 
     def get_transform_function(self):
         def transform_function(original_value: Any, field_schema: Dict[str, Any]) -> Any:
-            if original_value and "format" in field_schema and field_schema["format"] == "date":
+            if original_value and field_schema.get("format") == "date":
                 date_format = self.MARKETPLACE_DATE_FORMAT_MAP.get(self.marketplace_id)
                 if not date_format:
-                    raise KeyError(f"Date format not found for Markeplace ID: {self.marketplace_id}")
-                transformed_value = pendulum.from_format(original_value, date_format).to_date_string()
-                return transformed_value
+                    raise KeyError(f"Date format not found for Marketplace ID: {self.marketplace_id}")
+                try:
+                    transformed_value = pendulum.from_format(original_value, date_format).to_date_string()
+                    return transformed_value
+                except ValueError:
+                    pass
 
             return original_value
 

airbyte-integrations/connectors/source-amazon-seller-partner/unit_tests/integration/config.py

@@ -0,0 +1,36 @@
+#
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+#
+
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Dict
+
+_ACCESS_TOKEN = "test_access_token"
+_LWA_APP_ID = "amazon_app_id"
+_LWA_CLIENT_SECRET = "amazon_client_secret"
+_MARKETPLACE_ID = "ATVPDKIKX0DER"
+_REFRESH_TOKEN = "amazon_refresh_token"
+
+
+class ConfigBuilder:
+    def __init__(self) -> None:
+        self._config: Dict[str, str] = {
+            "refresh_token": _REFRESH_TOKEN,
+            "lwa_app_id": _LWA_APP_ID,
+            "lwa_client_secret": _LWA_CLIENT_SECRET,
+            "replication_start_date": "2023-01-01T00:00:00Z",
+            "replication_end_date": "2023-01-30T00:00:00Z",
+            "aws_environment": "PRODUCTION",
+            "region": "US",
+            "account_type": "Seller",
+        }
+
+    def with_start_date(self, start_date: datetime) -> ConfigBuilder:
+        self._config["replication_start_date"] = start_date.isoformat()[:-13] + "Z"
+        return self
+
+    def build(self) -> Dict[str, str]:
+        return self._config

airbyte-integrations/connectors/source-amazon-seller-partner/unit_tests/integration/request_builder.py

@@ -0,0 +1,90 @@
+#
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+#
+
+
+from __future__ import annotations
+
+import json
+from typing import Any, List, Mapping, Optional, Union
+
+import pendulum
+from airbyte_cdk.test.mock_http.request import ANY_QUERY_PARAMS, HttpRequest
+
+from .config import _ACCESS_TOKEN, _LWA_APP_ID, _LWA_CLIENT_SECRET, _MARKETPLACE_ID, _REFRESH_TOKEN
+
+
+class RequestBuilder:
+
+    @classmethod
+    def auth_endpoint(cls) -> RequestBuilder:
+        request_headers = {"Content-Type": "application/x-www-form-urlencoded"}
+        request_body = f"grant_type=refresh_token&client_id={_LWA_APP_ID}&client_secret={_LWA_CLIENT_SECRET}&refresh_token={_REFRESH_TOKEN}"
+        return cls("auth/o2/token").with_base_url("https://api.amazon.com").with_headers(request_headers).with_body(request_body)
+
+    @classmethod
+    def create_report_endpoint(cls, report_name: str) -> RequestBuilder:
+        # TODO: need to pass stream slice or what? Maybe with_stream_slice method?
+        request_body = {
+            "reportType": report_name,
+            "marketplaceIds": [_MARKETPLACE_ID],
+            "dataStartTime": "2023-01-01T00:00:00Z",
+            "dataEndTime": "2023-01-30T00:00:00Z",
+        }
+        return cls("reports/2021-06-30/reports").with_body(json.dumps(request_body))
+
+    @classmethod
+    def check_report_status_endpoint(cls, report_id: str) -> RequestBuilder:
+        return cls(f"reports/2021-06-30/reports/{report_id}")
+
+    @classmethod
+    def get_document_download_url_endpoint(cls, document_id: str) -> RequestBuilder:
+        return cls(f"reports/2021-06-30/documents/{document_id}")
+
+    @classmethod
+    def download_document_endpoint(cls, url: str) -> RequestBuilder:
+        return cls("").with_base_url(url).with_headers(None)
+
+    @classmethod
+    def vendor_direct_fulfillment_shipping_endpoint(cls) -> RequestBuilder:
+        return cls("vendor/directFulfillment/shipping/v1/shippingLabels")
+
+    def __init__(self, resource: str) -> None:
+        self._resource = resource
+        self._base_url = "https://sellingpartnerapi-na.amazon.com"
+        self._headers = {
+            "content-type": "application/json",
+            "host": self._base_url.replace("https://", ""),
+            "user-agent": "python-requests",
+            "x-amz-access-token": _ACCESS_TOKEN,
+            "x-amz-date": pendulum.now("utc").strftime("%Y%m%dT%H%M%SZ"),
+        }
+        self._query_params = ANY_QUERY_PARAMS
+        self._body = None
+
+    def with_base_url(self, base_url: str) -> RequestBuilder:
+        self._base_url = base_url
+        return self
+
+    def with_headers(self, headers: Optional[Union[str, Mapping[str, str]]]) -> RequestBuilder:
+        self._headers = headers
+        return self
+
+    def with_query_params(self, query_params: Union[str, Mapping[str, Union[str, List[str]]]]) -> RequestBuilder:
+        self._query_params = query_params
+        return self
+
+    def with_body(self, body: Union[str, bytes, Mapping[str, Any]]) -> RequestBuilder:
+        self._body = body
+        return self
+
+    @property
+    def url(self) -> str:
+        # TODO: make better?
+        url = self._base_url
+        if self._resource:
+            url = f"{url}/{self._resource}"
+        return url
+
+    def build(self) -> HttpRequest:
+        return HttpRequest(url=self.url, query_params=self._query_params, headers=self._headers, body=self._body)

airbyte-integrations/connectors/source-amazon-seller-partner/unit_tests/integration/response_builder.py

@@ -0,0 +1,18 @@
+#
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+#
+
+
+import json
+from typing import Any, Mapping
+
+from airbyte_cdk.test.mock_http import HttpResponse
+from airbyte_cdk.test.mock_http.response_builder import find_template
+
+
+def response_with_status(status_code: int) -> HttpResponse:
+    return HttpResponse(json.dumps(find_template(str(status_code), __file__)), status_code)
+
+
+def build_response(body: Mapping[str, Any], status_code: int) -> HttpResponse:
+    return HttpResponse(body=json.dumps(body), status_code=status_code)