diff --git a/tap_salesforce/__init__.py b/tap_salesforce/__init__.py index 5f8565d..f0d1d98 100644 --- a/tap_salesforce/__init__.py +++ b/tap_salesforce/__init__.py @@ -132,16 +132,19 @@ def create_property_schema(field, mdata, expected_pk_field): return (property_schema, mdata) + PK_OVERRIDES = { "LightningUriEvent": "EventIdentifier", } + # pylint: disable=too-many-branches,too-many-statements def do_discover(sf): """Describes a Salesforce instance's objects and generates a JSON schema for each field.""" global_description = sf.describe() - blacklisted = sf.get_blacklisted_objects() + all_objects = {o['name'] for o in global_description['sobjects']} + blacklisted = sf.get_blacklisted_objects(object_names=all_objects) objects_to_discover = [ o['name'] for o in global_description['sobjects'] if o['name'] not in blacklisted and not o['name'].endswith("ChangeEvent") @@ -300,6 +303,7 @@ def do_discover(sf): result = {'streams': entries} json.dump(result, sys.stdout, indent=4) + def do_sync(sf, catalog, state): starting_stream = state.get("current_stream") @@ -325,7 +329,7 @@ def do_sync(sf, catalog, state): LOGGER.info("%s: Skipping - not selected", stream_name) continue - if stream_name in sf.get_blacklisted_objects(): + if stream_name in sf.get_blacklisted_objects(object_names=[stream_name]): LOGGER.info("%s: Skipping - blacklisted", stream_name) continue diff --git a/tap_salesforce/salesforce/__init__.py b/tap_salesforce/salesforce/__init__.py index 30aa3ba..ad15f64 100644 --- a/tap_salesforce/salesforce/__init__.py +++ b/tap_salesforce/salesforce/__init__.py @@ -87,6 +87,28 @@ 'UndecidedEventRelation', 'OrderStatus']) +# Object name suffixes that are categorically unsupported by the Bulk API. +# This covers dynamically-named objects such as *Share, *Feed, *History, +# and *EventRelation (e.g. PardotEnvironment__Share, CaseHistory, CaseFeed, +# AcceptedEventRelation) which Salesforce reports as queryable=true via REST +# describe but fail at Bulk API prepare-time. +# Ref: https://help.salesforce.com/s/articleView?id=000383508&language=en_US&type=1 + +UNSUPPORTED_BULK_API_SALESFORCE_OBJECT_SUFFIXES = frozenset([ + 'Share', + 'Feed', + 'History', + 'EventRelation', +]) + + +def is_unsupported_bulk_object(name): + """Returns True if an object name is unsupported by the Bulk API because it + matches one of the documented unsupported suffix patterns.""" + return ( + any(name.endswith(suffix) for suffix in UNSUPPORTED_BULK_API_SALESFORCE_OBJECT_SUFFIXES) + ) + # The following objects have certain WHERE clause restrictions so we exclude them. QUERY_RESTRICTED_SALESFORCE_OBJECTS = set(['Announcement', 'ContentDocumentLink', @@ -121,7 +143,8 @@ 'RelatedListColumnDefinition', # A filter on a reified column is required [RelatedListDefinitionId,DurableId], 'RelatedListDefinition', # A filter on a reified column is required [ParentEntityDefinitionId,DurableId], 'ApexTypeImplementor', # A filter on a reified column is required [InterfaceName,DurableId] - 'IconDefinition',]) + 'IconDefinition' + ]) # The following objects are not supported by the query method being used. QUERY_INCOMPATIBLE_SALESFORCE_OBJECTS = set(['DataType', @@ -148,6 +171,7 @@ 'AttachedContentNote', 'QuoteTemplateRichTextData']) + def log_backoff_attempt(details): LOGGER.info("ConnectionError detected, triggering backoff: %d try", details.get("tries")) @@ -506,12 +530,30 @@ def query(self, catalog_entry, state): "api_type should be REST or BULK was: {}".format( self.api_type)) - def get_blacklisted_objects(self): + def get_blacklisted_objects(self, object_names=None): + # 1. Create a base set of blacklisted objects that includes those that are incompatible with both APIs + base_blacklisted_objects = QUERY_RESTRICTED_SALESFORCE_OBJECTS.union(QUERY_INCOMPATIBLE_SALESFORCE_OBJECTS) + + # 2. Filter based on API types if self.api_type == BULK_API_TYPE: - return UNSUPPORTED_BULK_API_SALESFORCE_OBJECTS.union( - QUERY_RESTRICTED_SALESFORCE_OBJECTS).union(QUERY_INCOMPATIBLE_SALESFORCE_OBJECTS) + # Extend the base set with static bulk unsupported objects + bulk_blacklisted_objects = base_blacklisted_objects.union(UNSUPPORTED_BULK_API_SALESFORCE_OBJECTS) + + if object_names: + if isinstance(object_names, str): + object_names = [object_names] + elif not isinstance(object_names, list): + object_names = list(object_names) + # Loop on the object_names and check if they have unsupported suffixes + # if they do, add them to the blacklisted objects set + suffix_blacklisted_objects = {name for name in object_names if is_unsupported_bulk_object(name)} + return bulk_blacklisted_objects.union(suffix_blacklisted_objects) + + return bulk_blacklisted_objects + elif self.api_type == REST_API_TYPE: - return QUERY_RESTRICTED_SALESFORCE_OBJECTS.union(QUERY_INCOMPATIBLE_SALESFORCE_OBJECTS) + return base_blacklisted_objects + else: raise TapSalesforceException( "api_type should be REST or BULK was: {}".format( diff --git a/tests/base.py b/tests/base.py index d93e8fd..48d6c8c 100644 --- a/tests/base.py +++ b/tests/base.py @@ -1123,12 +1123,27 @@ def rest_only_streams(self): 'UndecidedEventRelation', } + # 2026-05-29: *Share, *Feed, *History, and *EventRelation objects are + # unsupported by the Bulk API. Salesforce reports them as queryable=true + # via REST describe, but they fail at Bulk API prepare-time. They are now + # filtered out at discovery when api_type=BULK. See get_blacklisted_objects. + # Ref: https://trailhead.salesforce.com/trailblazer-community/feed/0D54V00007T48HCSAZ + BULK_UNSUPPORTED_SUFFIXES = ('Share', 'Feed', 'History', 'EventRelation') + + def bulk_unsupported_pattern_streams(self, streams): + """Returns the subset of stream names that are unsupported by the Bulk API + due to matching a documented unsupported suffix pattern (*Share, *Feed, + *History, *EventRelation). These are excluded from discovery when + api_type=BULK and should not be included in Bulk API test expectations.""" + return {s for s in streams if s.endswith(self.BULK_UNSUPPORTED_SUFFIXES)} + def expected_streams(self): """A set of expected stream names""" streams = set(self.expected_metadata().keys()) if self.salesforce_api == 'BULK': - return streams.difference(self.rest_only_streams()) + streams = streams.difference(self.rest_only_streams()) + streams = streams.difference(self.bulk_unsupported_pattern_streams(streams)) return streams def child_streams(self): diff --git a/tests/sfbase.py b/tests/sfbase.py index 9ecb81c..300d91a 100644 --- a/tests/sfbase.py +++ b/tests/sfbase.py @@ -82,6 +82,20 @@ def run_and_verify_check_mode(self, conn_id): return found_catalogs + # 2026-05-29: *Share, *Feed, *History, and *EventRelation objects are + # unsupported by the Bulk API. Salesforce reports them as queryable=true + # via REST describe, but they fail at Bulk API prepare-time. They are now + # filtered out at discovery when api_type=BULK. See get_blacklisted_objects. + # Ref: https://trailhead.salesforce.com/trailblazer-community/feed/0D54V00007T48HCSAZ + BULK_UNSUPPORTED_SUFFIXES = ('Share', 'Feed', 'History', 'EventRelation') + + def bulk_unsupported_pattern_streams(self, streams): + """Returns the subset of stream names that are unsupported by the Bulk API + due to matching a documented unsupported suffix pattern (*Share, *Feed, + *History, *EventRelation). These are excluded from discovery when + api_type=BULK and should not be included in Bulk API test expectations.""" + return {s for s in streams if s.endswith(self.BULK_UNSUPPORTED_SUFFIXES)} + @classmethod def expected_stream_names(cls): """A set of expected stream names""" @@ -1146,8 +1160,11 @@ def rest_only_streams(): def expected_stream_names(self): """A set of expected stream names""" streams = set(self.expected_metadata().keys()) + if self.salesforce_api == 'BULK': - return streams.difference(self.rest_only_streams()) + streams = streams.difference(self.rest_only_streams()) + streams = streams.difference(self.bulk_unsupported_pattern_streams(streams)) + return streams def set_replication_methods(self, conn_id, catalogs, replication_methods): diff --git a/tests/unittests/test_bulk_api_object_exclusions.py b/tests/unittests/test_bulk_api_object_exclusions.py new file mode 100644 index 0000000..74f58ab --- /dev/null +++ b/tests/unittests/test_bulk_api_object_exclusions.py @@ -0,0 +1,163 @@ +""" +Unit tests for Bulk API object exclusion logic introduced to filter objects +whose names match the *Share, *Feed, *History, or *EventRelation suffixes at +discovery time, preventing them from being offered to users when they would +fail at Bulk API prepare-time. +""" +import unittest + +from tap_salesforce.salesforce import (UNSUPPORTED_BULK_API_SALESFORCE_OBJECTS, + Salesforce, is_unsupported_bulk_object) + +START_DATE = "2022-01-01T00:00:00.000000Z" + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def make_sf(api_type="BULK"): + return Salesforce(default_start_date=START_DATE, api_type=api_type) + + +# --------------------------------------------------------------------------- +# Tests for is_unsupported_bulk_object() +# --------------------------------------------------------------------------- + +class TestIsUnsupportedBulkObject(unittest.TestCase): + """Tests for the is_unsupported_bulk_object helper function.""" + + # --- suffix: Share --- + def test_share_suffix_standard_object(self): + self.assertTrue(is_unsupported_bulk_object("AccountShare")) + + def test_share_suffix_custom_object(self): + self.assertTrue(is_unsupported_bulk_object("PardotEnvironment__Share")) + + # --- suffix: Feed --- + def test_feed_suffix_standard_object(self): + self.assertTrue(is_unsupported_bulk_object("CaseFeed")) + + def test_feed_suffix_custom_object(self): + self.assertTrue(is_unsupported_bulk_object("MyObject__Feed")) + + # --- suffix: History --- + def test_history_suffix_standard_object(self): + self.assertTrue(is_unsupported_bulk_object("ContactHistory")) + + def test_history_suffix_custom_object(self): + self.assertTrue(is_unsupported_bulk_object("MyObject__History")) + + # --- suffix: EventRelation --- + def test_eventrelation_suffix_standard(self): + self.assertTrue(is_unsupported_bulk_object("AcceptedEventRelation")) + + def test_eventrelation_suffix_declined(self): + self.assertTrue(is_unsupported_bulk_object("DeclinedEventRelation")) + + # --- objects that should NOT be excluded --- + def test_partial_suffix_match_not_excluded(self): + # "SharePoint" ends in "Point", not "Share" — must not be excluded + self.assertFalse(is_unsupported_bulk_object("SharePoint")) + + def test_object_containing_history_not_suffix(self): + # "HistoryLog" contains "History" but does not end with it + self.assertFalse(is_unsupported_bulk_object("HistoryLog")) + +# --------------------------------------------------------------------------- +# Tests for Salesforce.get_blacklisted_objects() — BULK API +# --------------------------------------------------------------------------- + + +class TestGetBlacklistedObjectsBulk(unittest.TestCase): + """get_blacklisted_objects with BULK api_type.""" + + def setUp(self): + self.sf = make_sf(api_type="BULK") + + def test_suffix_share_object_excluded(self): + names = ["Account", "AccountShare", "Contact"] + blacklisted = self.sf.get_blacklisted_objects(object_names=names) + self.assertIn("AccountShare", blacklisted) + + def test_suffix_feed_object_excluded(self): + names = ["Case", "CaseFeed"] + blacklisted = self.sf.get_blacklisted_objects(object_names=names) + self.assertIn("CaseFeed", blacklisted) + + def test_suffix_history_object_excluded(self): + names = ["Contact", "ContactHistory"] + blacklisted = self.sf.get_blacklisted_objects(object_names=names) + self.assertIn("ContactHistory", blacklisted) + + def test_suffix_eventrelation_object_excluded(self): + names = ["AcceptedEventRelation", "DeclinedEventRelation", "UndecidedEventRelation"] + blacklisted = self.sf.get_blacklisted_objects(object_names=names) + self.assertIn("AcceptedEventRelation", blacklisted) + self.assertIn("DeclinedEventRelation", blacklisted) + self.assertIn("UndecidedEventRelation", blacklisted) + + def test_custom_share_object_excluded(self): + names = ["PardotEnvironment__Share", "Account"] + blacklisted = self.sf.get_blacklisted_objects(object_names=names) + self.assertIn("PardotEnvironment__Share", blacklisted) + + def test_static_named_objects_excluded(self): + blacklisted = self.sf.get_blacklisted_objects() + for obj in UNSUPPORTED_BULK_API_SALESFORCE_OBJECTS: + self.assertIn(obj, blacklisted) + + def test_supported_objects_not_excluded(self): + names = ["Account", "Contact", "Opportunity", "Lead"] + blacklisted = self.sf.get_blacklisted_objects(object_names=names) + for name in names: + self.assertNotIn(name, blacklisted) + + def test_no_object_names_returns_static_set(self): + """Without object_names, only static blacklists are returned.""" + blacklisted_with = self.sf.get_blacklisted_objects(object_names=["AccountShare"]) + blacklisted_without = self.sf.get_blacklisted_objects() + # Static entries must be present in both + self.assertTrue(UNSUPPORTED_BULK_API_SALESFORCE_OBJECTS.issubset(blacklisted_without)) + # Pattern-matched entry only present when names provided + self.assertIn("AccountShare", blacklisted_with) + self.assertNotIn("AccountShare", blacklisted_without) + + def test_object_names_as_set_is_accepted(self): + """object_names can be passed as a set (not just a list).""" + names = {"AccountShare", "Contact"} + blacklisted = self.sf.get_blacklisted_objects(object_names=names) + self.assertIn("AccountShare", blacklisted) + self.assertNotIn("Contact", blacklisted) + + +# --------------------------------------------------------------------------- +# Tests for Salesforce.get_blacklisted_objects() — REST API +# --------------------------------------------------------------------------- + +class TestGetBlacklistedObjectsRest(unittest.TestCase): + """get_blacklisted_objects with REST api_type — suffix logic must NOT apply.""" + + def setUp(self): + self.sf = make_sf(api_type="REST") + + def test_share_suffix_not_excluded_for_rest(self): + names = ["AccountShare", "Account"] + blacklisted = self.sf.get_blacklisted_objects(object_names=names) + self.assertNotIn("AccountShare", blacklisted) + + def test_feed_suffix_not_excluded_for_rest(self): + names = ["CaseFeed", "Case"] + blacklisted = self.sf.get_blacklisted_objects(object_names=names) + self.assertNotIn("CaseFeed", blacklisted) + + def test_history_suffix_not_excluded_for_rest(self): + names = ["ContactHistory"] + blacklisted = self.sf.get_blacklisted_objects(object_names=names) + self.assertNotIn("ContactHistory", blacklisted) + + def test_static_bulk_objects_not_in_rest_blacklist(self): + """Objects in UNSUPPORTED_BULK_API_SALESFORCE_OBJECTS are bulk-only exclusions.""" + blacklisted = self.sf.get_blacklisted_objects() + # TaskStatus is bulk-only — it should NOT appear in the REST blacklist + self.assertNotIn("TaskStatus", blacklisted)