Merge pull request #429 from splitio/async-takss-impressions

added tasks.sync imps async classes

Author: chillaq

splitio/tasks/impressions_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 ImpressionsSyncTask(BaseSynchronizationTask):
+class ImpressionsSyncTaskBase(BaseSynchronizationTask):
+    """Impressions synchronization task uses an asynctask.AsyncTask to send impressions."""
+
+    def start(self):
+        """Start executing the impressions synchronization task."""
+        self._task.start()
+
+    def stop(self, event=None):
+        """Stop executing the impressions synchronization task."""
+        pass
+
+    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()
+
+    def flush(self):
+        """Flush impressions in storage."""
+        _LOGGER.debug('Forcing flush execution for impressions')
+        self._task.force_execution()
+
+
+class ImpressionsSyncTask(ImpressionsSyncTaskBase):
     """Impressions synchronization task uses an asynctask.AsyncTask to send impressions."""
 
     def __init__(self, synchronize_impressions, period):
@@ -25,13 +51,45 @@ def __init__(self, synchronize_impressions, period):
         self._task = AsyncTask(synchronize_impressions, self._period,
                                on_stop=synchronize_impressions)
 
+    def stop(self, event=None):
+        """Stop executing the impressions synchronization task."""
+        self._task.stop(event)
+
+
+class ImpressionsSyncTaskAsync(ImpressionsSyncTaskBase):
+    """Impressions synchronization task uses an asynctask.AsyncTask to send impressions."""
+
+    def __init__(self, synchronize_impressions, period):
+        """
+        Class constructor.
+
+        :param synchronize_impressions: sender
+        :type synchronize_impressions: func
+        :param period: How many seconds to wait between subsequent impressions pushes to the BE.
+        :type period: int
+
+        """
+        self._period = period
+        self._task = AsyncTaskAsync(synchronize_impressions, self._period,
+                               on_stop=synchronize_impressions)
+
+    async def stop(self, event=None):
+        """Stop executing the impressions synchronization task."""
+        await self._task.stop()
+
+
+class ImpressionsCountSyncTaskBase(BaseSynchronizationTask):
+    """Impressions synchronization task uses an asynctask.AsyncTask to send impressions."""
+
+    _PERIOD = 1800  # 30 * 60 # 30 minutes
+
     def start(self):
         """Start executing the impressions synchronization task."""
         self._task.start()
 
     def stop(self, event=None):
         """Stop executing the impressions synchronization task."""
-        self._task.stop(event)
+        pass
 
     def is_running(self):
         """
@@ -44,15 +102,12 @@ def is_running(self):
 
     def flush(self):
         """Flush impressions in storage."""
-        _LOGGER.debug('Forcing flush execution for impressions')
         self._task.force_execution()
 
 
-class ImpressionsCountSyncTask(BaseSynchronizationTask):
+class ImpressionsCountSyncTask(ImpressionsCountSyncTaskBase):
     """Impressions synchronization task uses an asynctask.AsyncTask to send impressions."""
 
-    _PERIOD = 1800  # 30 * 60 # 30 minutes
-
     def __init__(self, synchronize_counters):
         """
         Class constructor.
@@ -63,23 +118,24 @@ def __init__(self, synchronize_counters):
         """
         self._task = AsyncTask(synchronize_counters, self._PERIOD, on_stop=synchronize_counters)
 
-    def start(self):
-        """Start executing the impressions synchronization task."""
-        self._task.start()
-
     def stop(self, event=None):
         """Stop executing the impressions synchronization task."""
         self._task.stop(event)
 
-    def is_running(self):
+
+class ImpressionsCountSyncTaskAsync(ImpressionsCountSyncTaskBase):
+    """Impressions synchronization task uses an asynctask.AsyncTask to send impressions."""
+
+    def __init__(self, synchronize_counters):
         """
-        Return whether the task is running or not.
+        Class constructor.
+
+        :param synchronize_counters: Handler
+        :type synchronize_counters: func
 
-        :return: True if the task is running. False otherwise.
-        :rtype: bool
         """
-        return self._task.running()
+        self._task = AsyncTaskAsync(synchronize_counters, self._PERIOD, on_stop=synchronize_counters)
 
-    def flush(self):
-        """Flush impressions in storage."""
-        self._task.force_execution()
+    async def stop(self, event=None):
+        """Stop executing the impressions synchronization task."""
+        await self._task.stop()

tests/tasks/test_impressions_sync.py

@@ -2,15 +2,18 @@
 
 import threading
 import time
+import pytest
+
 from splitio.api.client import HttpResponse
 from splitio.tasks import impressions_sync
 from splitio.storage import ImpressionStorage
 from splitio.models.impressions import Impression
 from splitio.api.impressions import ImpressionsAPI
-from splitio.sync.impression import ImpressionSynchronizer, ImpressionsCountSynchronizer
+from splitio.sync.impression import ImpressionSynchronizer, ImpressionsCountSynchronizer, ImpressionSynchronizerAsync, ImpressionsCountSynchronizerAsync
 from splitio.engine.impressions.manager import Counter
+from splitio.optional.loaders import asyncio
 
-class ImpressionsSyncTests(object):
+class ImpressionsSyncTaskTests(object):
     """Impressions Syncrhonization task test cases."""
 
     def test_normal_operation(self, mocker):
@@ -44,7 +47,52 @@ def test_normal_operation(self, mocker):
         assert len(api.flush_impressions.mock_calls) > calls_now
 
 
-class ImpressionsCountSyncTests(object):
+class ImpressionsSyncTaskAsyncTests(object):
+    """Impressions Syncrhonization task test cases."""
+
+    @pytest.mark.asyncio
+    async def test_normal_operation(self, mocker):
+        """Test that the task works properly under normal circumstances."""
+        storage = mocker.Mock(spec=ImpressionStorage)
+        impressions = [
+            Impression('key1', 'split1', 'on', 'l1', 123456, 'b1', 321654),
+            Impression('key2', 'split1', 'on', 'l1', 123456, 'b1', 321654),
+            Impression('key3', 'split2', 'off', 'l1', 123456, 'b1', 321654),
+            Impression('key4', 'split2', 'on', 'l1', 123456, 'b1', 321654),
+            Impression('key5', 'split3', 'off', 'l1', 123456, 'b1', 321654)
+        ]
+        self.pop_called = 0
+        async def pop_many(*args):
+            self.pop_called += 1
+            return impressions
+        storage.pop_many = pop_many
+
+        api = mocker.Mock(spec=ImpressionsAPI)
+        self.flushed = None
+        self.called = 0
+        async def flush_impressions(imps):
+            self.called += 1
+            self.flushed = imps
+            return  HttpResponse(200, '', {})
+        api.flush_impressions = flush_impressions
+
+        impression_synchronizer = ImpressionSynchronizerAsync(api, storage, 5)
+        task = impressions_sync.ImpressionsSyncTaskAsync(
+            impression_synchronizer.synchronize_impressions,
+            1
+        )
+        task.start()
+        await asyncio.sleep(2)
+        assert task.is_running()
+        assert self.pop_called == 1
+        assert self.flushed == impressions
+
+        calls_now = self.called
+        await task.stop()
+        assert self.called > calls_now
+
+
+class ImpressionsCountSyncTaskTests(object):
     """Impressions Syncrhonization task test cases."""
 
     def test_normal_operation(self, mocker):
@@ -77,3 +125,48 @@ def test_normal_operation(self, mocker):
         stop_event.wait(5)
         assert stop_event.is_set()
         assert len(api.flush_counters.mock_calls) > calls_now
+
+
+class ImpressionsCountSyncTaskAsyncTests(object):
+    """Impressions Syncrhonization task test cases."""
+
+    @pytest.mark.asyncio
+    async def test_normal_operation(self, mocker):
+        """Test that the task works properly under normal circumstances."""
+        counter = mocker.Mock(spec=Counter)
+        counters = [
+            Counter.CountPerFeature('f1', 123, 2),
+            Counter.CountPerFeature('f2', 123, 123),
+            Counter.CountPerFeature('f1', 456, 111),
+            Counter.CountPerFeature('f2', 456, 222)
+        ]
+        self._pop_called = 0
+        async def pop_all():
+            self._pop_called += 1
+            return counters
+        counter.pop_all = pop_all
+
+        api = mocker.Mock(spec=ImpressionsAPI)
+        self.flushed = None
+        self.called = 0
+        async def flush_counters(imps):
+            self.called += 1
+            self.flushed = imps
+            return  HttpResponse(200, '', {})
+        api.flush_counters = flush_counters
+
+        impressions_sync.ImpressionsCountSyncTaskAsync._PERIOD = 1
+        impression_synchronizer = ImpressionsCountSynchronizerAsync(api, counter)
+        task = impressions_sync.ImpressionsCountSyncTaskAsync(
+            impression_synchronizer.synchronize_counters
+        )
+        task.start()
+        await asyncio.sleep(2)
+        assert task.is_running()
+
+        assert self._pop_called == 1
+        assert self.flushed == counters
+
+        calls_now = self.called
+        await task.stop()
+        assert self.called > calls_now