Updated redis, pluggable and localjson storages

splitio · chillaq · Jun 5, 2025 · Jan 17, 2025 · Mar 6, 2025 · Mar 6, 2025
commit 4d8327c84cdb0036899fa7f1639a7980b79ae7de
@@ -11,14 +11,14 @@
 class RuleBasedSegment(object):
     """RuleBasedSegment object class."""
 
-    def __init__(self, name, traffic_yype_Name, change_number, status, conditions, excluded):
+    def __init__(self, name, traffic_type_name, change_number, status, conditions, excluded):
         """
         Class constructor.
 
         :param name: Segment name.
         :type name: str
-        :param traffic_yype_Name: traffic type name.
-        :type traffic_yype_Name: str
+        :param traffic_type_name: traffic type name.
+        :type traffic_type_name: str
         :param change_number: change number.
         :type change_number: str
         :param status: status.
@@ -29,7 +29,7 @@ def __init__(self, name, traffic_yype_Name, change_number, status, conditions, e
         :type excluded: Excluded
         """
         self._name = name
-        self._traffic_yype_Name = traffic_yype_Name
+        self._traffic_type_name = traffic_type_name
         self._change_number = change_number
         self._status = status
         self._conditions = conditions
@@ -41,9 +41,9 @@ def name(self):
         return self._name
 
     @property
-    def traffic_yype_Name(self):
+    def traffic_type_name(self):
         """Return traffic type name."""
-        return self._traffic_yype_Name
+        return self._traffic_type_name
 
     @property
     def change_number(self):
@@ -65,6 +65,17 @@ def excluded(self):
         """Return excluded."""
         return self._excluded
 
+    def to_json(self):
+        """Return a JSON representation of this rule based segment."""
+        return {
+            'changeNumber': self.change_number,
+            'trafficTypeName': self.traffic_type_name,
+            'name': self.name,
+            'status': self.status,
+            'conditions': [c.to_json() for c in self.conditions],
+            'excluded': self.excluded.to_json()
+        }
+
 def from_raw(raw_rule_based_segment):
     """
     Parse a Rule based segment from a JSON portion of splitChanges.
@@ -111,3 +122,10 @@ def get_excluded_keys(self):
     def get_excluded_segments(self):
         """Return excluded segments"""
         return self._segments
+
+    def to_json(self):
+        """Return a JSON representation of this object."""
+        return {
+            'keys': self._keys,
+            'segments': self._segments
+        }
@@ -17,7 +17,6 @@
 class PluggableRuleBasedSegmentsStorageBase(RuleBasedSegmentsStorage):
     """RedPluggable storage for rule based segments."""
 
-    _RB_SEGMENT_NAME_LENGTH = 23
     _TILL_LENGTH = 4
 
     def __init__(self, pluggable_adapter, prefix=None):
@@ -28,9 +27,11 @@ def __init__(self, pluggable_adapter, prefix=None):
         :type redis_client: splitio.storage.adapters.redis.RedisAdapter
         """
         self._pluggable_adapter = pluggable_adapter
-        self._prefix = "SPLITIO.rbsegment.${segmen_name}"
+        self._prefix = "SPLITIO.rbsegment.{segment_name}"
         self._rb_segments_till_prefix = "SPLITIO.rbsegments.till"
+        self._rb_segment_name_length = 18
         if prefix is not None:
+            self._rb_segment_name_length += len(prefix) + 1
             self._prefix = prefix + "." + self._prefix
             self._rb_segments_till_prefix = prefix + "." + self._rb_segments_till_prefix
 
@@ -163,18 +164,21 @@ def get_segment_names(self):
         :rtype: list(str)
         """
         try:
+            _LOGGER.error(self._rb_segment_name_length)
+            _LOGGER.error(self._prefix)
+            _LOGGER.error(self._prefix[:self._rb_segment_name_length])
             keys = []
-            for key in self._pluggable_adapter.get_keys_by_prefix(self._prefix[:-self._RB_SEGMENT_NAME_LENGTH]):
+            for key in self._pluggable_adapter.get_keys_by_prefix(self._prefix[:self._rb_segment_name_length]):
                 if key[-self._TILL_LENGTH:] != 'till':
-                    keys.append(key[len(self._prefix[:-self._RB_SEGMENT_NAME_LENGTH]):])
+                    keys.append(key[len(self._prefix[:self._rb_segment_name_length]):])
             return keys
 
         except Exception:
             _LOGGER.error('Error getting rule based segments names from storage')
             _LOGGER.debug('Error: ', exc_info=True)
             return None
 
-class PluggableRuleBasedSegmentsStorageAsync(RuleBasedSegmentsStorage):
+class PluggableRuleBasedSegmentsStorageAsync(PluggableRuleBasedSegmentsStorageBase):
     """RedPluggable storage for rule based segments."""
 
     def __init__(self, pluggable_adapter, prefix=None):
@@ -231,7 +235,7 @@ async def contains(self, segment_names):
         :return: True if segment names exists. False otherwise.
         :rtype: bool
         """
-        return await set(segment_names).issubset(self.get_segment_names())
+        return set(segment_names).issubset(await self.get_segment_names())
 
     async def get_segment_names(self):
         """
@@ -242,9 +246,9 @@ async def get_segment_names(self):
         """
         try:
             keys = []
-            for key in await self._pluggable_adapter.get_keys_by_prefix(self._prefix[:-self._RB_SEGMENT_NAME_LENGTH]):
+            for key in await self._pluggable_adapter.get_keys_by_prefix(self._prefix[:self._rb_segment_name_length]):
                 if key[-self._TILL_LENGTH:] != 'till':
-                    keys.append(key[len(self._prefix[:-self._RB_SEGMENT_NAME_LENGTH]):])
+                    keys.append(key[len(self._prefix[:self._rb_segment_name_length]):])
             return keys
 
         except Exception:

@@ -19,7 +19,7 @@
 class RedisRuleBasedSegmentsStorage(RuleBasedSegmentsStorage):
     """Redis-based storage for rule based segments."""
 
-    _RB_SEGMENT_KEY = 'SPLITIO.rbsegment.${segmen_name}'
+    _RB_SEGMENT_KEY = 'SPLITIO.rbsegment.{segment_name}'
     _RB_SEGMENT_TILL_KEY = 'SPLITIO.rbsegments.till'
 
     def __init__(self, redis_client):
@@ -134,7 +134,7 @@ def get_large_segment_names(self):
 class RedisRuleBasedSegmentsStorageAsync(RuleBasedSegmentsStorage):
     """Redis-based storage for rule based segments."""
 
-    _RB_SEGMENT_KEY = 'SPLITIO.rbsegment.${segmen_name}'
+    _RB_SEGMENT_KEY = 'SPLITIO.rbsegment.{segment_name}'
     _RB_SEGMENT_TILL_KEY = 'SPLITIO.rbsegments.till'
 
     def __init__(self, redis_client):

@@ -545,6 +545,7 @@ def _sanitize_rb_segment_elements(self, parsed_rb_segments):
                             ('changeNumber', 0, 0, None, None, None)]:
                 rb_segment = util._sanitize_object_element(rb_segment, 'rule based segment', element[0], element[1], lower_value=element[2], upper_value=element[3], in_list=element[4], not_in_list=element[5])
             rb_segment = self._sanitize_condition(rb_segment)
+            rb_segment = self._remove_partition(rb_segment)
             sanitized_rb_segments.append(rb_segment)
         return sanitized_rb_segments
 
@@ -599,6 +600,15 @@ def _sanitize_condition(self, feature_flag):
         })
 
         return feature_flag
+
+    def _remove_partition(self, rb_segment):
+        sanitized = []
+        for condition in rb_segment['conditions']:
+            if 'partition' in condition:
+                del condition['partition']
+            sanitized.append(condition)
+        rb_segment['conditions'] = sanitized
+        return rb_segment
 
     @classmethod
     def _convert_yaml_to_feature_flag(cls, parsed):
@@ -769,8 +779,8 @@ def _read_feature_flags_from_json_file(self, filename):
             with open(filename, 'r') as flo:
                 parsed = json.load(flo)
             santitized = self._sanitize_json_elements(parsed)
-            santitized['ff'] = self._sanitize_feature_flag_elements(santitized['ff'])
-            santitized['rbs'] = self._sanitize_rb_segment_elements(santitized['rbs'])
+            santitized['ff']['d'] = self._sanitize_feature_flag_elements(santitized['ff']['d'])
+            santitized['rbs']['d'] = self._sanitize_rb_segment_elements(santitized['rbs']['d'])
             return santitized
 
         except Exception as exc:
@@ -903,7 +913,7 @@ async def _synchronize_json(self):
 
             if await self._rule_based_segment_storage.get_change_number() <= parsed['rbs']['t'] or parsed['rbs']['t'] == self._DEFAULT_FEATURE_FLAG_TILL:
                 fetched_rb_segments = [rule_based_segments.from_raw(rb_segment) for rb_segment in parsed['rbs']['d']]
-                segment_list.update(await update_rule_based_segment_storage(self._rule_based_segment_storage, fetched_rb_segments, parsed['rbs']['t']))
+                segment_list.update(await update_rule_based_segment_storage_async(self._rule_based_segment_storage, fetched_rb_segments, parsed['rbs']['t']))
 
             return segment_list
 
@@ -925,8 +935,8 @@ async def _read_feature_flags_from_json_file(self, filename):
             async with aiofiles.open(filename, 'r') as flo:
                 parsed = json.loads(await flo.read())
             santitized = self._sanitize_json_elements(parsed)
-            santitized['ff'] = self._sanitize_feature_flag_elements(santitized['ff'])
-            santitized['rbs'] = self._sanitize_rb_segment_elements(santitized['rbs'])
+            santitized['ff']['d'] = self._sanitize_feature_flag_elements(santitized['ff']['d'])
+            santitized['rbs']['d'] = self._sanitize_rb_segment_elements(santitized['rbs']['d'])
             return santitized
         except Exception as exc:
             _LOGGER.debug('Exception: ', exc_info=True)

@@ -47,3 +47,7 @@
     "splitChange6_2": split62,
     "splitChange6_3": split63,
 }
+
+rbsegments_json = {
+    "segment1": {"changeNumber": 12, "name": "some_segment", "status": "ACTIVE","trafficTypeName": "user","excluded":{"keys":[],"segments":[]},"conditions": []}
+}
@@ -1,20 +1,21 @@
 """Pluggable storage test module."""
 import json
 import threading
+import copy
 import pytest
 
 from splitio.optional.loaders import asyncio
 from splitio.models.splits import Split
-from splitio.models import splits, segments
+from splitio.models import splits, segments, rule_based_segments
 from splitio.models.segments import Segment
 from splitio.models.impressions import Impression
 from splitio.models.events import Event, EventWrapper
 from splitio.storage.pluggable import PluggableSplitStorage, PluggableSegmentStorage, PluggableImpressionsStorage, PluggableEventsStorage, \
     PluggableTelemetryStorage, PluggableEventsStorageAsync, PluggableSegmentStorageAsync, PluggableImpressionsStorageAsync,\
-    PluggableSplitStorageAsync, PluggableTelemetryStorageAsync
+    PluggableSplitStorageAsync, PluggableTelemetryStorageAsync, PluggableRuleBasedSegmentsStorage, PluggableRuleBasedSegmentsStorageAsync
 from splitio.client.util import get_metadata, SdkMetadata
 from splitio.models.telemetry import MAX_TAGS, MethodExceptionsAndLatencies, OperationMode
-from tests.integration import splits_json
+from tests.integration import splits_json, rbsegments_json
 
 class StorageMockAdapter(object):
     def __init__(self):
@@ -1372,3 +1373,124 @@ async def test_push_config_stats(self):
             await pluggable_telemetry_storage.record_active_and_redundant_factories(2, 1)
             await pluggable_telemetry_storage.push_config_stats()
             assert(self.mock_adapter._keys[pluggable_telemetry_storage._telemetry_config_key + "::" + pluggable_telemetry_storage._sdk_metadata] == '{"aF": 2, "rF": 1, "sT": "memory", "oM": 0, "t": []}')
+
+class PluggableRuleBasedSegmentStorageTests(object):
+    """In memory rule based segment storage test cases."""
+
+    def setup_method(self):
+        """Prepare storages with test data."""
+        self.mock_adapter = StorageMockAdapter()
+
+    def test_get(self):
+        self.mock_adapter._keys = {}
+        for sprefix in [None, 'myprefix']:
+            pluggable_rbs_storage = PluggableRuleBasedSegmentsStorage(self.mock_adapter, prefix=sprefix)
+
+            rbs1 = rule_based_segments.from_raw(rbsegments_json['segment1'])
+            rbs_name = rbsegments_json['segment1']['name']
+
+            self.mock_adapter.set(pluggable_rbs_storage._prefix.format(segment_name=rbs_name), rbs1.to_json())
+            assert(pluggable_rbs_storage.get(rbs_name).to_json() ==  rule_based_segments.from_raw(rbsegments_json['segment1']).to_json())
+            assert(pluggable_rbs_storage.get('not_existing') == None)
+
+    def test_get_change_number(self):
+        self.mock_adapter._keys = {}
+        for sprefix in [None, 'myprefix']:
+            pluggable_rbs_storage = PluggableRuleBasedSegmentsStorage(self.mock_adapter, prefix=sprefix)
+            if sprefix == 'myprefix':
+                prefix = 'myprefix.'
+            else:
+                prefix = ''
+            self.mock_adapter.set(prefix + "SPLITIO.rbsegments.till", 1234)
+            assert(pluggable_rbs_storage.get_change_number() == 1234)
+
+    def test_get_segment_names(self):
+        self.mock_adapter._keys = {}
+        for sprefix in [None, 'myprefix']:
+            pluggable_rbs_storage = PluggableRuleBasedSegmentsStorage(self.mock_adapter, prefix=sprefix)
+            rbs1 = rule_based_segments.from_raw(rbsegments_json['segment1'])
+            rbs2_temp = copy.deepcopy(rbsegments_json['segment1'])
+            rbs2_temp['name'] = 'another_segment'
+            rbs2 = rule_based_segments.from_raw(rbs2_temp)
+            self.mock_adapter.set(pluggable_rbs_storage._prefix.format(segment_name=rbs1.name), rbs1.to_json())
+            self.mock_adapter.set(pluggable_rbs_storage._prefix.format(segment_name=rbs2.name), rbs2.to_json())
+            assert(pluggable_rbs_storage.get_segment_names() == [rbs1.name, rbs2.name])
+
+    def test_contains(self):
+        self.mock_adapter._keys = {}
+        for sprefix in [None, 'myprefix']:
+            pluggable_rbs_storage = PluggableRuleBasedSegmentsStorage(self.mock_adapter, prefix=sprefix)
+            rbs1 = rule_based_segments.from_raw(rbsegments_json['segment1'])
+            rbs2_temp = copy.deepcopy(rbsegments_json['segment1'])
+            rbs2_temp['name'] = 'another_segment'
+            rbs2 = rule_based_segments.from_raw(rbs2_temp)
+            self.mock_adapter.set(pluggable_rbs_storage._prefix.format(segment_name=rbs1.name), rbs1.to_json())
+            self.mock_adapter.set(pluggable_rbs_storage._prefix.format(segment_name=rbs2.name), rbs2.to_json())
+
+            assert(pluggable_rbs_storage.contains([rbs1.name, rbs2.name]))
+            assert(pluggable_rbs_storage.contains([rbs2.name]))
+            assert(not pluggable_rbs_storage.contains(['none-exists', rbs2.name]))
+            assert(not pluggable_rbs_storage.contains(['none-exists', 'none-exists2']))
+
+class PluggableRuleBasedSegmentStorageAsyncTests(object):
+    """In memory rule based segment storage test cases."""
+
+    def setup_method(self):
+        """Prepare storages with test data."""
+        self.mock_adapter = StorageMockAdapterAsync()
+
+    @pytest.mark.asyncio
+    async def test_get(self):
+        self.mock_adapter._keys = {}
+        for sprefix in [None, 'myprefix']:
+            pluggable_rbs_storage = PluggableRuleBasedSegmentsStorageAsync(self.mock_adapter, prefix=sprefix)
+
+            rbs1 = rule_based_segments.from_raw(rbsegments_json['segment1'])
+            rbs_name = rbsegments_json['segment1']['name']
+
+            await self.mock_adapter.set(pluggable_rbs_storage._prefix.format(segment_name=rbs_name), rbs1.to_json())
+            rbs = await pluggable_rbs_storage.get(rbs_name)
+            assert(rbs.to_json() ==  rule_based_segments.from_raw(rbsegments_json['segment1']).to_json())
+            assert(await pluggable_rbs_storage.get('not_existing') == None)
+
+    @pytest.mark.asyncio
+    async def test_get_change_number(self):
+        self.mock_adapter._keys = {}
+        for sprefix in [None, 'myprefix']:
+            pluggable_rbs_storage = PluggableRuleBasedSegmentsStorageAsync(self.mock_adapter, prefix=sprefix)
+            if sprefix == 'myprefix':
+                prefix = 'myprefix.'
+            else:
+                prefix = ''
+            await self.mock_adapter.set(prefix + "SPLITIO.rbsegments.till", 1234)
+            assert(await pluggable_rbs_storage.get_change_number() == 1234)
+
+    @pytest.mark.asyncio
+    async def test_get_segment_names(self):
+        self.mock_adapter._keys = {}
+        for sprefix in [None, 'myprefix']:
+            pluggable_rbs_storage = PluggableRuleBasedSegmentsStorageAsync(self.mock_adapter, prefix=sprefix)
+            rbs1 = rule_based_segments.from_raw(rbsegments_json['segment1'])
+            rbs2_temp = copy.deepcopy(rbsegments_json['segment1'])
+            rbs2_temp['name'] = 'another_segment'
+            rbs2 = rule_based_segments.from_raw(rbs2_temp)
+            await self.mock_adapter.set(pluggable_rbs_storage._prefix.format(segment_name=rbs1.name), rbs1.to_json())
+            await self.mock_adapter.set(pluggable_rbs_storage._prefix.format(segment_name=rbs2.name), rbs2.to_json())
+            assert(await pluggable_rbs_storage.get_segment_names() == [rbs1.name, rbs2.name])
+
+    @pytest.mark.asyncio
+    async def test_contains(self):
+        self.mock_adapter._keys = {}
+        for sprefix in [None, 'myprefix']:
+            pluggable_rbs_storage = PluggableRuleBasedSegmentsStorageAsync(self.mock_adapter, prefix=sprefix)
+            rbs1 = rule_based_segments.from_raw(rbsegments_json['segment1'])
+            rbs2_temp = copy.deepcopy(rbsegments_json['segment1'])
+            rbs2_temp['name'] = 'another_segment'
+            rbs2 = rule_based_segments.from_raw(rbs2_temp)
+            await self.mock_adapter.set(pluggable_rbs_storage._prefix.format(segment_name=rbs1.name), rbs1.to_json())
+            await self.mock_adapter.set(pluggable_rbs_storage._prefix.format(segment_name=rbs2.name), rbs2.to_json())
+
+            assert(await pluggable_rbs_storage.contains([rbs1.name, rbs2.name]))
+            assert(await pluggable_rbs_storage.contains([rbs2.name]))
+            assert(not await pluggable_rbs_storage.contains(['none-exists', rbs2.name]))
+            assert(not await pluggable_rbs_storage.contains(['none-exists', 'none-exists2']))