TDL-15566: Data loss of child streams

.circleci/config.yml

@@ -11,7 +11,7 @@ jobs:
             python3 -mvenv /usr/local/share/virtualenvs/tap-pendo
             source /usr/local/share/virtualenvs/tap-pendo/bin/activate
             pip install -U pip setuptools
-            pip install .[dev]
+            pip install .[test]
       - run:
           name: 'JSON Validator'
           command: |
@@ -23,6 +23,17 @@ jobs:
             source /usr/local/share/virtualenvs/tap-pendo/bin/activate
             # TODO: Adjust the pylint disables
             pylint tap_pendo --disable 'broad-except,chained-comparison,empty-docstring,fixme,invalid-name,line-too-long,missing-class-docstring,missing-function-docstring,missing-module-docstring,no-else-raise,no-else-return,too-few-public-methods,too-many-arguments,too-many-branches,too-many-lines,too-many-locals,ungrouped-imports,wrong-spelling-in-comment,wrong-spelling-in-docstring,bad-whitespace,missing-class-docstring'
+      - run:
+          name: 'Unit Tests'
+          command: |
+            source /usr/local/share/virtualenvs/tap-pendo/bin/activate
+            pip install 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'

setup.py

@@ -17,9 +17,12 @@
         'ijson==3.1.4',
     ],
     extras_require={
-        'dev': [
-            'ipdb==0.11',
+        'test': [
             'pylint==2.5.3',
+            'nose'
+        ],
+        'dev': [
+            'ipdb==0.11'
         ]
     },
     entry_points="""

tap_pendo/streams.py

@@ -406,7 +406,8 @@ def sync_substream(self, state, parent, sub_stream, parent_response):
                             integer_datetime_fmt=
                             "unix-milliseconds-integer-datetime-parsing"
                         ) as transformer:
-                    stream_events = sub_stream.sync(state, new_bookmark,
+                    # syncing child streams from start date or state file date
+                    stream_events = sub_stream.sync(state, bookmark_dttm,
                                                     record.get(parent.key_properties[0]))
                     for event in stream_events:
                         counter.increment()
@@ -928,12 +929,10 @@ def get_params(self, start_time):
     def sync(self, state, start_date=None, key_id=None):
         update_currently_syncing(state, self.name)
 
-        bookmark_date = self.get_bookmark(state, self.name,
-                                          self.config.get('start_date'),
-                                          self.replication_key)
-        bookmark_dttm = strptime_to_utc(bookmark_date)
-
-        abs_start, abs_end = get_absolute_start_end_time(bookmark_dttm)
+        # using "start_date" that is passed and not using the bookmark
+        # value stored in the state file, as it will be updated after
+        # every sync of child stream for parent stream
+        abs_start, abs_end = get_absolute_start_end_time(start_date)
         lookback = abs_start - timedelta(days=self.lookback_window())
         window_next = lookback