Merge pull request #430 from splitio/sync-tasks-split

added tasks.sync split async class

Author: chillaq

splitio/tasks/split_sync.py

@@ -2,14 +2,36 @@
 
 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 SplitSynchronizationTask(BaseSynchronizationTask):
+class SplitSynchronizationTaskBase(BaseSynchronizationTask):
     """Split Synchronization task class."""
+
+    def start(self):
+        """Start the task."""
+        self._task.start()
+
+    def stop(self, event=None):
+        """Stop the task. Accept an optional event to set when the task has finished."""
+        pass
+
+    def is_running(self):
+        """
+        Return whether the task is running.
+
+        :return: True if the task is running. False otherwise.
+        :rtype bool
+        """
+        return self._task.running()
+
+
+class SplitSynchronizationTask(SplitSynchronizationTaskBase):
+    """Split Synchronization task class."""
+
     def __init__(self, synchronize_splits, period):
         """
         Class constructor.
@@ -22,19 +44,26 @@ def __init__(self, synchronize_splits, period):
         self._period = period
         self._task = AsyncTask(synchronize_splits, period, on_init=None)
 
-    def start(self):
-        """Start the task."""
-        self._task.start()
-
     def stop(self, event=None):
         """Stop the task. Accept an optional event to set when the task has finished."""
         self._task.stop(event)
 
-    def is_running(self):
+
+class SplitSynchronizationTaskAsync(SplitSynchronizationTaskBase):
+    """Split Synchronization async task class."""
+
+    def __init__(self, synchronize_splits, period):
         """
-        Return whether the task is running.
+        Class constructor.
 
-        :return: True if the task is running. False otherwise.
-        :rtype bool
+        :param synchronize_splits: Handler
+        :type synchronize_splits: func
+        :param period: Period of task
+        :type period: int
         """
-        return self._task.running()
+        self._period = period
+        self._task = AsyncTaskAsync(synchronize_splits, period, on_init=None)
+
+    async def stop(self, event=None):
+        """Stop the task. Accept an optional event to set when the task has finished."""
+        await self._task.stop()

tests/tasks/test_split_sync.py

@@ -1,13 +1,50 @@
 """Split syncrhonization task test module."""
-
 import threading
 import time
+import pytest
+
 from splitio.api import APIException
 from splitio.api.commons import FetchOptions
 from splitio.tasks import split_sync
 from splitio.storage import SplitStorage
 from splitio.models.splits import Split
-from splitio.sync.split import SplitSynchronizer
+from splitio.sync.split import SplitSynchronizer, SplitSynchronizerAsync
+from splitio.optional.loaders import asyncio
+
+splits = [{
+    'changeNumber': 123,
+    'trafficTypeName': 'user',
+    'name': 'some_name',
+    'trafficAllocation': 100,
+    'trafficAllocationSeed': 123456,
+    'seed': 321654,
+    'status': 'ACTIVE',
+    'killed': False,
+    'defaultTreatment': 'off',
+    'algo': 2,
+    'conditions': [
+        {
+            'partitions': [
+                {'treatment': 'on', 'size': 50},
+                {'treatment': 'off', 'size': 50}
+            ],
+            'contitionType': 'WHITELIST',
+            'label': 'some_label',
+            'matcherGroup': {
+                'matchers': [
+                    {
+                        'matcherType': 'WHITELIST',
+                        'whitelistMatcherData': {
+                            'whitelist': ['k1', 'k2', 'k3']
+                        },
+                        'negate': False,
+                    }
+                ],
+                'combiner': 'AND'
+            }
+        }
+    ]
+}]
 
 
 class SplitSynchronizationTests(object):
@@ -26,40 +63,6 @@ def change_number_mock():
         storage.get_change_number.side_effect = change_number_mock
 
         api = mocker.Mock()
-        splits = [{
-           'changeNumber': 123,
-           'trafficTypeName': 'user',
-           'name': 'some_name',
-           'trafficAllocation': 100,
-           'trafficAllocationSeed': 123456,
-           'seed': 321654,
-           'status': 'ACTIVE',
-           'killed': False,
-           'defaultTreatment': 'off',
-           'algo': 2,
-           'conditions': [
-               {
-                   'partitions': [
-                       {'treatment': 'on', 'size': 50},
-                       {'treatment': 'off', 'size': 50}
-                   ],
-                   'contitionType': 'WHITELIST',
-                   'label': 'some_label',
-                   'matcherGroup': {
-                       'matchers': [
-                           {
-                               'matcherType': 'WHITELIST',
-                               'whitelistMatcherData': {
-                                   'whitelist': ['k1', 'k2', 'k3']
-                               },
-                               'negate': False,
-                           }
-                       ],
-                       'combiner': 'AND'
-                   }
-               }
-            ]
-        }]
 
         def get_changes(*args, **kwargs):
             get_changes.called += 1
@@ -120,3 +123,88 @@ def run(x):
         time.sleep(1)
         assert task.is_running()
         task.stop()
+
+
+class SplitSynchronizationAsyncTests(object):
+    """Split synchronization task async test cases."""
+
+    @pytest.mark.asyncio
+    async def test_normal_operation(self, mocker):
+        """Test the normal operation flow."""
+        storage = mocker.Mock(spec=SplitStorage)
+
+        async def change_number_mock():
+            change_number_mock._calls += 1
+            if change_number_mock._calls == 1:
+                return -1
+            return 123
+        change_number_mock._calls = 0
+        storage.get_change_number = change_number_mock
+
+        api = mocker.Mock()
+        self.change_number = []
+        self.fetch_options = []
+        async def get_changes(change_number, fetch_options):
+            self.change_number.append(change_number)
+            self.fetch_options.append(fetch_options)
+            get_changes.called += 1
+            if get_changes.called == 1:
+                return {
+                    'splits': splits,
+                    'since': -1,
+                    'till': 123
+                }
+            else:
+                return {
+                    'splits': [],
+                    'since': 123,
+                    'till': 123
+                }
+        api.fetch_splits = get_changes
+        get_changes.called = 0
+        self.inserted_split = None
+        async def put(split):
+            self.inserted_split = split
+        storage.put = put
+
+        fetch_options = FetchOptions(True)
+        split_synchronizer = SplitSynchronizerAsync(api, storage)
+        task = split_sync.SplitSynchronizationTaskAsync(split_synchronizer.synchronize_splits, 0.5)
+        task.start()
+        await asyncio.sleep(0.7)
+        assert task.is_running()
+        await task.stop()
+        assert not task.is_running()
+        assert (self.change_number[0], self.fetch_options[0])  == (-1, fetch_options)
+        assert (self.change_number[1], self.fetch_options[1])  == (123, fetch_options)
+        assert isinstance(self.inserted_split, Split)
+        assert self.inserted_split.name == 'some_name'
+
+    @pytest.mark.asyncio
+    async def test_that_errors_dont_stop_task(self, mocker):
+        """Test that if fetching splits fails at some_point, the task will continue running."""
+        storage = mocker.Mock(spec=SplitStorage)
+        api = mocker.Mock()
+
+        async def run(x):
+            run._calls += 1
+            if run._calls == 1:
+                return {'splits': [], 'since': -1, 'till': -1}
+            if run._calls == 2:
+                return {'splits': [], 'since': -1, 'till': -1}
+            raise APIException("something broke")
+        run._calls = 0
+        api.fetch_splits = run
+
+        async def get_change_number():
+            return -1
+        storage.get_change_number = get_change_number
+
+        split_synchronizer = SplitSynchronizerAsync(api, storage)
+        task = split_sync.SplitSynchronizationTaskAsync(split_synchronizer.synchronize_splits, 0.5)
+        task.start()
+        await asyncio.sleep(0.1)
+        assert task.is_running()
+        await asyncio.sleep(1)
+        assert task.is_running()
+        await task.stop()