Merge branch 'development' into Feature/FlagSets

Author: chillaq

CHANGES.txt

@@ -1,3 +1,7 @@
+9.5.1 (Sep 5, 2023)
+- Exclude tests from when building the package
+- Fixed exception when fetching telemetry stats if no SSE Feature flags update events are stored
+
 9.5.0 (Jul 18, 2023)
 - Improved streaming architecture implementation to apply feature flag updates from the notification received which is now enhanced, improving efficiency and reliability of the whole update system.
 

setup.py

@@ -53,5 +53,5 @@
         'Programming Language :: Python :: 3',
         'Topic :: Software Development :: Libraries'
     ],
-    packages=find_packages()
+    packages=find_packages(exclude=('tests', 'tests.*'))
 )

splitio/models/token.py

@@ -58,25 +58,6 @@ def iat(self):
         return self._iat
 
 
-def decode_token(raw_token):
-    """Decode token"""
-    if not 'pushEnabled' in raw_token or not 'token' in raw_token:
-        return None, None, None
-
-    token = raw_token['token']
-    push_enabled = raw_token['pushEnabled']
-    if not push_enabled or len(token.strip()) == 0:
-        return None, None, None
-
-    token_parts = token.split('.')
-    if len(token_parts) < 2:
-        return None, None, None
-
-    to_decode = token_parts[1]
-    decoded_payload = base64.b64decode(to_decode + '='*(-len(to_decode) % 4))
-    return push_enabled, token, json.loads(decoded_payload)
-
-
 def from_raw(raw_token):
     """
     Parse a new token from a raw token response.
@@ -87,5 +68,15 @@ def from_raw(raw_token):
     :return: New token model object
     :rtype: splitio.models.token.Token
     """
-    push_enabled, token, decoded_token = decode_token(raw_token)
-    return None if push_enabled is None else Token(push_enabled, token, json.loads(decoded_token['x-ably-capability']), decoded_token['exp'], decoded_token['iat'])
+    if not 'pushEnabled' in raw_token or not 'token' in raw_token:
+        return Token(False, None, None, None, None)
+    token = raw_token['token']
+    push_enabled = raw_token['pushEnabled']
+    token_parts = token.strip().split('.')
+
+    if not push_enabled or len(token_parts) < 2:
+        return Token(False, None, None, None, None)
+
+    to_decode = token_parts[1]
+    decoded_token = json.loads(base64.b64decode(to_decode + '='*(-len(to_decode) % 4)))
+    return Token(push_enabled, token, json.loads(decoded_token['x-ably-capability']), decoded_token['exp'], decoded_token['iat'])

splitio/tasks/util/asynctask.py

@@ -94,7 +94,7 @@ def _execution_wrapper(self):
                         _LOGGER.debug("Force execution signal received. Running now")
                         if not _safe_run(self._main):
                             _LOGGER.error("An error occurred when executing the task. "
-                                          "Retrying after perio expires")
+                                          "Retrying after period expires")
                         continue
                 except queue.Empty:
                     # If no message was received, the timeout has expired
@@ -104,7 +104,7 @@ def _execution_wrapper(self):
                 if not _safe_run(self._main):
                     _LOGGER.error(
                         "An error occurred when executing the task. "
-                        "Retrying after perio expires"
+                        "Retrying after period expires"
                     )
         finally:
             self._cleanup()

splitio/version.py

@@ -1 +1 @@
-__version__ = '9.6.1-rc1'
+__version__ = '9.6.1'

tests/models/test_telemetry_model.py

@@ -44,6 +44,7 @@ def test_storage_type_and_operation_mode(self, mocker):
     def test_method_latencies(self, mocker):
         method_latencies = MethodLatencies()
 
+        method_latencies.pop_all() # should not raise exception
         for method in ModelTelemetry.MethodExceptionsAndLatencies:
             method_latencies.add_latency(method, 50)
             if method.value == 'treatment':
@@ -116,6 +117,7 @@ def test_method_latencies(self, mocker):
     def test_http_latencies(self, mocker):
         http_latencies = HTTPLatencies()
 
+        http_latencies.pop_all() # should not raise exception
         for resource in ModelTelemetry.HTTPExceptionsAndLatencies:
             if self._get_http_latency(resource, http_latencies) == None:
                 continue
@@ -169,6 +171,7 @@ def _get_http_latency(self, resource, storage):
     def test_method_exceptions(self, mocker):
         method_exception = MethodExceptions()
 
+        exceptions = method_exception.pop_all() # should not raise exception
         [method_exception.add_exception(ModelTelemetry.MethodExceptionsAndLatencies.TREATMENT) for i in range(2)]
         method_exception.add_exception(ModelTelemetry.MethodExceptionsAndLatencies.TREATMENTS)
         method_exception.add_exception(ModelTelemetry.MethodExceptionsAndLatencies.TREATMENT_WITH_CONFIG)
@@ -200,6 +203,7 @@ def test_method_exceptions(self, mocker):
 
     def test_http_errors(self, mocker):
         http_error = HTTPErrors()
+        errors = http_error.pop_all() # should not raise exception
         [http_error.add_error(ModelTelemetry.HTTPExceptionsAndLatencies.SEGMENT, str(i)) for i in [500, 501, 502]]
         [http_error.add_error(ModelTelemetry.HTTPExceptionsAndLatencies.SPLIT, str(i)) for i in [400, 401, 402]]
         http_error.add_error(ModelTelemetry.HTTPExceptionsAndLatencies.IMPRESSION, '502')
@@ -220,6 +224,7 @@ def test_http_errors(self, mocker):
 
     def test_last_synchronization(self, mocker):
         last_synchronization = LastSynchronization()
+        last_synchronization.get_all() # should not raise exception
         last_synchronization.add_latency(ModelTelemetry.HTTPExceptionsAndLatencies.SPLIT, 10)
         last_synchronization.add_latency(ModelTelemetry.HTTPExceptionsAndLatencies.IMPRESSION, 20)
         last_synchronization.add_latency(ModelTelemetry.HTTPExceptionsAndLatencies.SEGMENT, 40)
@@ -240,19 +245,27 @@ def test_telemetry_counters(self):
         assert(telemetry_counter._token_refreshes == 0)
         assert(telemetry_counter._update_from_sse == {})
 
+        assert(telemetry_counter.get_session_length() == 0)
         telemetry_counter.record_session_length(20)
         assert(telemetry_counter.get_session_length() == 20)
 
+        assert(telemetry_counter.pop_auth_rejections() == 0)
         [telemetry_counter.record_auth_rejections() for i in range(5)]
         auth_rejections = telemetry_counter.pop_auth_rejections()
         assert(telemetry_counter._auth_rejections == 0)
         assert(auth_rejections == 5)
 
+        assert(telemetry_counter.pop_token_refreshes() == 0)
         [telemetry_counter.record_token_refreshes() for i in range(3)]
         token_refreshes = telemetry_counter.pop_token_refreshes()
         assert(telemetry_counter._token_refreshes == 0)
         assert(token_refreshes == 3)
 
+        assert(telemetry_counter.get_counter_stats(ModelTelemetry.CounterConstants.IMPRESSIONS_QUEUED) == 0)
+        assert(telemetry_counter.get_counter_stats(ModelTelemetry.CounterConstants.IMPRESSIONS_DEDUPED) == 0)
+        assert(telemetry_counter.get_counter_stats(ModelTelemetry.CounterConstants.IMPRESSIONS_DROPPED) == 0)
+        assert(telemetry_counter.get_counter_stats(ModelTelemetry.CounterConstants.EVENTS_QUEUED) == 0)
+        assert(telemetry_counter.get_counter_stats(ModelTelemetry.CounterConstants.EVENTS_DROPPED) == 0)
         telemetry_counter.record_impressions_value(ModelTelemetry.CounterConstants.IMPRESSIONS_QUEUED, 10)
         assert(telemetry_counter._impressions_queued == 10)
         telemetry_counter.record_impressions_value(ModelTelemetry.CounterConstants.IMPRESSIONS_DEDUPED, 14)
@@ -263,6 +276,7 @@ def test_telemetry_counters(self):
         assert(telemetry_counter._events_queued == 30)
         telemetry_counter.record_events_value(ModelTelemetry.CounterConstants.EVENTS_DROPPED, 1)
         assert(telemetry_counter._events_dropped == 1)
+        assert(telemetry_counter.pop_update_from_sse(UpdateFromSSE.SPLIT_UPDATE) == 0)
         telemetry_counter.record_update_from_sse(UpdateFromSSE.SPLIT_UPDATE)
         assert(telemetry_counter._update_from_sse[UpdateFromSSE.SPLIT_UPDATE.value] == 1)
         updates = telemetry_counter.pop_update_from_sse(UpdateFromSSE.SPLIT_UPDATE)
@@ -277,6 +291,7 @@ def test_streaming_event(self, mocker):
 
     def test_streaming_events(self, mocker):
         streaming_events = StreamingEvents()
+        events = streaming_events.pop_streaming_events()  # should not raise exception
         streaming_events.record_streaming_event((ModelTelemetry.StreamingEventTypes.CONNECTION_ESTABLISHED, 'split', 1234))
         streaming_events.record_streaming_event((ModelTelemetry.StreamingEventTypes.STREAMING_STATUS, 'split', 1234))
         events = streaming_events.pop_streaming_events()
@@ -286,6 +301,7 @@ def test_streaming_events(self, mocker):
 
     def test_telemetry_config(self):
         telemetry_config = TelemetryConfig()
+        stats = telemetry_config.get_stats() # should not raise exception
         config = {'operationMode': 'standalone',
                   'streamingEnabled': True,
                   'impressionsQueueSize': 100,
@@ -326,9 +342,11 @@ def test_telemetry_config(self):
         telemetry_config.record_flag_sets(5)
         assert(telemetry_config._flag_sets == 5)
 
+        assert(telemetry_config.get_bur_time_outs() == 0)
         [telemetry_config.record_bur_time_out() for i in range(2)]
         assert(telemetry_config.get_bur_time_outs() == 2)
 
+        assert(telemetry_config.get_non_ready_usage() == 0)
         [telemetry_config.record_not_ready_usage() for i in range(5)]
         assert(telemetry_config.get_non_ready_usage() == 5)
 

tests/models/test_token.py

@@ -11,8 +11,12 @@ class TokenTests(object):
     def test_from_raw_false(self):
         """Test token model parsing."""
         parsed = token.from_raw(self.raw_false)
-        assert parsed == None
-    
+        assert parsed.push_enabled == False
+        assert parsed.iat == None
+        assert parsed.channels == None
+        assert parsed.exp == None
+        assert parsed.token == None
+
     raw_empty = {
         'pushEnabled': True,
         'token': '',
@@ -21,7 +25,11 @@ def test_from_raw_false(self):
     def test_from_raw_empty(self):
         """Test token model parsing."""
         parsed = token.from_raw(self.raw_empty)
-        assert parsed == None
+        assert parsed.push_enabled == False
+        assert parsed.iat == None
+        assert parsed.channels == None
+        assert parsed.exp == None
+        assert parsed.token == None
 
     raw_ok = {
         'pushEnabled': True,
@@ -39,4 +47,3 @@ def test_from_raw(self):
         assert parsed.channels['NzM2MDI5Mzc0_MTgyNTg1MTgwNg==_splits'] == ['subscribe']
         assert parsed.channels['control_pri'] == ['subscribe', 'channel-metadata:publishers']
         assert parsed.channels['control_sec'] == ['subscribe', 'channel-metadata:publishers']
-

tests/push/test_manager.py

@@ -90,6 +90,28 @@ def new_start(*args, **kwargs):  # pylint: disable=unused-argument
         assert feedback_loop.get() == Status.PUSH_RETRYABLE_ERROR
         assert timer_mock.mock_calls == [mocker.call(0, Any())]
 
+    def test_empty_auth_respnse(self, mocker):
+        """Test the initial status is ok and reset() works as expected."""
+        api_mock = mocker.Mock()
+        api_mock.authenticate.return_value = Token(False, None, None, None, None)
+
+        sse_mock = mocker.Mock(spec=SplitSSEClient)
+        sse_constructor_mock = mocker.Mock()
+        sse_constructor_mock.return_value = sse_mock
+        timer_mock = mocker.Mock()
+        mocker.patch('splitio.push.manager.Timer', new=timer_mock)
+        mocker.patch('splitio.push.manager.SplitSSEClient', new=sse_constructor_mock)
+        feedback_loop = Queue()
+        telemetry_storage = InMemoryTelemetryStorage()
+        telemetry_producer = TelemetryStorageProducer(telemetry_storage)
+        telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
+        manager = PushManager(api_mock, mocker.Mock(), feedback_loop, mocker.Mock(), telemetry_runtime_producer)
+        manager.start()
+        assert feedback_loop.get() == Status.PUSH_NONRETRYABLE_ERROR
+        assert timer_mock.mock_calls == [mocker.call(0, Any())]
+        assert sse_mock.mock_calls == []
+
+
     def test_push_disabled(self, mocker):
         """Test the initial status is ok and reset() works as expected."""
         api_mock = mocker.Mock()