Source Hubspot: add integration tests (#35945)

airbyte-integrations/connectors/source-hubspot/acceptance-test-config.yml

@@ -29,35 +29,35 @@ acceptance_tests:
         timeout_seconds: 3600
         empty_streams:
           - name: engagements_calls
-            bypass_reason: Unable to populate cost $20/month
+            bypass_reason: Unable to populate (cost $20/month) - covered by integration tests
           - name: owners_archived
-            bypass_reason: unable to populate
+            bypass_reason: Unable to populate - covered by integration tests
           - name: tickets_web_analytics
-            bypass_reason: Unable to populate
+            bypass_reason: Unable to populate - covered by integration tests
           - name: deals_web_analytics
-            bypass_reason: Unable to populate
+            bypass_reason: Unable to populate - covered by integration tests
           - name: companies_web_analytics
-            bypass_reason: Unable to populate
+            bypass_reason: Unable to populate - covered by integration tests
           - name: engagements_calls_web_analytics
-            bypass_reason: Unable to populate
+            bypass_reason: Unable to populate - covered by integration tests
           - name: engagements_emails_web_analytics
-            bypass_reason: Unable to populate
+            bypass_reason: Unable to populate - covered by integration tests
           - name: engagements_meetings_web_analytics
-            bypass_reason: Unable to populate
+            bypass_reason: Unable to populate - covered by integration tests
           - name: engagements_notes_web_analytics
-            bypass_reason: Unable to populate
+            bypass_reason: Unable to populate - covered by integration tests
           - name: engagements_tasks_web_analytics
-            bypass_reason: Unable to populate
+            bypass_reason: Unable to populate - covered by integration tests
           - name: goals_web_analytics
-            bypass_reason: Unable to populate
+            bypass_reason: Unable to populate - covered by integration tests
           - name: line_items_web_analytics
-            bypass_reason: Unable to populate
+            bypass_reason: Unable to populate - covered by integration tests
           - name: products_web_analytics
-            bypass_reason: Unable to populate
+            bypass_reason: Unable to populate - covered by integration tests
           - name: pets_web_analytics
-            bypass_reason: Unable to populate
+            bypass_reason: Unable to populate - covered by integration tests
           - name: cars_web_analytics
-            bypass_reason: Unable to populate
+            bypass_reason: Unable to populate - covered by integration tests
   full_refresh:
     tests:
       - config_path: secrets/config.json

airbyte-integrations/connectors/source-hubspot/metadata.yaml

@@ -10,7 +10,7 @@ data:
   connectorSubtype: api
   connectorType: source
   definitionId: 36c891d9-4bd9-43ac-bad2-10e12756272c
-  dockerImageTag: 4.1.0
+  dockerImageTag: 4.1.1
   dockerRepository: airbyte/source-hubspot
   documentationUrl: https://docs.airbyte.com/integrations/sources/hubspot
   githubIssueLabel: source-hubspot

airbyte-integrations/connectors/source-hubspot/pyproject.toml

@@ -3,7 +3,7 @@ requires = [ "poetry-core>=1.0.0",]
 build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
-version = "4.1.0"
+version = "4.1.1"
 name = "source-hubspot"
 description = "Source implementation for HubSpot."
 authors = [ "Airbyte <[email protected]>",]
@@ -27,3 +27,5 @@ requests-mock = "^1.9.3"
 mock = "^5.1.0"
 pytest-mock = "^3.6"
 pytest = "^6.2"
+pytz = "2024.1"
+freezegun = "0.3.4"

airbyte-integrations/connectors/source-hubspot/source_hubspot/source.py

@@ -197,11 +197,12 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
             self.logger.info("No scopes to grant when authenticating with API key.")
             available_streams = streams
 
-        available_streams.extend(self.get_custom_object_streams(api=api, common_params=common_params))
+        custom_object_streams = list(self.get_custom_object_streams(api=api, common_params=common_params))
+        available_streams.extend(custom_object_streams)
 
         if enable_experimental_streams:
             custom_objects_web_analytics_streams = self.get_web_analytics_custom_objects_stream(
-                custom_object_stream_instances=self.get_custom_object_streams(api=api, common_params=common_params),
+                custom_object_stream_instances=custom_object_streams,
                 common_params=common_params,
             )
             available_streams.extend(custom_objects_web_analytics_streams)

airbyte-integrations/connectors/source-hubspot/source_hubspot/streams.py

@@ -794,7 +794,7 @@ def _get_field_props(field_type: str) -> Mapping[str, List[str]]:
     @property
     @lru_cache()
     def properties(self) -> Mapping[str, Any]:
-        """Some entities has dynamic set of properties, so we trying to resolve those at runtime"""
+        """Some entities have dynamic set of properties, so we're trying to resolve those at runtime"""
         props = {}
         if not self.entity:
             return props

airbyte-integrations/connectors/source-hubspot/unit_tests/conftest.py

@@ -85,3 +85,8 @@ def fake_properties_list():
 @pytest.fixture(name="api")
 def api(some_credentials):
     return API(some_credentials)
+
+
+@pytest.fixture
+def http_mocker():
+    return None

airbyte-integrations/connectors/source-hubspot/unit_tests/integrations/__init__.py

@@ -0,0 +1,133 @@
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+
+import copy
+from datetime import datetime, timedelta
+from typing import Any, Dict, List, Optional
+
+import freezegun
+import pytz
+from airbyte_cdk.test.catalog_builder import CatalogBuilder
+from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput, read
+from airbyte_cdk.test.mock_http import HttpMocker
+from airbyte_cdk.test.mock_http.response_builder import FieldPath, HttpResponseBuilder, RecordBuilder, create_record_builder, find_template
+from airbyte_protocol.models import AirbyteStateMessage, SyncMode
+from source_hubspot import SourceHubspot
+
+from .config_builder import ConfigBuilder
+from .request_builders.api import CustomObjectsRequestBuilder, OAuthRequestBuilder, PropertiesRequestBuilder, ScopesRequestBuilder
+from .request_builders.streams import CRMStreamRequestBuilder, IncrementalCRMStreamRequestBuilder, WebAnalyticsRequestBuilder
+from .response_builder.helpers import RootHttpResponseBuilder
+from .response_builder.api import ScopesResponseBuilder
+from .response_builder.streams import GenericResponseBuilder, HubspotStreamResponseBuilder
+
+
+@freezegun.freeze_time("2024-03-03T14:42:00Z")
+class HubspotTestCase:
+    DT_FORMAT = '%Y-%m-%dT%H:%M:%SZ'
+    OBJECT_ID = "testID"
+    ACCESS_TOKEN = "new_access_token"
+    CURSOR_FIELD = "occurredAt"
+    PROPERTIES = {
+        "closed_date": "datetime",
+        "createdate": "datetime",
+    }
+
+    @classmethod
+    def now(cls):
+        return datetime.now(pytz.utc)
+
+    @classmethod
+    def start_date(cls):
+        return cls.now() - timedelta(days=30)
+
+    @classmethod
+    def updated_at(cls):
+        return cls.now() - timedelta(days=1)
+
+    @classmethod
+    def dt_str(cls, dt: datetime.date) -> str:
+        return dt.strftime(cls.DT_FORMAT)
+
+    @classmethod
+    def oauth_config(cls, start_date: Optional[str] = None) -> Dict[str, Any]:
+        start_date = start_date or cls.dt_str(cls.start_date())
+        return ConfigBuilder().with_start_date(start_date).with_auth(
+            {
+                "credentials_title": "OAuth Credentials",
+                "redirect_uri": "https://airbyte.io",
+                "client_id": "client_id",
+                "client_secret": "client_secret",
+                "refresh_token": "refresh_token",
+            }
+        ).build()
+
+    @classmethod
+    def private_token_config(cls, token: str, start_date: Optional[str] = None) -> Dict[str, Any]:
+        start_date = start_date or cls.dt_str(cls.start_date())
+        return ConfigBuilder().with_start_date(start_date).with_auth(
+            {
+                "credentials_title": "Private App Credentials",
+                "access_token": token,
+            }
+        ).build()
+
+    @classmethod
+    def mock_oauth(cls, http_mocker: HttpMocker, token: str):
+        creds = cls.oauth_config()["credentials"]
+        req = OAuthRequestBuilder().with_client_id(
+            creds["client_id"]
+        ).with_client_secret(
+            creds["client_secret"]
+        ).with_refresh_token(
+            creds["refresh_token"]
+        ).build()
+        response = GenericResponseBuilder().with_value("access_token", token).with_value("expires_in", 7200).build()
+        http_mocker.post(req, response)
+
+    @classmethod
+    def mock_scopes(cls, http_mocker: HttpMocker, token: str, scopes: List[str]):
+        http_mocker.get(ScopesRequestBuilder().with_access_token(token).build(), ScopesResponseBuilder(scopes).build())
+
+    @classmethod
+    def mock_custom_objects(cls, http_mocker: HttpMocker):
+        http_mocker.get(
+            CustomObjectsRequestBuilder().build(),
+            HttpResponseBuilder({}, records_path=FieldPath("results"), pagination_strategy=None).build()
+        )
+
+    @classmethod
+    def mock_properties(cls, http_mocker: HttpMocker, object_type: str, properties: Dict[str, str]):
+        templates = find_template("properties", __file__)
+        record_builder = lambda: RecordBuilder(copy.deepcopy(templates[0]), id_path=None, cursor_path=None)
+
+        response_builder = RootHttpResponseBuilder(templates)
+        for name, type in properties.items():
+            record = record_builder().with_field(FieldPath("name"), name).with_field(FieldPath("type"), type)
+            response_builder = response_builder.with_record(record)
+
+        http_mocker.get(
+            PropertiesRequestBuilder().for_entity(object_type).build(),
+            response_builder.build()
+        )
+
+    @classmethod
+    def mock_response(cls, http_mocker: HttpMocker, request, responses, method: str = "get"):
+        if not isinstance(responses, (list, tuple)):
+            responses = [responses]
+        getattr(http_mocker, method)(request, responses)
+
+    @classmethod
+    def record_builder(cls, stream: str, record_cursor_path):
+        return create_record_builder(
+            find_template(stream, __file__), records_path=FieldPath("results"), record_id_path=None, record_cursor_path=record_cursor_path
+        )
+
+    @classmethod
+    def catalog(cls, stream: str, sync_mode: SyncMode):
+        return CatalogBuilder().with_stream(stream, sync_mode).build()
+
+    @classmethod
+    def read_from_stream(
+        cls, cfg, stream: str, sync_mode: SyncMode, state: Optional[List[AirbyteStateMessage]] = None, expecting_exception: bool = False
+    ) -> EntrypointOutput:
+        return read(SourceHubspot(), cfg, cls.catalog(stream, sync_mode), state, expecting_exception)

airbyte-integrations/connectors/source-hubspot/unit_tests/integrations/config_builder.py

@@ -0,0 +1,21 @@
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+
+from typing import Any, Mapping
+
+
+class ConfigBuilder:
+    def __init__(self):
+        self._config = {
+            "enable_experimental_streams": True
+        }
+
+    def with_start_date(self, start_date: str):
+        self._config["start_date"] = start_date
+        return self
+
+    def with_auth(self, credentials: Mapping[str, str]):
+        self._config["credentials"] = credentials
+        return self
+
+    def build(self) -> Mapping[str, Any]:
+        return self._config

airbyte-integrations/connectors/source-hubspot/unit_tests/integrations/request_builders/__init__.py

@@ -0,0 +1,9 @@
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+
+import abc
+
+
+class AbstractRequestBuilder:
+    @abc.abstractmethod
+    def build(self):
+        pass

airbyte-integrations/connectors/source-hubspot/unit_tests/integrations/request_builders/api.py

@@ -0,0 +1,66 @@
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+
+from airbyte_cdk.test.mock_http import HttpRequest
+
+from . import AbstractRequestBuilder
+
+
+class OAuthRequestBuilder(AbstractRequestBuilder):
+    URL = "https://api.hubapi.com/oauth/v1/token"
+
+    def __init__(self):
+        self._params = {}
+
+    def with_client_id(self, client_id: str):
+        self._params["client_id"] = client_id
+        return self
+
+    def with_client_secret(self, client_secret: str):
+        self._params["client_secret"] = client_secret
+        return self
+
+    def with_refresh_token(self, refresh_token: str):
+        self._params["refresh_token"] = refresh_token
+        return self
+
+    def build(self) -> HttpRequest:
+        client_id, client_secret, refresh_token = self._params["client_id"], self._params["client_secret"], self._params["refresh_token"]
+        return HttpRequest(
+            url=self.URL,
+            body=f"grant_type=refresh_token&client_id={client_id}&client_secret={client_secret}&refresh_token={refresh_token}"
+        )
+
+
+class ScopesRequestBuilder(AbstractRequestBuilder):
+    URL = "https://api.hubapi.com/oauth/v1/access-tokens/{token}"
+
+    def __init__(self):
+        self._token = None
+
+    def with_access_token(self, token: str):
+        self._token = token
+        return self
+
+    def build(self) -> HttpRequest:
+        return HttpRequest(url=self.URL.format(token=self._token))
+
+
+class CustomObjectsRequestBuilder(AbstractRequestBuilder):
+    URL = "https://api.hubapi.com/crm/v3/schemas"
+
+    def build(self) -> HttpRequest:
+        return HttpRequest(url=self.URL)
+
+
+class PropertiesRequestBuilder(AbstractRequestBuilder):
+    URL = "https://api.hubapi.com/properties/v2/{resource}/properties"
+
+    def __init__(self):
+        self._resource = None
+
+    def for_entity(self, entity):
+        self._resource = entity
+        return self
+
+    def build(self) -> HttpRequest:
+        return HttpRequest(url=self.URL.format(resource=self._resource))