more tests

Author: mredolatti

splitio/push/manager.py

@@ -141,9 +141,7 @@ def _trigger_connection_flow(self):
             return
 
         self._status_tracker.reset()
-        res = self._sse_client.start(token)
-        if res:
-            print(res)
+        if self._sse_client.start(token):
             self._setup_next_token_refresh(token)
             self._running = True
 

splitio/push/parser.py

@@ -37,9 +37,9 @@ class UpdateType(Enum):
 class ControlType(Enum):
     """Control type enumeration."""
 
-    STREAMING_ENABLED = 0
-    STREAMING_PAUSED = 1
-    STREAMING_DISABLED = 2
+    STREAMING_ENABLED = 'STREAMING_ENABLED'
+    STREAMING_PAUSED = 'STREAMING_PAUSED'
+    STREAMING_DISABLED = 'STREAMING_DISABLED'
 
 
 TAG_OCCUPANCY = '[meta]occupancy'

splitio/push/status_tracker.py

@@ -1,5 +1,4 @@
 """NotificationManagerKeeper implementation."""
-from collections import defaultdict
 from enum import Enum
 import logging
 import six
@@ -90,7 +89,7 @@ def handle_control_message(self, event):
         :type event: splitio.push.parser.ControlMessage
         """
         # we don't care about control messages if a disconnection is expected
-        if self._shutdown_expected:  
+        if self._shutdown_expected:
             return None
 
         if self._timestamps.control > event.timestamp:

splitio/sync/manager.py

@@ -103,11 +103,14 @@ def _streaming_feedback_handler(self):
                 self._push.stop(True)
                 self._synchronizer.sync_all()
                 self._synchronizer.start_periodic_fetching()
-                time.sleep(self._backoff.get())
+                how_long = self._backoff.get()
+                _LOGGER.info('error in streaming. restarting flow in %d seconds', how_long)
+                time.sleep(how_long)
                 self._push.start()
-                _LOGGER.info('error in streaming. restarting flow')
             elif status == Status.PUSH_NONRETRYABLE_ERROR:
-                self._synchronizer.start_periodic_fetching()
+                self._push.update_workers_status(False)
                 self._push.stop(False)
+                self._synchronizer.sync_all()
+                self._synchronizer.start_periodic_fetching()
                 _LOGGER.info('non-recoverable error in streaming. switching to polling.')
                 return