fix(source-hubspot): bump to latest version of CDK to get the bug fix for missing custom properties during incremental syncs (#68159)

## What

We had a bug where during incremental syncs Hubspot CRM Search streams
were not including the custom properties in the json body of the POST
request so they were not getting received and emitted with records.

## How

The bug was in the CDK and it was fixed in version `7.3.7` in this
airbytehq/airbyte-python-cdk#797

We need to bump the version of SDM to get the fix, but in addition, we
need to upgrade the unit_test `pyproject.toml` which is still on v6.
I've also added a new test that validates that properties are indeed
populated in the outbound request. And with the bump from v6 to v7 I
fixed the tests which have now changed.

**Note**: It does feel like we have something of a gap where our unit
tests don't properly test CDK changes since the two are independently
versioned... This is something we may want to investigate and solve so
these types of things don't happen again

## Can this PR be safely reverted and rolled back?

- [ ] YES 💚
- [ ] NO ❌

Kind of... If we do this wrong then we have to reset customers back to
their previous state, but this is no different than the state we were
previously in

Author: brianjlai

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

@@ -6,11 +6,11 @@ data:
     hosts:
       - api.hubapi.com
   connectorBuildOptions:
-    baseImage: docker.io/airbyte/source-declarative-manifest:7.3.4@sha256:ebf5c159101614ec9ab3a5183c92781c397b998d15c0b7e2e561ec1e00397fa4
+    baseImage: docker.io/airbyte/source-declarative-manifest:7.3.7@sha256:7832dad42cabf1fb0c0c3f5239f256d3e4eca6191d3c8d0bff3b5b708dac46d4
   connectorSubtype: api
   connectorType: source
   definitionId: 36c891d9-4bd9-43ac-bad2-10e12756272c
-  dockerImageTag: 6.0.6
+  dockerImageTag: 6.0.7-rc.1
   dockerRepository: airbyte/source-hubspot
   documentationUrl: https://docs.airbyte.com/integrations/sources/hubspot
   resourceRequirements:
@@ -43,7 +43,7 @@ data:
   releaseStage: generally_available
   releases:
     rolloutConfiguration:
-      enableProgressiveRollout: false
+      enableProgressiveRollout: true
     breakingChanges:
       6.0.0:
         message: >-

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

@@ -145,7 +145,7 @@ def mock_response(cls, http_mocker: HttpMocker, request, responses, method: str
 
     @classmethod
     def mock_dynamic_schema_requests(cls, http_mocker: HttpMocker, entities: Optional[List[str]] = None):
-        entities = entities or OBJECTS_WITH_DYNAMIC_SCHEMA
+        entities = entities if entities is not None else OBJECTS_WITH_DYNAMIC_SCHEMA
 
         # figure out which entities are already mocked
         existing = set()

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

@@ -68,7 +68,7 @@ def test_given_oauth_authentication_when_read_then_perform_authenticated_queries
             http_mocker,
             with_oauth=True,
             with_dynamic_schemas=True,
-            entities=OBJECTS_WITH_DYNAMIC_SCHEMA,
+            entities=["calls"],
         )
         self.mock_response(http_mocker, self.request(), self.response())
         self.read_from_stream(self.oauth_config(), self.STREAM_NAME, SyncMode.full_refresh)
@@ -79,30 +79,30 @@ def test_given_records_when_read_extract_desired_records(self, http_mocker: Http
             http_mocker,
             with_oauth=True,
             with_dynamic_schemas=True,
-            entities=OBJECTS_WITH_DYNAMIC_SCHEMA,
+            entities=["calls"],
         )
         self.mock_response(http_mocker, self.request(), self.response())
         output = self.read_from_stream(self.oauth_config(), self.STREAM_NAME, SyncMode.full_refresh)
         assert len(output.records) == 1
 
     @HttpMocker()
     def test_given_one_page_when_read_stream_private_token_then_return_records(self, http_mocker: HttpMocker):
-        self._set_up_requests(http_mocker)
+        self._set_up_requests(http_mocker, with_dynamic_schemas=True, entities=["calls"])
         self.mock_response(http_mocker, self.request(), self.response())
         output = self.read_from_stream(self.private_token_config(self.ACCESS_TOKEN), self.STREAM_NAME, SyncMode.full_refresh)
         assert len(output.records) == 1
 
     @HttpMocker()
     def test_given_two_pages_when_read_then_return_records(self, http_mocker: HttpMocker):
-        self._set_up_requests(http_mocker)
+        self._set_up_requests(http_mocker, with_dynamic_schemas=True, entities=["calls"])
         self.mock_response(http_mocker, self.request(), self.response(with_pagination=True))
         self.mock_response(http_mocker, self.request(page_token=self.response_builder.pagination_strategy.NEXT_PAGE_TOKEN), self.response())
         output = self.read_from_stream(self.private_token_config(self.ACCESS_TOKEN), self.STREAM_NAME, SyncMode.full_refresh)
         assert len(output.records) == 2
 
     @HttpMocker()
     def test_given_error_response_when_read_analytics_then_get_trace_message(self, http_mocker: HttpMocker):
-        self._set_up_requests(http_mocker)
+        self._set_up_requests(http_mocker, with_dynamic_schemas=True, entities=["calls"])
         self.mock_response(http_mocker, self.request(), HttpResponse(status_code=500, body="{}"))
         with mock.patch("time.sleep"):
             output = self.read_from_stream(self.private_token_config(self.ACCESS_TOKEN), self.STREAM_NAME, SyncMode.full_refresh)
@@ -112,7 +112,7 @@ def test_given_error_response_when_read_analytics_then_get_trace_message(self, h
 
     @HttpMocker()
     def test_given_500_then_200_when_read_then_return_records(self, http_mocker: HttpMocker):
-        self._set_up_requests(http_mocker)
+        self._set_up_requests(http_mocker, with_dynamic_schemas=True, entities=["calls"])
         self.mock_response(http_mocker, self.request(), [HttpResponse(status_code=500, body="{}"), self.response()])
         with mock.patch("time.sleep"):
             output = self.read_from_stream(self.private_token_config(self.ACCESS_TOKEN), self.STREAM_NAME, SyncMode.full_refresh)
@@ -128,7 +128,7 @@ def test_given_missing_scopes_error_when_read_then_stop_sync(self, http_mocker:
 
     @HttpMocker()
     def test_given_unauthorized_error_when_read_then_stop_sync(self, http_mocker: HttpMocker):
-        self._set_up_requests(http_mocker)
+        self._set_up_requests(http_mocker, with_dynamic_schemas=True, entities=["calls"])
         self.mock_response(http_mocker, self.request(), HttpResponse(status_code=http.HTTPStatus.UNAUTHORIZED, body="{}"))
         with mock.patch("time.sleep"):
             output = self.read_from_stream(self.private_token_config(self.ACCESS_TOKEN), self.STREAM_NAME, SyncMode.full_refresh)
@@ -138,7 +138,7 @@ def test_given_unauthorized_error_when_read_then_stop_sync(self, http_mocker: Ht
 
     @HttpMocker()
     def test_given_one_page_when_read_then_get_records_with_flattened_properties(self, http_mocker: HttpMocker):
-        self._set_up_requests(http_mocker)
+        self._set_up_requests(http_mocker, with_dynamic_schemas=True, entities=["calls"])
         self.mock_response(http_mocker, self.request(), self.response())
         output = self.read_from_stream(self.private_token_config(self.ACCESS_TOKEN), self.STREAM_NAME, SyncMode.full_refresh)
         record = output.records[0].record.data
@@ -148,7 +148,7 @@ def test_given_one_page_when_read_then_get_records_with_flattened_properties(sel
 
     @HttpMocker()
     def test_given_incremental_sync_when_read_then_state_message_produced_and_state_match_latest_record(self, http_mocker: HttpMocker):
-        self._set_up_requests(http_mocker)
+        self._set_up_requests(http_mocker, with_dynamic_schemas=True, entities=["calls"])
         self.mock_response(http_mocker, self.request(), self.response())
         output = self.read_from_stream(self.private_token_config(self.ACCESS_TOKEN), self.STREAM_NAME, SyncMode.incremental)
         assert len(output.state_messages) == 2

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

@@ -1,7 +1,7 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 
 import http
-from typing import Dict, Optional
+from typing import Dict, List, Optional
 
 import freezegun
 import mock
@@ -52,44 +52,46 @@ def response(self, with_pagination: bool = False):
     def _set_up_oauth(self, http_mocker: HttpMocker):
         self.mock_oauth(http_mocker, self.ACCESS_TOKEN)
 
-    def _set_up_requests(self, http_mocker: HttpMocker, with_oauth: bool = False, with_dynamic_schema: bool = True):
+    def _set_up_requests(
+        self, http_mocker: HttpMocker, with_oauth: bool = False, with_dynamic_schema: bool = True, entities: Optional[List[str]] = None
+    ):
         if with_oauth:
             self._set_up_oauth(http_mocker)
         self.mock_custom_objects(http_mocker)
         self.mock_properties(http_mocker, self.OBJECT_TYPE, self.MOCK_PROPERTIES_FOR_SCHEMA_LOADER)
         if with_dynamic_schema:
-            self.mock_dynamic_schema_requests(http_mocker)
+            self.mock_dynamic_schema_requests(http_mocker, entities)
 
     @HttpMocker()
     def test_given_oauth_authentication_when_read_then_perform_authenticated_queries(self, http_mocker: HttpMocker):
-        self._set_up_requests(http_mocker, with_oauth=True, with_dynamic_schema=True)
+        self._set_up_requests(http_mocker, with_oauth=True, with_dynamic_schema=True, entities=["leads"])
         self.read_from_stream(self.oauth_config(), self.STREAM_NAME, SyncMode.full_refresh)
 
     @HttpMocker()
     def test_given_records_when_read_extract_desired_records(self, http_mocker: HttpMocker):
-        self._set_up_requests(http_mocker, with_oauth=True, with_dynamic_schema=True)
+        self._set_up_requests(http_mocker, with_oauth=True, with_dynamic_schema=True, entities=["leads"])
         self.mock_response(http_mocker, self.request(), self.response())
         output = self.read_from_stream(self.oauth_config(), self.STREAM_NAME, SyncMode.full_refresh)
         assert len(output.records) == 1
 
     @HttpMocker()
     def test_given_one_page_when_read_stream_private_token_then_return_records(self, http_mocker: HttpMocker):
-        self._set_up_requests(http_mocker)
+        self._set_up_requests(http_mocker, with_dynamic_schema=True, entities=["leads"])
         self.mock_response(http_mocker, self.request(), self.response())
         output = self.read_from_stream(self.private_token_config(self.ACCESS_TOKEN), self.STREAM_NAME, SyncMode.full_refresh)
         assert len(output.records) == 1
 
     @HttpMocker()
     def test_given_two_pages_when_read_then_return_records(self, http_mocker: HttpMocker):
-        self._set_up_requests(http_mocker)
+        self._set_up_requests(http_mocker, with_dynamic_schema=True, entities=["leads"])
         self.mock_response(http_mocker, self.request(), self.response(with_pagination=True))
         self.mock_response(http_mocker, self.request(page_token=self.response_builder.pagination_strategy.NEXT_PAGE_TOKEN), self.response())
         output = self.read_from_stream(self.private_token_config(self.ACCESS_TOKEN), self.STREAM_NAME, SyncMode.full_refresh)
         assert len(output.records) == 2
 
     @HttpMocker()
     def test_given_error_response_when_read_analytics_then_get_trace_message(self, http_mocker: HttpMocker):
-        self._set_up_requests(http_mocker)
+        self._set_up_requests(http_mocker, with_dynamic_schema=True, entities=["leads"])
         self.mock_response(http_mocker, self.request(), HttpResponse(status_code=500, body="{}"))
         with mock.patch("time.sleep"):
             output = self.read_from_stream(self.private_token_config(self.ACCESS_TOKEN), self.STREAM_NAME, SyncMode.full_refresh)
@@ -99,7 +101,7 @@ def test_given_error_response_when_read_analytics_then_get_trace_message(self, h
 
     @HttpMocker()
     def test_given_500_then_200_when_read_then_return_records(self, http_mocker: HttpMocker):
-        self._set_up_requests(http_mocker)
+        self._set_up_requests(http_mocker, with_dynamic_schema=True, entities=["leads"])
         self.mock_response(http_mocker, self.request(), [HttpResponse(status_code=500, body="{}"), self.response()])
         with mock.patch("time.sleep"):
             output = self.read_from_stream(self.private_token_config(self.ACCESS_TOKEN), self.STREAM_NAME, SyncMode.full_refresh)
@@ -115,7 +117,7 @@ def test_given_missing_scopes_error_when_read_then_stop_sync(self, http_mocker:
 
     @HttpMocker()
     def test_given_unauthorized_error_when_read_then_stop_sync(self, http_mocker: HttpMocker):
-        self._set_up_requests(http_mocker)
+        self._set_up_requests(http_mocker, with_dynamic_schema=True, entities=["leads"])
         self.mock_response(http_mocker, self.request(), HttpResponse(status_code=http.HTTPStatus.UNAUTHORIZED, body="{}"))
         with mock.patch("time.sleep"):
             output = self.read_from_stream(self.private_token_config(self.ACCESS_TOKEN), self.STREAM_NAME, SyncMode.full_refresh)
@@ -125,7 +127,7 @@ def test_given_unauthorized_error_when_read_then_stop_sync(self, http_mocker: Ht
 
     @HttpMocker()
     def test_given_one_page_when_read_then_get_records_with_flattened_properties(self, http_mocker: HttpMocker):
-        self._set_up_requests(http_mocker)
+        self._set_up_requests(http_mocker, with_dynamic_schema=True, entities=["leads"])
         self.mock_response(http_mocker, self.request(), self.response())
         output = self.read_from_stream(self.private_token_config(self.ACCESS_TOKEN), self.STREAM_NAME, SyncMode.full_refresh)
         record = output.records[0].record.data
@@ -135,7 +137,7 @@ def test_given_one_page_when_read_then_get_records_with_flattened_properties(sel
 
     @HttpMocker()
     def test_given_incremental_sync_when_read_then_state_message_produced_and_state_match_latest_record(self, http_mocker: HttpMocker):
-        self._set_up_requests(http_mocker)
+        self._set_up_requests(http_mocker, with_dynamic_schema=True, entities=["leads"])
         self.mock_response(http_mocker, self.request(), self.response())
         output = self.read_from_stream(self.private_token_config(self.ACCESS_TOKEN), self.STREAM_NAME, SyncMode.incremental)
         assert len(output.state_messages) == 2

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

@@ -44,15 +44,15 @@ def response(self, with_pagination: bool = False):
     def test_given_one_page_when_read_stream_oauth_then_return_records(self, http_mocker: HttpMocker):
         self.mock_oauth(http_mocker, self.ACCESS_TOKEN)
         self.mock_custom_objects(http_mocker)
-        self.mock_dynamic_schema_requests(http_mocker)
+        self.mock_dynamic_schema_requests(http_mocker, entities=[])
         self.mock_response(http_mocker, self.request().build(), self.response().build())
         output = self.read_from_stream(self.oauth_config(), self.STREAM_NAME, SyncMode.full_refresh)
         assert len(output.records) == 1
 
     @HttpMocker()
     def test_given_two_pages_when_read_stream_private_token_then_return_records(self, http_mocker: HttpMocker):
         self.mock_custom_objects(http_mocker)
-        self.mock_dynamic_schema_requests(http_mocker)
+        self.mock_dynamic_schema_requests(http_mocker, entities=[])
         self.mock_response(http_mocker, self.request().build(), self.response(with_pagination=True).build())
         self.mock_response(
             http_mocker,