"""Streaming integration tests."""
# pylint:disable=no-self-use,invalid-name,too-many-arguments,too-few-public-methods,line-too-long
# pylint:disable=too-many-statements,too-many-locals,too-many-lines
import threading
import time
import json
from queue import Queue
from splitio.client.factory import get_factory
from tests.helpers.mockserver import SSEMockServer, SplitMockServer

try:  # try to import python3 names. fallback to python2
    from urllib.parse import parse_qs
except ImportError:
    from urlparse import parse_qs


class StreamingIntegrationTests(object):
    """Test streaming operation and failover."""

    def test_happiness(self):
        """Test initialization & splits/segment updates."""
        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, 'on', '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}
        }

        factory = get_factory('some_apikey', **kwargs)
        factory.block_until_ready(1)
        assert factory.ready
        assert factory.client().get_treatment('maldo', 'split1') == 'on'

        time.sleep(1)
        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_split_change_event(2))
        time.sleep(1)
        assert factory.client().get_treatment('maldo', 'split1') == 'off'

        split_changes[2] = {
            'since': 2,
            'till': 3,
            'splits': [make_split_with_segment('split2', 2, True, False,
                                               'off', 'user', 'off', 'segment1')]
        }
        split_changes[3] = {'since': 3, 'till': 3, 'splits': []}
        segment_changes[('segment1', -1)] = {
            'name': 'segment1',
            'added': ['maldo'],
            'removed': [],
            'since': -1,
            'till': 1
        }
        segment_changes[('segment1', 1)] = {'name': 'segment1', 'added': [],
                                            'removed': [], 'since': 1, 'till': 1}

        sse_server.publish(make_split_change_event(3))
        time.sleep(1)
        sse_server.publish(make_segment_change_event('segment1', 1))
        time.sleep(1)

        assert factory.client().get_treatment('pindon', 'split2') == 'off'
        assert factory.client().get_treatment('maldo', 'split2') == 'on'

        # Validate the SSE request
        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'

        # 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'

        # Fetch after first notification
        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'

        # Fetch after second 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'

        # Segment change notification
        req = split_backend_requests.get()
        assert req.method == 'GET'
        assert req.path == '/api/segmentChanges/segment1?since=-1'
        assert req.headers['authorization'] == 'Bearer some_apikey'

        # Iteration until segment1 since == till
        req = split_backend_requests.get()
        assert req.method == 'GET'
        assert req.path == '/api/segmentChanges/segment1?since=1'
        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 test_occupancy_flicker(self):
        """Test that changes in occupancy switch between polling & streaming properly."""
        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.
        # After dropping occupancy, the sdk should switch to polling
        # and perform a syncAll that gets this change
        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_occupancy('control_pri', 0))
        sse_server.publish(make_occupancy('control_sec', 0))
        time.sleep(2)
        assert factory.client().get_treatment('maldo', 'split1') == 'off'
        assert task.running()

        # We make another chagne in the BE and don't send the event.
        # We restore occupancy, and it should be fetched by the
        # sync all after streaming is restored.
        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_occupancy('control_pri', 1))
        time.sleep(2)
        assert factory.client().get_treatment('maldo', 'split1') == 'on'
        assert not task.running()

        # Now we make another change and send an event so it's propagated
        split_changes[3] = {
            'since': 3,
            'till': 4,
            'splits': [make_simple_split('split1', 4, True, False, 'off', 'user', False)]
        }
        split_changes[4] = {'since': 4, 'till': 4, 'splits': []}
        sse_server.publish(make_split_change_event(4))
        time.sleep(2)
        assert factory.client().get_treatment('maldo', 'split1') == 'off'

        # Kill the split
        split_changes[4] = {
            'since': 4,
            'till': 5,
            'splits': [make_simple_split('split1', 5, True, True, 'frula', 'user', False)]
        }
        split_changes[5] = {'since': 5, 'till': 5, 'splits': []}
        sse_server.publish(make_split_kill_event('split1', 'frula', 5))
        time.sleep(2)
        assert factory.client().get_treatment('maldo', 'split1') == 'frula'

        # Validate the SSE request
        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'

        # 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'

        # Fetch after first notification
        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'

        # Fetch after second 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'

        # 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'

        # Iteration until since == till
        req = split_backend_requests.get()
        assert req.method == 'GET'
        assert req.path == '/api/splitChanges?since=4'
        assert req.headers['authorization'] == 'Bearer some_apikey'

        # Split kill
        req = split_backend_requests.get()
        assert req.method == 'GET'
        assert req.path == '/api/splitChanges?since=4'
        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=5'
        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 test_start_without_occupancy(self):
        """Test an SDK starting with occupancy on 0 and switching to streamin afterwards."""
        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', 0))
        sse_server.publish(make_occupancy('control_sec', 0))

        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 task.running()
        assert factory.client().get_treatment('maldo', 'split1') == 'on'

        # Make a change in the BE but don't send the event.
        # After restoring occupancy, the sdk should switch to polling
        # and perform a syncAll that gets this change
        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_occupancy('control_sec', 1))
        time.sleep(2)
        assert factory.client().get_treatment('maldo', 'split1') == 'off'
        assert not task.running()

        # Validate the SSE request
        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'

        # 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 after push down
        req = split_backend_requests.get()
        assert req.method == 'GET'
        assert req.path == '/api/splitChanges?since=1'
        assert req.headers['authorization'] == 'Bearer some_apikey'

        # SyncAll after push restored
        req = split_backend_requests.get()
        assert req.method == 'GET'
        assert req.path == '/api/splitChanges?since=1'
        assert req.headers['authorization'] == 'Bearer some_apikey'

        # Second iteration of previous syncAll
        req = split_backend_requests.get()
        assert req.method == 'GET'
        assert req.path == '/api/splitChanges?since=2'
        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 test_streaming_status_changes(self):
        """Test changes between streaming enabled, paused and disabled."""
        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.
        # After dropping occupancy, the sdk should switch to polling
        # and perform a syncAll that gets this change
        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_control_event('STREAMING_PAUSED', 1))
        time.sleep(2)
        assert factory.client().get_treatment('maldo', 'split1') == 'off'
        assert task.running()

        # We make another chagne in the BE and don't send the event.
        # We restore occupancy, and it should be fetched by the
        # sync all after streaming is restored.
        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_control_event('STREAMING_ENABLED', 2))
        time.sleep(2)
        assert factory.client().get_treatment('maldo', 'split1') == 'on'
        assert not task.running()

        # Now we make another change and send an event so it's propagated
        split_changes[3] = {
            'since': 3,
            'till': 4,
            'splits': [make_simple_split('split1', 4, True, False, 'off', 'user', False)]
        }
        split_changes[4] = {'since': 4, 'till': 4, 'splits': []}
        sse_server.publish(make_split_change_event(4))
        time.sleep(2)
        assert factory.client().get_treatment('maldo', 'split1') == 'off'
        assert not task.running()

        split_changes[4] = {
            'since': 4,
            'till': 5,
            'splits': [make_simple_split('split1', 5, True, False, 'off', 'user', True)]
        }
        split_changes[5] = {'since': 5, 'till': 5, 'splits': []}
        sse_server.publish(make_control_event('STREAMING_DISABLED', 2))
        time.sleep(2)
        assert factory.client().get_treatment('maldo', 'split1') == 'on'
        assert task.running()
        assert 'PushStatusHandler' not in [t.name for t in threading.enumerate()]

        # Validate the SSE request
        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'

        # 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 on push down
        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'

        # 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'

        # 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'

        # Fetch after notification
        req = split_backend_requests.get()
        assert req.method == 'GET'
        assert req.path == '/api/splitChanges?since=3'
        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=4'
        assert req.headers['authorization'] == 'Bearer some_apikey'

        # SyncAll after streaming disabled
        req = split_backend_requests.get()
        assert req.method == 'GET'
        assert req.path == '/api/splitChanges?since=4'
        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=5'
        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 test_server_closes_connection(self):
        """Test that if the server closes the connection, the whole flow is retried with BO."""
        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, 'on', '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': 100,
                       'segmentsRefreshRate': 100, 'metricsRefreshRate': 100,
                       'impressionsRefreshRate': 100, 'eventsPushRate': 100}
        }

        factory = get_factory('some_apikey', **kwargs)
        factory.block_until_ready(1)
        assert factory.ready
        assert factory.client().get_treatment('maldo', 'split1') == 'on'
        task = factory._sync_manager._synchronizer._split_tasks.split_task._task  # pylint:disable=protected-access
        assert not task.running()

        time.sleep(1)
        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_split_change_event(2))
        time.sleep(1)
        assert factory.client().get_treatment('maldo', 'split1') == 'off'

        sse_server.publish(SSEMockServer.GRACEFUL_REQUEST_END)
        time.sleep(1)
        assert factory.client().get_treatment('maldo', 'split1') == 'off'
        assert task.running()

        time.sleep(2)  # wait for the backoff to expire so streaming gets re-attached

        # re-send initial event AND occupancy
        sse_server.publish(make_initial_event())
        sse_server.publish(make_occupancy('control_pri', 2))
        sse_server.publish(make_occupancy('control_sec', 2))
        time.sleep(2)

        assert not task.running()
        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(1)
        assert factory.client().get_treatment('maldo', 'split1') == 'on'
        assert not task.running()

        # 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'

        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'

        # 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'

        # Fetch after first notification
        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'

        # SyncAll on retryable error handling
        req = split_backend_requests.get()
        assert req.method == 'GET'
        assert req.path == '/api/splitChanges?since=2'
        assert req.headers['authorization'] == 'Bearer some_apikey'

        # Auth after connection breaks
        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 again
        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 new 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'

        # 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 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."""
    return {
        'event': 'message',
        'data': json.dumps({
            'id':'TVUsxaabHs:0:0',
            'clientId':'pri:MzM0ODI1MTkxMw==',
            'timestamp': change_number-1,
            'encoding':'json',
            'channel':'MTYyMTcxOTQ4Mw==_MjA4MzczNDU1Mg==_splits',
            'data': json.dumps({
                'type': 'SPLIT_UPDATE',
                'changeNumber': change_number
            })
        })
    }

def make_split_kill_event(name, default_treatment, change_number):
    """Make a split change event."""
    return {
        'event': 'message',
        'data': json.dumps({
            'id':'TVUsxaabHs:0:0',
            'clientId':'pri:MzM0ODI1MTkxMw==',
            'timestamp': change_number-1,
            'encoding':'json',
            'channel':'MTYyMTcxOTQ4Mw==_MjA4MzczNDU1Mg==_splits',
            'data': json.dumps({
                'type': 'SPLIT_KILL',
                'splitName': name,
                'defaultTreatment': default_treatment,
                'changeNumber': change_number
            })
        })
    }

def make_initial_event():
    """Make a split change event."""
    return {'id':'TVUsxaabHs:0:0'}

def make_occupancy(channel, publishers):
    """Make an occupancy event."""
    return {
        'event': 'message',
        'data': json.dumps({
            'id':'aP6EuhrcUm:0:0',
            'timestamp':1604325712734,
            'encoding': 'json',
            'channel': "[?occupancy=metrics.publishers]%s" % channel,
            'data': json.dumps({'metrics': {'publishers': publishers}}),
            'name':'[meta]occupancy'
        })
    }

def make_segment_change_event(name, change_number):
    """Make a split change event."""
    return {
        'event': 'message',
        'data': json.dumps({
            'id':'TVUsxaabHs:0:0',
            'clientId':'pri:MzM0ODI1MTkxMw==',
            'timestamp': change_number-1,
            'encoding':'json',
            'channel':'MTYyMTcxOTQ4Mw==_MjA4MzczNDU1Mg==_segments',
            'data': json.dumps({
                'type': 'SEGMENT_UPDATE',
                'segmentName': name,
                'changeNumber': change_number
            })
        })
    }

def make_control_event(control_type, timestamp):
    """Make a control event."""
    return {
        'event': 'message',
        'data': json.dumps({
            'id':'TVUsxaabHs:0:0',
            'clientId':'pri:MzM0ODI1MTkxMw==',
            'timestamp': timestamp,
            'encoding':'json',
            'channel':'[?occupancy=metrics.publishers]control_pri',
            'data': json.dumps({
                'type': 'CONTROL',
                'controlType': control_type,
            })
        })
    }

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 {
        'trafficTypeName': tt,
        'name': name,
        'seed': 1699838640,
        'status': 'ACTIVE' if active else 'ARCHIVED',
        'changeNumber': cn,
        'killed': killed,
        'defaultTreatment': default_treatment,
        'conditions': [
            {
                'matcherGroup': {
                    'combiner': 'AND',
                    'matchers': [
                        {
                            'matcherType': 'ALL_KEYS',
                            'negate': False,
                            'userDefinedSegmentMatcherData': None,
                            'whitelistMatcherData': None
                        }
                    ]
                },
                'partitions': [
                    {'treatment': 'on' if on else 'off', 'size': 100},
                    {'treatment': 'off' if on else 'on', 'size': 0}
                ]
            }
        ]
    }

def make_split_with_segment(name, cn, active, killed, default_treatment,
                            tt, on, segment):
    """Make a split with a segment."""
    return {
        'trafficTypeName': tt,
        'name': name,
        'seed': cn,
        'status': 'ACTIVE' if active else 'ARCHIVED',
        'changeNumber': cn,
        'killed': killed,
        'defaultTreatment': default_treatment,
        'configurations': {
            'on': '{\'size\':15,\'test\':20}'
        },
        'conditions': [
            {
                'matcherGroup': {
                    'combiner': 'AND',
                    'matchers': [
                        {
                            'matcherType': 'IN_SEGMENT',
                            'negate': False,
                            'userDefinedSegmentMatcherData': {'segmentName': segment},
                            'whitelistMatcherData': None
                        }
                    ]
                },
                'partitions': [{
                    'treatment': 'on' if on else 'off',
                    'size': 100
                }]
            }
        ]
    }

[((49, 33, 49, 40), 'queue.Queue', 'Queue', ({}, {}), '()', False, 'from queue import Queue\n'), ((50, 24, 51, 61), 'tests.helpers.mockserver.SplitMockServer', 'SplitMockServer', ({(50, 40, 50, 53): 'split_changes', (50, 55, 50, 70): 'segment_changes', (50, 72, 50, 94): 'split_backend_requests', (51, 40, 51, 60): 'auth_server_response'}, {}), '(split_changes, segment_changes, split_backend_requests,\n    auth_server_response)', False, 'from tests.helpers.mockserver import SSEMockServer, SplitMockServer\n'), ((52, 23, 52, 30), 'queue.Queue', 'Queue', ({}, {}), '()', False, 'from queue import Queue\n'), ((53, 21, 53, 48), 'tests.helpers.mockserver.SSEMockServer', 'SSEMockServer', ({(53, 35, 53, 47): 'sse_requests'}, {}), '(sse_requests)', False, 'from tests.helpers.mockserver import SSEMockServer, SplitMockServer\n'), ((69, 18, 69, 54), 'splitio.client.factory.get_factory', 'get_factory', ({(69, 30, 69, 43): '"""some_apikey"""'}, {}), "('some_apikey', **kwargs)", False, 'from splitio.client.factory import get_factory\n'), ((74, 8, 74, 21), 'time.sleep', 'time.sleep', ({(74, 19, 74, 20): '(1)'}, {}), '(1)', False, 'import time\n'), ((82, 8, 82, 21), 'time.sleep', 'time.sleep', ({(82, 19, 82, 20): '(1)'}, {}), '(1)', False, 'import time\n'), ((103, 8, 103, 21), 'time.sleep', 'time.sleep', ({(103, 19, 103, 20): '(1)'}, {}), '(1)', False, 'import time\n'), ((105, 8, 105, 21), 'time.sleep', 'time.sleep', ({(105, 19, 105, 20): '(1)'}, {}), '(1)', False, 'import time\n'), ((115, 13, 115, 25), 'urlparse.parse_qs', 'parse_qs', ({(115, 22, 115, 24): 'qs'}, {}), '(qs)', False, 'from urlparse import parse_qs\n'), ((201, 24, 201, 41), 'threading.Event', 'threading.Event', ({}, {}), '()', False, 'import threading\n'), ((233, 33, 233, 40), 'queue.Queue', 'Queue', ({}, {}), '()', False, 'from queue import Queue\n'), ((234, 24, 235, 61), 'tests.helpers.mockserver.SplitMockServer', 'SplitMockServer', ({(234, 40, 234, 53): 'split_changes', (234, 55, 234, 70): 'segment_changes', (234, 72, 234, 94): 'split_backend_requests', (235, 40, 235, 60): 'auth_server_response'}, {}), '(split_changes, segment_changes, split_backend_requests,\n    auth_server_response)', False, 'from tests.helpers.mockserver import SSEMockServer, SplitMockServer\n'), ((236, 23, 236, 30), 'queue.Queue', 'Queue', ({}, {}), '()', False, 'from queue import Queue\n'), ((237, 21, 237, 48), 'tests.helpers.mockserver.SSEMockServer', 'SSEMockServer', ({(237, 35, 237, 47): 'sse_requests'}, {}), '(sse_requests)', False, 'from tests.helpers.mockserver import SSEMockServer, SplitMockServer\n'), ((253, 18, 253, 54), 'splitio.client.factory.get_factory', 'get_factory', ({(253, 30, 253, 43): '"""some_apikey"""'}, {}), "('some_apikey', **kwargs)", False, 'from splitio.client.factory import get_factory\n'), ((256, 8, 256, 21), 'time.sleep', 'time.sleep', ({(256, 19, 256, 20): '(2)'}, {}), '(2)', False, 'import time\n'), ((276, 8, 276, 21), 'time.sleep', 'time.sleep', ({(276, 19, 276, 20): '(2)'}, {}), '(2)', False, 'import time\n'), ((291, 8, 291, 21), 'time.sleep', 'time.sleep', ({(291, 19, 291, 20): '(2)'}, {}), '(2)', False, 'import time\n'), ((303, 8, 303, 21), 'time.sleep', 'time.sleep', ({(303, 19, 303, 20): '(2)'}, {}), '(2)', False, 'import time\n'), ((314, 8, 314, 21), 'time.sleep', 'time.sleep', ({(314, 19, 314, 20): '(2)'}, {}), '(2)', False, 'import time\n'), ((322, 13, 322, 25), 'urlparse.parse_qs', 'parse_qs', ({(322, 22, 322, 24): 'qs'}, {}), '(qs)', False, 'from urlparse import parse_qs\n'), ((420, 24, 420, 41), 'threading.Event', 'threading.Event', ({}, {}), '()', False, 'import threading\n'), ((452, 33, 452, 40), 'queue.Queue', 'Queue', ({}, {}), '()', False, 'from queue import Queue\n'), ((453, 24, 454, 61), 'tests.helpers.mockserver.SplitMockServer', 'SplitMockServer', ({(453, 40, 453, 53): 'split_changes', (453, 55, 453, 70): 'segment_changes', (453, 72, 453, 94): 'split_backend_requests', (454, 40, 454, 60): 'auth_server_response'}, {}), '(split_changes, segment_changes, split_backend_requests,\n    auth_server_response)', False, 'from tests.helpers.mockserver import SSEMockServer, SplitMockServer\n'), ((455, 23, 455, 30), 'queue.Queue', 'Queue', ({}, {}), '()', False, 'from queue import Queue\n'), ((456, 21, 456, 48), 'tests.helpers.mockserver.SSEMockServer', 'SSEMockServer', ({(456, 35, 456, 47): 'sse_requests'}, {}), '(sse_requests)', False, 'from tests.helpers.mockserver import SSEMockServer, SplitMockServer\n'), ((472, 18, 472, 54), 'splitio.client.factory.get_factory', 'get_factory', ({(472, 30, 472, 43): '"""some_apikey"""'}, {}), "('some_apikey', **kwargs)", False, 'from splitio.client.factory import get_factory\n'), ((475, 8, 475, 21), 'time.sleep', 'time.sleep', ({(475, 19, 475, 20): '(2)'}, {}), '(2)', False, 'import time\n'), ((493, 8, 493, 21), 'time.sleep', 'time.sleep', ({(493, 19, 493, 20): '(2)'}, {}), '(2)', False, 'import time\n'), ((502, 13, 502, 25), 'urlparse.parse_qs', 'parse_qs', ({(502, 22, 502, 24): 'qs'}, {}), '(qs)', False, 'from urlparse import parse_qs\n'), ((570, 24, 570, 41), 'threading.Event', 'threading.Event', ({}, {}), '()', False, 'import threading\n'), ((602, 33, 602, 40), 'queue.Queue', 'Queue', ({}, {}), '()', False, 'from queue import Queue\n'), ((603, 24, 604, 61), 'tests.helpers.mockserver.SplitMockServer', 'SplitMockServer', ({(603, 40, 603, 53): 'split_changes', (603, 55, 603, 70): 'segment_changes', (603, 72, 603, 94): 'split_backend_requests', (604, 40, 604, 60): 'auth_server_response'}, {}), '(split_changes, segment_changes, split_backend_requests,\n    auth_server_response)', False, 'from tests.helpers.mockserver import SSEMockServer, SplitMockServer\n'), ((605, 23, 605, 30), 'queue.Queue', 'Queue', ({}, {}), '()', False, 'from queue import Queue\n'), ((606, 21, 606, 48), 'tests.helpers.mockserver.SSEMockServer', 'SSEMockServer', ({(606, 35, 606, 47): 'sse_requests'}, {}), '(sse_requests)', False, 'from tests.helpers.mockserver import SSEMockServer, SplitMockServer\n'), ((622, 18, 622, 54), 'splitio.client.factory.get_factory', 'get_factory', ({(622, 30, 622, 43): '"""some_apikey"""'}, {}), "('some_apikey', **kwargs)", False, 'from splitio.client.factory import get_factory\n'), ((625, 8, 625, 21), 'time.sleep', 'time.sleep', ({(625, 19, 625, 20): '(2)'}, {}), '(2)', False, 'import time\n'), ((644, 8, 644, 21), 'time.sleep', 'time.sleep', ({(644, 19, 644, 20): '(2)'}, {}), '(2)', False, 'import time\n'), ((659, 8, 659, 21), 'time.sleep', 'time.sleep', ({(659, 19, 659, 20): '(2)'}, {}), '(2)', False, 'import time\n'), ((671, 8, 671, 21), 'time.sleep', 'time.sleep', ({(671, 19, 671, 20): '(2)'}, {}), '(2)', False, 'import time\n'), ((682, 8, 682, 21), 'time.sleep', 'time.sleep', ({(682, 19, 682, 20): '(2)'}, {}), '(2)', False, 'import time\n'), ((692, 13, 692, 25), 'urlparse.parse_qs', 'parse_qs', ({(692, 22, 692, 24): 'qs'}, {}), '(qs)', False, 'from urlparse import parse_qs\n'), ((790, 24, 790, 41), 'threading.Event', 'threading.Event', ({}, {}), '()', False, 'import threading\n'), ((826, 33, 826, 40), 'queue.Queue', 'Queue', ({}, {}), '()', False, 'from queue import Queue\n'), ((827, 24, 828, 61), 'tests.helpers.mockserver.SplitMockServer', 'SplitMockServer', ({(827, 40, 827, 53): 'split_changes', (827, 55, 827, 70): 'segment_changes', (827, 72, 827, 94): 'split_backend_requests', (828, 40, 828, 60): 'auth_server_response'}, {}), '(split_changes, segment_changes, split_backend_requests,\n    auth_server_response)', False, 'from tests.helpers.mockserver import SSEMockServer, SplitMockServer\n'), ((829, 23, 829, 30), 'queue.Queue', 'Queue', ({}, {}), '()', False, 'from queue import Queue\n'), ((830, 21, 830, 48), 'tests.helpers.mockserver.SSEMockServer', 'SSEMockServer', ({(830, 35, 830, 47): 'sse_requests'}, {}), '(sse_requests)', False, 'from tests.helpers.mockserver import SSEMockServer, SplitMockServer\n'), ((848, 18, 848, 54), 'splitio.client.factory.get_factory', 'get_factory', ({(848, 30, 848, 43): '"""some_apikey"""'}, {}), "('some_apikey', **kwargs)", False, 'from splitio.client.factory import get_factory\n'), ((855, 8, 855, 21), 'time.sleep', 'time.sleep', ({(855, 19, 855, 20): '(1)'}, {}), '(1)', False, 'import time\n'), ((863, 8, 863, 21), 'time.sleep', 'time.sleep', ({(863, 19, 863, 20): '(1)'}, {}), '(1)', False, 'import time\n'), ((867, 8, 867, 21), 'time.sleep', 'time.sleep', ({(867, 19, 867, 20): '(1)'}, {}), '(1)', False, 'import time\n'), ((871, 8, 871, 21), 'time.sleep', 'time.sleep', ({(871, 19, 871, 20): '(2)'}, {}), '(2)', False, 'import time\n'), ((877, 8, 877, 21), 'time.sleep', 'time.sleep', ({(877, 19, 877, 20): '(2)'}, {}), '(2)', False, 'import time\n'), ((887, 8, 887, 21), 'time.sleep', 'time.sleep', ({(887, 19, 887, 20): '(1)'}, {}), '(1)', False, 'import time\n'), ((896, 13, 896, 25), 'urlparse.parse_qs', 'parse_qs', ({(896, 22, 896, 24): 'qs'}, {}), '(qs)', False, 'from urlparse import parse_qs\n'), ((919, 13, 919, 25), 'urlparse.parse_qs', 'parse_qs', ({(919, 22, 919, 24): 'qs'}, {}), '(qs)', False, 'from urlparse import parse_qs\n'), ((1011, 24, 1011, 41), 'threading.Event', 'threading.Event', ({}, {}), '()', False, 'import threading\n'), ((1021, 17, 1021, 45), 'logging.getLogger', 'logging.getLogger', ({(1021, 35, 1021, 44): '"""splitio"""'}, {}), "('splitio')", False, 'import logging\n'), ((1022, 18, 1022, 41), 'logging.StreamHandler', 'logging.StreamHandler', ({}, {}), '()', False, 'import logging\n'), ((1023, 20, 1023, 92), 'logging.Formatter', 'logging.Formatter', ({(1023, 38, 1023, 91): '"""%(asctime)s %(name)-12s %(levelname)-8s %(message)s"""'}, {}), "('%(asctime)s %(name)-12s %(levelname)-8s %(message)s')", False, 'import logging\n'), ((1050, 33, 1050, 40), 'queue.Queue', 'Queue', ({}, {}), '()', False, 'from queue import Queue\n'), ((1051, 24, 1052, 61), 'tests.helpers.mockserver.SplitMockServer', 'SplitMockServer', ({(1051, 40, 1051, 53): 'split_changes', (1051, 55, 1051, 70): 'segment_changes', (1051, 72, 1051, 94): 'split_backend_requests', (1052, 40, 1052, 60): 'auth_server_response'}, {}), '(split_changes, segment_changes, split_backend_requests,\n    auth_server_response)', False, 'from tests.helpers.mockserver import SSEMockServer, SplitMockServer\n'), ((1053, 23, 1053, 30), 'queue.Queue', 'Queue', ({}, {}), '()', False, 'from queue import Queue\n'), ((1054, 21, 1054, 48), 'tests.helpers.mockserver.SSEMockServer', 'SSEMockServer', ({(1054, 35, 1054, 47): 'sse_requests'}, {}), '(sse_requests)', False, 'from tests.helpers.mockserver import SSEMockServer, SplitMockServer\n'), ((1070, 18, 1070, 54), 'splitio.client.factory.get_factory', 'get_factory', ({(1070, 30, 1070, 43): '"""some_apikey"""'}, {}), "('some_apikey', **kwargs)", False, 'from splitio.client.factory import get_factory\n'), ((1073, 8, 1073, 21), 'time.sleep', 'time.sleep', ({(1073, 19, 1073, 20): '(2)'}, {}), '(2)', False, 'import time\n'), ((1091, 8, 1091, 21), 'time.sleep', 'time.sleep', ({(1091, 19, 1091, 20): '(1)'}, {}), '(1)', False, 'import time\n'), ((1097, 8, 1097, 21), 'time.sleep', 'time.sleep', ({(1097, 19, 1097, 20): '(3)'}, {}), '(3)', False, 'import time\n'), ((1105, 8, 1105, 21), 'time.sleep', 'time.sleep', ({(1105, 19, 1105, 20): '(3)'}, {}), '(3)', False, 'import time\n'), ((1116, 8, 1116, 21), 'time.sleep', 'time.sleep', ({(1116, 19, 1116, 20): '(2)'}, {}), '(2)', False, 'import time\n'), ((1123, 8, 1123, 21), 'time.sleep', 'time.sleep', ({(1123, 19, 1123, 20): '(3)'}, {}), '(3)', False, 'import time\n'), ((1134, 13, 1134, 25), 'urlparse.parse_qs', 'parse_qs', ({(1134, 22, 1134, 24): 'qs'}, {}), '(qs)', False, 'from urlparse import parse_qs\n'), ((1156, 13, 1156, 25), 'urlparse.parse_qs', 'parse_qs', ({(1156, 22, 1156, 24): 'qs'}, {}), '(qs)', False, 'from urlparse import parse_qs\n'), ((1248, 24, 1248, 41), 'threading.Event', 'threading.Event', ({}, {}), '()', False, 'import threading\n'), ((1349, 16, 1354, 10), 'json.dumps', 'json.dumps', ({(1349, 27, 1354, 9): "{'message': 'Invalid accessToken in request: sarasa', 'code': code,\n    'statusCode': status, 'href': 'https://help.ably.io/error/%d' % code}"}, {}), "({'message': 'Invalid accessToken in request: sarasa', 'code':\n    code, 'statusCode': status, 'href': 'https://help.ably.io/error/%d' % code}\n    )", False, 'import json\n'), ((1266, 20, 1269, 14), 'json.dumps', 'json.dumps', ({(1266, 31, 1269, 13): "{'type': 'SPLIT_UPDATE', 'changeNumber': change_number}"}, {}), "({'type': 'SPLIT_UPDATE', 'changeNumber': change_number})", False, 'import json\n'), ((1283, 20, 1288, 14), 'json.dumps', 'json.dumps', ({(1283, 31, 1288, 13): "{'type': 'SPLIT_KILL', 'splitName': name, 'defaultTreatment':\n    default_treatment, 'changeNumber': change_number}"}, {}), "({'type': 'SPLIT_KILL', 'splitName': name, 'defaultTreatment':\n    default_treatment, 'changeNumber': change_number})", False, 'import json\n'), ((1305, 20, 1305, 71), 'json.dumps', 'json.dumps', ({(1305, 31, 1305, 70): "{'metrics': {'publishers': publishers}}"}, {}), "({'metrics': {'publishers': publishers}})", False, 'import json\n'), ((1320, 20, 1324, 14), 'json.dumps', 'json.dumps', ({(1320, 31, 1324, 13): "{'type': 'SEGMENT_UPDATE', 'segmentName': name, 'changeNumber': change_number}"}, {}), "({'type': 'SEGMENT_UPDATE', 'segmentName': name, 'changeNumber':\n    change_number})", False, 'import json\n'), ((1338, 20, 1341, 14), 'json.dumps', 'json.dumps', ({(1338, 31, 1341, 13): "{'type': 'CONTROL', 'controlType': control_type}"}, {}), "({'type': 'CONTROL', 'controlType': control_type})", False, 'import json\n'), ((685, 59, 685, 80), 'threading.enumerate', 'threading.enumerate', ({}, {}), '()', False, 'import threading\n'), ((1127, 59, 1127, 80), 'threading.enumerate', 'threading.enumerate', ({}, {}), '()', False, 'import threading\n')]