TDL-7596: Add tap-tester test for attribution window

.circleci/config.yml

@@ -166,6 +166,18 @@ workflows:
           test_name: "test_facebook_archived_data.py"
           requires:
             - ensure_env
+      - run_integration_tests:
+          context: circleci-user
+          name: "test_facebook_attribution_window.py"
+          test_name: "test_facebook_attribution_window.py"
+          requires:
+            - ensure_env
+      - run_integration_tests:
+          context: circleci-user
+          name: "test_facebook_invalid_attribution_window.py"
+          test_name: "test_facebook_invalid_attribution_window.py"
+          requires:
+            - ensure_env
       - build:
           context: circleci-user
           requires:
@@ -178,6 +190,8 @@ workflows:
             - test_facebook_automatic_fields.py
             - test_facebook_tests_run.py
             - test_facebook_archived_data.py
+            - test_facebook_attribution_window.py
+            - test_facebook_invalid_attribution_window.py
       - deploy:
           context: circleci-user
           requires:
@@ -239,6 +253,18 @@ workflows:
           test_name: "test_facebook_archived_data.py"
           requires:
             - ensure_env
+      - run_integration_tests:
+          context: circleci-user
+          name: "test_facebook_attribution_window.py"
+          test_name: "test_facebook_attribution_window.py"
+          requires:
+            - ensure_env
+      - run_integration_tests:
+          context: circleci-user
+          name: "test_facebook_invalid_attribution_window.py"
+          test_name: "test_facebook_invalid_attribution_window.py"
+          requires:
+            - ensure_env
       - build:
           context: circleci-user
           requires:
@@ -251,6 +277,8 @@ workflows:
             - test_facebook_automatic_fields.py
             - test_facebook_tests_run.py
             - test_facebook_archived_data.py
+            - test_facebook_attribution_window.py
+            - test_facebook_invalid_attribution_window.py
     triggers:
       - schedule:
           cron: "0 6 * * *"

tap_facebook/__init__.py

@@ -588,14 +588,17 @@ def __attrs_post_init__(self):
         if self.options.get('primary-keys'):
             self.key_properties.extend(self.options['primary-keys'])
 
+        self.buffer_days = 28
+        if CONFIG.get('insights_buffer_days'):
+            self.buffer_days = int(CONFIG.get('insights_buffer_days'))
+            # attribution window should only be 1, 7 or 28
+            if self.buffer_days not in [1, 7, 28]:
+                raise Exception("The attribution window must be 1, 7 or 28.")
+
     def job_params(self):
         start_date = get_start(self, self.bookmark_key)
 
-        buffer_days = 28
-        if CONFIG.get('insights_buffer_days'):
-            buffer_days = int(CONFIG.get('insights_buffer_days'))
-
-        buffered_start_date = start_date.subtract(days=buffer_days)
+        buffered_start_date = start_date.subtract(days=self.buffer_days)
         min_start_date = pendulum.today().subtract(months=self.FACEBOOK_INSIGHTS_RETENTION_PERIOD)
         if buffered_start_date < min_start_date:
             LOGGER.warning("%s: Start date is earlier than %s months ago, using %s instead. "

tests/test_facebook_attribution_window.py

@@ -0,0 +1,89 @@
+import os
+
+from tap_tester import runner, connections
+
+from base import FacebookBaseTest
+
+class FacebookAttributionWindow(FacebookBaseTest):
+
+    # set attribution window
+    ATTRIBUTION_WINDOW = 7
+
+    @staticmethod
+    def name():
+        return "tap_tester_facebook_attribution_window"
+
+    def get_properties(self, original: bool = True):
+        """Configuration properties required for the tap."""
+        return_value = {
+            'account_id': os.getenv('TAP_FACEBOOK_ACCOUNT_ID'),
+            'start_date' : '2019-07-24T00:00:00Z',
+            'end_date' : '2019-07-26T00:00:00Z',
+            'insights_buffer_days': str(self.ATTRIBUTION_WINDOW)
+        }
+        if original:
+            return return_value
+
+        return_value["start_date"] = self.start_date
+        return return_value
+
+    def test_run(self):
+        self.run_test(self.ATTRIBUTION_WINDOW) # attribution window: 7
+
+        self.ATTRIBUTION_WINDOW = 28
+        self.run_test(self.ATTRIBUTION_WINDOW) # attribution window: 28
+
+    def run_test(self, attr_window):
+        """
+            Test to check the attribution window
+        """
+
+        conn_id = connections.ensure_connection(self)
+
+        # get start date
+        start_date = self.get_properties()['start_date']
+        # calculate start date with attribution window
+        start_date_with_attribution_window = self.timedelta_formatted(start_date, days=-attr_window)
+
+        # 'attribution window' is only supported for 'ads_insights' streams
+        expected_streams = []
+        for stream in self.expected_streams():
+            if self.is_insight(stream):
+                expected_streams.append(stream)
+
+        # Run in check mode
+        found_catalogs = self.run_and_verify_check_mode(conn_id)
+
+        # Select only the expected streams tables
+        catalog_entries = [ce for ce in found_catalogs if ce['tap_stream_id'] in expected_streams]
+        self.perform_and_verify_table_and_field_selection(conn_id, catalog_entries, select_all_fields=True)
+
+        # Run a sync job using orchestrator
+        self.run_and_verify_sync(conn_id)
+        sync_records = runner.get_records_from_target_output()
+
+        expected_replication_keys = self.expected_replication_keys()
+
+        for stream in expected_streams:
+            with self.subTest(stream=stream):
+
+                replication_key = next(iter(expected_replication_keys[stream]))
+
+                # get records
+                records = [record.get('data') for record in sync_records.get(stream).get('messages')]
+
+                # check for the record is between attribution date and start date
+                is_between = False
+
+                for record in records:
+                    replication_key_value = record.get(replication_key)
+
+                    # Verify the sync records respect the (simulated) start date value
+                    self.assertGreaterEqual(self.parse_date(replication_key_value), self.parse_date(start_date_with_attribution_window),
+                                            msg="The record does not respect the attribution window.")
+
+                    # verify if the record's bookmark value is between start date and (simulated) start date value
+                    if self.parse_date(start_date_with_attribution_window) <= self.parse_date(replication_key_value) < self.parse_date(start_date):
+                        is_between = True
+
+                    self.assertTrue(is_between, msg="No record found between start date and attribution date.")

tests/test_facebook_invalid_attribution_window.py

@@ -0,0 +1,40 @@
+import os
+
+from tap_tester import runner, connections, menagerie
+
+from base import FacebookBaseTest
+
+class FacebookInvalidAttributionWindow(FacebookBaseTest):
+
+    @staticmethod
+    def name():
+        return "tap_tester_facebook_invalid_attribution_window"
+
+    def get_properties(self, original: bool = True):
+        """Configuration properties required for the tap."""
+        return_value = {
+            'account_id': os.getenv('TAP_FACEBOOK_ACCOUNT_ID'),
+            'start_date' : '2019-07-24T00:00:00Z',
+            'end_date' : '2019-07-26T00:00:00Z',
+            'insights_buffer_days': '10' # set attribution window other than 1, 7 or 28
+        }
+        if original:
+            return return_value
+
+        return_value["start_date"] = self.start_date
+        return return_value
+
+    def test_run(self):
+        """
+            Test to verify that the error is raise when passing attribution window other than 1, 7 or 28
+        """
+        # create connection
+        conn_id = connections.ensure_connection(self)
+        # run check mode
+        check_job_name = runner.run_check_mode(self, conn_id)
+        # get exit status
+        exit_status = menagerie.get_exit_status(conn_id, check_job_name)
+        # get discovery error message
+        discovery_error_message = exit_status.get('discovery_error_message')
+        # validate the error message
+        self.assertEquals(discovery_error_message, "The attribution window must be 1, 7 or 28.")

tests/unittests/test_attribution_window.py

@@ -0,0 +1,29 @@
+import unittest
+import tap_facebook.__init__ as tap_facebook
+
+class TestAttributionWindow(unittest.TestCase):
+    """
+        Test case to verify that proper error message is raise
+        when user enters attribution window other than 1, 7 and 28
+    """
+
+    def test_invalid_attribution_window(self):
+        error_message = None
+
+        # set config
+        tap_facebook.CONFIG = {
+            "start_date": "2019-01-01T00:00:00Z",
+            "account_id": "test_account_id",
+            "access_token": "test_access_token",
+            "insights_buffer_days": 30
+        }
+
+        try:
+            # initialize 'AdsInsights' stream as attribution window is only supported in those streams
+            tap_facebook.AdsInsights("test", "test", "test", None, {}, {})
+        except Exception as e:
+            # save error message for assertion
+            error_message = str(e)
+
+        # verify the error message was as expected
+        self.assertEquals(error_message, "The attribution window must be 1, 7 or 28.")