test_factory.py

"""Split factory test module."""
# pylint: disable=no-self-use,protected-access,line-too-long,too-many-statements
# pylint: disable=too-many-locals, too-many-arguments

import os
import time
import threading
import pytest
import queue

from splitio.optional.loaders import asyncio
from splitio.client.factory import get_factory, get_factory_async, SplitFactory, _INSTANTIATED_FACTORIES, Status,\
    _LOGGER as _logger, SplitFactoryAsync
from splitio.client.config import DEFAULT_CONFIG
from splitio.engine.telemetry import TelemetryStorageConsumer, TelemetryStorageProducer, TelemetryStorageProducerAsync
from splitio.engine.impressions.impressions import Manager as ImpressionsManager
from splitio.engine.impressions.manager import Counter as ImpressionsCounter
from splitio.engine.impressions.unique_keys_tracker import UniqueKeysTracker, UniqueKeysTrackerAsync
from splitio.engine.telemetry import TelemetryStorageConsumer, TelemetryStorageProducer, TelemetryStorageProducerAsync
from splitio.engine.evaluator import Evaluator, EvaluationContext
from splitio.engine.impressions.strategies import StrategyDebugMode, StrategyNoneMode, StrategyOptimizedMode
from splitio.events.events_task import EventsTask
from splitio.events.events_manager import EventsManagerAsync
from splitio.models.splits import from_raw
from splitio.models.fallback_config import FallbackTreatmentsConfiguration, FallbackTreatmentCalculator
from splitio.models.fallback_treatment import FallbackTreatment
from splitio.models.events import SdkInternalEvent
from splitio.recorder.recorder import PipelinedRecorder, StandardRecorder, StandardRecorderAsync
from splitio.storage import redis, inmemmory, pluggable, EventStorage
from splitio.storage.inmemmory import InMemorySplitStorage, InMemorySegmentStorage, \
    InMemoryImpressionStorage, InMemoryTelemetryStorage, InMemorySplitStorageAsync, \
    InMemoryImpressionStorageAsync, InMemorySegmentStorageAsync, InMemoryTelemetryStorageAsync, InMemoryEventStorageAsync, \
    InMemoryRuleBasedSegmentStorage, InMemoryRuleBasedSegmentStorageAsync
from splitio.sync.manager import Manager, ManagerAsync
from splitio.sync.synchronizer import Synchronizer, SynchronizerAsync, SplitSynchronizers, SplitTasks
from splitio.sync.split import SplitSynchronizer, SplitSynchronizerAsync
from splitio.sync.segment import SegmentSynchronizer, SegmentSynchronizerAsync
from splitio.storage.adapters.redis import RedisAdapter, RedisPipelineAdapter
from splitio.tasks.util import asynctask
from tests.storage.test_pluggable import StorageMockAdapter, StorageMockAdapterAsync
from tests.integration import splits_json


class SplitFactoryTests(object):
    """Split factory test cases."""

    def test_flag_sets_counts(self):      
        factory = get_factory("none", config={
            'flagSetsFilter': ['set1', 'set2', 'set3']
        })

        assert factory._telemetry_init_producer._telemetry_storage._tel_config._flag_sets == 3
        assert factory._telemetry_init_producer._telemetry_storage._tel_config._flag_sets_invalid == 0
        event = threading.Event()
        factory.destroy(event)
        event.wait()
        
        factory = get_factory("none", config={
            'flagSetsFilter': ['s#et1', 'set2', 'set3']
        })
        assert factory._telemetry_init_producer._telemetry_storage._tel_config._flag_sets == 3
        assert factory._telemetry_init_producer._telemetry_storage._tel_config._flag_sets_invalid == 1
        event = threading.Event()
        factory.destroy(event)
        event.wait()

        factory = get_factory("none", config={
            'flagSetsFilter': ['s#et1', 22, 'set3']
        })
        assert factory._telemetry_init_producer._telemetry_storage._tel_config._flag_sets == 3
        assert factory._telemetry_init_producer._telemetry_storage._tel_config._flag_sets_invalid == 2
        event = threading.Event()
        factory.destroy(event)
        event.wait()

    def test_inmemory_client_creation_streaming_false(self, mocker):
        """Test that a client with in-memory storage is created correctly."""
        # Setup synchronizer
        def _split_synchronizer(self, ready_flag, some, auth_api, streaming_enabled, sdk_matadata, telemetry_runtime_producer, sse_url=None, client_key=None):
            synchronizer = mocker.Mock(spec=Synchronizer)
            synchronizer.sync_all.return_values = None
            self._ready_flag = ready_flag
            self._synchronizer = synchronizer
            self._streaming_enabled = False
            self._telemetry_runtime_producer = telemetry_runtime_producer

        mocker.patch('splitio.sync.manager.Manager.__init__', new=_split_synchronizer)

        # Start factory and make assertions
        factory = get_factory('some_api_key')
        class TelemetrySubmitterMock():
            def synchronize_config(*_):
                pass
        factory._telemetry_submitter = TelemetrySubmitterMock()

        assert isinstance(factory._storages['splits'], inmemmory.InMemorySplitStorage)
        assert isinstance(factory._storages['segments'], inmemmory.InMemorySegmentStorage)
        assert isinstance(factory._storages['impressions'], inmemmory.InMemoryImpressionStorage)
        assert factory._storages['impressions']._impressions.maxsize == 10000
        assert isinstance(factory._storages['events'], inmemmory.InMemoryEventStorage)
        assert factory._storages['events']._events.maxsize == 10000

        assert isinstance(factory._sync_manager, Manager)
        assert isinstance(factory._recorder, StandardRecorder)
        assert isinstance(factory._recorder._impressions_manager, ImpressionsManager)
        assert isinstance(factory._recorder._event_sotrage, inmemmory.EventStorage)
        assert isinstance(factory._recorder._impression_storage, inmemmory.ImpressionStorage)

        assert factory._labels_enabled is True
        try:
            factory.block_until_ready(1)
        except:
            pass
        assert factory.ready
        factory.destroy()

    def test_redis_client_creation(self, mocker):
        """Test that a client with redis storage is created correctly."""
        strict_redis_mock = mocker.Mock()
        mocker.patch('splitio.storage.adapters.redis.StrictRedis', new=strict_redis_mock)
        fallback_treatments_configuration = FallbackTreatmentsConfiguration(FallbackTreatment("on"))
        config = {
            'labelsEnabled': False,
            'impressionListener': 123,
            'redisHost': 'some_host',
            'redisPort': 1234,
            'redisDb': 1,
            'redisPassword': 'some_password',
            'redisUsername': 'redis_user',
            'redisSocketTimeout': 123,
            'redisSocketConnectTimeout': 123,
            'redisSocketKeepalive': 123,
            'redisSocketKeepaliveOptions': False,
            'redisConnectionPool': False,
            'redisUnixSocketPath': '/some_path',
            'redisEncodingErrors': 'non-strict',
            'redisDecodeResponses': True,
            'redisRetryOnTimeout': True,
            'redisSsl': True,
            'redisSslKeyfile': 'some_file',
            'redisSslCertfile': 'some_cert_file',
            'redisSslCertReqs': 'some_cert_req',
            'redisSslCaCerts': 'some_ca_cert',
            'redisMaxConnections': 999,
            'flagSetsFilter': ['set_1'],
            'fallbackTreatments': fallback_treatments_configuration
        }
        factory = get_factory('some_api_key', config=config)
        class TelemetrySubmitterMock():
            def synchronize_config(*_):
                pass
        factory._telemetry_submitter = TelemetrySubmitterMock()

        assert isinstance(factory._get_storage('splits'), redis.RedisSplitStorage)
        assert isinstance(factory._get_storage('segments'), redis.RedisSegmentStorage)
        assert isinstance(factory._get_storage('impressions'), redis.RedisImpressionsStorage)
        assert isinstance(factory._get_storage('events'), redis.RedisEventsStorage)

        assert factory._get_storage('splits').flag_set_filter.flag_sets == set([])
        assert factory._fallback_treatment_calculator.fallback_treatments_configuration.global_fallback_treatment.treatment == fallback_treatments_configuration.global_fallback_treatment.treatment

        adapter = factory._get_storage('splits')._redis
        assert adapter == factory._get_storage('segments')._redis
        assert adapter == factory._get_storage('impressions')._redis
        assert adapter == factory._get_storage('events')._redis

        assert strict_redis_mock.mock_calls[0] == mocker.call(
            host='some_host',
            port=1234,
            db=1,
            username='redis_user',
            password='some_password',
            socket_timeout=123,
            socket_connect_timeout=123,
            socket_keepalive=123,
            socket_keepalive_options=False,
            connection_pool=False,
            unix_socket_path='/some_path',
            encoding='utf-8',
            encoding_errors='non-strict',
            decode_responses=True,
            retry_on_timeout=True,
            ssl=True,
            ssl_keyfile='some_file',
            ssl_certfile='some_cert_file',
            ssl_cert_reqs='some_cert_req',
            ssl_ca_certs='some_ca_cert',
            max_connections=999,
        )
        assert factory._labels_enabled is False
        assert isinstance(factory._recorder, PipelinedRecorder)
        assert isinstance(factory._recorder._impressions_manager, ImpressionsManager)
        assert isinstance(factory._recorder._make_pipe(), RedisPipelineAdapter)
        assert isinstance(factory._recorder._event_sotrage, redis.RedisEventsStorage)
        assert isinstance(factory._recorder._impression_storage, redis.RedisImpressionsStorage)

        try:
            factory.block_until_ready(1)
        except:
            pass
        assert factory.ready
        factory.destroy()

    def test_destroy(self, mocker):
        """Test that tasks are shutdown and data is flushed when destroy is called."""

        def stop_mock():
            return

        split_async_task_mock = mocker.Mock(spec=asynctask.AsyncTask)
        split_async_task_mock.stop.side_effect = stop_mock

        def _split_task_init_mock(self, synchronize_splits, period):
            self._task = split_async_task_mock
            self._period = period
        mocker.patch('splitio.client.factory.SplitSynchronizationTask.__init__',
                     new=_split_task_init_mock)

        segment_async_task_mock = mocker.Mock(spec=asynctask.AsyncTask)
        segment_async_task_mock.stop.side_effect = stop_mock

        def _segment_task_init_mock(self, synchronize_segments, period):
            self._task = segment_async_task_mock
            self._period = period
        mocker.patch('splitio.client.factory.SegmentSynchronizationTask.__init__',
                     new=_segment_task_init_mock)

        imp_async_task_mock = mocker.Mock(spec=asynctask.AsyncTask)
        imp_async_task_mock.stop.side_effect = stop_mock

        def _imppression_task_init_mock(self, synchronize_impressions, period):
            self._period = period
            self._task = imp_async_task_mock
        mocker.patch('splitio.client.factory.ImpressionsSyncTask.__init__',
                     new=_imppression_task_init_mock)

        evt_async_task_mock = mocker.Mock(spec=asynctask.AsyncTask)
        evt_async_task_mock.stop.side_effect = stop_mock

        def _event_task_init_mock(self, synchronize_events, period):
            self._period = period
            self._task = evt_async_task_mock
        mocker.patch('splitio.client.factory.EventsSyncTask.__init__', new=_event_task_init_mock)

        imp_count_async_task_mock = mocker.Mock(spec=asynctask.AsyncTask)
        imp_count_async_task_mock.stop.side_effect = stop_mock

        def _imppression_count_task_init_mock(self, synchronize_counters):
            self._task = imp_count_async_task_mock
        mocker.patch('splitio.client.factory.ImpressionsCountSyncTask.__init__',
                     new=_imppression_count_task_init_mock)

        telemetry_async_task_mock = mocker.Mock(spec=asynctask.AsyncTask)
        telemetry_async_task_mock.stop.side_effect = stop_mock

        def _telemetry_task_init_mock(self, synchronize_telemetry, synchronize_telemetry2):
            self._task = telemetry_async_task_mock
        mocker.patch('splitio.client.factory.TelemetrySyncTask.__init__',
                     new=_telemetry_task_init_mock)

        split_sync = mocker.Mock(spec=SplitSynchronizer)
        split_sync.synchronize_splits.return_value = []
        segment_sync = mocker.Mock(spec=SegmentSynchronizer)
        segment_sync.synchronize_segments.return_values = None
        syncs = SplitSynchronizers(split_sync, segment_sync, mocker.Mock(),
                                   mocker.Mock(), mocker.Mock(), mocker.Mock())
        tasks = SplitTasks(split_async_task_mock, segment_async_task_mock, imp_async_task_mock,
                           evt_async_task_mock, imp_count_async_task_mock, telemetry_async_task_mock)

        # Setup synchronizer
        def _split_synchronizer(self, ready_flag, some, auth_api, streaming_enabled, sdk_matadata, telemetry_runtime_producer, sse_url=None, client_key=None):
            synchronizer = Synchronizer(syncs, tasks)
            self._ready_flag = ready_flag
            self._synchronizer = synchronizer
            self._streaming_enabled = False
            self._telemetry_runtime_producer = telemetry_runtime_producer
        mocker.patch('splitio.sync.manager.Manager.__init__', new=_split_synchronizer)

        # Start factory and make assertions
        # Using invalid key should result in a timeout exception
        factory = get_factory('some_api_key')
        class TelemetrySubmitterMock():
            def synchronize_config(*_):
                pass
        factory._telemetry_submitter = TelemetrySubmitterMock()

        try:
            factory.block_until_ready(1)
        except:
            pass
        assert factory.ready
        assert factory.destroyed is False

        factory.destroy()
        assert len(imp_async_task_mock.stop.mock_calls) == 1
        assert len(evt_async_task_mock.stop.mock_calls) == 1
        assert len(imp_count_async_task_mock.stop.mock_calls) == 1
        assert factory.destroyed is True

    def test_destroy_with_event(self, mocker):
        """Test that tasks are shutdown and data is flushed when destroy is called."""

        def stop_mock(event):
            time.sleep(0.1)
            event.set()
            return

        def stop_mock_2():
            return

        split_async_task_mock = mocker.Mock(spec=asynctask.AsyncTask)
        split_async_task_mock.stop.side_effect = stop_mock_2

        def _split_task_init_mock(self, synchronize_splits, period):
            self._task = split_async_task_mock
            self._period = period
        mocker.patch('splitio.client.factory.SplitSynchronizationTask.__init__',
                     new=_split_task_init_mock)

        segment_async_task_mock = mocker.Mock(spec=asynctask.AsyncTask)
        segment_async_task_mock.stop.side_effect = stop_mock_2

        def _segment_task_init_mock(self, synchronize_segments, period):
            self._task = segment_async_task_mock
            self._period = period
        mocker.patch('splitio.client.factory.SegmentSynchronizationTask.__init__',
                     new=_segment_task_init_mock)

        imp_async_task_mock = mocker.Mock(spec=asynctask.AsyncTask)
        imp_async_task_mock.stop.side_effect = stop_mock

        def _imppression_task_init_mock(self, synchronize_impressions, period):
            self._period = period
            self._task = imp_async_task_mock
        mocker.patch('splitio.client.factory.ImpressionsSyncTask.__init__',
                     new=_imppression_task_init_mock)

        evt_async_task_mock = mocker.Mock(spec=asynctask.AsyncTask)
        evt_async_task_mock.stop.side_effect = stop_mock

        def _event_task_init_mock(self, synchronize_events, period):
            self._period = period
            self._task = evt_async_task_mock
        mocker.patch('splitio.client.factory.EventsSyncTask.__init__', new=_event_task_init_mock)

        imp_count_async_task_mock = mocker.Mock(spec=asynctask.AsyncTask)
        imp_count_async_task_mock.stop.side_effect = stop_mock

        def _imppression_count_task_init_mock(self, synchronize_counters):
            self._task = imp_count_async_task_mock
        mocker.patch('splitio.client.factory.ImpressionsCountSyncTask.__init__',
                     new=_imppression_count_task_init_mock)

        telemetry_async_task_mock = mocker.Mock(spec=asynctask.AsyncTask)
        telemetry_async_task_mock.stop.side_effect = stop_mock

        def _telemetry_task_init_mock(self, synchronize_telemetry, synchronize_telemetry2):
            self._task = telemetry_async_task_mock
        mocker.patch('splitio.client.factory.TelemetrySyncTask.__init__',
                     new=_telemetry_task_init_mock)

        internal_event_task_mock = mocker.Mock(spec=EventsTask)
        internal_event_task_mock.stop.side_effect = stop_mock_2
        internal_event_task_mock.start.side_effect = stop_mock_2

        split_sync = mocker.Mock(spec=SplitSynchronizer)
        split_sync.synchronize_splits.return_value = []
        segment_sync = mocker.Mock(spec=SegmentSynchronizer)
        segment_sync.synchronize_segments.return_values = None
        syncs = SplitSynchronizers(split_sync, segment_sync, mocker.Mock(),
                                   mocker.Mock(), mocker.Mock(), mocker.Mock())
        tasks = SplitTasks(split_async_task_mock, segment_async_task_mock, imp_async_task_mock,
                           evt_async_task_mock, imp_count_async_task_mock, telemetry_async_task_mock, None, None, internal_event_task_mock)

        # Setup synchronizer
        def _split_synchronizer(self, ready_flag, some, auth_api, streaming_enabled, sdk_matadata, telemetry_runtime_producer, sse_url=None, client_key=None):
            synchronizer = Synchronizer(syncs, tasks)
            self._ready_flag = ready_flag
            self._synchronizer = synchronizer
            self._streaming_enabled = False
            self._telemetry_runtime_producer = telemetry_runtime_producer
        mocker.patch('splitio.sync.manager.Manager.__init__', new=_split_synchronizer)

        # Start factory and make assertions
        factory = get_factory('some_api_key')
        class TelemetrySubmitterMock():
            def synchronize_config(*_):
                pass
        factory._telemetry_submitter = TelemetrySubmitterMock()

        try:
            factory.block_until_ready(1)
        except:
            pass
        assert factory._status == Status.READY
        assert factory.destroyed is False

        event = threading.Event()
        factory.destroy(event)
        time.sleep(1)
        assert event.is_set()
        assert len(imp_async_task_mock.stop.mock_calls) == 1
        assert len(evt_async_task_mock.stop.mock_calls) == 1
        assert len(imp_count_async_task_mock.stop.mock_calls) == 1
        assert factory.destroyed is True

    def test_destroy_with_event_redis(self, mocker):
        def _make_factory_with_apikey(apikey, *_, **__):
            return SplitFactory(apikey, {}, True, mocker.Mock(spec=ImpressionsManager), None, mocker.Mock(), mocker.Mock(), mocker.Mock(), mocker.Mock(), mocker.Mock(), mocker.Mock())

        factory_module_logger = mocker.Mock()
        build_redis = mocker.Mock()
        build_redis.side_effect = _make_factory_with_apikey
        mocker.patch('splitio.client.factory._LOGGER', new=factory_module_logger)
        mocker.patch('splitio.client.factory._build_redis_factory', new=build_redis)

        config = {
            'redisDb': 0,
            'redisHost': 'localhost',
            'redisPosrt': 6379,
        }

        factory = get_factory("none", config=config)
        class TelemetrySubmitterMock():
            def synchronize_config(*_):
                pass
        factory._telemetry_submitter = TelemetrySubmitterMock()

        event = threading.Event()
        factory.destroy(event)
        event.wait()
        assert factory.destroyed
        assert len(build_redis.mock_calls) == 1

        factory = get_factory("none", config=config)
        class TelemetrySubmitterMock():
            def synchronize_config(*_):
                pass
        factory._telemetry_submitter = TelemetrySubmitterMock()

        factory.destroy(None)
        time.sleep(0.1)
        assert factory.destroyed
        assert len(build_redis.mock_calls) == 2

    def test_multiple_factories(self, mocker):
        """Test multiple factories instantiation and tracking."""
        sdk_ready_flag = threading.Event()

        def _init(self, ready_flag, some, auth_api, streaming_enabled, telemetry_runtime_producer, telemetry_init_consumer, sse_url=None):
            self._ready_flag = ready_flag
            self._synchronizer = mocker.Mock(spec=Synchronizer)
            self._streaming_enabled = False
            self._telemetry_runtime_producer = telemetry_runtime_producer
            self._telemetry_init_consumer = telemetry_init_consumer
        mocker.patch('splitio.sync.manager.Manager.__init__', new=_init)

        def _start(self, *args, **kwargs):
            sdk_ready_flag.set()
        mocker.patch('splitio.sync.manager.Manager.start', new=_start)

        def _stop(self, *args, **kwargs):
            pass
        mocker.patch('splitio.sync.manager.Manager.stop', new=_stop)

        mockManager = Manager(sdk_ready_flag, mocker.Mock(), mocker.Mock(), False, mocker.Mock(), mocker.Mock())

        def _make_factory_with_apikey(apikey, *_, **__):
            return SplitFactory(apikey, {}, True, mocker.Mock(spec=StandardRecorder), mocker.Mock(), mocker.Mock(), mockManager, mocker.Mock(), mocker.Mock(), mocker.Mock())

        factory_module_logger = mocker.Mock()
        build_in_memory = mocker.Mock()
        build_in_memory.side_effect = _make_factory_with_apikey
        build_redis = mocker.Mock()
        build_redis.side_effect = _make_factory_with_apikey
        build_localhost = mocker.Mock()
        build_localhost.side_effect = _make_factory_with_apikey
        mocker.patch('splitio.client.factory._LOGGER', new=factory_module_logger)
        mocker.patch('splitio.client.factory._build_in_memory_factory', new=build_in_memory)
        mocker.patch('splitio.client.factory._build_redis_factory', new=build_redis)
        mocker.patch('splitio.client.factory._build_localhost_factory', new=build_localhost)

        _INSTANTIATED_FACTORIES.clear()  # Clear all factory counters for testing purposes

        factory1 = get_factory('some_api_key')
        class TelemetrySubmitterMock():
            def synchronize_config(*_):
                pass
        factory1._telemetry_submitter = TelemetrySubmitterMock()

        assert _INSTANTIATED_FACTORIES['some_api_key'] == 1
        assert factory_module_logger.warning.mock_calls == []

        factory2 = get_factory('some_api_key')
        class TelemetrySubmitterMock():
            def synchronize_config(*_):
                pass
        factory2._telemetry_submitter = TelemetrySubmitterMock()

        assert _INSTANTIATED_FACTORIES['some_api_key'] == 2
        assert factory_module_logger.warning.mock_calls == [mocker.call(
            "factory instantiation: You already have %d %s with this SDK Key. "
            "We recommend keeping only one instance of the factory at all times "
            "(Singleton pattern) and reusing it throughout your application.",
            1,
            'factory'
        )]

        factory_module_logger.reset_mock()
        factory3 = get_factory('some_api_key')
        class TelemetrySubmitterMock():
            def synchronize_config(*_):
                pass
        factory3._telemetry_submitter = TelemetrySubmitterMock()

        assert _INSTANTIATED_FACTORIES['some_api_key'] == 3
        assert factory_module_logger.warning.mock_calls == [mocker.call(
            "factory instantiation: You already have %d %s with this SDK Key. "
            "We recommend keeping only one instance of the factory at all times "
            "(Singleton pattern) and reusing it throughout your application.",
            2,
            'factories'
        )]

        factory_module_logger.reset_mock()
        factory4 = get_factory('some_other_api_key')
        class TelemetrySubmitterMock():
            def synchronize_config(*_):
                pass
        factory4._telemetry_submitter = TelemetrySubmitterMock()

        assert _INSTANTIATED_FACTORIES['some_api_key'] == 3
        assert _INSTANTIATED_FACTORIES['some_other_api_key'] == 1
        assert factory_module_logger.warning.mock_calls == [mocker.call(
            "factory instantiation: You already have an instance of the Split factory. "
            "Make sure you definitely want this additional instance. "
            "We recommend keeping only one instance of the factory at all times "
            "(Singleton pattern) and reusing it throughout your application."
        )]

        event = threading.Event()
        factory1.destroy(event)
        event.wait()
        assert _INSTANTIATED_FACTORIES['some_other_api_key'] == 1
        assert _INSTANTIATED_FACTORIES['some_api_key'] == 2
        event = threading.Event()
        factory2.destroy(event)
        event.wait()
        event = threading.Event()
        factory3.destroy(event)
        event.wait()
        event = threading.Event()
        factory4.destroy(event)
        event.wait()

    def test_uwsgi_preforked(self, mocker):
        """Test preforked initializations."""

        def clear_impressions():
            clear_impressions._called += 1

        def clear_events():
            clear_events._called += 1

        clear_impressions._called = 0
        clear_events._called = 0
        split_storage = mocker.Mock(spec=inmemmory.SplitStorage)
        segment_storage = mocker.Mock(spec=inmemmory.SegmentStorage)
        impression_storage = mocker.Mock(spec=inmemmory.ImpressionStorage)
        impression_storage.clear.side_effect = clear_impressions
        event_storage = mocker.Mock(spec=inmemmory.EventStorage)
        event_storage.clear.side_effect = clear_events

        def _get_storage_mock(self, name):
            return {
                'splits': split_storage,
                'segments': segment_storage,
                'impressions': impression_storage,
                'events': event_storage,
            }[name]

        mocker.patch('splitio.client.factory.SplitFactory._get_storage', new=_get_storage_mock)

        sync_all_mock = mocker.Mock()
        mocker.patch('splitio.sync.synchronizer.Synchronizer.sync_all', new=sync_all_mock)

        start_mock = mocker.Mock()
        mocker.patch('splitio.sync.manager.Manager.start', new=start_mock)

        recreate_mock = mocker.Mock()
        mocker.patch('splitio.sync.manager.Manager.recreate', new=recreate_mock)

        config = {
            'preforkedInitialization': True,
        }
        factory = get_factory("none", config=config)
        class TelemetrySubmitterMock():
            def synchronize_config(*_):
                pass
        factory._telemetry_submitter = TelemetrySubmitterMock()

        try:
            factory.block_until_ready(10)
        except:
            pass
        assert factory._status == Status.WAITING_FORK
        assert len(sync_all_mock.mock_calls) == 1
        assert len(start_mock.mock_calls) == 0

        factory.resume()
        assert len(recreate_mock.mock_calls) == 1
        assert len(start_mock.mock_calls) == 1

        assert clear_impressions._called == 1
        assert clear_events._called == 1
        factory.destroy()
        time.sleep(0.1)
        assert factory.destroyed

    def test_error_prefork(self, mocker):
        """Test not handling fork."""
        expected_msg = [
            mocker.call('Cannot call resume')
        ]

        filename = os.path.join(os.path.dirname(__file__), '../integration/files', 'file2.yaml')
        factory = get_factory('localhost', config={'splitFile': filename})
        class TelemetrySubmitterMock():
            def synchronize_config(*_):
                pass
        factory._telemetry_submitter = TelemetrySubmitterMock()

        try:
            factory.block_until_ready(1)
        except:
            pass
        _logger = mocker.Mock()
        mocker.patch('splitio.client.factory._LOGGER', new=_logger)
        factory.resume()
        assert _logger.warning.mock_calls == expected_msg
        factory.destroy()

    def test_pluggable_client_creation(self, mocker):
        """Test that a client with pluggable storage is created correctly."""
        config = {
            'labelsEnabled': False,
            'impressionListener': 123,
            'storageType': 'pluggable',
            'storageWrapper': StorageMockAdapter(),
            'flagSetsFilter': ['set_1']
        }
        factory = get_factory('some_api_key', config=config)
        class TelemetrySubmitterMock():
            def synchronize_config(*_):
                pass
        factory._telemetry_submitter = TelemetrySubmitterMock()

        assert isinstance(factory._get_storage('splits'), pluggable.PluggableSplitStorage)
        assert isinstance(factory._get_storage('segments'), pluggable.PluggableSegmentStorage)
        assert isinstance(factory._get_storage('impressions'), pluggable.PluggableImpressionsStorage)
        assert isinstance(factory._get_storage('events'), pluggable.PluggableEventsStorage)
        assert factory._get_storage('splits').flag_set_filter.flag_sets == set([])

        adapter = factory._get_storage('splits')._pluggable_adapter
        assert adapter == factory._get_storage('segments')._pluggable_adapter
        assert adapter == factory._get_storage('impressions')._pluggable_adapter
        assert adapter == factory._get_storage('events')._pluggable_adapter

        assert factory._labels_enabled is False
        assert isinstance(factory._recorder, StandardRecorder)
        assert isinstance(factory._recorder._impressions_manager, ImpressionsManager)
        assert isinstance(factory._recorder._event_sotrage, pluggable.PluggableEventsStorage)
        assert isinstance(factory._recorder._impression_storage, pluggable.PluggableImpressionsStorage)

        try:
            factory.block_until_ready(1)
        except:
            pass
        assert factory.ready
        factory.destroy()
        time.sleep(0.1)
        assert factory.destroyed

    def test_destroy_with_event_pluggable(self, mocker):
        config = {
            'labelsEnabled': False,
            'impressionListener': 123,
            'storageType': 'pluggable',
            'storageWrapper': StorageMockAdapter()
        }

        factory = get_factory("none", config=config)
        class TelemetrySubmitterMock():
            def synchronize_config(*_):
                pass
        factory._telemetry_submitter = TelemetrySubmitterMock()

        event = threading.Event()
        factory.destroy(event)
        event.wait()
        assert factory.destroyed

        factory = get_factory("none", config=config)
        class TelemetrySubmitterMock():
            def synchronize_config(*_):
                pass
        factory._telemetry_submitter = TelemetrySubmitterMock()

        factory.destroy(None)
        time.sleep(0.1)
        assert factory.destroyed
        
    def test_internal_ready_event_notification(self, mocker):
        """Test that a client with in-memory storage is sending internal events correctly."""
        # Setup synchronizer
        def _split_synchronizer(self, ready_flag, some, auth_api, streaming_enabled, sdk_matadata, telemetry_runtime_producer, sse_url=None, client_key=None):
            synchronizer = mocker.Mock(spec=Synchronizer)
            synchronizer.sync_all.return_values = None
            self._ready_flag = ready_flag
            self._synchronizer = synchronizer
            self._streaming_enabled = False
            self._telemetry_runtime_producer = telemetry_runtime_producer

        mocker.patch('splitio.sync.manager.Manager.__init__', new=_split_synchronizer)

        # Start factory and make assertions
        
        telemetry_storage = InMemoryTelemetryStorage()
        telemetry_producer = TelemetryStorageProducer(telemetry_storage)
        events_queue = queue.Queue()
        split_storage = InMemorySplitStorage(events_queue)
        segment_storage = InMemorySegmentStorage(events_queue)
        rb_segment_storage = InMemoryRuleBasedSegmentStorage(events_queue)
        telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
        impression_storage = InMemoryImpressionStorage(10, telemetry_runtime_producer)
        impmanager = ImpressionsManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer)
        event_storage = mocker.Mock(spec=EventStorage)

        destroyed_property = mocker.PropertyMock()
        destroyed_property.return_value = False
        recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer())
        factory = SplitFactory("some key",
            {'splits': split_storage,
            'segments': segment_storage,
            'rule_based_segments': rb_segment_storage,
            'impressions': impression_storage,
            'events': event_storage},
            mocker.Mock(),
            recorder,
            events_queue,
            mocker.Mock(),
            mocker.Mock(),
            mocker.Mock(),
            telemetry_producer,
            telemetry_producer.get_telemetry_init_producer(),
            mocker.Mock()
        )

        class TelemetrySubmitterMock():
            def synchronize_config(*_):
                pass
        factory._telemetry_submitter = TelemetrySubmitterMock()

        try:
            factory.block_until_ready(1)
        except:
            pass

        assert factory.ready
        event = events_queue.get()
        assert event.internal_event == SdkInternalEvent.SDK_READY
        assert event.metadata == None
        factory.destroy()
        
    def test_uwsgi_forked_client_creation(self):
        """Test client with preforked initialization."""
        # Invalid API Key with preforked should exit after 3 attempts.
        factory = get_factory('some_api_key', config={'preforkedInitialization': True})
        class TelemetrySubmitterMock():
            def synchronize_config(*_):
                pass
        factory._telemetry_submitter = TelemetrySubmitterMock()

        assert isinstance(factory._storages['splits'], inmemmory.InMemorySplitStorage)
        assert isinstance(factory._storages['segments'], inmemmory.InMemorySegmentStorage)
        assert isinstance(factory._storages['impressions'], inmemmory.InMemoryImpressionStorage)
        assert factory._storages['impressions']._impressions.maxsize == 10000
        assert isinstance(factory._storages['events'], inmemmory.InMemoryEventStorage)
        assert factory._storages['events']._events.maxsize == 10000

        assert isinstance(factory._sync_manager, Manager)

        assert isinstance(factory._recorder, StandardRecorder)
        assert isinstance(factory._recorder._impressions_manager, ImpressionsManager)
        assert isinstance(factory._recorder._event_sotrage, inmemmory.EventStorage)
        assert isinstance(factory._recorder._impression_storage, inmemmory.ImpressionStorage)

        assert factory._status == Status.WAITING_FORK
        factory.destroy()
        time.sleep(0.1)
        assert factory.destroyed

class SplitFactoryAsyncTests(object):
    """Split factory async test cases."""

    @pytest.mark.asyncio
    async def test_flag_sets_counts(self):
        fallback_treatments_configuration = FallbackTreatmentsConfiguration(FallbackTreatment("on"))
        factory = await get_factory_async("none", config={
            'flagSetsFilter': ['set1', 'set2', 'set3'],
            'streamEnabled': False,
            'fallbackTreatments': fallback_treatments_configuration
        })
        assert factory._fallback_treatment_calculator.fallback_treatments_configuration.global_fallback_treatment.treatment == fallback_treatments_configuration.global_fallback_treatment.treatment
        assert factory._telemetry_init_producer._telemetry_storage._tel_config._flag_sets == 3
        assert factory._telemetry_init_producer._telemetry_storage._tel_config._flag_sets_invalid == 0
        await factory.destroy()

        factory = await get_factory_async("none", config={
            'flagSetsFilter': ['s#et1', 'set2', 'set3']
        })
        assert factory._telemetry_init_producer._telemetry_storage._tel_config._flag_sets == 3
        assert factory._telemetry_init_producer._telemetry_storage._tel_config._flag_sets_invalid == 1
        await factory.destroy()

        factory = await get_factory_async("none", config={
            'flagSetsFilter': ['s#et1', 22, 'set3']
        })
        assert factory._telemetry_init_producer._telemetry_storage._tel_config._flag_sets == 3
        assert factory._telemetry_init_producer._telemetry_storage._tel_config._flag_sets_invalid == 2
        await factory.destroy()

    @pytest.mark.asyncio
    async def test_inmemory_client_creation_streaming_false_async(self, mocker):
        """Test that a client with in-memory storage is created correctly for async."""
        # Setup synchronizer
        def _split_synchronizer(self, ready_flag, some, auth_api, streaming_enabled, sdk_matadata, telemetry_runtime_producer, sse_url=None, client_key=None):
            synchronizer = mocker.Mock(spec=SynchronizerAsync)
            async def sync_all(*_):
                return None
            synchronizer.sync_all = sync_all

            def start_periodic_fetching():
                pass        
            synchronizer.start_periodic_fetching = start_periodic_fetching

            self._ready_flag = ready_flag
            self._synchronizer = synchronizer
            self._streaming_enabled = False
            self._telemetry_runtime_producer = telemetry_runtime_producer
            
        mocker.patch('splitio.sync.manager.ManagerAsync.__init__', new=_split_synchronizer)

        async def synchronize_config(*_):
            pass
        mocker.patch('splitio.sync.telemetry.InMemoryTelemetrySubmitterAsync.synchronize_config', new=synchronize_config)

        # Start factory and make assertions
        factory2 = await get_factory_async('some_api_key', config={'streamingEmabled': False})

        assert isinstance(factory2, SplitFactoryAsync)
        assert isinstance(factory2._storages['splits'], inmemmory.InMemorySplitStorageAsync)
        assert isinstance(factory2._storages['segments'], inmemmory.InMemorySegmentStorageAsync)
        assert isinstance(factory2._storages['impressions'], inmemmory.InMemoryImpressionStorageAsync)
        assert factory2._storages['impressions']._impressions.maxsize == 10000
        assert isinstance(factory2._storages['events'], inmemmory.InMemoryEventStorageAsync)
        assert factory2._storages['events']._events.maxsize == 10000

        assert isinstance(factory2._sync_manager, ManagerAsync)

        assert isinstance(factory2._recorder, StandardRecorderAsync)
        assert isinstance(factory2._recorder._impressions_manager, ImpressionsManager)
        assert isinstance(factory2._recorder._event_sotrage, inmemmory.EventStorage)
        assert isinstance(factory2._recorder._impression_storage, inmemmory.ImpressionStorage)

        assert factory2._labels_enabled is True
        try:
            await factory2.block_until_ready(1)
        except:
            pass
        assert factory2._status == Status.READY
        await factory2.destroy()

    @pytest.mark.asyncio
    async def test_destroy_async(self, mocker):
        """Test that tasks are shutdown and data is flushed when destroy is called."""

        async def stop_mock():
            return

        split_async_task_mock = mocker.Mock(spec=asynctask.AsyncTaskAsync)
        split_async_task_mock.stop.side_effect = stop_mock

        def _split_task_init_mock(self, synchronize_splits, period):
            self._task = split_async_task_mock
            self._period = period
        mocker.patch('splitio.client.factory.SplitSynchronizationTaskAsync.__init__',
                     new=_split_task_init_mock)

        segment_async_task_mock = mocker.Mock(spec=asynctask.AsyncTaskAsync)
        segment_async_task_mock.stop.side_effect = stop_mock

        def _segment_task_init_mock(self, synchronize_segments, period):
            self._task = segment_async_task_mock
            self._period = period
        mocker.patch('splitio.client.factory.SegmentSynchronizationTaskAsync.__init__',
                     new=_segment_task_init_mock)

        imp_async_task_mock = mocker.Mock(spec=asynctask.AsyncTaskAsync)
        imp_async_task_mock.stop.side_effect = stop_mock

        def _imppression_task_init_mock(self, synchronize_impressions, period):
            self._period = period
            self._task = imp_async_task_mock
        mocker.patch('splitio.client.factory.ImpressionsSyncTaskAsync.__init__',
                     new=_imppression_task_init_mock)

        evt_async_task_mock = mocker.Mock(spec=asynctask.AsyncTaskAsync)
        evt_async_task_mock.stop.side_effect = stop_mock

        def _event_task_init_mock(self, synchronize_events, period):
            self._period = period
            self._task = evt_async_task_mock
        mocker.patch('splitio.client.factory.EventsSyncTaskAsync.__init__', new=_event_task_init_mock)

        imp_count_async_task_mock = mocker.Mock(spec=asynctask.AsyncTaskAsync)
        imp_count_async_task_mock.stop.side_effect = stop_mock

        def _imppression_count_task_init_mock(self, synchronize_counters):
            self._task = imp_count_async_task_mock
        mocker.patch('splitio.client.factory.ImpressionsCountSyncTaskAsync.__init__',
                     new=_imppression_count_task_init_mock)

        telemetry_async_task_mock = mocker.Mock(spec=asynctask.AsyncTaskAsync)
        telemetry_async_task_mock.stop.side_effect = stop_mock

        def _telemetry_task_init_mock(self, synchronize_telemetry, synchronize_telemetry2):
            self._task = telemetry_async_task_mock
        mocker.patch('splitio.client.factory.TelemetrySyncTaskAsync.__init__',
                     new=_telemetry_task_init_mock)

        split_sync = mocker.Mock(spec=SplitSynchronizerAsync)
        async def synchronize_splits(*_):
            return []
        split_sync.synchronize_splits = synchronize_splits

        segment_sync = mocker.Mock(spec=SegmentSynchronizerAsync)
        async def synchronize_segments(*_):
            return True
        segment_sync.synchronize_segments = synchronize_segments

        syncs = SplitSynchronizers(split_sync, segment_sync, mocker.Mock(),
                                   mocker.Mock(), mocker.Mock(), mocker.Mock())
        tasks = SplitTasks(split_async_task_mock, segment_async_task_mock, imp_async_task_mock,
                           evt_async_task_mock, imp_count_async_task_mock, telemetry_async_task_mock)

        # Setup synchronizer
        def _split_synchronizer(self, ready_flag, some, auth_api, streaming_enabled, sdk_matadata, telemetry_runtime_producer, sse_url=None, client_key=None):
            synchronizer = SynchronizerAsync(syncs, tasks)
            self._ready_flag = ready_flag
            self._synchronizer = synchronizer
            self._streaming_enabled = False
            self._telemetry_runtime_producer = telemetry_runtime_producer
        mocker.patch('splitio.sync.manager.ManagerAsync.__init__', new=_split_synchronizer)

        async def synchronize_config(*_):
            pass
        mocker.patch('splitio.sync.telemetry.InMemoryTelemetrySubmitterAsync.synchronize_config', new=synchronize_config)
        # Start factory and make assertions
        # Using invalid key should result in a timeout exception
        factory = await get_factory_async('some_api_key')
        self.manager_called = False
        async def stop(*_):
            self.manager_called = True
            pass
        factory._sync_manager.stop = stop

        async def start(*_):
            pass
        factory._sync_manager.start = start

        try:
            await factory.block_until_ready(1)
        except:
            pass
        assert factory._status == Status.READY
        assert factory.destroyed is False

        await factory.destroy()
        assert self.manager_called
        assert factory.destroyed is True

    @pytest.mark.asyncio
    async def test_pluggable_client_creation_async(self, mocker):
        """Test that a client with pluggable storage is created correctly."""
        config = {
            'labelsEnabled': False,
            'impressionListener': 123,
            'featuresRefreshRate': 1,
            'segmentsRefreshRate': 1,
            'metricsRefreshRate': 1,
            'impressionsRefreshRate': 1,
            'eventsPushRate': 1,
            'storageType': 'pluggable',
            'storageWrapper': StorageMockAdapterAsync()
        }
        factory = await get_factory_async('some_api_key', config=config)
        assert isinstance(factory._get_storage('splits'), pluggable.PluggableSplitStorageAsync)
        assert isinstance(factory._get_storage('segments'), pluggable.PluggableSegmentStorageAsync)
        assert isinstance(factory._get_storage('impressions'), pluggable.PluggableImpressionsStorageAsync)
        assert isinstance(factory._get_storage('events'), pluggable.PluggableEventsStorageAsync)

        adapter = factory._get_storage('splits')._pluggable_adapter
        assert adapter == factory._get_storage('segments')._pluggable_adapter
        assert adapter == factory._get_storage('impressions')._pluggable_adapter
        assert adapter == factory._get_storage('events')._pluggable_adapter

        assert factory._labels_enabled is False
        assert isinstance(factory._recorder, StandardRecorderAsync)
        assert isinstance(factory._recorder._impressions_manager, ImpressionsManager)
        assert isinstance(factory._recorder._event_sotrage, pluggable.PluggableEventsStorageAsync)
        assert isinstance(factory._recorder._impression_storage, pluggable.PluggableImpressionsStorageAsync)
        try:
            await factory.block_until_ready(1)
        except:
            pass
        assert factory._status == Status.READY
        await factory.destroy()

    @pytest.mark.asyncio
    async def test_destroy_redis_async(self, mocker):
        internal_events_queue = asyncio.Queue()
        events_manager = mocker.Mock(EventsManagerAsync)
        async def notify_internal_event(sdk_internal_event, event_metadata):
            pass
        events_manager.notify_internal_event = notify_internal_event

        async def _make_factory_with_apikey(apikey, *_, **__):
            return SplitFactoryAsync(apikey, {}, True, mocker.Mock(), internal_events_queue, events_manager, mocker.Mock(spec=ManagerAsync), mocker.Mock(), mocker.Mock(), mocker.Mock(), mocker.Mock())

        factory_module_logger = mocker.Mock()
        build_redis = mocker.Mock()
        build_redis.side_effect = _make_factory_with_apikey
        mocker.patch('splitio.client.factory._LOGGER', new=factory_module_logger)
        mocker.patch('splitio.client.factory._build_redis_factory_async', new=build_redis)

        config = {
            'redisDb': 0,
            'redisHost': 'localhost',
            'redisPosrt': 6379,
        }
        factory = await get_factory_async("none", config=config)
        await factory.destroy()
        assert factory.destroyed
        assert len(build_redis.mock_calls) == 1

        factory = await get_factory_async("none", config=config)
        await factory.destroy()
        await asyncio.sleep(0.5)
        assert factory.destroyed
        assert len(build_redis.mock_calls) == 2
        
    @pytest.mark.asyncio
    async def test_internal_ready_event_notification(self, mocker):
        """Test that a client with in-memory storage is sending internal events correctly."""
        # Setup synchronizer
        def _split_synchronizer(self, ready_flag, some, auth_api, streaming_enabled, sdk_matadata, telemetry_runtime_producer, sse_url=None, client_key=None):
            synchronizer = mocker.Mock(spec=SynchronizerAsync)
            async def sync_all(*_):
                return None
            synchronizer.sync_all = sync_all

            def start_periodic_fetching():
                pass        
            synchronizer.start_periodic_fetching = start_periodic_fetching

            def start_periodic_data_recording():
                pass        
            synchronizer.start_periodic_data_recording = start_periodic_data_recording

            self._ready_flag = ready_flag
            self._synchronizer = synchronizer
            self._streaming_enabled = False
            self._telemetry_runtime_producer = telemetry_runtime_producer
            
        mocker.patch('splitio.sync.manager.ManagerAsync.__init__', new=_split_synchronizer)

        async def synchronize_config(*_):
            await asyncio.sleep(2)
            pass
        mocker.patch('splitio.sync.telemetry.InMemoryTelemetrySubmitterAsync.synchronize_config', new=synchronize_config)

        async def record_ready_time(*_):
            pass
        mocker.patch('splitio.models.telemetry.TelemetryConfigAsync.record_ready_time', new=record_ready_time)

        async def record_active_and_redundant_factories(*_):
            pass
        mocker.patch('splitio.models.telemetry.TelemetryConfigAsync.record_active_and_redundant_factories', new=record_active_and_redundant_factories)

        # Start factory and make assertions
        factory = await get_factory_async('some_api_key', config={'streamingEmabled': False})
        for task in asyncio.all_tasks():
            if task._coro.__qualname__ == "EventsTaskAsync._run":
                task.cancel()
        try:
            await factory.block_until_ready(3)
        except:
            pass
        await asyncio.sleep(.2)
        event = await factory._internal_events_queue.get()
        assert event.internal_event == SdkInternalEvent.SDK_READY
        assert event.metadata == None
        await factory.destroy()