Merge pull request #379 from splitio/iff-change-number-check

Added discard split event if change number is null

Author: chillaq

splitio/push/parser.py

@@ -503,9 +503,9 @@ def _parse_update(channel, timestamp, data):
     """
     update_type = UpdateType(data['type'])
     change_number = data['changeNumber']
-    if update_type == UpdateType.SPLIT_UPDATE:
+    if update_type == UpdateType.SPLIT_UPDATE and change_number is not None:
         return SplitChangeUpdate(channel, timestamp, change_number, data.get('pcn'), data.get('d'), data.get('c'))
-    elif update_type == UpdateType.SPLIT_KILL:
+    elif update_type == UpdateType.SPLIT_KILL and change_number is not None:
         return SplitKillUpdate(channel, timestamp, change_number,
                                data['splitName'], data['defaultTreatment'])
     elif update_type == UpdateType.SEGMENT_UPDATE:

splitio/push/splitworker.py

@@ -54,6 +54,12 @@ def _get_feature_flag_definition(self, event):
         cm = CompressionMode(event.compression) # will throw if the number is not defined in compression mode
         return self._compression_handlers[cm](event)
 
+    def _check_instant_ff_update(self, event):
+        if event.update_type == UpdateType.SPLIT_UPDATE and event.compression is not None and event.previous_change_number == self._feature_flag_storage.get_change_number():
+            return True
+        return False
+
+
     def _run(self):
         """Run worker handler."""
         while self.is_running():
@@ -64,17 +70,16 @@ def _run(self):
                 continue
             _LOGGER.debug('Processing feature flag update %d', event.change_number)
             try:
-                if event.update_type == UpdateType.SPLIT_UPDATE:
-                    if event.compression is not None and event.previous_change_number == self._feature_flag_storage.get_change_number():
-                        try:
-                            self._feature_flag_storage.put(from_raw(json.loads(self._get_feature_flag_definition(event))))
-                            self._feature_flag_storage.set_change_number(event.change_number)
-                            continue
-                        except Exception as e:
-                            _LOGGER.error('Exception raised in updating feature flag')
-                            _LOGGER.debug(str(e))
-                            _LOGGER.debug('Exception information: ', exc_info=True)
-                            pass
+                if self._check_instant_ff_update(event):
+                    try:
+                        self._feature_flag_storage.put(from_raw(json.loads(self._get_feature_flag_definition(event))))
+                        self._feature_flag_storage.set_change_number(event.change_number)
+                        continue
+                    except Exception as e:
+                        _LOGGER.error('Exception raised in updating feature flag')
+                        _LOGGER.debug(str(e))
+                        _LOGGER.debug('Exception information: ', exc_info=True)
+                        pass
                 self._handler(event.change_number)
             except Exception as e:  # pylint: disable=broad-except
                 _LOGGER.error('Exception raised in feature flag synchronization')

tests/integration/test_streaming_e2e.py

@@ -109,20 +109,10 @@ def test_happiness(self):
         assert factory.client().get_treatment('pindon', 'split2') == 'off'
         assert factory.client().get_treatment('maldo', 'split2') == 'on'
 
-        # test if changeNumber is missing
-#        split_changes = make_split_fast_change_event(4)
-#        data = json.loads(split_changes['data'])
-#        inner_data = json.loads(data['data'])
-#        inner_data['changeNumber'] = None
-#        data['data'] = json.dumps(inner_data)
-#        split_changes['data'] = json.dumps(data)
-#        sse_server.publish(split_changes)
-#        time.sleep(1)
-#        assert factory.client().get_treatment('maldo', 'split1') == 'off'
-
         sse_server.publish(make_split_fast_change_event(4))
         time.sleep(1)
-        assert factory.client().get_treatment('maldo', 'split1') == 'on'
+        assert factory.client().get_treatment('maldo', 'split5') == 'on'
+
 
         # Validate the SSE request
         sse_request = sse_requests.get()
@@ -1233,6 +1223,66 @@ def test_ably_errors_handling(self):
         sse_server.stop()
         split_backend.stop()
 
+    def test_change_number(mocker):
+        # test if changeNumber is missing
+        auth_server_response = {
+            'pushEnabled': True,
+            'token': ('eyJhbGciOiJIUzI1NiIsImtpZCI6IjVZOU05US45QnJtR0EiLCJ0eXAiOiJKV1QifQ.'
+                      'eyJ4LWFibHktY2FwYWJpbGl0eSI6IntcIk1UWXlNVGN4T1RRNE13PT1fTWpBNE16Y3pO'
+                      'RFUxTWc9PV9zZWdtZW50c1wiOltcInN1YnNjcmliZVwiXSxcIk1UWXlNVGN4T1RRNE13P'
+                      'T1fTWpBNE16Y3pORFUxTWc9PV9zcGxpdHNcIjpbXCJzdWJzY3JpYmVcIl0sXCJjb250cm'
+                      '9sX3ByaVwiOltcInN1YnNjcmliZVwiLFwiY2hhbm5lbC1tZXRhZGF0YTpwdWJsaXNoZXJ'
+                      'zXCJdLFwiY29udHJvbF9zZWNcIjpbXCJzdWJzY3JpYmVcIixcImNoYW5uZWwtbWV0YWRh'
+                      'dGE6cHVibGlzaGVyc1wiXX0iLCJ4LWFibHktY2xpZW50SWQiOiJjbGllbnRJZCIsImV4c'
+                      'CI6MTYwNDEwMDU5MSwiaWF0IjoxNjA0MDk2OTkxfQ.aP9BfR534K6J9h8gfDWg_CQgpz5E'
+                      'vJh17WlOlAKhcD0')
+        }
+
+        split_changes = {
+            -1: {
+                'since': -1,
+                'till': 1,
+                'splits': [make_simple_split('split1', 1, True, False, 'off', 'user', True)]
+            },
+            1: {'since': 1, 'till': 1, 'splits': []}
+        }
+
+        segment_changes = {}
+        split_backend_requests = Queue()
+        split_backend = SplitMockServer(split_changes, segment_changes, split_backend_requests,
+                                        auth_server_response)
+        sse_requests = Queue()
+        sse_server = SSEMockServer(sse_requests)
+
+        split_backend.start()
+        sse_server.start()
+        sse_server.publish(make_initial_event())
+        sse_server.publish(make_occupancy('control_pri', 2))
+        sse_server.publish(make_occupancy('control_sec', 2))
+
+        kwargs = {
+            'sdk_api_base_url': 'http://localhost:%d/api' % split_backend.port(),
+            'events_api_base_url': 'http://localhost:%d/api' % split_backend.port(),
+            'auth_api_base_url': 'http://localhost:%d/api' % split_backend.port(),
+            'streaming_api_base_url': 'http://localhost:%d' % sse_server.port(),
+            'config': {'connectTimeout': 10000, 'featuresRefreshRate': 10}
+        }
+
+        factory = get_factory('some_apikey', **kwargs)
+        factory.block_until_ready(1)
+        assert factory.ready
+        time.sleep(2)
+
+        split_changes = make_split_fast_change_event(5).copy()
+        data = json.loads(split_changes['data'])
+        inner_data = json.loads(data['data'])
+        inner_data['changeNumber'] = None
+        data['data'] = json.dumps(inner_data)
+        split_changes['data'] = json.dumps(data)
+        sse_server.publish(split_changes)
+        time.sleep(1)
+        assert factory._storages['splits'].get_change_number() == 1
+
 
 def make_split_change_event(change_number):
     """Make a split change event."""
@@ -1253,7 +1303,7 @@ def make_split_change_event(change_number):
 
 def make_split_fast_change_event(change_number):
     """Make a split change event."""
-    json1 = make_simple_split('split1', 1, True, False, 'off', 'user', True)
+    json1 = make_simple_split('split5', 1, True, False, 'off', 'user', True)
     str1 = json.dumps(json1)
     byt1 = bytes(str1, encoding='utf-8')
     compressed = base64.b64encode(byt1)