diff --git a/splitio/tasks/events_sync.py b/splitio/tasks/events_sync.py index bddcfd2c..b6b374e6 100644 --- a/splitio/tasks/events_sync.py +++ b/splitio/tasks/events_sync.py @@ -2,13 +2,39 @@ import logging from splitio.tasks import BaseSynchronizationTask -from splitio.tasks.util.asynctask import AsyncTask +from splitio.tasks.util.asynctask import AsyncTask, AsyncTaskAsync _LOGGER = logging.getLogger(__name__) -class EventsSyncTask(BaseSynchronizationTask): +class EventsSyncTaskBase(BaseSynchronizationTask): + """Events synchronization task base uses an asynctask.AsyncTask to send events.""" + + def start(self): + """Start executing the events synchronization task.""" + self._task.start() + + def stop(self, event=None): + """Stop executing the events synchronization task.""" + pass + + def flush(self): + """Flush events in storage.""" + _LOGGER.debug('Forcing flush execution for events') + self._task.force_execution() + + def is_running(self): + """ + Return whether the task is running or not. + + :return: True if the task is running. False otherwise. + :rtype: bool + """ + return self._task.running() + + +class EventsSyncTask(EventsSyncTaskBase): """Events synchronization task uses an asynctask.AsyncTask to send events.""" def __init__(self, synchronize_events, period): @@ -24,24 +50,27 @@ def __init__(self, synchronize_events, period): self._period = period self._task = AsyncTask(synchronize_events, self._period, on_stop=synchronize_events) - def start(self): - """Start executing the events synchronization task.""" - self._task.start() - def stop(self, event=None): """Stop executing the events synchronization task.""" self._task.stop(event) - def flush(self): - """Flush events in storage.""" - _LOGGER.debug('Forcing flush execution for events') - self._task.force_execution() - def is_running(self): +class EventsSyncTaskAsync(EventsSyncTaskBase): + """Events synchronization task uses an asynctask.AsyncTaskAsync to send events.""" + + def __init__(self, synchronize_events, period): """ - Return whether the task is running or not. + Class constructor. + + :param synchronize_events: Events Api object to send data to the backend + :type synchronize_events: splitio.api.events.EventsAPIAsync + :param period: How many seconds to wait between subsequent event pushes to the BE. + :type period: int - :return: True if the task is running. False otherwise. - :rtype: bool """ - return self._task.running() + self._period = period + self._task = AsyncTaskAsync(synchronize_events, self._period, on_stop=synchronize_events) + + async def stop(self, event=None): + """Stop executing the events synchronization task.""" + await self._task.stop() diff --git a/tests/tasks/test_events_sync.py b/tests/tasks/test_events_sync.py index 24f4173a..b2ea500d 100644 --- a/tests/tasks/test_events_sync.py +++ b/tests/tasks/test_events_sync.py @@ -2,12 +2,15 @@ import threading import time +import pytest + from splitio.api.client import HttpResponse from splitio.tasks import events_sync from splitio.storage import EventStorage from splitio.models.events import Event from splitio.api.events import EventsAPI -from splitio.sync.event import EventSynchronizer +from splitio.sync.event import EventSynchronizer, EventSynchronizerAsync +from splitio.optional.loaders import asyncio class EventsSyncTests(object): @@ -40,3 +43,47 @@ def test_normal_operation(self, mocker): stop_event.wait(5) assert stop_event.is_set() assert len(api.flush_events.mock_calls) > calls_now + + +class EventsSyncAsyncTests(object): + """Impressions Syncrhonization task async test cases.""" + + @pytest.mark.asyncio + async def test_normal_operation(self, mocker): + """Test that the task works properly under normal circumstances.""" + self.events = [ + Event('key1', 'user', 'purchase', 5.3, 123456, None), + Event('key2', 'user', 'purchase', 5.3, 123456, None), + Event('key3', 'user', 'purchase', 5.3, 123456, None), + Event('key4', 'user', 'purchase', 5.3, 123456, None), + Event('key5', 'user', 'purchase', 5.3, 123456, None), + ] + storage = mocker.Mock(spec=EventStorage) + self.called = False + async def pop_many(*args): + self.called = True + return self.events + storage.pop_many = pop_many + + api = mocker.Mock(spec=EventsAPI) + self.flushed_events = None + self.count = 0 + async def flush_events(events): + self.count += 1 + self.flushed_events = events + return HttpResponse(200, '', {}) + api.flush_events = flush_events + + event_synchronizer = EventSynchronizerAsync(api, storage, 5) + task = events_sync.EventsSyncTaskAsync(event_synchronizer.synchronize_events, 1) + task.start() + await asyncio.sleep(2) + + assert task.is_running() + assert self.called + assert self.flushed_events == self.events + + calls_now = self.count + await task.stop() + assert not task.is_running() + assert self.count > calls_now