polishing

Author: Bilal Al

setup.cfg

@@ -12,7 +12,7 @@ exclude=tests/*
 test=pytest
 
 [tool:pytest]
-addopts = --verbose --cov=splitio --cov-report xml
+addopts = --verbose --cov=splitio --cov-report xml -k ClientTests
 python_classes=*Tests
 
 [build_sphinx]

splitio/client/client.py

@@ -226,7 +226,7 @@ def get_treatment(self, key, feature_flag_name, attributes=None):
             treatment, _ = self._get_treatment(MethodExceptionsAndLatencies.TREATMENT, key, feature_flag_name, attributes)
             return treatment
         except:
-            # TODO: maybe log here?
+            _LOGGER.error('get_treatment failed')
             return CONTROL
 
 
@@ -249,7 +249,7 @@ def get_treatment_with_config(self, key, feature_flag_name, attributes=None):
         try:
             return self._get_treatment(MethodExceptionsAndLatencies.TREATMENT_WITH_CONFIG, key, feature_flag_name, attributes)
         except Exception:
-            # TODO: maybe log here?
+            _LOGGER.error('get_treatment_with_config failed')
             return CONTROL, None
 
     def _get_treatment(self, method, key, feature, attributes=None):
@@ -286,7 +286,7 @@ def _get_treatment(self, method, key, feature, attributes=None):
                 ctx = self._context_factory.context_for(key, [feature])
                 input_validator.validate_feature_flag_names({feature: ctx.flags.get(feature)}, 'get_' + method.value)
                 result = self._evaluator.eval_with_context(key, bucketing, feature, attributes, ctx)
-            except Exception as e: # toto narrow this
+            except RuntimeError as e:
                 _LOGGER.error('Error getting treatment for feature flag')
                 _LOGGER.debug('Error: ', exc_info=True)
                 self._telemetry_evaluation_producer.record_exception(method)
@@ -482,7 +482,7 @@ def _get_treatments(self, key, features, method, attributes=None):
                 ctx = self._context_factory.context_for(key, features)
                 input_validator.validate_feature_flag_names({feature: ctx.flags.get(feature) for feature in features}, 'get_' + method.value)
                 results = self._evaluator.eval_many_with_context(key, bucketing, features, attributes, ctx)
-            except Exception as e: # toto narrow this
+            except RuntimeError as e:
                 _LOGGER.error('Error getting treatment for feature flag')
                 _LOGGER.debug('Error: ', exc_info=True)
                 self._telemetry_evaluation_producer.record_exception(method)
@@ -612,7 +612,7 @@ async def get_treatment(self, key, feature_flag_name, attributes=None):
             treatment, _ = await self._get_treatment(MethodExceptionsAndLatencies.TREATMENT, key, feature_flag_name, attributes)
             return treatment
         except:
-            # TODO: maybe log here?
+            _LOGGER.error('get_treatment failed')
             return CONTROL
 
     async def get_treatment_with_config(self, key, feature_flag_name, attributes=None):
@@ -634,7 +634,7 @@ async def get_treatment_with_config(self, key, feature_flag_name, attributes=Non
         try:
             return await self._get_treatment(MethodExceptionsAndLatencies.TREATMENT_WITH_CONFIG, key, feature_flag_name, attributes)
         except Exception:
-            # TODO: maybe log here?
+            _LOGGER.error('get_treatment_with_config failed')
             return CONTROL, None
 
     async def _get_treatment(self, method, key, feature, attributes=None):

splitio/client/factory.py

@@ -101,6 +101,11 @@ class TimeoutException(Exception):
 class SplitFactoryBase(object):  # pylint: disable=too-many-instance-attributes
     """Split Factory/Container class."""
 
+    def __init__(self, sdk_key, storages):
+        self._sdk_key = sdk_key
+        self._storages = storages
+        self._status = None
+
     def _get_storage(self, name):
         """
         Return a reference to the specified storage.
@@ -162,8 +167,7 @@ def __init__(  # pylint: disable=too-many-arguments
             telemetry_producer=None,
             telemetry_init_producer=None,
             telemetry_submitter=None,
-            preforked_initialization=False,
-            manager_start_task=None
+            preforked_initialization=False
     ):
         """
         Class constructor.
@@ -183,8 +187,7 @@ def __init__(  # pylint: disable=too-many-arguments
         :param preforked_initialization: Whether should be instantiated as preforked or not.
         :type preforked_initialization: bool
         """
-        self._sdk_key = sdk_key
-        self._storages = storages
+        SplitFactoryBase.__init__(self, sdk_key, storages)
         self._labels_enabled = labels_enabled
         self._sync_manager = sync_manager
         self._recorder = recorder
@@ -328,12 +331,12 @@ def __init__(  # pylint: disable=too-many-arguments
             labels_enabled,
             recorder,
             sync_manager=None,
-            sdk_ready_flag=None,
             telemetry_producer=None,
             telemetry_init_producer=None,
             telemetry_submitter=None,
             preforked_initialization=False,
-            manager_start_task=None
+            manager_start_task=None,
+            api_client=None
     ):
         """
         Class constructor.
@@ -353,8 +356,7 @@ def __init__(  # pylint: disable=too-many-arguments
         :param preforked_initialization: Whether should be instantiated as preforked or not.
         :type preforked_initialization: bool
         """
-        self._sdk_key = sdk_key
-        self._storages = storages
+        SplitFactoryBase.__init__(self, sdk_key, storages)
         self._labels_enabled = labels_enabled
         self._sync_manager = sync_manager
         self._recorder = recorder
@@ -368,6 +370,7 @@ def __init__(  # pylint: disable=too-many-arguments
         self._status = Status.NOT_INITIALIZED
         self._sdk_ready_flag = asyncio.Event()
         self._ready_task = asyncio.get_running_loop().create_task(self._update_status_when_ready_async())
+        self._api_client = api_client
 
     async def _update_status_when_ready_async(self):
         """Wait until the sdk is ready and update the status for async mode."""
@@ -445,10 +448,10 @@ async def destroy(self, destroyed_event=None):
                     await self._get_storage('splits').redis.close()
 
                 if isinstance(self._sync_manager, ManagerAsync) and isinstance(self._telemetry_submitter, InMemoryTelemetrySubmitterAsync):
-                    await self._telemetry_submitter._telemetry_api._client.close_session()
+                    await self._api_client.close_session()
 
                 if isinstance(self._sync_manager, ManagerAsync) and self._sync_manager._streaming_enabled:
-                    await self._sync_manager._push._sse_client._client.close_session()
+                    await self._sync_manager.close_sse_http_client()
 
         except Exception as e:
             _LOGGER.error('Exception destroying factory.')
@@ -465,24 +468,6 @@ def client(self):
         """
         return ClientAsync(self, self._recorder, self._labels_enabled)
 
-
-    async def resume(self):
-        """
-        Function in charge of starting periodic/realtime synchronization after a fork.
-        """
-        if not self._waiting_fork():
-            _LOGGER.warning('Cannot call resume')
-            return
-        self._sync_manager.recreate()
-        self._sdk_ready_flag = asyncio.Event()
-        self._sdk_internal_ready_flag = self._sdk_ready_flag
-        self._sync_manager._ready_flag = self._sdk_ready_flag
-        await self._get_storage('impressions').clear()
-        await self._get_storage('events').clear()
-        self._preforked_initialization = False  # reset for status updater
-        asyncio.get_running_loop().create_task(self._update_status_when_ready_async())
-
-
 def _wrap_impression_listener(listener, metadata):
     """
     Wrap the impression listener if any.
@@ -749,19 +734,13 @@ async def _build_in_memory_factory_async(api_key, cfg, sdk_url=None, events_url=
 
     await telemetry_init_producer.record_config(cfg, extra_cfg, total_flag_sets, invalid_flag_sets)
 
-    if preforked_initialization:
-        await synchronizer.sync_all(max_retry_attempts=_MAX_RETRY_SYNC_ALL)
-        await synchronizer._split_synchronizers._segment_sync.shutdown()
-
-        return SplitFactoryAsync(api_key, storages, cfg['labelsEnabled'],
-                            recorder, manager, None, telemetry_producer, telemetry_init_producer, telemetry_submitter, preforked_initialization=preforked_initialization)
-
     manager_start_task = asyncio.get_running_loop().create_task(manager.start())
 
     return SplitFactoryAsync(api_key, storages, cfg['labelsEnabled'],
-                        recorder, manager, manager_start_task,
+                        recorder, manager,
                         telemetry_producer, telemetry_init_producer,
-                        telemetry_submitter, manager_start_task=manager_start_task)
+                        telemetry_submitter, manager_start_task=manager_start_task,
+                        api_client=http_client)
 
 def _build_redis_factory(api_key, cfg):
     """Build and return a split factory with redis-based storage."""

splitio/push/manager.py

@@ -356,6 +356,9 @@ async def stop(self, blocking=False):
         else:
             asyncio.get_running_loop().create_task(self._stop_current_conn())
 
+    async def close_sse_http_client(self):
+        await self._sse_client.close_sse_http_client()
+
     async def _event_handler(self, event):
         """
         Process an incoming event.

splitio/push/splitsse.py

@@ -237,3 +237,6 @@ async def stop(self):
             _LOGGER.error("Exception waiting for event source ended")
             _LOGGER.debug('stack trace: ', exc_info=True)
             pass
+
+    async def close_sse_http_client(self):
+        await self._client.close_session()

splitio/push/status_tracker.py

@@ -115,7 +115,7 @@ class PushStatusTracker(PushStatusTrackerBase):
 
     def __init__(self, telemetry_runtime_producer):
         """Class constructor."""
-        super().__init__(telemetry_runtime_producer)
+        PushStatusTrackerBase.__init__(self, telemetry_runtime_producer)
 
     def handle_occupancy(self, event):
         """
@@ -237,7 +237,7 @@ class PushStatusTrackerAsync(PushStatusTrackerBase):
 
     def __init__(self, telemetry_runtime_producer):
         """Class constructor."""
-        super().__init__(telemetry_runtime_producer)
+        PushStatusTrackerBase.__init__(self, telemetry_runtime_producer)
 
     async def handle_occupancy(self, event):
         """

splitio/sync/manager.py

@@ -206,6 +206,9 @@ async def stop(self, blocking):
         await self._synchronizer.shutdown(blocking)
         self._stopped = True
 
+    async def close_sse_http_client(self):
+        await self._push.close_sse_http_client()
+
     async def _streaming_feedback_handler(self):
         """
         Handle status updates from the streaming subsystem.