Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion splitio/push/segmentworker.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down
6 changes: 5 additions & 1 deletion splitio/push/splitworker.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down
2 changes: 1 addition & 1 deletion splitio/tasks/impressions_sync.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
"""
Expand Down
39 changes: 31 additions & 8 deletions tests/push/test_segment_worker.py
Original file line number Diff line number Diff line change
@@ -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

Expand All @@ -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()
41 changes: 33 additions & 8 deletions tests/push/test_split_worker.py
Original file line number Diff line number Diff line change
@@ -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()