last integration tests

Author: mredolatti

splitio/client/config.py

@@ -16,8 +16,8 @@
     'splitSdkMachineName': None,
     'splitSdkMachineIp': None,
     'streamingEnabled': True,
-    'featuresRefreshRate': 5,
-    'segmentsRefreshRate': 60,
+    'featuresRefreshRate': 30,
+    'segmentsRefreshRate': 30,
     'metricsRefreshRate': 60,
     'impressionsRefreshRate': 5 * 60,
     'impressionsBulkSize': 5000,

splitio/push/manager.py

@@ -140,8 +140,10 @@ def _trigger_connection_flow(self):
             self._feedback_loop.put(Status.PUSH_NONRETRYABLE_ERROR)
             return
 
+        _LOGGER.debug("auth token fetched. connecting to streaming.")
         self._status_tracker.reset()
         if self._sse_client.start(token):
+            _LOGGER.debug("connected to streaming, scheduling next refresh")
             self._setup_next_token_refresh(token)
             self._running = True
 

splitio/push/splitworker.py

@@ -41,7 +41,7 @@ def _run(self):
             _LOGGER.debug('Processing split_update %d', event.change_number)
             try:
                 self._handler(event.change_number)
-            except Exception:
+            except Exception:  # pylint: disable=broad-except
                 _LOGGER.error('Exception raised in split synchronization')
                 _LOGGER.debug('Exception information: ', exc_info=True)
 

splitio/version.py

@@ -1 +1 @@
-__version__ = '8.3.0-rc5'
+__version__ = '8.3.0'

tests/integration/test_streaming_e2e.py

@@ -1015,6 +1015,243 @@ def test_server_closes_connection(self):
         sse_server.stop()
         split_backend.stop()
 
+    def test_ably_errors_handling(self):
+        """Test incoming ably errors and validate its handling."""
+        import logging
+        logger = logging.getLogger('splitio')
+        handler = logging.StreamHandler()
+        formatter = logging.Formatter('%(asctime)s %(name)-12s %(levelname)-8s %(message)s')
+        handler.setFormatter(formatter)
+        logger.addHandler(handler)
+        logger.setLevel(logging.DEBUG)
+        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)
+
+        # Get a hook of the task so we can query its status
+        task = factory._sync_manager._synchronizer._split_tasks.split_task._task  # pylint:disable=protected-access
+        assert not task.running()
+
+        assert factory.client().get_treatment('maldo', 'split1') == 'on'
+
+        # Make a change in the BE but don't send the event.
+        # We'll send an ignorable error and check it has nothing happened
+        split_changes[1] = {
+            'since': 1,
+            'till': 2,
+            'splits': [make_simple_split('split1', 2, True, False, 'off', 'user', False)]
+        }
+        split_changes[2] = {'since': 2, 'till': 2, 'splits': []}
+
+        sse_server.publish(make_ably_error_event(60000, 600))
+        time.sleep(1)
+        assert factory.client().get_treatment('maldo', 'split1') == 'on'
+        assert not task.running()
+
+        sse_server.publish(make_ably_error_event(40145, 401))
+        sse_server.publish(sse_server.GRACEFUL_REQUEST_END)
+        time.sleep(3)
+        assert task.running()
+        assert factory.client().get_treatment('maldo', 'split1') == 'off'
+
+        # Re-publish initial events so that the retry succeeds
+        sse_server.publish(make_initial_event())
+        sse_server.publish(make_occupancy('control_pri', 2))
+        sse_server.publish(make_occupancy('control_sec', 2))
+        time.sleep(3)
+        assert not task.running()
+
+        # Assert streaming is working properly
+        split_changes[2] = {
+            'since': 2,
+            'till': 3,
+            'splits': [make_simple_split('split1', 3, True, False, 'off', 'user', True)]
+        }
+        split_changes[3] = {'since': 3, 'till': 3, 'splits': []}
+        sse_server.publish(make_split_change_event(3))
+        time.sleep(2)
+        assert factory.client().get_treatment('maldo', 'split1') == 'on'
+        assert not task.running()
+
+        # Send a non-retryable ably error
+        sse_server.publish(make_ably_error_event(40200, 402))
+        sse_server.publish(sse_server.GRACEFUL_REQUEST_END)
+        time.sleep(3)
+
+        # Assert sync-task is running and the streaming status handler thread is over
+        assert task.running()
+        assert 'PushStatusHandler' not in [t.name for t in threading.enumerate()]
+
+        # Validate the SSE requests
+        sse_request = sse_requests.get()
+        assert sse_request.method == 'GET'
+        path, qs = sse_request.path.split('?', 1)
+        assert path == '/event-stream'
+        qs = parse_qs(qs)
+        assert qs['accessToken'][0] == (
+            'eyJhbGciOiJIUzI1NiIsImtpZCI6IjVZOU05'
+            'US45QnJtR0EiLCJ0eXAiOiJKV1QifQ.eyJ4LWFibHktY2FwYWJpbGl0eSI6IntcIk1UW'
+            'XlNVGN4T1RRNE13PT1fTWpBNE16Y3pORFUxTWc9PV9zZWdtZW50c1wiOltcInN1YnNjc'
+            'mliZVwiXSxcIk1UWXlNVGN4T1RRNE13PT1fTWpBNE16Y3pORFUxTWc9PV9zcGxpdHNcI'
+            'jpbXCJzdWJzY3JpYmVcIl0sXCJjb250cm9sX3ByaVwiOltcInN1YnNjcmliZVwiLFwiY'
+            '2hhbm5lbC1tZXRhZGF0YTpwdWJsaXNoZXJzXCJdLFwiY29udHJvbF9zZWNcIjpbXCJzd'
+            'WJzY3JpYmVcIixcImNoYW5uZWwtbWV0YWRhdGE6cHVibGlzaGVyc1wiXX0iLCJ4LWFib'
+            'HktY2xpZW50SWQiOiJjbGllbnRJZCIsImV4cCI6MTYwNDEwMDU5MSwiaWF0IjoxNjA0M'
+            'Dk2OTkxfQ.aP9BfR534K6J9h8gfDWg_CQgpz5EvJh17WlOlAKhcD0'
+        )
+
+        assert set(qs['channels'][0].split(',')) == set(['MTYyMTcxOTQ4Mw==_MjA4MzczNDU1Mg==_splits',
+                                                         'MTYyMTcxOTQ4Mw==_MjA4MzczNDU1Mg==_segments',
+                                                         '[?occupancy=metrics.publishers]control_pri',
+                                                         '[?occupancy=metrics.publishers]control_sec'])
+        assert qs['v'][0] == '1.1'
+
+        assert sse_request.method == 'GET'
+        path, qs = sse_request.path.split('?', 1)
+        assert path == '/event-stream'
+        qs = parse_qs(qs)
+        assert qs['accessToken'][0] == (
+            'eyJhbGciOiJIUzI1NiIsImtpZCI6IjVZOU05'
+            'US45QnJtR0EiLCJ0eXAiOiJKV1QifQ.eyJ4LWFibHktY2FwYWJpbGl0eSI6IntcIk1UW'
+            'XlNVGN4T1RRNE13PT1fTWpBNE16Y3pORFUxTWc9PV9zZWdtZW50c1wiOltcInN1YnNjc'
+            'mliZVwiXSxcIk1UWXlNVGN4T1RRNE13PT1fTWpBNE16Y3pORFUxTWc9PV9zcGxpdHNcI'
+            'jpbXCJzdWJzY3JpYmVcIl0sXCJjb250cm9sX3ByaVwiOltcInN1YnNjcmliZVwiLFwiY'
+            '2hhbm5lbC1tZXRhZGF0YTpwdWJsaXNoZXJzXCJdLFwiY29udHJvbF9zZWNcIjpbXCJzd'
+            'WJzY3JpYmVcIixcImNoYW5uZWwtbWV0YWRhdGE6cHVibGlzaGVyc1wiXX0iLCJ4LWFib'
+            'HktY2xpZW50SWQiOiJjbGllbnRJZCIsImV4cCI6MTYwNDEwMDU5MSwiaWF0IjoxNjA0M'
+            'Dk2OTkxfQ.aP9BfR534K6J9h8gfDWg_CQgpz5EvJh17WlOlAKhcD0'
+        )
+
+        assert set(qs['channels'][0].split(',')) == set(['MTYyMTcxOTQ4Mw==_MjA4MzczNDU1Mg==_splits',
+                                                         'MTYyMTcxOTQ4Mw==_MjA4MzczNDU1Mg==_segments',
+                                                         '[?occupancy=metrics.publishers]control_pri',
+                                                         '[?occupancy=metrics.publishers]control_sec'])
+        assert qs['v'][0] == '1.1'
+
+        # Initial apikey validation
+        req = split_backend_requests.get()
+        assert req.method == 'GET'
+        assert req.path == '/api/segmentChanges/__SOME_INVALID_SEGMENT__?since=-1'
+        assert req.headers['authorization'] == 'Bearer some_apikey'
+
+        # Initial splits fetch
+        req = split_backend_requests.get()
+        assert req.method == 'GET'
+        assert req.path == '/api/splitChanges?since=-1'
+        assert req.headers['authorization'] == 'Bearer some_apikey'
+
+        # Iteration until since == till
+        req = split_backend_requests.get()
+        assert req.method == 'GET'
+        assert req.path == '/api/splitChanges?since=1'
+        assert req.headers['authorization'] == 'Bearer some_apikey'
+
+        # Auth
+        req = split_backend_requests.get()
+        assert req.method == 'GET'
+        assert req.path == '/api/auth'
+        assert req.headers['authorization'] == 'Bearer some_apikey'
+
+        # SyncAll after streaming connected
+        req = split_backend_requests.get()
+        assert req.method == 'GET'
+        assert req.path == '/api/splitChanges?since=1'
+        assert req.headers['authorization'] == 'Bearer some_apikey'
+
+        # SyncAll retriable error
+        req = split_backend_requests.get()
+        assert req.method == 'GET'
+        assert req.path == '/api/splitChanges?since=1'
+        assert req.headers['authorization'] == 'Bearer some_apikey'
+
+        # Iteration until since == till
+        req = split_backend_requests.get()
+        assert req.method == 'GET'
+        assert req.path == '/api/splitChanges?since=2'
+        assert req.headers['authorization'] == 'Bearer some_apikey'
+
+        # Auth again
+        req = split_backend_requests.get()
+        assert req.method == 'GET'
+        assert req.path == '/api/auth'
+        assert req.headers['authorization'] == 'Bearer some_apikey'
+
+        # SyncAll after push is up
+        req = split_backend_requests.get()
+        assert req.method == 'GET'
+        assert req.path == '/api/splitChanges?since=2'
+        assert req.headers['authorization'] == 'Bearer some_apikey'
+
+        # Fetch after notification
+        req = split_backend_requests.get()
+        assert req.method == 'GET'
+        assert req.path == '/api/splitChanges?since=2'
+        assert req.headers['authorization'] == 'Bearer some_apikey'
+
+        # Iteration until since == till
+        req = split_backend_requests.get()
+        assert req.method == 'GET'
+        assert req.path == '/api/splitChanges?since=3'
+        assert req.headers['authorization'] == 'Bearer some_apikey'
+
+        # SyncAll after non recoverable ably error
+        req = split_backend_requests.get()
+        assert req.method == 'GET'
+        assert req.path == '/api/splitChanges?since=3'
+        assert req.headers['authorization'] == 'Bearer some_apikey'
+
+        # Cleanup
+        destroy_event = threading.Event()
+        factory.destroy(destroy_event)
+        destroy_event.wait()
+        sse_server.publish(sse_server.GRACEFUL_REQUEST_END)
+        sse_server.stop()
+        split_backend.stop()
+
 
 def make_split_change_event(change_number):
     """Make a split change event."""
@@ -1105,6 +1342,18 @@ def make_control_event(control_type, timestamp):
         })
     }
 
+def make_ably_error_event(code, status):
+    """Make a control event."""
+    return {
+        'event': 'error',
+        'data': json.dumps({
+            'message':'Invalid accessToken in request: sarasa',
+            'code': code,
+            'statusCode': status,
+            'href':"https://help.ably.io/error/%d" % code
+        })
+    }
+
 def make_simple_split(name, cn, active, killed, default_treatment, tt, on):
     """Make a simple split."""
     return {