Merge branch 'feature/streaming' of github.com:splitio/python-client into tests/push_integration

Author: mredolatti

splitio/push/segmentworker.py

@@ -40,7 +40,11 @@ def _run(self):
                 continue
             _LOGGER.debug('Processing segment_update: %s, change_number: %d',
                           event.segment_name, event.change_number)
-            self._handler(event.segment_name, event.change_number)
+            try:
+                self._handler(event.segment_name, event.change_number)
+            except Exception:
+                _LOGGER.error('Exception raised in segment synchronization')
+                _LOGGER.debug('Exception information: ', exc_info=True)
 
     def start(self):
         """Start worker."""

splitio/push/splitworker.py

@@ -39,7 +39,11 @@ def _run(self):
             if event == self._centinel:
                 continue
             _LOGGER.debug('Processing split_update %d', event.change_number)
-            self._handler(event.change_number)
+            try:
+                self._handler(event.change_number)
+            except Exception:
+                _LOGGER.error('Exception raised in split synchronization')
+                _LOGGER.debug('Exception information: ', exc_info=True)
 
     def start(self):
         """Start worker."""

splitio/tasks/impressions_sync.py

@@ -56,7 +56,7 @@ def flush(self):
 class ImpressionsCountSyncTask(BaseSynchronizationTask):
     """Impressions synchronization task uses an asynctask.AsyncTask to send impressions."""
 
-    _PERIOD = 5  # 30 * 60 # 30 minutes
+    _PERIOD = 1800  # 30 * 60 # 30 minutes
 
     def __init__(self, synchronize_counters):
         """

tests/push/test_segment_worker.py

@@ -1,7 +1,9 @@
 """Split Worker tests."""
 import time
 import queue
+import pytest
 
+from splitio.api import APIException
 from splitio.push.segmentworker import SegmentWorker
 from splitio.models.notification import SegmentChangeNotification
 
@@ -18,20 +20,41 @@ def handler_sync(segment_name, change_number):
 
 
 class SegmentWorkerTests(object):
-    q = queue.Queue()
-    segment_worker = SegmentWorker(handler_sync, q)
+    def test_on_error(self):
+        q = queue.Queue()
+
+        def handler_sync(change_number):
+            raise APIException('some')
+
+        segment_worker = SegmentWorker(handler_sync, q)
+        segment_worker.start()
+        assert segment_worker.is_running()
+
+        q.put(SegmentChangeNotification('some', 'SEGMENT_UPDATE', 123456789, 'some'))
+
+        with pytest.raises(Exception):
+            segment_worker._handler()
+
+        assert segment_worker.is_running()
+        assert segment_worker._worker.is_alive()
+        segment_worker.stop()
+        time.sleep(1)
+        assert not segment_worker.is_running()
+        assert not segment_worker._worker.is_alive()
 
     def test_handler(self):
+        q = queue.Queue()
+        segment_worker = SegmentWorker(handler_sync, q)
         global change_number_received
-        assert not self.segment_worker.is_running()
-        self.segment_worker.start()
-        assert self.segment_worker.is_running()
+        assert not segment_worker.is_running()
+        segment_worker.start()
+        assert segment_worker.is_running()
 
-        self.q.put(SegmentChangeNotification('some', 'SEGMENT_UPDATE', 123456789, 'some'))
+        q.put(SegmentChangeNotification('some', 'SEGMENT_UPDATE', 123456789, 'some'))
 
         time.sleep(0.1)
         assert change_number_received == 123456789
         assert segment_name_received == 'some'
 
-        self.segment_worker.stop()
-        assert not self.segment_worker.is_running()
+        segment_worker.stop()
+        assert not segment_worker.is_running()

tests/push/test_split_worker.py

@@ -1,32 +1,57 @@
 """Split Worker tests."""
 import time
 import queue
+import pytest
 
+from splitio.api import APIException
 from splitio.push.splitworker import SplitWorker
 from splitio.models.notification import SplitChangeNotification
 
 change_number_received = None
 
+
 def handler_sync(change_number):
     global change_number_received
     change_number_received = change_number
     return
 
 
 class SplitWorkerTests(object):
-    q = queue.Queue()
-    split_worker = SplitWorker(handler_sync, q)
+
+    def test_on_error(self):
+        q = queue.Queue()
+
+        def handler_sync(change_number):
+            raise APIException('some')
+
+        split_worker = SplitWorker(handler_sync, q)
+        split_worker.start()
+        assert split_worker.is_running()
+
+        q.put(SplitChangeNotification('some', 'SPLIT_UPDATE', 123456789))
+        with pytest.raises(Exception):
+            split_worker._handler()
+
+        assert split_worker.is_running()
+        assert split_worker._worker.is_alive()
+        split_worker.stop()
+        time.sleep(1)
+        assert not split_worker.is_running()
+        assert not split_worker._worker.is_alive()
 
     def test_handler(self):
+        q = queue.Queue()
+        split_worker = SplitWorker(handler_sync, q)
+
         global change_number_received
-        assert self.split_worker.is_running() == False
-        self.split_worker.start()
-        assert self.split_worker.is_running() == True
+        assert not split_worker.is_running()
+        split_worker.start()
+        assert split_worker.is_running()
 
-        self.q.put(SplitChangeNotification('some', 'SPLIT_UPDATE', 123456789))
+        q.put(SplitChangeNotification('some', 'SPLIT_UPDATE', 123456789))
 
         time.sleep(0.1)
         assert change_number_received == 123456789
 
-        self.split_worker.stop()
-        assert self.split_worker.is_running() == False
+        split_worker.stop()
+        assert not split_worker.is_running()