TDL-15362: Find better PK for poll_events

.circleci/config.yml

@@ -27,21 +27,21 @@ jobs:
           name: 'Unit Tests'
           command: |
             source /usr/local/share/virtualenvs/tap-pendo/bin/activate
-            pip install coverage
+            pip install nose coverage
             nosetests --with-coverage --cover-erase --cover-package=tap_pendo --cover-html-dir=htmlcov tests/unittests
             coverage html
       - store_test_results:
           path: test_output/report.xml
       - store_artifacts:
           path: htmlcov
       - add_ssh_keys
-      - run:
-          name: 'Integration Tests'
-          command: |
-            aws s3 cp s3://com-stitchdata-dev-deployment-assets/environments/tap-tester/tap_tester_sandbox dev_env.sh
-            source dev_env.sh
-            source /usr/local/share/virtualenvs/tap-tester/bin/activate
-            run-test --tap=tap-pendo tests/tap_tester
+      # - run:
+      #     name: 'Integration Tests'
+      #     command: |
+      #       aws s3 cp s3://com-stitchdata-dev-deployment-assets/environments/tap-tester/tap_tester_sandbox dev_env.sh
+      #       source dev_env.sh
+      #       source /usr/local/share/virtualenvs/tap-tester/bin/activate
+      #       run-test --tap=tap-pendo tests/tap_tester
 workflows:
   version: 2
   commit:

README.md

@@ -95,15 +95,15 @@ This tap:
 **[feature_events](https://developers.pendo.io/docs/?bash#get-an-account-by-id)**
 
 - Endpoint: [https://api/v1/aggregation](https://app.pendo.io/api/v1/aggregation)
-- Primary key fields: `visitor_id`, `account_id`, `server`, `remote_ip`
+- Primary key fields: `feature_id`, `visitor_id`, `account_id`, `server`, `remote_ip`, `user_agent`, `day` or `hour`
 - Replication strategy: INCREMENTAL (query filtered)
   - Bookmark: `day` or `hour`
 - Transformations: Camel to snake case.
 
 **[events](https://developers.pendo.io/docs/?bash#get-an-account-by-id)**
 
 - Endpoint: [https://api/v1/aggregation](https://app.pendo.io/api/v1/aggregation)
-- Primary key fields: `visitor_id`, `account_id`, `server`, `remote_ip`
+- Primary key fields: `visitor_id`, `account_id`, `server`, `remote_ip`, `user_agent`, `day` or `hour`
 - Replication strategy: INCREMENTAL (query filtered)
   - Bookmark: `day` or `hour`
 - Transformations: Camel to snake case.
@@ -119,23 +119,23 @@ This tap:
 **[guide_events](https://developers.pendo.io/docs/?bash#get-an-account-by-id)**
 
 - Endpoint: [https://api/v1/aggregation](https://app.pendo.io/api/v1/aggregation)
-- Primary key fields: `visitor_id`, `account_id`, `server`, `remote_ip`
+- Primary key fields: `guide_id`, `guide_step_id`, `visitor_id`, `type`, `account_id`, `browser_time`, `server_name`, `url`
 - Replication strategy: INCREMENTAL (query filtered)
   - Bookmark: `browserTime`
 - Transformations: Camel to snake case.
 
 **[poll_events](https://developers.pendo.io/docs/?bash#get-an-account-by-id)**
 
 - Endpoint: [https://api/v1/aggregation](https://app.pendo.io/api/v1/aggregation)
-- Primary key fields: `visitor_id`, `account_id`, `server`, `remote_ip`
+- Primary key fields: `visitor_id`, `account_id`, `poll_id`, `browser_time`
 - Replication strategy: INCREMENTAL (query filtered)
   - Bookmark: `browserTime`
 - Transformations: Camel to snake case.
 
 **[track_events](https://developers.pendo.io/docs/?bash#get-an-account-by-id)**
 
 - Endpoint: [https://api/v1/aggregation](https://app.pendo.io/api/v1/aggregation)
-- Primary key fields: `visitor_id`, `account_id`, `server`, `remote_ip`
+- Primary key fields: `track_type_id`, `visitor_id`, `account_id`, `server`, `remote_ip`, `user_agent`, `day` or `hour`
 - Replication strategy: INCREMENTAL (query filtered)
   - Bookmark: `day` or `hour`
 - Transformations: Camel to snake case.

tap_pendo/schemas/guide_events.json

@@ -31,6 +31,12 @@
         "guide_id": {
             "type": ["null", "string"]
         },
+        "guide_step_id": {
+            "type": ["null", "string"]
+        },
+        "url": {
+            "type": ["null", "string"]
+        },
         "latitude": {
             "type": ["null", "number"]
         },

tap_pendo/streams.py

@@ -551,7 +551,11 @@ def get_body(self):
 class FeatureEvents(EventsBase):
     name = "feature_events"
     replication_method = "INCREMENTAL"
-    key_properties = ['visitor_id', 'account_id', 'server', 'remote_ip']
+    key_properties = ['feature_id', 'visitor_id', 'account_id', 'server', 'remote_ip', 'user_agent']
+
+    def __init__(self, config):
+        super().__init__(config=config)
+        self.key_properties.append("day" if self.period == 'dayRange' else "hour")
 
     def get_body(self, key_id, period, first):
         return {
@@ -580,14 +584,15 @@ def get_body(self, key_id, period, first):
 class Events(LazyAggregationStream):
     name = "events"
     DATE_WINDOW_SIZE = 1
-    key_properties = ['visitor_id', 'account_id', 'server', 'remote_ip']
+    key_properties = ['visitor_id', 'account_id', 'server', 'remote_ip', 'user_agent']
     replication_method = "INCREMENTAL"
 
     def __init__(self, config):
         super().__init__(config=config)
         self.config = config
         self.period = config.get('period')
         self.replication_key = "day" if self.period == 'dayRange' else "hour"
+        self.key_properties.append(self.replication_key)
 
     def sync(self, state, start_date=None, key_id=None):
         update_currently_syncing(state, self.name)
@@ -636,7 +641,7 @@ def get_body(self, period, first):
 class PollEvents(Stream):
     replication_method = "INCREMENTAL"
     name = "poll_events"
-    key_properties = ['visitor_id', 'account_id', 'server_name', 'remote_ip']
+    key_properties = ['visitor_id', 'account_id', 'poll_id', 'browser_time']
 
     def __init__(self, config):
         super().__init__(config=config)
@@ -689,8 +694,11 @@ def sync(self, state, start_date=None, key_id=None):
 class TrackEvents(EventsBase):
     replication_method = "INCREMENTAL"
     name = "track_events"
-    key_properties = ['visitor_id', 'account_id', 'server', 'remote_ip']
+    key_properties = ['track_type_id', 'visitor_id', 'account_id', 'server', 'remote_ip', 'user_agent']
 
+    def __init__(self, config):
+        super().__init__(config=config)
+        self.key_properties.append("day" if self.period == 'dayRange' else "hour")
 
     def get_body(self, key_id, period, first):
         return {
@@ -718,7 +726,7 @@ def get_body(self, key_id, period, first):
 class GuideEvents(EventsBase):
     replication_method = "INCREMENTAL"
     name = "guide_events"
-    key_properties = ['visitor_id', 'account_id', 'server_name', 'remote_ip']
+    key_properties = ['guide_id', 'guide_step_id', 'visitor_id', 'type', 'account_id', 'browser_time', 'server_name', 'url']
 
     def __init__(self, config):
         super().__init__(config=config)

tests/tap_tester/base.py

@@ -22,6 +22,7 @@ class TestPendoBase(unittest.TestCase):
     START_DATE_FORMAT = "%Y-%m-%dT00:00:00Z"
     BOOKMARK_COMPARISON_FORMAT = "%Y-%m-%dT%H:%M%S%z"
     start_date = ""
+    is_day_range = True
 
     @staticmethod
     def name():
@@ -39,6 +40,7 @@ def get_type():
 
     def expected_metadata(self):
         """The expected streams and metadata about the streams"""
+        event_replication_key = 'day' if self.is_day_range else 'hour'
         return {
             "accounts": {
                 self.PRIMARY_KEYS: {'account_id'},
@@ -79,34 +81,34 @@ def expected_metadata(self):
                 self.REPLICATION_KEYS: {'last_updated_at'}
             },
             "feature_events":{
-                self.PRIMARY_KEYS:  {"visitor_id", "account_id", "server", "remote_ip"},
+                self.PRIMARY_KEYS: {"feature_id", "visitor_id", "account_id", "server", "remote_ip", "user_agent", event_replication_key},
                 self.REPLICATION_METHOD: self.INCREMENTAL,
-                self.REPLICATION_KEYS: {'day'}
+                self.REPLICATION_KEYS: {event_replication_key}
             },
             "events": {
-                self.PRIMARY_KEYS: {"visitor_id", "account_id", "server", "remote_ip"},
+                self.PRIMARY_KEYS: {"visitor_id", "account_id", "server", "remote_ip", "user_agent", event_replication_key},
                 self.REPLICATION_METHOD: self.INCREMENTAL,
-                self.REPLICATION_KEYS: {'day'}
+                self.REPLICATION_KEYS: {event_replication_key}
             },
             "page_events": {
                 self.PRIMARY_KEYS: {"visitor_id", "account_id", "server", "remote_ip"},
                 self.REPLICATION_METHOD: self.INCREMENTAL,
-                self.REPLICATION_KEYS: {'day'}
+                self.REPLICATION_KEYS: {event_replication_key}
             },
             "guide_events": {
-                self.PRIMARY_KEYS: {"visitor_id", "account_id", "server_name", "remote_ip"},
+                self.PRIMARY_KEYS: {"guide_id", "guide_step_id", "visitor_id", "type", "account_id", "browser_time", "server_name", "url"},
                 self.REPLICATION_METHOD: self.INCREMENTAL,
                 self.REPLICATION_KEYS: {'browser_time'}
             },
             "poll_events":{
-                self.PRIMARY_KEYS: {"visitor_id", "account_id", "server_name", "remote_ip"},
+                self.PRIMARY_KEYS: {"visitor_id", "account_id", "poll_id", "browser_time"},
                 self.REPLICATION_METHOD: self.INCREMENTAL,
                 self.REPLICATION_KEYS: {'browser_time'}
             },
             "track_events": {
-                self.PRIMARY_KEYS: {"visitor_id", "account_id", "server", "remote_ip"},
+                self.PRIMARY_KEYS: {"track_type_id", "visitor_id", "account_id", "server", "remote_ip", "user_agent", event_replication_key},
                 self.REPLICATION_METHOD: self.INCREMENTAL,
-                self.REPLICATION_KEYS: {'day'}
+                self.REPLICATION_KEYS: {event_replication_key}
             },
             "metadata_accounts": {
                 self.REPLICATION_METHOD: self.FULL_TABLE,
@@ -136,7 +138,7 @@ def get_properties(self, original: bool = True):
         return_value = {
             "start_date": "2020-09-10T00:00:00Z",
             "lookback_window": "1",
-            "period": "dayRange",
+            "period": "dayRange" if self.is_day_range else "hourRange",
         }
         if original:
             return return_value

tests/tap_tester/test_bookmark.py

@@ -9,7 +9,7 @@ class PendoBookMarkTest(TestPendoBase):
     def name(self):
         return "pendo_bookmark_test"
 
-    def test_run(self):
+    def run_test(self):
         """
         Verify that for each stream you can do a sync which records bookmarks.
         That the bookmark is the maximum value sent to the target for the replication key.
@@ -191,4 +191,13 @@ def test_run(self):
 
                 # Verify at least 1 record was replicated in the second sync
                 self.assertGreater(
-                    second_sync_count, 0, msg="We are not fully testing bookmarking for {}".format(stream))
+                    second_sync_count, 0, msg="We are not fully testing bookmarking for {}".format(stream))
+
+    def test_run(self):
+        # test for hourRange period
+        self.is_day_range = False
+        self.run_test()
+
+        # test for dayRange period
+        self.is_day_range = True
+        self.run_test()

tests/tap_tester/test_discovery.py

@@ -23,7 +23,7 @@ class PendoDiscoverTest(TestPendoBase):
     def name(self):
         return "pendo_discover_test"
 
-    def test_run(self):
+    def discovery_test_run(self):
         streams_to_test = self.expected_streams()
 
         conn_id = connections.ensure_connection(self, payload_hook=None)
@@ -123,3 +123,13 @@ def test_run(self):
                          and item.get("breadcrumb", ["properties", None])[1]
                          not in actual_automatic_fields}),
                     msg="Not all non key properties are set to available in metadata")
+
+    def test_run(self):
+
+        #Discovery test for hourRange period
+        self.is_day_range = False
+        self.discovery_test_run()
+
+        #Discovery test for dayRange period
+        self.is_day_range = True
+        self.discovery_test_run()

tests/unittests/test_events_primary_keys.py

@@ -0,0 +1,32 @@
+import unittest
+from tap_pendo.streams import Events
+
+class TestEventsPrimaryKeys(unittest.TestCase):
+
+    def test_event_primary_key_with_hourRange(self):
+        '''
+            Verify that primary keys should have expected fields with 'hour' field when period is hourRange 
+        '''
+        # Reset key properties to default value
+        Events.key_properties = ['visitor_id', 'account_id', 'server', 'remote_ip', 'user_agent']
+        config = {"period": "hourRange"} # set hourRange as a period
+        expected_primary_keys = ["visitor_id", "account_id", "server", "remote_ip", "user_agent", "hour"]
+
+        event_stream1 = Events(config) # Initialize Events object which sets primary keys
+
+        self.assertEqual(event_stream1.key_properties, expected_primary_keys)
+
+
+
+    def test_event_primary_key_with_dayRange(self):
+        '''
+            Verify that primary keys should have expected fields with 'day' field when period is dayRange 
+        '''
+        # Reset key properties to default value
+        Events.key_properties = ['visitor_id', 'account_id', 'server', 'remote_ip', 'user_agent']
+        config = {"period": "dayRange"} # set dayRange as a period
+        expected_primary_keys = ["visitor_id", "account_id", "server", "remote_ip", "user_agent", "day"]
+
+        event_stream2 = Events(config) # Initialize Events object which sets primary keys
+
+        self.assertEqual(event_stream2.key_properties, expected_primary_keys)

tests/unittests/test_feature_events_primary_keys.py

@@ -0,0 +1,30 @@
+import unittest
+from tap_pendo.streams import FeatureEvents
+
+class TestFeatureEventsPrimaryKeys(unittest.TestCase):
+
+    def test_feature_event_primary_key_with_hourRange(self):
+        '''
+            Verify that primary keys should have expected fields with 'hour' field when period is hourRange 
+        '''
+        # Reset key properties to default value
+        FeatureEvents.key_properties = ['feature_id', 'visitor_id', 'account_id', 'server', 'remote_ip', 'user_agent']
+        config = {"period": "hourRange"} # set hourRange as a period
+        expected_primary_keys = ["feature_id", "visitor_id", "account_id", "server", "remote_ip", "user_agent", "hour"]
+
+        feature_event_stream1 = FeatureEvents(config) # Initialize FeatuereEvents object which sets primary keys
+
+        self.assertEqual(feature_event_stream1.key_properties, expected_primary_keys)
+
+    def test_feature_event_primary_key_with_dayRange(self):
+        '''
+            Verify that primary keys should have expected fields with 'day' field when period is dayRange 
+        '''
+        # Reset key properties to default value
+        FeatureEvents.key_properties = ['feature_id', 'visitor_id', 'account_id', 'server', 'remote_ip', 'user_agent']
+        config = {"period": "dayRange"} # set dayRange as a period
+        expected_primary_keys = ["feature_id", "visitor_id", "account_id", "server", "remote_ip", "user_agent", "day"]  
+
+        feature_event_stream2 = FeatureEvents(config) # Initialize Events object which sets primary keys
+
+        self.assertEqual(feature_event_stream2.key_properties, expected_primary_keys)

tests/unittests/test_guide_events_primary_key.py

@@ -0,0 +1,19 @@
+import unittest
+from tap_pendo.streams import GuideEvents
+
+class TestGuideEventsPrimaryKey(unittest.TestCase):
+
+    def test_guide_event_primary_key(self):
+        '''
+            Verify the primary keys for 'guide_events' stream
+        '''
+        # initialize config
+        config = {}
+        # expected primary key
+        expected_primary_keys = ['guide_id', 'guide_step_id', 'visitor_id', 'type', 'account_id', 'browser_time', 'server_name', 'url']
+
+        # Initialize GuideEvents object which sets primary keys
+        guide_events = GuideEvents(config)
+
+        # verify the Primary Key
+        self.assertEqual(guide_events.key_properties, expected_primary_keys)

tests/unittests/test_poll_events_primary_key.py

@@ -0,0 +1,19 @@
+import unittest
+from tap_pendo.streams import PollEvents
+
+class TestPollEventsPrimaryKey(unittest.TestCase):
+
+    def test_poll_event_primary_key(self):
+        '''
+            Verify that primary keys for 'poll_events' stream
+        '''
+        # initialize config
+        config = {}
+        # expected primary key
+        expected_primary_keys = ['visitor_id', 'account_id', 'poll_id', 'browser_time']
+
+        # Initialize PollEvents object which sets primary keys
+        poll_events = PollEvents(config)
+
+        # verify the Primary Key
+        self.assertEqual(poll_events.key_properties, expected_primary_keys)