Update factory and tests

Author: chillaq

splitio/client/client.py

@@ -63,7 +63,7 @@ def __init__(self, factory, recorder, labels_enabled=True, fallback_treatments_c
         self._feature_flag_storage = factory._get_storage('splits')  # pylint: disable=protected-access
         self._segment_storage = factory._get_storage('segments')  # pylint: disable=protected-access
         self._events_storage = factory._get_storage('events')  # pylint: disable=protected-access
-        self._evaluator = Evaluator(self._splitter)
+        self._evaluator = Evaluator(self._splitter, fallback_treatments_configuration)
         self._telemetry_evaluation_producer = self._factory._telemetry_evaluation_producer
         self._telemetry_init_producer = self._factory._telemetry_init_producer
         self._fallback_treatments_configuration = fallback_treatments_configuration
@@ -344,7 +344,7 @@ def _get_treatment(self, method, key, feature, attributes=None, evaluation_optio
                 result = self._get_fallback_eval_results(self._FAILED_EVAL_RESULT, feature)
 
         properties = self._get_properties(evaluation_options)
-        if result['impression']['label'].find(Label.SPLIT_NOT_FOUND) == -1:
+        if result['impression']['label'] == None or (result['impression']['label'] != None and result['impression']['label'].find(Label.SPLIT_NOT_FOUND) == -1):
             impression_decorated = self._build_impression(key, bucketing, feature, result, properties)
             self._record_stats([(impression_decorated, attributes)], start, method)
 
@@ -647,7 +647,7 @@ def _get_treatments(self, key, features, method, attributes=None, evaluation_opt
         properties = self._get_properties(evaluation_options)
         imp_decorated_attrs = [
             (i, attributes) for i in self._build_impressions(key, bucketing, results, properties)
-            if i.Impression.label.find(Label.SPLIT_NOT_FOUND) == -1
+            if i.Impression.label == None or (i.Impression.label != None and i.Impression.label.find(Label.SPLIT_NOT_FOUND)) == -1
         ]
         self._record_stats(imp_decorated_attrs, start, method)
 
@@ -844,7 +844,7 @@ async def _get_treatment(self, method, key, feature, attributes=None, evaluation
                 result = self._get_fallback_eval_results(self._FAILED_EVAL_RESULT, feature)
 
         properties = self._get_properties(evaluation_options)
-        if result['impression']['label'].find(Label.SPLIT_NOT_FOUND) == -1:
+        if result['impression']['label'] == None or (result['impression']['label'] != None and result['impression']['label'].find(Label.SPLIT_NOT_FOUND) == -1):
             impression_decorated = self._build_impression(key, bucketing, feature, result, properties)
             await self._record_stats([(impression_decorated, attributes)], start, method)
         return result['treatment'], result['configurations']
@@ -1060,7 +1060,7 @@ async def _get_treatments(self, key, features, method, attributes=None, evaluati
         properties = self._get_properties(evaluation_options)
         imp_decorated_attrs = [
             (i, attributes) for i in self._build_impressions(key, bucketing, results, properties)
-            if i.Impression.label != Label.SPLIT_NOT_FOUND
+            if i.Impression.label == None or (i.Impression.label != None and i.Impression.label.find(Label.SPLIT_NOT_FOUND)) == -1
         ]
         await self._record_stats(imp_decorated_attrs, start, method)
 

splitio/client/factory.py

@@ -170,7 +170,8 @@ def __init__(  # pylint: disable=too-many-arguments
             telemetry_producer=None,
             telemetry_init_producer=None,
             telemetry_submitter=None,
-            preforked_initialization=False
+            preforked_initialization=False,
+            fallback_treatments_configuration=None
     ):
         """
         Class constructor.
@@ -201,6 +202,7 @@ def __init__(  # pylint: disable=too-many-arguments
         self._ready_time = get_current_epoch_time_ms()
         _LOGGER.debug("Running in threading mode")
         self._sdk_internal_ready_flag = sdk_ready_flag
+        self._fallback_treatments_configuration = fallback_treatments_configuration
         self._start_status_updater()
 
     def _start_status_updater(self):
@@ -242,7 +244,7 @@ def client(self):
         This client is only a set of references to structures hold by the factory.
         Creating one a fast operation and safe to be used anywhere.
         """
-        return Client(self, self._recorder, self._labels_enabled)
+        return Client(self, self._recorder, self._labels_enabled, self._fallback_treatments_configuration)
 
     def manager(self):
         """
@@ -338,7 +340,8 @@ def __init__(  # pylint: disable=too-many-arguments
             telemetry_init_producer=None,
             telemetry_submitter=None,
             manager_start_task=None,
-            api_client=None
+            api_client=None,
+            fallback_treatments_configuration=None
     ):
         """
         Class constructor.
@@ -372,6 +375,7 @@ def __init__(  # pylint: disable=too-many-arguments
         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
+        self._fallback_treatments_configuration = fallback_treatments_configuration
 
     async def _update_status_when_ready_async(self):
         """Wait until the sdk is ready and update the status for async mode."""
@@ -460,7 +464,7 @@ def client(self):
         This client is only a set of references to structures hold by the factory.
         Creating one a fast operation and safe to be used anywhere.
         """
-        return ClientAsync(self, self._recorder, self._labels_enabled)
+        return ClientAsync(self, self._recorder, self._labels_enabled, self._fallback_treatments_configuration)
 
 def _wrap_impression_listener(listener, metadata):
     """
@@ -623,15 +627,16 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         synchronizer._split_synchronizers._segment_sync.shutdown()
 
         return SplitFactory(api_key, storages, cfg['labelsEnabled'],
-                            recorder, manager, None, telemetry_producer, telemetry_init_producer, telemetry_submitter, preforked_initialization=preforked_initialization)
+                            recorder, manager, None, telemetry_producer, telemetry_init_producer, telemetry_submitter, preforked_initialization=preforked_initialization,
+                            fallback_treatments_configuration=cfg['fallbackTreatmentsConfiguration'])
 
     initialization_thread = threading.Thread(target=manager.start, name="SDKInitializer", daemon=True)
     initialization_thread.start()
 
     return SplitFactory(api_key, storages, cfg['labelsEnabled'],
                         recorder, manager, sdk_ready_flag,
                         telemetry_producer, telemetry_init_producer,
-                        telemetry_submitter)
+                        telemetry_submitter, fallback_treatments_configuration=cfg['fallbackTreatmentsConfiguration'])
 
 async def _build_in_memory_factory_async(api_key, cfg, sdk_url=None, events_url=None,  # pylint:disable=too-many-arguments,too-many-localsa
                              auth_api_base_url=None, streaming_api_base_url=None, telemetry_api_base_url=None,
@@ -750,7 +755,7 @@ async def _build_in_memory_factory_async(api_key, cfg, sdk_url=None, events_url=
                         recorder, manager,
                         telemetry_producer, telemetry_init_producer,
                         telemetry_submitter, manager_start_task=manager_start_task,
-                        api_client=http_client)
+                        api_client=http_client, fallback_treatments_configuration=cfg['fallbackTreatmentsConfiguration'])
 
 def _build_redis_factory(api_key, cfg):
     """Build and return a split factory with redis-based storage."""
@@ -828,7 +833,8 @@ def _build_redis_factory(api_key, cfg):
         manager,
         sdk_ready_flag=None,
         telemetry_producer=telemetry_producer,
-        telemetry_init_producer=telemetry_init_producer
+        telemetry_init_producer=telemetry_init_producer,
+        fallback_treatments_configuration=cfg['fallbackTreatmentsConfiguration']
     )
     redundant_factory_count, active_factory_count = _get_active_and_redundant_count()
     storages['telemetry'].record_active_and_redundant_factories(active_factory_count, redundant_factory_count)
@@ -910,7 +916,8 @@ async def _build_redis_factory_async(api_key, cfg):
         manager,
         telemetry_producer=telemetry_producer,
         telemetry_init_producer=telemetry_init_producer,
-        telemetry_submitter=telemetry_submitter
+        telemetry_submitter=telemetry_submitter,
+        fallback_treatments_configuration=cfg['fallbackTreatmentsConfiguration']
     )
     redundant_factory_count, active_factory_count = _get_active_and_redundant_count()
     await storages['telemetry'].record_active_and_redundant_factories(active_factory_count, redundant_factory_count)
@@ -992,7 +999,8 @@ def _build_pluggable_factory(api_key, cfg):
         manager,
         sdk_ready_flag=None,
         telemetry_producer=telemetry_producer,
-        telemetry_init_producer=telemetry_init_producer
+        telemetry_init_producer=telemetry_init_producer,
+        fallback_treatments_configuration=cfg['fallbackTreatmentsConfiguration']
     )
     redundant_factory_count, active_factory_count = _get_active_and_redundant_count()
     storages['telemetry'].record_active_and_redundant_factories(active_factory_count, redundant_factory_count)
@@ -1072,7 +1080,8 @@ async def _build_pluggable_factory_async(api_key, cfg):
         manager,
         telemetry_producer=telemetry_producer,
         telemetry_init_producer=telemetry_init_producer,
-        telemetry_submitter=telemetry_submitter
+        telemetry_submitter=telemetry_submitter,
+        fallback_treatments_configuration=cfg['fallbackTreatmentsConfiguration']
     )
     redundant_factory_count, active_factory_count = _get_active_and_redundant_count()
     await storages['telemetry'].record_active_and_redundant_factories(active_factory_count, redundant_factory_count)
@@ -1150,6 +1159,7 @@ def _build_localhost_factory(cfg):
         telemetry_producer=telemetry_producer,
         telemetry_init_producer=telemetry_producer.get_telemetry_init_producer(),
         telemetry_submitter=LocalhostTelemetrySubmitter(),
+        fallback_treatments_configuration=cfg['fallbackTreatmentsConfiguration']
     )
 
 async def _build_localhost_factory_async(cfg):
@@ -1220,7 +1230,8 @@ async def _build_localhost_factory_async(cfg):
         telemetry_producer=telemetry_producer,
         telemetry_init_producer=telemetry_producer.get_telemetry_init_producer(),
         telemetry_submitter=LocalhostTelemetrySubmitterAsync(),
-        manager_start_task=manager_start_task
+        manager_start_task=manager_start_task,
+        fallback_treatments_configuration=cfg['fallbackTreatmentsConfiguration']
     )
 
 def get_factory(api_key, **kwargs):

splitio/storage/adapters/redis.py

@@ -715,7 +715,7 @@ def _build_default_client(config):  # pylint: disable=too-many-locals
     unix_socket_path = config.get('redisUnixSocketPath', None)
     encoding = config.get('redisEncoding', 'utf-8')
     encoding_errors = config.get('redisEncodingErrors', 'strict')
-    errors = config.get('redisErrors', None)
+#    errors = config.get('redisErrors', None)
     decode_responses = config.get('redisDecodeResponses', True)
     retry_on_timeout = config.get('redisRetryOnTimeout', False)
     ssl = config.get('redisSsl', False)
@@ -740,7 +740,7 @@ def _build_default_client(config):  # pylint: disable=too-many-locals
         unix_socket_path=unix_socket_path,
         encoding=encoding,
         encoding_errors=encoding_errors,
-        errors=errors,
+#        errors=errors, Starting from redis 6.0.0 errors argument is removed
         decode_responses=decode_responses,
         retry_on_timeout=retry_on_timeout,
         ssl=ssl,

tests/client/test_factory.py

@@ -13,6 +13,8 @@
 from splitio.storage import redis, inmemmory, pluggable
 from splitio.tasks.util import asynctask
 from splitio.engine.impressions.impressions import Manager as ImpressionsManager
+from splitio.models.fallback_config import FallbackConfig, FallbackTreatmentsConfiguration
+from splitio.models.fallback_treatment import FallbackTreatment
 from splitio.sync.manager import Manager, ManagerAsync
 from splitio.sync.synchronizer import Synchronizer, SynchronizerAsync, SplitSynchronizers, SplitTasks
 from splitio.sync.split import SplitSynchronizer, SplitSynchronizerAsync
@@ -94,7 +96,7 @@ def test_redis_client_creation(self, mocker):
         """Test that a client with redis storage is created correctly."""
         strict_redis_mock = mocker.Mock()
         mocker.patch('splitio.storage.adapters.redis.StrictRedis', new=strict_redis_mock)
-
+        fallback_treatments_configuration = FallbackTreatmentsConfiguration(FallbackConfig(FallbackTreatment("on")))
         config = {
             'labelsEnabled': False,
             'impressionListener': 123,
@@ -119,7 +121,8 @@ def test_redis_client_creation(self, mocker):
             'redisSslCertReqs': 'some_cert_req',
             'redisSslCaCerts': 'some_ca_cert',
             'redisMaxConnections': 999,
-            'flagSetsFilter': ['set_1']
+            'flagSetsFilter': ['set_1'],
+            'fallbackTreatmentsConfiguration': fallback_treatments_configuration
         }
         factory = get_factory('some_api_key', config=config)
         class TelemetrySubmitterMock():
@@ -133,6 +136,7 @@ def synchronize_config(*_):
         assert isinstance(factory._get_storage('events'), redis.RedisEventsStorage)
 
         assert factory._get_storage('splits').flag_set_filter.flag_sets == set([])
+        assert factory._fallback_treatments_configuration == fallback_treatments_configuration
 
         adapter = factory._get_storage('splits')._redis
         assert adapter == factory._get_storage('segments')._redis
@@ -153,7 +157,7 @@ def synchronize_config(*_):
             unix_socket_path='/some_path',
             encoding='utf-8',
             encoding_errors='non-strict',
-            errors=True,
+#            errors=True,
             decode_responses=True,
             retry_on_timeout=True,
             ssl=True,
@@ -705,10 +709,13 @@ class SplitFactoryAsyncTests(object):
 
     @pytest.mark.asyncio
     async def test_flag_sets_counts(self):
+        fallback_treatments_configuration = FallbackTreatmentsConfiguration(FallbackConfig(FallbackTreatment("on")))
         factory = await get_factory_async("none", config={
             'flagSetsFilter': ['set1', 'set2', 'set3'],
-            'streamEnabled': False
+            'streamEnabled': False,
+            'fallbackTreatmentsConfiguration': fallback_treatments_configuration
         })
+        assert factory._fallback_treatments_configuration == fallback_treatments_configuration
         assert factory._telemetry_init_producer._telemetry_storage._tel_config._flag_sets == 3
         assert factory._telemetry_init_producer._telemetry_storage._tel_config._flag_sets_invalid == 0
         await factory.destroy()

tests/client/test_input_validator.py

@@ -1,5 +1,6 @@
 """Unit tests for the input_validator module."""
 import pytest
+import logging
 
 from splitio.client.factory import SplitFactory, get_factory, SplitFactoryAsync, get_factory_async
 from splitio.client.client import CONTROL, Client, _LOGGER as _logger, ClientAsync
@@ -9,6 +10,7 @@
     InMemorySplitStorage, InMemorySplitStorageAsync, InMemoryRuleBasedSegmentStorage, InMemoryRuleBasedSegmentStorageAsync
 from splitio.models.splits import Split
 from splitio.client import input_validator
+from splitio.client.manager import SplitManager, SplitManagerAsync 
 from splitio.recorder.recorder import StandardRecorder, StandardRecorderAsync
 from splitio.engine.telemetry import TelemetryStorageProducer, TelemetryStorageProducerAsync
 from splitio.engine.impressions.impressions import Manager as ImpressionManager