Source Salesforce: perform call to objects list in connection check (#…

…17094) * #629 oncall - source salesforce: perform call to objects list in connection check * source salesforce: upd changelog * auto-bump connector version [ci skip] Co-authored-by: Octavia Squidington III <[email protected]>
airbytehq · Sep 23, 2022 · d62f881 · d62f881
1 parent d4b7f4e
commit d62f881
Show file tree

Hide file tree

Showing 6 changed files with 82 additions and 58 deletions.
diff --git a/airbyte-config/init/src/main/resources/seed/source_definitions.yaml b/airbyte-config/init/src/main/resources/seed/source_definitions.yaml
@@ -908,7 +908,7 @@
 - name: Salesforce
   sourceDefinitionId: b117307c-14b6-41aa-9422-947e34922962
   dockerRepository: airbyte/source-salesforce
-  dockerImageTag: 1.0.16
+  dockerImageTag: 1.0.17
   documentationUrl: https://docs.airbyte.io/integrations/sources/salesforce
   icon: salesforce.svg
   sourceType: api

diff --git a/airbyte-config/init/src/main/resources/seed/source_specs.yaml b/airbyte-config/init/src/main/resources/seed/source_specs.yaml
@@ -9379,7 +9379,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-salesforce:1.0.16"
+- dockerImage: "airbyte/source-salesforce:1.0.17"
   spec:
     documentationUrl: "https://docs.airbyte.com/integrations/sources/salesforce"
     connectionSpecification:

diff --git a/airbyte-integrations/connectors/source-salesforce/Dockerfile b/airbyte-integrations/connectors/source-salesforce/Dockerfile
@@ -13,5 +13,5 @@ RUN pip install .
 
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=1.0.16
+LABEL io.airbyte.version=1.0.17
 LABEL io.airbyte.name=airbyte/source-salesforce
diff --git a/airbyte-integrations/connectors/source-salesforce/source_salesforce/source.py b/airbyte-integrations/connectors/source-salesforce/source_salesforce/source.py
@@ -4,6 +4,7 @@
 
 from typing import Any, Iterator, List, Mapping, MutableMapping, Optional, Tuple, Union
 
+import requests
 from airbyte_cdk import AirbyteLogger
 from airbyte_cdk.models import AirbyteMessage, AirbyteStateMessage, ConfiguredAirbyteCatalog
 from airbyte_cdk.sources import AbstractSource
@@ -26,14 +27,20 @@ def _get_sf_object(config: Mapping[str, Any]) -> Salesforce:
 
     def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) -> Tuple[bool, Optional[str]]:
         try:
-            _ = self._get_sf_object(config)
+            salesforce = self._get_sf_object(config)
+            salesforce.describe()
         except exceptions.HTTPError as error:
-            error_data = error.response.json()[0]
-            error_code = error_data.get("errorCode")
-            if error.response.status_code == codes.FORBIDDEN and error_code == "REQUEST_LIMIT_EXCEEDED":
-                logger.warn(f"API Call limit is exceeded. Error message: '{error_data.get('message')}'")
-                return False, "API Call limit is exceeded"
-
+            error_msg = f"An error occurred: {error.response.text}"
+            try:
+                error_data = error.response.json()[0]
+            except (KeyError, requests.exceptions.JSONDecodeError):
+                pass
+            else:
+                error_code = error_data.get("errorCode")
+                if error.response.status_code == codes.FORBIDDEN and error_code == "REQUEST_LIMIT_EXCEEDED":
+                    logger.warn(f"API Call limit is exceeded. Error message: '{error_data.get('message')}'")
+                    error_msg = "API Call limit is exceeded"
+            return False, error_msg
         return True, None
 
     @classmethod

diff --git a/airbyte-integrations/connectors/source-salesforce/unit_tests/api_test.py b/airbyte-integrations/connectors/source-salesforce/unit_tests/api_test.py
@@ -192,17 +192,33 @@ def test_download_data_filter_null_bytes(stream_config, stream_api):
         assert res == [{"Id": "0014W000027f6UwQAI", "IsDeleted": False}]
 
 
-def test_check_connection_rate_limit(stream_config):
+@pytest.mark.parametrize(
+    "login_status_code, login_json_resp, discovery_status_code, discovery_resp_json, expected_error_msg",
+    (
+        (403, [{"errorCode": "REQUEST_LIMIT_EXCEEDED", "message": "TotalRequests Limit exceeded."}], 200, {}, "API Call limit is exceeded"),
+        (
+            200,
+            {"access_token": "access_token", "instance_url": "https://instance_url"},
+            403,
+            [{"errorCode": "FORBIDDEN", "message": "You do not have enough permissions"}],
+            'An error occurred: [{"errorCode": "FORBIDDEN", "message": "You do not have enough permissions"}]',
+        ),
+    ),
+)
+def test_check_connection_rate_limit(
+    stream_config, login_status_code, login_json_resp, discovery_status_code, discovery_resp_json, expected_error_msg
+):
     source = SourceSalesforce()
     logger = logging.getLogger("airbyte")
 
-    json_response = [{"errorCode": "REQUEST_LIMIT_EXCEEDED", "message": "TotalRequests Limit exceeded."}]
-    url = "https://login.salesforce.com/services/oauth2/token"
     with requests_mock.Mocker() as m:
-        m.register_uri("POST", url, json=json_response, status_code=403)
+        m.register_uri("POST", "https://login.salesforce.com/services/oauth2/token", json=login_json_resp, status_code=login_status_code)
+        m.register_uri(
+            "GET", "https://instance_url/services/data/v52.0/sobjects", json=discovery_resp_json, status_code=discovery_status_code
+        )
         result, msg = source.check_connection(logger, stream_config)
         assert result is False
-        assert msg == "API Call limit is exceeded"
+        assert msg == expected_error_msg
 
 
 def configure_request_params_mock(stream_1, stream_2):