updated localhost classes and tests

Author: chillaq

splitio/client/factory.py

@@ -1145,11 +1145,11 @@ def _build_localhost_factory(cfg):
     telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
     telemetry_evaluation_producer = telemetry_producer.get_telemetry_evaluation_producer()
 
-    events_queue = queue.Queue()
+    internal_events_queue = queue.Queue()
     storages = {
-        'splits': InMemorySplitStorage(events_queue, cfg['flagSetsFilter'] if cfg['flagSetsFilter'] is not None else []),
-        'segments': InMemorySegmentStorage(events_queue),  # not used, just to avoid possible future errors.
-        'rule_based_segments': InMemoryRuleBasedSegmentStorage(events_queue),   
+        'splits': InMemorySplitStorage(internal_events_queue, cfg['flagSetsFilter'] if cfg['flagSetsFilter'] is not None else []),
+        'segments': InMemorySegmentStorage(internal_events_queue),  # not used, just to avoid possible future errors.
+        'rule_based_segments': InMemoryRuleBasedSegmentStorage(internal_events_queue),   
         'impressions': LocalhostImpressionsStorage(),
         'events': LocalhostEventsStorage(),
     }
@@ -1162,6 +1162,8 @@ def _build_localhost_factory(cfg):
         LocalSegmentSynchronizer(cfg['segmentDirectory'], storages['splits'], storages['segments']),
         None, None, None,
     )
+    events_manager = EventsManager(EventsManagerConfig(), EventsDelivery())
+    internal_events_task = EventsTask(events_manager.notify_internal_event, internal_events_queue)
 
     feature_flag_sync_task = None
     segment_sync_task = None
@@ -1178,6 +1180,7 @@ def _build_localhost_factory(cfg):
         feature_flag_sync_task,
         segment_sync_task,
         None, None, None,
+        internal_events_task=internal_events_task
     )
 
     sdk_metadata = util.get_metadata(cfg)
@@ -1199,8 +1202,7 @@ def _build_localhost_factory(cfg):
         telemetry_evaluation_producer,
         telemetry_runtime_producer
     )
-    internal_events_queue = queue.Queue()
-    events_manager = EventsManager(EventsManagerConfig(), EventsDelivery())
+    internal_events_task.start()
 
     return SplitFactory(
         'localhost',
@@ -1226,6 +1228,8 @@ async def _build_localhost_factory_async(cfg):
 
     internal_events_queue = asyncio.Queue()
     events_manager = EventsManagerAsync(EventsManagerConfig(), EventsDelivery())
+    internal_events_task = EventsTaskAsync(events_manager.notify_internal_event, internal_events_queue)
+
     storages = {
         'splits': InMemorySplitStorageAsync(internal_events_queue),
         'segments': InMemorySegmentStorageAsync(internal_events_queue),  # not used, just to avoid possible future errors.
@@ -1258,6 +1262,7 @@ async def _build_localhost_factory_async(cfg):
         feature_flag_sync_task,
         segment_sync_task,
         None, None, None,
+        internal_events_task=internal_events_task
     )
 
     sdk_metadata = util.get_metadata(cfg)
@@ -1277,8 +1282,9 @@ async def _build_localhost_factory_async(cfg):
         storages['impressions'],
         telemetry_evaluation_producer,
         telemetry_runtime_producer
-    )
-        
+    )        
+    internal_events_task.start()
+
     return SplitFactoryAsync(
         'localhost',
         storages,

splitio/sync/synchronizer.py

@@ -955,12 +955,13 @@ def sync_all(self, till=None):
 
     def stop_periodic_fetching(self):
         """Stop fetchers for feature flags and segments."""
+        _LOGGER.debug('Stopping periodic fetching')
         if self._split_tasks.split_task is not None:
-            _LOGGER.debug('Stopping periodic fetching')
             self._split_tasks.split_task.stop()
         if self._split_tasks.segment_task is not None:
             self._split_tasks.segment_task.stop()
         if self._split_tasks.internal_events_task:
+            _LOGGER.debug('Stopping internal events notification')
             self._split_tasks.internal_events_task.stop()
 
     def synchronize_splits(self):
@@ -1031,12 +1032,15 @@ async def sync_all(self, till=None):
 
     async def stop_periodic_fetching(self):
         """Stop fetchers for feature flags and segments."""
+        _LOGGER.debug('Stopping periodic fetching')
         if self._split_tasks.split_task is not None:
-            _LOGGER.debug('Stopping periodic fetching')
             await self._split_tasks.split_task.stop()
         if self._split_tasks.segment_task is not None:
-            await self._split_tasks.segment_task.stop()
-
+            await self._split_tasks.segment_task.stop()        
+        if self._split_tasks.internal_events_task is not None:
+            _LOGGER.debug('Stopping internal events notification')
+            await self._split_tasks.internal_events_task.stop()
+        
     async def synchronize_splits(self):
         """Synchronize all feature flags."""
         try:

tests/integration/test_streaming_e2e.py

@@ -1367,6 +1367,9 @@ def test_change_number(mocker):
 class StreamingIntegrationAsyncTests(object):
     """Test streaming operation and failover."""
 
+    update_flag = False
+    metadata = []
+
     @pytest.mark.asyncio
     async def test_happiness(self):
         """Test initialization & splits/segment updates."""
@@ -1421,6 +1424,7 @@ async def test_happiness(self):
 
         factory = await get_factory_async('some_apikey', **kwargs)
         await factory.block_until_ready(1)
+        await factory.client().on(SdkEvent.SDK_UPDATE, self._update_callcack)
         assert factory.ready
         assert await factory.client().get_treatment('maldo', 'split1') == 'on'
 
@@ -1437,6 +1441,13 @@ async def test_happiness(self):
             'rbs': {'t': -1, 's': -1, 'd': []}}
         sse_server.publish(make_split_change_event(2))
         await asyncio.sleep(1)
+        flag = False
+        for meta in self.metadata:
+            if 'split1' in meta.get_names():
+                assert meta.get_type() == SdkEventType.FLAG_UPDATE
+                flag = True
+        assert flag
+        
         assert await factory.client().get_treatment('maldo', 'split1') == 'off'
 
         split_changes[2] = {'ff': {
@@ -1556,6 +1567,10 @@ async def test_happiness(self):
         sse_server.stop()
         split_backend.stop()
 
+    async def _update_callcack(self, metadata):
+        self.update_flag = True
+        self.metadata.append(metadata)
+
     @pytest.mark.asyncio
     async def test_occupancy_flicker(self):
         """Test that changes in occupancy switch between polling & streaming properly."""

tests/sync/test_synchronizer.py

@@ -210,7 +210,6 @@ def intersect(sets):
                                                  mocker.Mock(), mocker.Mock())
 
         synchronizer = Synchronizer(split_synchronizers, mocker.Mock(spec=SplitTasks))
-#        pytest.set_trace()
         self.clear = False
         def clear():
             self.clear = True