Updated push.splitworker and sync.split

Author: chillaq

splitio/push/splitworker.py

@@ -10,7 +10,7 @@
 from splitio.models.splits import from_raw, Status
 from splitio.models.telemetry import UpdateFromSSE
 from splitio.push.parser import UpdateType
-
+from splitio.util.storage_helper import update_feature_flag_storage
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -88,17 +88,20 @@ def _run(self):
             try:
                 if self._check_instant_ff_update(event):
                     try:
-                        new_split = from_raw(json.loads(self._get_feature_flag_definition(event)))
+                        new_feature_flag = from_raw(json.loads(self._get_feature_flag_definition(event)))
+                        segment_list = update_feature_flag_storage(self._feature_flag_storage, [new_feature_flag], event.change_number)
+                        '''
                         if new_split.status == Status.ACTIVE:
                             self._feature_flag_storage.put(new_split)
                             _LOGGER.debug('Feature flag %s is updated', new_split.name)
-                            for segment_name in new_split.get_segment_names():
-                                if self._segment_storage.get(segment_name) is None:
-                                    _LOGGER.debug('Fetching new segment %s', segment_name)
-                                    self._segment_handler(segment_name, event.change_number)
                         else:
                             self._feature_flag_storage.remove(new_split.name)
                         self._feature_flag_storage.set_change_number(event.change_number)
+                        '''
+                        for segment_name in segment_list:
+                            if self._segment_storage.get(segment_name) is None:
+                                _LOGGER.debug('Fetching new segment %s', segment_name)
+                                self._segment_handler(segment_name, event.change_number)
                         self._telemetry_runtime_producer.record_update_from_sse(UpdateFromSSE.SPLIT_UPDATE)
                         continue
                     except Exception as e:

splitio/sync/split.py

@@ -13,6 +13,7 @@
 from splitio.models import splits
 from splitio.util.backoff import Backoff
 from splitio.util.time import get_current_epoch_time_ms
+from splitio.util.storage_helper import update_feature_flag_storage
 from splitio.sync import util
 
 _LEGACY_COMMENT_LINE_RE = re.compile(r'^#.*$')
@@ -79,7 +80,10 @@ def _fetch_until(self, fetch_options, till=None):
                 _LOGGER.error('Exception raised while fetching feature flags')
                 _LOGGER.debug('Exception information: ', exc_info=True)
                 raise exc
-
+            fetched_feature_flags = []
+            [fetched_feature_flags.append(splits.from_raw(feature_flag)) for feature_flag in feature_flag_changes.get('splits', [])]
+            segment_list = update_feature_flag_storage(self._feature_flag_storage, fetched_feature_flags, feature_flag_changes['till'])
+            '''
             to_add = []
             to_delete = []
             for feature_flag in feature_flag_changes.get('splits', []):
@@ -93,25 +97,10 @@ def _fetch_until(self, fetch_options, till=None):
                         to_delete.append(feature_flag['name'])
 
             self._feature_flag_storage.update(to_add, to_delete, feature_flag_changes['till'])
+            '''
             if feature_flag_changes['till'] == feature_flag_changes['since']:
                 return feature_flag_changes['till'], segment_list
 
-    def _check_flag_sets(self, feature_flag):
-        """
-        Check all flag sets in a feature flag, return True if any of sets exist in storage
-
-        :param feature_flag: Flag set to validate.
-        :type feature_flag: json
-
-        :return: True if any of its flag_set exist. False otherwise.
-        :rtype: bool
-        """
-        for flag_set in feature_flag['sets']:
-            if self._feature_flag_storage.is_flag_set_exist(flag_set):
-                return True
-        return False
-
-
     def _attempt_feature_flag_sync(self, fetch_options, till=None):
         """
         Hit endpoint, update storage and return True if sync is complete.

splitio/util/storage_helper.py

@@ -0,0 +1,49 @@
+"""Storage Helper."""
+
+from splitio.models import splits
+
+def update_feature_flag_storage(feature_flag_storage, feature_flags, change_number):
+    """
+    Update feature flag storage from given list of feature flags while checking the flag set logic
+
+    :param feature_flag_storage: Feature flag storage instance
+    :type feature_flag_storage: splitio.storage.inmemory.InMemorySplitStorage
+    :param feature_flag: Feature flag instance to validate.
+    :type feature_flag: splitio.models.splits.Split
+    :param: last change number
+    :type: int
+
+    :return: segments list from feature flags list
+    :rtype: list(str)
+    """
+    segment_list = set()
+    to_add = []
+    to_delete = []
+    for feature_flag in feature_flags:
+        if (feature_flag_storage.config_flag_sets_used == 0 and feature_flag.status == splits.Status.ACTIVE) or \
+        (feature_flag.status == splits.Status.ACTIVE and _check_flag_sets(feature_flag_storage, feature_flag)):
+            to_add.append(feature_flag)
+            segment_list.update(set(feature_flag.get_segment_names()))
+        else:
+            if feature_flag_storage.get(feature_flag.name) is not None:
+                to_delete.append(feature_flag.name)
+
+    feature_flag_storage.update(to_add, to_delete, change_number)
+    return segment_list
+
+def _check_flag_sets(feature_flag_storage, feature_flag):
+    """
+    Check all flag sets in a feature flag, return True if any of sets exist in storage
+
+    :param feature_flag_storage: Feature flag storage instance
+    :type feature_flag_storage: splitio.storage.inmemory.InMemorySplitStorage
+    :param feature_flag: Feature flag instance to validate.
+    :type feature_flag: splitio.models.splits.Split
+
+    :return: True if any of its flag_set exist. False otherwise.
+    :rtype: bool
+    """
+    for flag_set in feature_flag.sets:
+        if feature_flag_storage.is_flag_set_exist(flag_set):
+            return True
+    return False

tests/push/test_split_worker.py

@@ -55,18 +55,13 @@ def test_handler(self, mocker):
 
         def get_change_number():
             return 2345
-
-        self._feature_flag = None
-        def put(feature_flag):
-            self._feature_flag = feature_flag
+        split_worker._feature_flag_storage.get_change_number = get_change_number
 
         self.new_change_number = 0
-        def set_change_number(new_change_number):
-            self.new_change_number = new_change_number
-
-        split_worker._feature_flag_storage.get_change_number = get_change_number
-        split_worker._feature_flag_storage.set_change_number = set_change_number
-        split_worker._feature_flag_storage.put = put
+        def update(to_add, to_delete, change_number):
+            self.new_change_number = change_number
+        split_worker._feature_flag_storage.update = update
+        split_worker._feature_flag_storage.config_flag_sets_used = 0
 
         # should call the handler
         q.put(SplitChangeUpdate('some', 'SPLIT_UPDATE', 123456790, 12345,  "{}", 1))
@@ -98,45 +93,45 @@ def test_compression(self, mocker):
         split_worker.start()
         def get_change_number():
             return 2345
-
-        def put(feature_flag):
-            self._feature_flag = feature_flag
-
-        def remove(feature_flag):
-            self._feature_flag_delete = feature_flag
-
         split_worker._feature_flag_storage.get_change_number = get_change_number
-        split_worker._feature_flag_storage.put = put
-        split_worker._feature_flag_storage.remove = remove
+
+        self._feature_flag_added = None
+        self._feature_flag_deleted = None
+        def update(feature_flag_add, feature_flag_delete, change_number):
+            self._feature_flag_added = feature_flag_add
+            self._feature_flag_deleted = feature_flag_delete
+        split_worker._feature_flag_storage.update = update
+        split_worker._feature_flag_storage.config_flag_sets_used = 0
 
         # compression 0
-        self._feature_flag = None
+        self._feature_flag_added = None
         q.put(SplitChangeUpdate('some', 'SPLIT_UPDATE', 123456790, 2345,  'eyJ0cmFmZmljVHlwZU5hbWUiOiJ1c2VyIiwiaWQiOiIzM2VhZmE1MC0xYTY1LTExZWQtOTBkZi1mYTMwZDk2OTA0NDUiLCJuYW1lIjoiYmlsYWxfc3BsaXQiLCJ0cmFmZmljQWxsb2NhdGlvbiI6MTAwLCJ0cmFmZmljQWxsb2NhdGlvblNlZWQiOi0xMzY0MTE5MjgyLCJzZWVkIjotNjA1OTM4ODQzLCJzdGF0dXMiOiJBQ1RJVkUiLCJraWxsZWQiOmZhbHNlLCJkZWZhdWx0VHJlYXRtZW50Ijoib2ZmIiwiY2hhbmdlTnVtYmVyIjoxNjg0MzQwOTA4NDc1LCJhbGdvIjoyLCJjb25maWd1cmF0aW9ucyI6e30sImNvbmRpdGlvbnMiOlt7ImNvbmRpdGlvblR5cGUiOiJST0xMT1VUIiwibWF0Y2hlckdyb3VwIjp7ImNvbWJpbmVyIjoiQU5EIiwibWF0Y2hlcnMiOlt7ImtleVNlbGVjdG9yIjp7InRyYWZmaWNUeXBlIjoidXNlciJ9LCJtYXRjaGVyVHlwZSI6IklOX1NFR01FTlQiLCJuZWdhdGUiOmZhbHNlLCJ1c2VyRGVmaW5lZFNlZ21lbnRNYXRjaGVyRGF0YSI6eyJzZWdtZW50TmFtZSI6ImJpbGFsX3NlZ21lbnQifX1dfSwicGFydGl0aW9ucyI6W3sidHJlYXRtZW50Ijoib24iLCJzaXplIjowfSx7InRyZWF0bWVudCI6Im9mZiIsInNpemUiOjEwMH1dLCJsYWJlbCI6ImluIHNlZ21lbnQgYmlsYWxfc2VnbWVudCJ9LHsiY29uZGl0aW9uVHlwZSI6IlJPTExPVVQiLCJtYXRjaGVyR3JvdXAiOnsiY29tYmluZXIiOiJBTkQiLCJtYXRjaGVycyI6W3sia2V5U2VsZWN0b3IiOnsidHJhZmZpY1R5cGUiOiJ1c2VyIn0sIm1hdGNoZXJUeXBlIjoiQUxMX0tFWVMiLCJuZWdhdGUiOmZhbHNlfV19LCJwYXJ0aXRpb25zIjpbeyJ0cmVhdG1lbnQiOiJvbiIsInNpemUiOjB9LHsidHJlYXRtZW50Ijoib2ZmIiwic2l6ZSI6MTAwfV0sImxhYmVsIjoiZGVmYXVsdCBydWxlIn1dfQ==', 0))
         time.sleep(0.1)
-        assert self._feature_flag.name == 'bilal_split'
+#        pytest.set_trace()
+        assert self._feature_flag_added[0].name == 'bilal_split'
         assert telemetry_storage._counters._update_from_sse['sp'] == 1
 
         # compression 2
-        self._feature_flag = None
+        self._feature_flag_added = None
         q.put(SplitChangeUpdate('some', 'SPLIT_UPDATE', 123456790, 2345,  'eJzEUtFq20AQ/JUwz2c4WZZr3ZupTQh1FKjcQinGrKU95cjpZE6nh9To34ssJ3FNX0sfd3Zm53b2TgietDbF9vXIGdUMha5lDwFTQiGOmTQlchLRPJlEEZeTVJZ6oimWZTpP5WyWQMCNyoOxZPft0ZoA8TZ5aW1TUDCNg4qk/AueM5dQkyiez6IonS6mAu0IzWWSxovFLBZoA4WuhcLy8/bh+xoCL8bagaXJtixQsqbOhq1nCjW7AIVGawgUz+Qqzrr6wB4qmi9m00/JIk7TZCpAtmqgpgJF47SpOn9+UQt16s9YaS71z9NHOYQFha9Pm83Tty0EagrFM/t733RHqIFZH4wb7LDMVh+Ecc4Lv+ZsuQiNH8hXF3hLv39XXNCHbJ+v7x/X2eDmuKLA74sPihVr47jMuRpWfxy1Kwo0GLQjmv1xpBFD3+96gSP5cLVouM7QQaA1vxhK9uKmd853bEZS9jsBSwe2UDDu7mJxd2Mo/muQy81m/2X9I7+N8R/FcPmUd76zjH7X/w4AAP//90glTw==', 2))
         time.sleep(0.1)
-        assert self._feature_flag.name == 'bilal_split'
+        assert self._feature_flag_added[0].name == 'bilal_split'
         assert telemetry_storage._counters._update_from_sse['sp'] == 2
 
         # compression 1
-        self._feature_flag = None
+        self._feature_flag_added = None
         q.put(SplitChangeUpdate('some', 'SPLIT_UPDATE', 123456790, 2345,  'H4sIAAkVZWQC/8WST0+DQBDFv0qzZ0ig/BF6a2xjGismUk2MaZopzOKmy9Isy0EbvrtDwbY2Xo233Tdv5se85cCMBs5FtvrYYwIlsglratTMYiKns+chcAgc24UwsF0Xczt2cm5z8Jw8DmPH9wPyqr5zKyTITb2XwpA4TJ5KWWVgRKXYxHWcX/QUkVi264W+68bjaGyxupdCJ4i9KPI9UgyYpibI9Ha1eJnT/J2QsnNxkDVaLEcOjTQrjWBKVIasFefky95BFZg05Zb2mrhh5I9vgsiL44BAIIuKTeiQVYqLotHHLyLOoT1quRjub4fztQuLxj89LpePzytClGCyd9R3umr21ErOcitUh2PTZHY29HN2+JGixMxUujNfvMB3+u2pY1AXySad3z3Mk46msACDp8W7jhly4uUpFt3qD33vDAx0gLpXkx+P1GusbdcE24M2F4uaywwVEWvxSa1Oa13Vjvn2RXradm0xCVuUVBJqNCBGV0DrX4OcLpeb+/lreh3jH8Uw/JQj3UhkxPgCCurdEnADAAA=', 1))
         time.sleep(0.1)
-        assert self._feature_flag.name == 'bilal_split'
+        assert self._feature_flag_added[0].name == 'bilal_split'
         assert telemetry_storage._counters._update_from_sse['sp'] == 3
 
         # should call delete split
-        self._feature_flag = None
-        self._feature_flag_delete = None
+        self._feature_flag_added = None
+        self._feature_flag_deleted = None
         q.put(SplitChangeUpdate('some', 'SPLIT_UPDATE', 123456790, 2345,  'eyJ0cmFmZmljVHlwZU5hbWUiOiAidXNlciIsICJpZCI6ICIzM2VhZmE1MC0xYTY1LTExZWQtOTBkZi1mYTMwZDk2OTA0NDUiLCAibmFtZSI6ICJiaWxhbF9zcGxpdCIsICJ0cmFmZmljQWxsb2NhdGlvbiI6IDEwMCwgInRyYWZmaWNBbGxvY2F0aW9uU2VlZCI6IC0xMzY0MTE5MjgyLCAic2VlZCI6IC02MDU5Mzg4NDMsICJzdGF0dXMiOiAiQVJDSElWRUQiLCAia2lsbGVkIjogZmFsc2UsICJkZWZhdWx0VHJlYXRtZW50IjogIm9mZiIsICJjaGFuZ2VOdW1iZXIiOiAxNjg0Mjc1ODM5OTUyLCAiYWxnbyI6IDIsICJjb25maWd1cmF0aW9ucyI6IHt9LCAiY29uZGl0aW9ucyI6IFt7ImNvbmRpdGlvblR5cGUiOiAiUk9MTE9VVCIsICJtYXRjaGVyR3JvdXAiOiB7ImNvbWJpbmVyIjogIkFORCIsICJtYXRjaGVycyI6IFt7ImtleVNlbGVjdG9yIjogeyJ0cmFmZmljVHlwZSI6ICJ1c2VyIn0sICJtYXRjaGVyVHlwZSI6ICJJTl9TRUdNRU5UIiwgIm5lZ2F0ZSI6IGZhbHNlLCAidXNlckRlZmluZWRTZWdtZW50TWF0Y2hlckRhdGEiOiB7InNlZ21lbnROYW1lIjogImJpbGFsX3NlZ21lbnQifX1dfSwgInBhcnRpdGlvbnMiOiBbeyJ0cmVhdG1lbnQiOiAib24iLCAic2l6ZSI6IDB9LCB7InRyZWF0bWVudCI6ICJvZmYiLCAic2l6ZSI6IDEwMH1dLCAibGFiZWwiOiAiaW4gc2VnbWVudCBiaWxhbF9zZWdtZW50In0sIHsiY29uZGl0aW9uVHlwZSI6ICJST0xMT1VUIiwgIm1hdGNoZXJHcm91cCI6IHsiY29tYmluZXIiOiAiQU5EIiwgIm1hdGNoZXJzIjogW3sia2V5U2VsZWN0b3IiOiB7InRyYWZmaWNUeXBlIjogInVzZXIifSwgIm1hdGNoZXJUeXBlIjogIkFMTF9LRVlTIiwgIm5lZ2F0ZSI6IGZhbHNlfV19LCAicGFydGl0aW9ucyI6IFt7InRyZWF0bWVudCI6ICJvbiIsICJzaXplIjogMH0sIHsidHJlYXRtZW50IjogIm9mZiIsICJzaXplIjogMTAwfV0sICJsYWJlbCI6ICJkZWZhdWx0IHJ1bGUifV19', 0))
         time.sleep(0.1)
-        assert self._feature_flag_delete == 'bilal_split'
-        assert self._feature_flag == None
+        assert self._feature_flag_deleted[0] == 'bilal_split'
+        self._feature_flag_added = None
 
     def test_edge_cases(self, mocker):
         q = queue.Queue()

tests/util/test_storage_helper.py

@@ -0,0 +1,75 @@
+"""Storage Helper tests."""
+
+from splitio.util.storage_helper import update_feature_flag_storage
+from splitio.storage.inmemmory import InMemorySplitStorage
+from splitio.models import splits
+from tests.sync.test_splits_synchronizer import splits as split_sample
+
+class StorageHelperTests(object):
+
+    def test_helper_scenarios(self, mocker):
+        storage = mocker.Mock(spec=InMemorySplitStorage)
+        split = splits.from_raw(split_sample[0])
+
+        self.added = []
+        self.deleted = []
+        self.change_number = 0
+        def update(to_add, to_delete, change_number):
+            self.added = to_add
+            self.deleted = to_delete
+            self.change_number = change_number
+        storage.update = update
+
+        def is_flag_set_exist(flag_set):
+            return False
+        storage.is_flag_set_exist = is_flag_set_exist
+
+        storage.config_flag_sets_used = 0
+        update_feature_flag_storage(storage, [split], 123)
+        assert self.added[0] == split
+        assert self.deleted == []
+        assert self.change_number == 123
+
+        storage.config_flag_sets_used = 2
+        update_feature_flag_storage(storage, [split], 123)
+        assert self.added == []
+        assert self.deleted[0] == split.name
+
+        def is_flag_set_exist2(flag_set):
+            return True
+        storage.is_flag_set_exist = is_flag_set_exist2
+        update_feature_flag_storage(storage, [split], 123)
+        assert self.added[0] == split
+        assert self.deleted == []
+
+        split_json = split_sample[0]
+        split_json['conditions'].append({
+                    "matcherGroup": {
+                        "combiner": "AND",
+                        "matchers": [
+                            {
+                                "matcherType": "IN_SEGMENT",
+                                "negate": False,
+                                "userDefinedSegmentMatcherData": {
+                                    "segmentName": "segment1"
+                                },
+                                "whitelistMatcherData": None
+                            }
+                        ]
+                    },
+                    "partitions": [
+                        {
+                            "treatment": "on",
+                            "size": 30
+                        },
+                        {
+                            "treatment": "off",
+                            "size": 70
+                        }
+                    ]
+                }
+        )
+
+        split = splits.from_raw(split_json)
+        storage.config_flag_sets_used = 0
+        assert update_feature_flag_storage(storage, [split], 123) == {'segment1'}