Merge pull request #428 from splitio/async-tasks-events

added tasks.event_sync async class

Author: chillaq

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()

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