🐛Source Amazon Seller Partner: add integration tests

airbyte-integrations/connectors/source-amazon-seller-partner/metadata.yaml

@@ -15,7 +15,7 @@ data:
   connectorSubtype: api
   connectorType: source
   definitionId: e55879a8-0ef8-4557-abcf-ab34c53ec460
-  dockerImageTag: 3.2.1
+  dockerImageTag: 3.2.2
   dockerRepository: airbyte/source-amazon-seller-partner
   documentationUrl: https://docs.airbyte.com/integrations/sources/amazon-seller-partner
   githubIssueLabel: source-amazon-seller-partner

airbyte-integrations/connectors/source-amazon-seller-partner/setup.py

@@ -7,7 +7,7 @@
 
 MAIN_REQUIREMENTS = ["airbyte-cdk", "xmltodict~=0.12", "dateparser==1.2.0"]
 
-TEST_REQUIREMENTS = ["requests-mock~=1.9.3", "pytest~=6.1", "pytest-mock"]
+TEST_REQUIREMENTS = ["requests-mock~=1.9.3", "pytest~=6.1", "pytest-mock", "freezegun==1.2.2"]
 
 setup(
     entry_points={

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

@@ -5,7 +5,8 @@
 
 import csv
 import gzip
-import json as json_lib
+import json
+import os
 import time
 from abc import ABC, abstractmethod
 from enum import Enum
@@ -32,6 +33,8 @@
 DATE_TIME_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
 DATE_FORMAT = "%Y-%m-%d"
 
+IS_TESTING = os.environ.get("DEPLOYMENT_MODE") == "testing"
+
 
 class AmazonSPStream(HttpStream, ABC):
     data_field = "payload"
@@ -64,6 +67,12 @@ def request_headers(self, *args, **kwargs) -> Mapping[str, Any]:
     def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
         return None
 
+    def retry_factor(self) -> float:
+        """
+        Override for testing purposes
+        """
+        return 0 if IS_TESTING else super().retry_factor
+
 
 class IncrementalAmazonSPStream(AmazonSPStream, ABC):
     page_size = 100
@@ -242,7 +251,7 @@ def _create_report(
         create_report_request = self._create_prepared_request(
             path=f"{self.path_prefix}/reports",
             headers=dict(request_headers, **self.authenticator.get_auth_header()),
-            data=json_lib.dumps(report_data),
+            data=json.dumps(report_data),
         )
         report_response = self._send_request(create_report_request, {})
         self.http_method = "GET"  # rollback
@@ -264,8 +273,9 @@ def download_and_decompress_report_document(self, payload: dict) -> str:
         """
         Unpacks a report document
         """
-        report = requests.get(payload.get("url"))
-        report.raise_for_status()
+
+        download_report_request = self._create_prepared_request(path=payload.get("url"))
+        report = self._send_request(download_report_request, {})
         if "compressionAlgorithm" in payload:
             return gzip.decompress(report.content).decode("iso-8859-1")
         return report.content.decode("iso-8859-1")
@@ -371,6 +381,12 @@ def read_records(
         else:
             raise Exception(f"Unknown response for stream '{self.name}'. Response body {report_payload}")
 
+    def retry_factor(self) -> float:
+        """
+        Override for testing purposes
+        """
+        return 0 if IS_TESTING else super().retry_factor
+
 
 class IncrementalReportsAmazonSPStream(ReportsAmazonSPStream):
     @property
@@ -636,7 +652,7 @@ class FbaInventoryPlaningReport(IncrementalReportsAmazonSPStream):
 
 class AnalyticsStream(ReportsAmazonSPStream):
     def parse_document(self, document):
-        parsed = json_lib.loads(document)
+        parsed = json.loads(document)
         return parsed.get(self.result_key, [])
 
     def _report_data(
@@ -875,12 +891,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 Marketplace ID: {self.marketplace_id}")
-                transformed_value = pendulum.from_format(original_value, date_format).to_date_string()
-                return transformed_value
+                try:
+                    transformed_value = pendulum.from_format(original_value, date_format).to_date_string()
+                    return transformed_value
+                except ValueError:
+                    pass
 
             return original_value
 
@@ -1088,6 +1107,12 @@ class VendorDirectFulfillmentShipping(IncrementalAmazonSPStream):
     def path(self, **kwargs) -> str:
         return f"vendor/directFulfillment/shipping/{VENDORS_API_VERSION}/shippingLabels"
 
+    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
+        stream_data = response.json()
+        next_page_token = stream_data.get("payload", {}).get("pagination", {}).get(self.next_page_token_field)
+        if next_page_token:
+            return {self.next_page_token_field: next_page_token}
+
     def request_params(
         self, stream_state: Mapping[str, Any], next_page_token: Mapping[str, Any] = None, **kwargs
     ) -> MutableMapping[str, Any]:

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

@@ -3,10 +3,13 @@
 #
 
 
+import os
 from typing import Any, Dict
 
 import pytest
 
+os.environ["DEPLOYMENT_MODE"] = "testing"
+
 
 @pytest.fixture
 def report_init_kwargs() -> Dict[str, Any]:
@@ -18,3 +21,8 @@ def report_init_kwargs() -> Dict[str, Any]:
         "report_options": None,
         "replication_end_date": None,
     }
+
+
+@pytest.fixture
+def http_mocker() -> None:
+    """This fixture is needed to pass http_mocker parameter from the @HttpMocker decorator to a test"""

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

@@ -0,0 +1,46 @@
+#
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+#
+
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Dict
+
+import pendulum
+
+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"
+
+CONFIG_START_DATE = "2023-01-01T00:00:00Z"
+CONFIG_END_DATE = "2023-01-30T00:00:00Z"
+NOW = pendulum.now(tz="utc")
+
+
+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": CONFIG_START_DATE,
+            "replication_end_date": CONFIG_END_DATE,
+            "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 with_end_date(self, end_date: datetime) -> ConfigBuilder:
+        self._config["replication_end_date"] = end_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/pagination.py

@@ -0,0 +1,16 @@
+#
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+#
+
+
+from typing import Any, Dict
+
+from airbyte_cdk.test.mock_http.response_builder import PaginationStrategy
+
+NEXT_TOKEN_STRING = "MDAwMDAwMDAwMQ=="
+
+
+class VendorDirectFulfillmentShippingPaginationStrategy(PaginationStrategy):
+    def update(self, response: Dict[str, Any]) -> None:
+        response["payload"]["pagination"] = {}
+        response["payload"]["pagination"]["nextToken"] = NEXT_TOKEN_STRING

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

@@ -0,0 +1,83 @@
+#
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+#
+
+
+from __future__ import annotations
+
+import json
+from typing import Any, List, Mapping, Optional, Union
+
+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, NOW, 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:
+        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": NOW.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
+
+    def _url(self) -> str:
+        return f"{self._base_url}/{self._resource}" if self._resource else self._base_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,19 @@
+#
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+#
+
+
+import json
+from http import HTTPStatus
+from typing import Any, Mapping, Optional
+
+from airbyte_cdk.test.mock_http import HttpResponse
+
+
+def response_with_status(status_code: HTTPStatus, body: Optional[Mapping[str, Any]] = None) -> HttpResponse:
+    body = body or {}
+    return HttpResponse(body=json.dumps(body), status_code=status_code)
+
+
+def build_response(body: Mapping[str, Any], status_code: HTTPStatus) -> HttpResponse:
+    return HttpResponse(body=json.dumps(body), status_code=status_code)