Merge branch 'Feature/FlagSets' into flagsets-tests-e2e

Author: chillaq

splitio/client/client.py

@@ -430,7 +430,12 @@ def _get_feature_flag_names_by_flag_sets(self, flag_sets):
         """
         sanitized_flag_sets = config.sanitize_flag_sets(flag_sets)
         feature_flags = []
-        [feature_flags.extend(self._split_storage.get_feature_flags_by_set(flag_set)) for flag_set in sanitized_flag_sets]
+        for flag_set in sanitized_flag_sets:
+            feature_flags_by_set = self._split_storage.get_feature_flags_by_sets(flag_set)
+            if feature_flags_by_set is None:
+                _LOGGER.warning("Fetching feature flags for flag set %s encountered an error, skipping this flag set." % (flag_set))
+                continue
+            feature_flags.extend(feature_flags_by_set)
         feature_flags_names = []
         [feature_flags_names.append(feature_flag) for feature_flag in feature_flags]
         return feature_flags_names

splitio/client/config.py

@@ -60,7 +60,7 @@
     'storageWrapper': None,
     'storagePrefix': None,
     'storageType': None,
-    'FlagSetsFilter': None
+    'flagSetsFilter': None
 }
 
 def _parse_operation_mode(sdk_key, config):
@@ -179,6 +179,6 @@ def sanitize(sdk_key, config):
         _LOGGER.warning('metricRefreshRate parameter minimum value is 60 seconds, defaulting to 3600 seconds.')
         processed['metricsRefreshRate'] = 3600
 
-    processed['FlagSetsFilter'] = sanitize_flag_sets(processed['FlagSetsFilter']) if processed['FlagSetsFilter'] is not None else None
+    processed['flagSetsFilter'] = sanitize_flag_sets(processed['flagSetsFilter']) if processed['flagSetsFilter'] is not None else None
 
     return processed

splitio/client/factory.py

@@ -350,7 +350,7 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
     }
 
     storages = {
-        'splits': InMemorySplitStorage(),
+        'splits': InMemorySplitStorage(cfg['flagSetsFilter'] if cfg['flagSetsFilter'] is not None else []),
         'segments': InMemorySegmentStorage(),
         'impressions': InMemoryImpressionStorage(cfg['impressionsQueueSize'], telemetry_runtime_producer),
         'events': InMemoryEventStorage(cfg['eventsQueueSize'], telemetry_runtime_producer),
@@ -523,7 +523,7 @@ def _build_pluggable_factory(api_key, cfg):
     pluggable_adapter = cfg.get('storageWrapper')
     storage_prefix = cfg.get('storagePrefix')
     storages = {
-        'splits': PluggableSplitStorage(pluggable_adapter, storage_prefix),
+        'splits': PluggableSplitStorage(pluggable_adapter, storage_prefix, cfg['flagSetsFilter'] if cfg['flagSetsFilter'] is not None else []),
         'segments': PluggableSegmentStorage(pluggable_adapter, storage_prefix),
         'impressions': PluggableImpressionsStorage(pluggable_adapter, sdk_metadata, storage_prefix),
         'events': PluggableEventsStorage(pluggable_adapter, sdk_metadata, storage_prefix),

splitio/models/telemetry.py

@@ -27,7 +27,7 @@ class CounterConstants(Enum):
     EVENTS_QUEUED = 'eventsQueued'
     EVENTS_DROPPED = 'eventsDropped'
 
-class ConfigParams(Enum):
+class _ConfigParams(Enum):
     """Config parameters constants"""
     SPLITS_REFRESH_RATE = 'featuresRefreshRate'
     SEGMENTS_REFRESH_RATE = 'segmentsRefreshRate'
@@ -41,8 +41,9 @@ class ConfigParams(Enum):
     EVENTS_QUEUE_SIZE = 'eventsQueueSize'
     IMPRESSIONS_MODE = 'impressionsMode'
     IMPRESSIONS_LISTENER = 'impressionListener'
+    FLAG_SETS = 'flagSetsFilter'
 
-class ExtraConfig(Enum):
+class _ExtraConfig(Enum):
     """Extra config constants"""
     ACTIVE_FACTORY_COUNT = 'activeFactoryCount'
     REDUNDANT_FACTORY_COUNT = 'redundantFactoryCount'
@@ -53,7 +54,7 @@ class ExtraConfig(Enum):
     HTTP_PROXY = 'httpProxy'
     HTTPS_PROXY_ENV = 'HTTPS_PROXY'
 
-class ApiURLs(Enum):
+class _ApiURLs(Enum):
     """Api URL constants"""
     SDK_URL = 'sdk_url'
     EVENTS_URL = 'events_url'
@@ -88,7 +89,7 @@ class MethodExceptionsAndLatencies(Enum):
     TREATMENTS_WITH_CONFIG_BY_FLAG_SETS = 'treatments_with_config_by_flag_sets'
     TRACK = 'track'
 
-class LastSynchronizationConstants(Enum):
+class _LastSynchronizationConstants(Enum):
     """Last sync constants"""
     LAST_SYNCHRONIZATIONS = 'lastSynchronizations'
 
@@ -108,7 +109,7 @@ class SSESyncMode(Enum):
     STREAMING = 0
     POLLING = 1
 
-class StreamingEventsConstant(Enum):
+class _StreamingEventsConstant(Enum):
     """Storage types constant"""
     STREAMING_EVENTS = 'streamingEvents'
 
@@ -426,7 +427,7 @@ def get_all(self):
         :rtype: dict
         """
         with self._lock:
-            return {LastSynchronizationConstants.LAST_SYNCHRONIZATIONS.value: {HTTPExceptionsAndLatencies.SPLIT.value: self._split, HTTPExceptionsAndLatencies.SEGMENT.value: self._segment, HTTPExceptionsAndLatencies.IMPRESSION.value: self._impression,
+            return {_LastSynchronizationConstants.LAST_SYNCHRONIZATIONS.value: {HTTPExceptionsAndLatencies.SPLIT.value: self._split, HTTPExceptionsAndLatencies.SEGMENT.value: self._segment, HTTPExceptionsAndLatencies.IMPRESSION.value: self._impression,
                                         HTTPExceptionsAndLatencies.IMPRESSION_COUNT.value: self._impression_count, HTTPExceptionsAndLatencies.EVENT.value: self._event,
                                         HTTPExceptionsAndLatencies.TELEMETRY.value: self._telemetry, HTTPExceptionsAndLatencies.TOKEN.value: self._token}
                     }
@@ -671,6 +672,8 @@ def pop_update_from_sse(self, event):
         :rtype: int
         """
         with self._lock:
+            if self._update_from_sse.get(event.value) is None:
+                return 0
             update_from_sse = self._update_from_sse[event.value]
             self._update_from_sse[event.value] = 0
             return update_from_sse
@@ -758,7 +761,7 @@ def pop_streaming_events(self):
         with self._lock:
             streaming_events = self._streaming_events
             self._streaming_events = []
-            return {StreamingEventsConstant.STREAMING_EVENTS.value: [{'e': streaming_event.type, 'd': streaming_event.data,
+            return {_StreamingEventsConstant.STREAMING_EVENTS.value: [{'e': streaming_event.type, 'd': streaming_event.data,
                                          't': streaming_event.time} for streaming_event in streaming_events]}
 
 class TelemetryConfig(object):
@@ -780,10 +783,10 @@ def _reset_all(self):
             self._operation_mode = None
             self._storage_type = None
             self._streaming_enabled = None
-            self._refresh_rate = {ConfigParams.SPLITS_REFRESH_RATE.value: 0, ConfigParams.SEGMENTS_REFRESH_RATE.value: 0,
-                ConfigParams.IMPRESSIONS_REFRESH_RATE.value: 0, ConfigParams.EVENTS_REFRESH_RATE.value: 0, ConfigParams.TELEMETRY_REFRESH_RATE.value: 0}
-            self._url_override = {ApiURLs.SDK_URL.value: False, ApiURLs.EVENTS_URL.value: False, ApiURLs.AUTH_URL.value: False,
-                                  ApiURLs.STREAMING_URL.value: False, ApiURLs.TELEMETRY_URL.value: False}
+            self._refresh_rate = {_ConfigParams.SPLITS_REFRESH_RATE.value: 0, _ConfigParams.SEGMENTS_REFRESH_RATE.value: 0,
+                _ConfigParams.IMPRESSIONS_REFRESH_RATE.value: 0, _ConfigParams.EVENTS_REFRESH_RATE.value: 0, _ConfigParams.TELEMETRY_REFRESH_RATE.value: 0}
+            self._url_override = {_ApiURLs.SDK_URL.value: False, _ApiURLs.EVENTS_URL.value: False, _ApiURLs.AUTH_URL.value: False,
+                                  _ApiURLs.STREAMING_URL.value: False, _ApiURLs.TELEMETRY_URL.value: False}
             self._impressions_queue_size = 0
             self._events_queue_size = 0
             self._impressions_mode = None
@@ -816,16 +819,17 @@ def record_config(self, config, extra_config):
         :type config: dict
         """
         with self._lock:
-            self._operation_mode = self._get_operation_mode(config[ConfigParams.OPERATION_MODE.value])
-            self._storage_type = self._get_storage_type(config[ConfigParams.OPERATION_MODE.value], config[ConfigParams.STORAGE_TYPE.value])
-            self._streaming_enabled = config[ConfigParams.STREAMING_ENABLED.value]
+            self._operation_mode = self._get_operation_mode(config[_ConfigParams.OPERATION_MODE.value])
+            self._storage_type = self._get_storage_type(config[_ConfigParams.OPERATION_MODE.value], config[_ConfigParams.STORAGE_TYPE.value])
+            self._streaming_enabled = config[_ConfigParams.STREAMING_ENABLED.value]
             self._refresh_rate = self._get_refresh_rates(config)
             self._url_override = self._get_url_overrides(extra_config)
-            self._impressions_queue_size = config[ConfigParams.IMPRESSIONS_QUEUE_SIZE.value]
-            self._events_queue_size = config[ConfigParams.EVENTS_QUEUE_SIZE.value]
-            self._impressions_mode = self._get_impressions_mode(config[ConfigParams.IMPRESSIONS_MODE.value])
-            self._impression_listener = True if config[ConfigParams.IMPRESSIONS_LISTENER.value] is not None else False
+            self._impressions_queue_size = config[_ConfigParams.IMPRESSIONS_QUEUE_SIZE.value]
+            self._events_queue_size = config[_ConfigParams.EVENTS_QUEUE_SIZE.value]
+            self._impressions_mode = self._get_impressions_mode(config[_ConfigParams.IMPRESSIONS_MODE.value])
+            self._impression_listener = True if config[_ConfigParams.IMPRESSIONS_LISTENER.value] is not None else False
             self._http_proxy = self._check_if_proxy_detected()
+            self._flag_sets = len(config[_ConfigParams.FLAG_SETS.value]) if config[_ConfigParams.FLAG_SETS.value] is not None else 0
 
     def record_active_and_redundant_factories(self, active_factory_count, redundant_factory_count):
         with self._lock:
@@ -911,16 +915,16 @@ def get_stats(self):
                 'oM': self._operation_mode,
                 'sT': self._storage_type,
                 'sE': self._streaming_enabled,
-                'rR': {'sp': self._refresh_rate[ConfigParams.SPLITS_REFRESH_RATE.value],
-                                'se': self._refresh_rate[ConfigParams.SEGMENTS_REFRESH_RATE.value],
-                                'im': self._refresh_rate[ConfigParams.IMPRESSIONS_REFRESH_RATE.value],
-                                'ev': self._refresh_rate[ConfigParams.EVENTS_REFRESH_RATE.value],
-                                'te': self._refresh_rate[ConfigParams.TELEMETRY_REFRESH_RATE.value]},
-                'uO': {'s': self._url_override[ApiURLs.SDK_URL.value],
-                                'e': self._url_override[ApiURLs.EVENTS_URL.value],
-                                'a': self._url_override[ApiURLs.AUTH_URL.value],
-                                'st': self._url_override[ApiURLs.STREAMING_URL.value],
-                                't': self._url_override[ApiURLs.TELEMETRY_URL.value]},
+                'rR': {'sp': self._refresh_rate[_ConfigParams.SPLITS_REFRESH_RATE.value],
+                                'se': self._refresh_rate[_ConfigParams.SEGMENTS_REFRESH_RATE.value],
+                                'im': self._refresh_rate[_ConfigParams.IMPRESSIONS_REFRESH_RATE.value],
+                                'ev': self._refresh_rate[_ConfigParams.EVENTS_REFRESH_RATE.value],
+                                'te': self._refresh_rate[_ConfigParams.TELEMETRY_REFRESH_RATE.value]},
+                'uO': {'s': self._url_override[_ApiURLs.SDK_URL.value],
+                                'e': self._url_override[_ApiURLs.EVENTS_URL.value],
+                                'a': self._url_override[_ApiURLs.AUTH_URL.value],
+                                'st': self._url_override[_ApiURLs.STREAMING_URL.value],
+                                't': self._url_override[_ApiURLs.TELEMETRY_URL.value]},
                 'iQ': self._impressions_queue_size,
                 'eQ': self._events_queue_size,
                 'iM': self._impressions_mode,
@@ -979,11 +983,11 @@ def _get_refresh_rates(self, config):
         """
         with self._lock:
             return {
-                ConfigParams.SPLITS_REFRESH_RATE.value: config[ConfigParams.SPLITS_REFRESH_RATE.value],
-                ConfigParams.SEGMENTS_REFRESH_RATE.value: config[ConfigParams.SEGMENTS_REFRESH_RATE.value],
-                ConfigParams.IMPRESSIONS_REFRESH_RATE.value: config[ConfigParams.IMPRESSIONS_REFRESH_RATE.value],
-                ConfigParams.EVENTS_REFRESH_RATE.value: config[ConfigParams.EVENTS_REFRESH_RATE.value],
-                ConfigParams.TELEMETRY_REFRESH_RATE.value: config[ConfigParams.TELEMETRY_REFRESH_RATE.value]
+                _ConfigParams.SPLITS_REFRESH_RATE.value: config[_ConfigParams.SPLITS_REFRESH_RATE.value],
+                _ConfigParams.SEGMENTS_REFRESH_RATE.value: config[_ConfigParams.SEGMENTS_REFRESH_RATE.value],
+                _ConfigParams.IMPRESSIONS_REFRESH_RATE.value: config[_ConfigParams.IMPRESSIONS_REFRESH_RATE.value],
+                _ConfigParams.EVENTS_REFRESH_RATE.value: config[_ConfigParams.EVENTS_REFRESH_RATE.value],
+                _ConfigParams.TELEMETRY_REFRESH_RATE.value: config[_ConfigParams.TELEMETRY_REFRESH_RATE.value]
             }
 
     def _get_url_overrides(self, config):
@@ -998,11 +1002,11 @@ def _get_url_overrides(self, config):
         """
         with self._lock:
             return  {
-                ApiURLs.SDK_URL.value: True if ApiURLs.SDK_URL.value in config else False,
-                ApiURLs.EVENTS_URL.value: True if ApiURLs.EVENTS_URL.value in config else False,
-                ApiURLs.AUTH_URL.value: True if ApiURLs.AUTH_URL.value in config else False,
-                ApiURLs.STREAMING_URL.value: True if ApiURLs.STREAMING_URL.value in config else False,
-                ApiURLs.TELEMETRY_URL.value: True if ApiURLs.TELEMETRY_URL.value in config else False
+                _ApiURLs.SDK_URL.value: True if _ApiURLs.SDK_URL.value in config else False,
+                _ApiURLs.EVENTS_URL.value: True if _ApiURLs.EVENTS_URL.value in config else False,
+                _ApiURLs.AUTH_URL.value: True if _ApiURLs.AUTH_URL.value in config else False,
+                _ApiURLs.STREAMING_URL.value: True if _ApiURLs.STREAMING_URL.value in config else False,
+                _ApiURLs.TELEMETRY_URL.value: True if _ApiURLs.TELEMETRY_URL.value in config else False
             }
 
     def _get_impressions_mode(self, imp_mode):
@@ -1032,6 +1036,6 @@ def _check_if_proxy_detected(self):
         """
         with self._lock:
             for x in os.environ:
-                if x.upper() == ExtraConfig.HTTPS_PROXY_ENV.value:
+                if x.upper() == _ExtraConfig.HTTPS_PROXY_ENV.value:
                     return True
             return False

tests/models/test_telemetry_model.py

@@ -297,7 +297,8 @@ def test_telemetry_config(self):
                   'impressionsRefreshRate': 60,
                   'eventsPushRate': 60,
                   'metricsRefreshRate': 10,
-                  'storageType': None
+                  'storageType': None,
+                  'flagSetsFilter': None
                 }
         telemetry_config.record_config(config, {})
         assert(telemetry_config.get_stats() == {'oM': 0,

tests/storage/test_pluggable.py

@@ -141,9 +141,9 @@ def test_init(self):
                 prefix = 'myprefix.'
             else:
                 prefix = ''
-        assert(pluggable_split_storage._prefix == prefix + "SPLITIO.split.{split_name}")
+        assert(pluggable_split_storage._prefix == prefix + "SPLITIO.split.{feature_flag_name}")
         assert(pluggable_split_storage._traffic_type_prefix == prefix + "SPLITIO.trafficType.{traffic_type_name}")
-        assert(pluggable_split_storage._split_till_prefix == prefix + "SPLITIO.splits.till")
+        assert(pluggable_split_storage._feature_flag_till_prefix == prefix + "SPLITIO.splits.till")
 
     # TODO: To be added when producer mode is aupported
 #    def test_put_many(self):
@@ -166,10 +166,10 @@ def test_get(self):
             pluggable_split_storage = PluggableSplitStorage(self.mock_adapter, prefix=sprefix)
 
             split1 = splits.from_raw(splits_json['splitChange1_2']['splits'][0])
-            split_name = splits_json['splitChange1_2']['splits'][0]['name']
+            feature_flag_name = splits_json['splitChange1_2']['splits'][0]['name']
 
-            self.mock_adapter.set(pluggable_split_storage._prefix.format(split_name=split_name), split1.to_json())
-            assert(pluggable_split_storage.get(split_name).to_json() ==  splits.from_raw(splits_json['splitChange1_2']['splits'][0]).to_json())
+            self.mock_adapter.set(pluggable_split_storage._prefix.format(feature_flag_name=feature_flag_name), split1.to_json())
+            assert(pluggable_split_storage.get(feature_flag_name).to_json() ==  splits.from_raw(splits_json['splitChange1_2']['splits'][0]).to_json())
             assert(pluggable_split_storage.get('not_existing') == None)
 
     def test_fetch_many(self):
@@ -181,8 +181,8 @@ def test_fetch_many(self):
             split2_temp['name'] = 'another_split'
             split2 = splits.from_raw(split2_temp)
 
-            self.mock_adapter.set(pluggable_split_storage._prefix.format(split_name=split1.name), split1.to_json())
-            self.mock_adapter.set(pluggable_split_storage._prefix.format(split_name=split2.name), split2.to_json())
+            self.mock_adapter.set(pluggable_split_storage._prefix.format(feature_flag_name=split1.name), split1.to_json())
+            self.mock_adapter.set(pluggable_split_storage._prefix.format(feature_flag_name=split2.name), split2.to_json())
             fetched = pluggable_split_storage.fetch_many([split1.name, split2.name])
             assert(fetched[split1.name].to_json() == split1.to_json())
             assert(fetched[split2.name].to_json() == split2.to_json())
@@ -220,8 +220,8 @@ def test_get_split_names(self):
             split2_temp = splits_json['splitChange1_2']['splits'][0].copy()
             split2_temp['name'] = 'another_split'
             split2 = splits.from_raw(split2_temp)
-            self.mock_adapter.set(pluggable_split_storage._prefix.format(split_name=split1.name), split1.to_json())
-            self.mock_adapter.set(pluggable_split_storage._prefix.format(split_name=split2.name), split2.to_json())
+            self.mock_adapter.set(pluggable_split_storage._prefix.format(feature_flag_name=split1.name), split1.to_json())
+            self.mock_adapter.set(pluggable_split_storage._prefix.format(feature_flag_name=split2.name), split2.to_json())
             assert(pluggable_split_storage.get_split_names() == [split1.name, split2.name])
 
     def test_get_all(self):
@@ -233,8 +233,8 @@ def test_get_all(self):
             split2_temp['name'] = 'another_split'
             split2 = splits.from_raw(split2_temp)
 
-            self.mock_adapter.set(pluggable_split_storage._prefix.format(split_name=split1.name), split1.to_json())
-            self.mock_adapter.set(pluggable_split_storage._prefix.format(split_name=split2.name), split2.to_json())
+            self.mock_adapter.set(pluggable_split_storage._prefix.format(feature_flag_name=split1.name), split1.to_json())
+            self.mock_adapter.set(pluggable_split_storage._prefix.format(feature_flag_name=split2.name), split2.to_json())
             all_splits = pluggable_split_storage.get_all()
             assert([all_splits[0].to_json(), all_splits[1].to_json()] == [split1.to_json(), split2.to_json()])