TDL-15168: Use unique_line_item_id for invoice updates' lines value instead of id

tap_stripe/__init__.py

@@ -705,6 +705,42 @@ def sync_sub_stream(sub_stream_name, parent_obj, updates=False):
             obj_ad_dict = sub_stream_obj.to_dict_recursive()
 
             if sub_stream_name == "invoice_line_items":
+                # we will get "unique_id" for default API versions older than "2019-12-03"
+                # ie. for API version older than "2019-12-03", the value in the field
+                # "unique_id" is moved to "id" field in the newer API version
+                # For example:
+                #   OLDER API VERSION
+                #     {
+                #         "id": "ii_testinvoiceitem",
+                #         "object": "line_item",
+                #         "invoice_item": "ii_testinvoiceitem",
+                #         "subscription": "sub_testsubscription",
+                #         "type": "invoiceitem",
+                #         "unique_id": "il_testlineitem"
+                #     }
+
+                #   NEWER API VERSION
+                #     {
+                #         "id": "il_testlineitem",
+                #         "object": "line_item",
+                #         "invoice_item": "ii_testinvoiceitem",
+                #         "subscription": "sub_testsubscription",
+                #         "type": "invoiceitem",
+                #     }
+                if updates and obj_ad_dict.get("unique_id"):
+                    # get "unique_id"
+                    object_unique_id = obj_ad_dict.get("unique_id")
+                    # get "id"
+                    object_id = obj_ad_dict.get("id")
+                    # update "id" field with "unique_id" value
+                    obj_ad_dict["id"] = object_unique_id
+                    # if type is invoiceitem, update 'invoice_item' field with 'id' if not present
+                    if obj_ad_dict.get("type") == "invoiceitem" and not obj_ad_dict.get("invoice_item"):
+                        obj_ad_dict["invoice_item"] = object_id
+                    # if type is subscription, update 'subscription' field with 'id' if not present
+                    elif obj_ad_dict.get("type") == "subscription" and not obj_ad_dict.get("subscription"):
+                        obj_ad_dict["subscription"] = object_id
+
                 # Synthetic addition of a key to the record we sync
                 obj_ad_dict["invoice"] = parent_obj.id
             elif sub_stream_name == "payout_transactions":

tests/test_all_fields.py

@@ -165,6 +165,28 @@
     'payment_intents': set()
 }
 
+# we have observed that the SDK object creation returns some new fields intermittently
+SCHEMA_MISSING_FIELDS = {
+    'customers': {
+        'test_clock'
+    },
+    'subscriptions': {
+        'test_clock',
+    },
+    'products':set(),
+    'invoice_items':{
+        'test_clock',
+    },
+    'payouts':set(),
+    'charges': set(),
+    'subscription_items': set(),
+    'plans': set(),
+    'invoice_line_items': set(),
+    'invoices': {
+        'test_clock',
+    }
+}
+
 KNOWN_FAILING_FIELDS = {
     'coupons': {
         'percent_off', # BUG_9720 | Decimal('67') != Decimal('66.6') (value is changing in duplicate records)
@@ -254,7 +276,10 @@
 # As for the `price` field added in the schema, the API doc doesn't mention any
 # `trial_period_days` in the field, hence skipping the assertion error for the same.
 KNOWN_NESTED_MISSING_FIELDS = {
-    'subscription_items': {'price': 'recurring.trial_period_days'}
+    'subscription_items': {'price': 'recurring.trial_period_days'},
+    'charges': {'payment_method_details': 'card.mandate'},
+    'payment_intents': {'charges': 'payment_method_details.card.mandate',
+                        'payment_method_options': 'card.mandate_options'}
 }
 
 class ALlFieldsTest(BaseTapTest):
@@ -373,6 +398,19 @@ def find_nested_key(self, nested_key, actual_field_value, field):
                         if keys[-1] in temp_value:
                             return True
 
+    def handle_list_data(self, expected_field_value, field, nested_key):
+        """
+        Find the nested key that is failing in the list and ignore the assertion error, if any.
+        """
+        is_fickle = True
+        for each_expected_field_value in expected_field_value:
+            if self.find_nested_key(nested_key, each_expected_field_value, field):
+                continue
+            else:
+                is_fickle = False
+                break
+        return is_fickle
+
     def all_fields_test(self, streams_to_test):
         """
         Verify that for each stream data is synced when all fields are selected.
@@ -456,7 +494,6 @@ def all_fields_test(self, streams_to_test):
                 adjusted_actual_keys = actual_records_keys.union(  # BUG_12478
                     KNOWN_MISSING_FIELDS.get(stream, set())
                 ).union(SCHEMA_MISSING_FIELDS.get(stream, set()))
-
                 if stream == 'invoice_items':
                     adjusted_actual_keys = adjusted_actual_keys.union({'subscription_item'})  # BUG_13666
 
@@ -548,8 +585,14 @@ def all_fields_test(self, streams_to_test):
                                         f"AssertionError({failure_1})")
 
                                     nested_key = KNOWN_NESTED_MISSING_FIELDS.get(stream, {})
-                                    if self.find_nested_key(nested_key, expected_field_value, field):
-                                        continue
+                                    # Check whether expected_field_value is list or not.
+                                    # If expected_field_value is list then loop through each item of list
+                                    if type(expected_field_value) == list:
+                                        if self.handle_list_data(expected_field_value, field, nested_key):
+                                            continue
+                                    else:
+                                        if self.find_nested_key(nested_key, expected_field_value, field):
+                                            continue
 
                                     if field in KNOWN_FAILING_FIELDS[stream] or field in FIELDS_TO_NOT_CHECK[stream]:
                                         continue # skip the following wokaround