Merge pull request #314 from splitio/release-9.4.0

Release 9.4.0

Author: chillaq

CHANGES.txt

@@ -1,4 +1,4 @@
-9.4.0 (Feb 14, 2023)
+9.4.0 (Mar 1, 2023)
 - Added support to use JSON files in localhost mode.
 - Updated default periodic telemetry post time to one hour.
 - Fixed unhandeled exception in push.manager.py class when SDK is connected to split proxy

splitio/sync/segment.py

@@ -1,6 +1,7 @@
 import logging
 import time
 import json
+import os
 
 from splitio.api import APIException
 from splitio.api.commons import FetchOptions
@@ -254,24 +255,22 @@ def synchronize_segment(self, segment_name, till=None):
         """
         try:
             fetched = self._read_segment_from_json_file(segment_name)
-            if fetched is None:
-                return False
             fetched_sha = util._get_sha(json.dumps(fetched))
             if not self.segment_exist_in_storage(segment_name):
                     self._segment_sha[segment_name] = fetched_sha
                     self._segment_storage.put(segments.from_raw(fetched))
                     _LOGGER.debug("segment %s is added to storage", segment_name)
-            else:
-                if fetched_sha != self._segment_sha[segment_name]:
-                    self._segment_sha[segment_name] = fetched_sha
-                    if self._segment_storage.get_change_number(segment_name) <= fetched['till'] or fetched['till'] == self._DEFAULT_SEGMENT_TILL:
-                        self._segment_storage.update(
-                            segment_name,
-                            fetched['added'],
-                            fetched['removed'],
-                            fetched['till']
-                        )
-                        _LOGGER.debug("segment %s is updated", segment_name)
+                    return True
+
+            if fetched_sha == self._segment_sha[segment_name]:
+                return True
+
+            self._segment_sha[segment_name] = fetched_sha
+            if self._segment_storage.get_change_number(segment_name) > fetched['till'] and fetched['till'] != self._DEFAULT_SEGMENT_TILL:
+                return True
+
+            self._segment_storage.update(segment_name, fetched['added'], fetched['removed'], fetched['till'])
+            _LOGGER.debug("segment %s is updated", segment_name)
         except Exception as e:
             _LOGGER.error("Could not fetch segment: %s \n" + str(e), segment_name)
             return False
@@ -289,7 +288,7 @@ def _read_segment_from_json_file(self, filename):
         :rtype: Dict
         """
         try:
-            with open(self._segment_folder + '/' + filename + '.json', 'r') as flo:
+            with open(os.path.join(self._segment_folder, "%s.json" % filename), 'r') as flo:
                 parsed = json.load(flo)
             santitized_segment = self._sanitize_segment(parsed)
             return santitized_segment
@@ -308,10 +307,10 @@ def _sanitize_segment(self, parsed):
         """
         if 'name' not in parsed or parsed['name'] is None:
             _LOGGER.warning("Segment does not have [name] element, skipping")
-            return None
+            raise Exception("Segment does not have [name] element")
         if parsed['name'].strip() == '':
             _LOGGER.warning("Segment [name] element is blank, skipping")
-            return None
+            raise Exception("Segment [name] element is blank")
 
         for element in [('till', -1, -1, None, None, [0]),
                         ('added', [], None, None, None, None),

splitio/sync/split.py

@@ -323,18 +323,11 @@ def synchronize_splits(self, till=None):  # pylint:disable=unused-argument
         """Update splits in storage."""
         _LOGGER.info('Synchronizing splits now.')
         try:
-            if self._localhost_mode == LocalhostMode.JSON:
-                return self._synchronize_json()
-            else:
-                return self._synchronize_legacy()
+            return self._synchronize_json() if self._localhost_mode == LocalhostMode.JSON else self._synchronize_legacy()
         except Exception as exc:
             _LOGGER.error(str(exc))
             raise APIException("Error fetching splits information") from exc
 
-#            _LOGGER.error("Error fetching splits information")
-#            _LOGGER.error(str(e))
-#        return []
-
     def _synchronize_legacy(self):
         """
         Update splits in storage for legacy mode.
@@ -368,19 +361,21 @@ def _synchronize_json(self):
             fetched, till = self._read_splits_from_json_file(self._filename)
             segment_list = set()
             fecthed_sha = util._get_sha(json.dumps(fetched))
-            if fecthed_sha != self._current_json_sha:
-                self._current_json_sha = fecthed_sha
-                if self._split_storage.get_change_number() <= till or till == self._DEFAULT_SPLIT_TILL:
-                    for split in fetched:
-                        if split['status'] == splits.Status.ACTIVE.value:
-                            parsed = splits.from_raw(split)
-                            self._split_storage.put(parsed)
-                            _LOGGER.debug("split %s is updated", parsed.name)
-                            segment_list.update(set(parsed.get_segment_names()))
-                        else:
-                            self._split_storage.remove(split['name'])
-
-                    self._split_storage.set_change_number(till)
+            if fecthed_sha == self._current_json_sha:
+                return []
+            self._current_json_sha = fecthed_sha
+            if self._split_storage.get_change_number() > till and till != self._DEFAULT_SPLIT_TILL:
+                return []
+            for split in fetched:
+                if split['status'] == splits.Status.ACTIVE.value:
+                    parsed = splits.from_raw(split)
+                    self._split_storage.put(parsed)
+                    _LOGGER.debug("split %s is updated", parsed.name)
+                    segment_list.update(set(parsed.get_segment_names()))
+                else:
+                    self._split_storage.remove(split['name'])
+
+                self._split_storage.set_change_number(till)
             return segment_list
         except Exception as exc:
             raise ValueError("Error reading splits from json.") from exc
@@ -392,14 +387,13 @@ def _read_splits_from_json_file(self, filename):
         :param filename: Path of the file containing split
         :type filename: str.
 
-        :return: Tuple: sanitized split structure dict, since and till
-        :rtype: Tuple(Dict, int, int)
+        :return: Tuple: sanitized split structure dict and till
+        :rtype: Tuple(Dict, int)
         """
         try:
             with open(filename, 'r') as flo:
                 parsed = json.load(flo)
             santitized = self._sanitize_split(parsed)
-            flo.close
             return santitized['splits'], santitized['till']
         except Exception as exc:
             _LOGGER.error(str(exc))
@@ -464,11 +458,11 @@ def _sanitize_split_elements(self, parsed_splits):
                             ('changeNumber', 0, 0, None, None, None),
                             ('algo', 2, 2, 2, None, None)]:
                 split = util._sanitize_object_element(split, 'split', element[0], element[1], lower_value=element[2], upper_value=element[3], in_list=element[4], not_in_list=element[5])
-            split = self._santizie_condition(split)
+            split = self._sanitize_condition(split)
             sanitized_splits.append(split)
         return sanitized_splits
 
-    def _santizie_condition(self, split):
+    def _sanitize_condition(self, split):
         """
         Sanitize split and ensure a condition type ROLLOUT and matcher exist with ALL_KEYS elements.
 
@@ -479,8 +473,7 @@ def _santizie_condition(self, split):
         :rtype: Dict
         """
         found_all_keys_matcher = False
-        if 'conditions' not in split or split['conditions'] is None:
-            split['conditions'] = []
+        split['conditions'] = split.get('conditions', [])
         if len(split['conditions']) > 0:
             last_condition = split['conditions'][-1]
             if 'conditionType' in last_condition:

splitio/version.py

@@ -1 +1 @@
-__version__ = '9.4.0-rc2'
+__version__ = '9.4.0'

tests/sync/test_manager.py

@@ -3,6 +3,7 @@
 import threading
 import unittest.mock as mock
 import time
+import pytest
 
 from splitio.api.auth import AuthAPI
 from splitio.api import auth, client, APIException
@@ -73,26 +74,23 @@ def test_start_streaming_false(self, mocker):
         assert len(synchronizer.start_periodic_data_recording.mock_calls) == 1
 
     def test_telemetry(self, mocker):
-        httpclient = mocker.Mock(spec=client.HttpClient)
-        token = "eyJhbGciOiJIUzI1NiIsImtpZCI6IjVZOU05US45QnJtR0EiLCJ0eXAiOiJKV1QifQ.eyJ4LWFibHktY2FwYWJpbGl0eSI6IntcIk56TTJNREk1TXpjMF9NVGd5TlRnMU1UZ3dOZz09X3NlZ21lbnRzXCI6W1wic3Vic2NyaWJlXCJdLFwiTnpNMk1ESTVNemMwX01UZ3lOVGcxTVRnd05nPT1fc3BsaXRzXCI6W1wic3Vic2NyaWJlXCJdLFwiY29udHJvbF9wcmlcIjpbXCJzdWJzY3JpYmVcIixcImNoYW5uZWwtbWV0YWRhdGE6cHVibGlzaGVyc1wiXSxcImNvbnRyb2xfc2VjXCI6W1wic3Vic2NyaWJlXCIsXCJjaGFubmVsLW1ldGFkYXRhOnB1Ymxpc2hlcnNcIl19IiwieC1hYmx5LWNsaWVudElkIjoiY2xpZW50SWQiLCJleHAiOjE2MDIwODgxMjcsImlhdCI6MTYwMjA4NDUyN30.5_MjWonhs6yoFhw44hNJm3H7_YMjXpSW105DwjjppqE"
-        payload = '{{"pushEnabled": true, "token": "{token}"}}'.format(token=token)
-        cfg = DEFAULT_CONFIG.copy()
-        cfg.update({'IPAddressesEnabled': True, 'machineName': 'some_machine_name', 'machineIp': '123.123.123.123'})
-        sdk_metadata = get_metadata(cfg)
-        httpclient.get.return_value = client.HttpResponse(200, payload)
-        telemetry_storage = InMemoryTelemetryStorage()
-        telemetry_producer = TelemetryStorageProducer(telemetry_storage)
-        telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
-        auth_api = auth.AuthAPI(httpclient, 'some_api_key', sdk_metadata, telemetry_runtime_producer)
         splits_ready_event = threading.Event()
         synchronizer = mocker.Mock(spec=Synchronizer)
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
         telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
-        manager = Manager(splits_ready_event, synchronizer, auth_api, True, sdk_metadata, telemetry_runtime_producer)
-        manager.start()
+        manager = Manager(splits_ready_event, synchronizer, mocker.Mock(), True, SdkMetadata('1.0', 'some', '1.2.3.4'), telemetry_runtime_producer)
+        try:
+            manager.start()
+        except:
+            pass
+        splits_ready_event.wait(2)
+
+        manager._queue.put(Status.PUSH_SUBSYSTEM_UP)
+        manager._queue.put(Status.PUSH_NONRETRYABLE_ERROR)
         time.sleep(1)
-        manager._push_status_handler_active = True
+        assert(telemetry_storage._streaming_events._streaming_events[len(telemetry_storage._streaming_events._streaming_events)-2]._type == StreamingEventTypes.SYNC_MODE_UPDATE.value)
+        assert(telemetry_storage._streaming_events._streaming_events[len(telemetry_storage._streaming_events._streaming_events)-2]._data == SSESyncMode.STREAMING.value)
         assert(telemetry_storage._streaming_events._streaming_events[len(telemetry_storage._streaming_events._streaming_events)-1]._type == StreamingEventTypes.SYNC_MODE_UPDATE.value)
         assert(telemetry_storage._streaming_events._streaming_events[len(telemetry_storage._streaming_events._streaming_events)-1]._data == SSESyncMode.POLLING.value)
 

tests/sync/test_segments_synchronizer.py

@@ -305,11 +305,21 @@ def test_json_elements_sanitization(self, mocker):
 
         # should reject segment if 'name' is null
         segment2 = {"name": None, "added": [], "removed": [], "since": -1, "till": 12}
-        assert(segment_synchronizer._sanitize_segment(segment2) == None)
+        exception_called = False
+        try:
+            segment_synchronizer._sanitize_segment(segment2)
+        except:
+            exception_called = True
+        assert(exception_called)
 
         # should reject segment if 'name' does not exist
         segment2 = {"added": [], "removed": [], "since": -1, "till": 12}
-        assert(segment_synchronizer._sanitize_segment(segment2) == None)
+        exception_called = False
+        try:
+            segment_synchronizer._sanitize_segment(segment2)
+        except:
+            exception_called = True
+        assert(exception_called)
 
         # should add missing 'added' element
         segment2 = {"name": 'seg', "removed": [], "since": -1, "till": 12}

tests/sync/test_splits_synchronizer.py

@@ -500,7 +500,7 @@ def test_split_condition_sanitization(self, mocker):
         target_split = splits_json["splitChange1_1"]["splits"].copy()
         target_split[0]["conditions"][0]['partitions'][0]['size'] = 0
         target_split[0]["conditions"][0]['partitions'][1]['size'] = 100
-        split[0]["conditions"] = None
+        del split[0]["conditions"]
         assert (split_synchronizer._sanitize_split_elements(split) == target_split)
 
         # test missing ALL_KEYS condition matcher with default rule set to 100% off