added telemetry sync task async class and tests

Author: chillaq

splitio/tasks/telemetry_sync.py

@@ -7,7 +7,7 @@
 _LOGGER = logging.getLogger(__name__)
 
 class TelemetrySyncTaskBase(BaseSynchronizationTask):
-    """Unique Keys synchronization task uses an asynctask.AsyncTask to send MTKs."""
+    """Telemetry synchronization task uses an asynctask.AsyncTask to send MTKs."""
 
     def start(self):
         """Start executing the telemetry synchronization task."""
@@ -32,8 +32,8 @@ def flush(self):
         self._task.force_execution()
 
 
-class TelemetrySyncTask(BaseSynchronizationTask):
-    """Unique Keys synchronization task uses an asynctask.AsyncTask to send MTKs."""
+class TelemetrySyncTask(TelemetrySyncTaskBase):
+    """Unique Telemetry task uses an asynctask.AsyncTask to send MTKs."""
 
     def __init__(self, synchronize_telemetry, period):
         """
@@ -53,8 +53,8 @@ def stop(self, event=None):
         self._task.stop(event)
 
 
-class TelemetrySyncTaskAsync(BaseSynchronizationTask):
-    """Unique Keys synchronization task uses an asynctask.AsyncTask to send MTKs."""
+class TelemetrySyncTaskAsync(TelemetrySyncTaskBase):
+    """Telemetry synchronization task uses an asynctask.AsyncTask to send MTKs."""
 
     def __init__(self, synchronize_telemetry, period):
         """

splitio/tasks/util/asynctask.py

@@ -218,7 +218,7 @@ def __init__(self, main, period, on_init=None, on_stop=None):
         self._period = period
         self._messages = asyncio.Queue()
         self._running = False
-        self._completion_event = None
+        self._completion_event = asyncio.Event()
 
     async def _execution_wrapper(self):
         """
@@ -284,7 +284,7 @@ def start(self):
             _LOGGER.warning("Task is already running. Ignoring .start() call")
             return
         # Start execution
-        self._completion_event = asyncio.Event()
+        self._completion_event.clear()
         asyncio.get_running_loop().create_task(self._execution_wrapper())
 
     async def stop(self, wait_for_completion=False):

tests/tasks/test_telemetry_sync.py

@@ -0,0 +1,63 @@
+"""Impressions synchronization task test module."""
+import pytest
+import threading
+import time
+from splitio.api.client import HttpResponse
+from splitio.tasks.telemetry_sync import TelemetrySyncTask, TelemetrySyncTaskAsync
+from splitio.api.telemetry import TelemetryAPI, TelemetryAPIAsync
+from splitio.sync.telemetry import TelemetrySynchronizer, TelemetrySynchronizerAsync, InMemoryTelemetrySubmitter, InMemoryTelemetrySubmitterAsync
+from splitio.storage.inmemmory import InMemoryTelemetryStorage, InMemoryTelemetryStorageAsync
+from splitio.engine.telemetry import TelemetryStorageConsumer, TelemetryStorageConsumerAsync
+from splitio.optional.loaders import asyncio
+
+
+class TelemetrySyncTaskTests(object):
+    """Unique Keys Syncrhonization task test cases."""
+
+    def test_record_stats(self, mocker):
+        """Test that the task works properly under normal circumstances."""
+        api = mocker.Mock(spec=TelemetryAPI)
+        api.record_stats.return_value = HttpResponse(200, '', {})
+        telemetry_storage = InMemoryTelemetryStorage()
+        telemetry_consumer = TelemetryStorageConsumer(telemetry_storage)
+
+        telemetry_synchronizer = TelemetrySynchronizer(InMemoryTelemetrySubmitter(telemetry_consumer, mocker.Mock(), mocker.Mock(),api))
+        task = TelemetrySyncTask(telemetry_synchronizer.synchronize_stats, 1)
+        task.start()
+        time.sleep(2)
+        assert task.is_running()
+        assert len(api.record_stats.mock_calls) == 1
+        stop_event = threading.Event()
+        task.stop(stop_event)
+        stop_event.wait(5)
+        assert stop_event.is_set()
+
+
+class TelemetrySyncTaskAsyncTests(object):
+    """Unique Keys Syncrhonization task test cases."""
+
+    @pytest.mark.asyncio
+    async def test_record_stats(self, mocker):
+        """Test that the task works properly under normal circumstances."""
+        api = mocker.Mock(spec=TelemetryAPIAsync)
+        self.called = False
+        async def record_stats(stats):
+            self.called = True
+            return HttpResponse(200, '', {})
+        api.record_stats = record_stats
+
+        telemetry_storage = await InMemoryTelemetryStorageAsync.create()
+        telemetry_consumer = TelemetryStorageConsumerAsync(telemetry_storage)
+        telemetry_submitter = InMemoryTelemetrySubmitterAsync(telemetry_consumer, mocker.Mock(), mocker.Mock(),api)
+        async def _build_stats():
+            return {}
+        telemetry_submitter._build_stats = _build_stats
+
+        telemetry_synchronizer = TelemetrySynchronizerAsync(telemetry_submitter)
+        task = TelemetrySyncTaskAsync(telemetry_synchronizer.synchronize_stats, 1)
+        task.start()
+        await asyncio.sleep(2)
+        assert task.is_running()
+        assert self.called
+        await task.stop()
+        assert not task.is_running()