diff --git a/splitio/client/client.py b/splitio/client/client.py index 7dcb3edc..a0d0e5f0 100644 --- a/splitio/client/client.py +++ b/splitio/client/client.py @@ -313,6 +313,8 @@ def get_treatment(self, key, feature_flag_name, attributes=None, evaluation_opti except: _LOGGER.error('get_treatment failed') + _LOGGER.error('Error: ', exc_info=True) + treatment, _ = self._get_fallback_treatment_with_config(feature_flag_name) return treatment @@ -759,7 +761,7 @@ def track(self, key, traffic_type, event_type, value=None, properties=None): except Exception: # pylint: disable=broad-except self._telemetry_evaluation_producer.record_exception(MethodExceptionsAndLatencies.TRACK) _LOGGER.error('Error processing track event') - _LOGGER.debug('Error: ', exc_info=True) + _LOGGER.error('Error: ', exc_info=True) return False diff --git a/splitio/client/factory.py b/splitio/client/factory.py index 5b6c8494..e1d12918 100644 --- a/splitio/client/factory.py +++ b/splitio/client/factory.py @@ -835,7 +835,7 @@ def _build_push_classes(threading_mode, synchronizer, cfg, telemetry_runtime_pro return push_manager, push_queue -def _build_recorder(threading_mode, cfg, imp_manager, storages, telemetry_evaluation_producer, telemetry_runtime_producer, sdk_metadata, imp_counter, unique_keys_tracker, redis_adapter=None): +def _build_recorder(threading_mode, synchronizer, cfg, imp_manager, storages, telemetry_evaluation_producer, telemetry_runtime_producer, sdk_metadata, imp_counter, unique_keys_tracker, redis_adapter=None): if cfg['storageType'] == 'redis': data_sampling = cfg.get('dataSampling', DEFAULT_DATA_SAMPLING) if data_sampling < _MIN_DEFAULT_DATA_SAMPLING_ALLOWED: @@ -870,6 +870,8 @@ def _build_recorder(threading_mode, cfg, imp_manager, storages, telemetry_evalua if threading_mode == ThreadingMode.ASYNC: return StandardRecorderAsync( + synchronizer._harness_synchronizers.impressions_sync.synchronize_impressions, + synchronizer._harness_synchronizers.events_sync.synchronize_events, imp_manager, storages['events'], storages['impressions'], @@ -881,6 +883,8 @@ def _build_recorder(threading_mode, cfg, imp_manager, storages, telemetry_evalua ) return StandardRecorder( + synchronizer._harness_synchronizers.impressions_sync.synchronize_impressions, + synchronizer._harness_synchronizers.events_sync.synchronize_events, imp_manager, storages['events'], storages['impressions'], @@ -936,10 +940,8 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None, # pyl push_manager, push_queue = _build_push_classes(ThreadingMode.THREADED, synchronizer, cfg, telemetry_runtime_producer, apis, sdk_metadata, streaming_api_base_url, api_key) manager = Manager(sdk_ready_flag, synchronizer, apis['auth'], cfg['streamingEnabled'], sdk_metadata, telemetry_runtime_producer, streaming_api_base_url, api_key[-4:], push_manager, push_queue) - recorder = _build_recorder(ThreadingMode.THREADED, cfg, imp_manager, storages, telemetry_evaluation_producer, telemetry_runtime_producer, sdk_metadata, imp_counter, unique_keys_tracker) + recorder = _build_recorder(ThreadingMode.THREADED, synchronizer, cfg, imp_manager, storages, telemetry_evaluation_producer, telemetry_runtime_producer, sdk_metadata, imp_counter, unique_keys_tracker) - storages['events'].set_queue_full_hook(tasks.events_task.flush) - storages['impressions'].set_queue_full_hook(tasks.impressions_task.flush) telemetry_init_producer.record_config(cfg, extra_cfg, total_flag_sets, invalid_flag_sets) internal_events_task.start() @@ -992,9 +994,7 @@ async def _build_in_memory_factory_async(api_key, cfg, sdk_url=None, events_url= manager = ManagerAsync(synchronizer, apis['auth'], cfg['streamingEnabled'], sdk_metadata, telemetry_runtime_producer, streaming_api_base_url, api_key[-4:], push_manager, push_queue) - storages['events'].set_queue_full_hook(tasks.events_task.flush) - storages['impressions'].set_queue_full_hook(tasks.impressions_task.flush) - recorder = _build_recorder(ThreadingMode.ASYNC, cfg, imp_manager, storages, telemetry_evaluation_producer, telemetry_runtime_producer, sdk_metadata, imp_counter, unique_keys_tracker) + recorder = _build_recorder(ThreadingMode.ASYNC, synchronizer, cfg, imp_manager, storages, telemetry_evaluation_producer, telemetry_runtime_producer, sdk_metadata, imp_counter, unique_keys_tracker) await telemetry_init_producer.record_config(cfg, extra_cfg, total_flag_sets, invalid_flag_sets) internal_events_task.start() @@ -1022,7 +1022,7 @@ def _build_redis_factory(api_key, cfg): tasks = _build_sync_tasks(ThreadingMode.THREADED, synchronizers, cfg, None) synchronizer = RedisSynchronizer(synchronizers, tasks) - recorder = _build_recorder(ThreadingMode.THREADED, cfg, imp_manager, storages, None, None, sdk_metadata, imp_counter, unique_keys_tracker, redis_adapter) + recorder = _build_recorder(ThreadingMode.THREADED, synchronizer, cfg, imp_manager, storages, None, None, sdk_metadata, imp_counter, unique_keys_tracker, redis_adapter) manager = RedisManager(synchronizer) initialization_thread = threading.Thread(target=manager.start, name="SDKInitializer", daemon=True) initialization_thread.start() @@ -1065,7 +1065,7 @@ async def _build_redis_factory_async(api_key, cfg): tasks = _build_sync_tasks(ThreadingMode.ASYNC, synchronizers, cfg, None) synchronizer = RedisSynchronizerAsync(synchronizers, tasks) - recorder = _build_recorder(ThreadingMode.ASYNC, cfg, imp_manager, storages, None, None, sdk_metadata, imp_counter, unique_keys_tracker, redis_adapter) + recorder = _build_recorder(ThreadingMode.ASYNC, synchronizer, cfg, imp_manager, storages, None, None, sdk_metadata, imp_counter, unique_keys_tracker, redis_adapter) manager = RedisManagerAsync(synchronizer) await telemetry_init_producer.record_config(cfg, {}, 0, 0) manager.start() @@ -1109,7 +1109,19 @@ def _build_pluggable_factory(api_key, cfg): tasks = _build_sync_tasks(ThreadingMode.THREADED, synchronizers, cfg, None) synchronizer = RedisSynchronizer(synchronizers, tasks) - recorder = _build_recorder(ThreadingMode.THREADED, cfg, imp_manager, storages, NoOpClass(), NoOpClass(), sdk_metadata, imp_counter, unique_keys_tracker, pluggable_adapter) + recorder = StandardRecorder( + None, + None, + imp_manager, + storages['events'], + storages['impressions'], + NoOpClass(), + NoOpClass(), + _wrap_impression_listener(cfg['impressionListener'], sdk_metadata), + imp_counter=imp_counter, + unique_keys_tracker=unique_keys_tracker + ) + # Using same class as redis for consumer mode only manager = RedisManager(synchronizer) initialization_thread = threading.Thread(target=manager.start, name="SDKInitializer", daemon=True) @@ -1157,7 +1169,19 @@ async def _build_pluggable_factory_async(api_key, cfg): tasks = _build_sync_tasks(ThreadingMode.ASYNC, synchronizers, cfg, None) synchronizer = RedisSynchronizerAsync(synchronizers, tasks) - recorder = _build_recorder(ThreadingMode.ASYNC, cfg, imp_manager, storages, NoOpClassAsync(), NoOpClassAsync(), sdk_metadata, imp_counter, unique_keys_tracker, pluggable_adapter) + recorder = StandardRecorderAsync( + None, + None, + imp_manager, + storages['events'], + storages['impressions'], + telemetry_producer.get_telemetry_evaluation_producer(), + telemetry_runtime_producer, + _wrap_impression_listener(cfg['impressionListener'], sdk_metadata), + imp_counter=imp_counter, + unique_keys_tracker=unique_keys_tracker + ) + # Using same class as redis for consumer mode only manager = RedisManagerAsync(synchronizer) manager.start() @@ -1241,6 +1265,8 @@ def _build_localhost_factory(cfg): manager.start() recorder = StandardRecorder( + None, + None, ImpressionsManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer), storages['events'], storages['impressions'], @@ -1322,6 +1348,8 @@ async def _build_localhost_factory_async(cfg): await manager.start() recorder = StandardRecorderAsync( + None, + None, ImpressionsManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer), storages['events'], storages['impressions'], diff --git a/splitio/storage/pluggable.py b/splitio/storage/pluggable.py index 25ce30cb..724c6f1a 100644 --- a/splitio/storage/pluggable.py +++ b/splitio/storage/pluggable.py @@ -3,6 +3,7 @@ import logging import json import threading +import queue from splitio.optional.loaders import asyncio from splitio.models import splits @@ -1327,7 +1328,6 @@ def expire_key(self, total_keys, inserted): if total_keys == inserted: self._pluggable_adapter.expire(self._impressions_queue_key, self.IMPRESSIONS_KEY_DEFAULT_TTL) - class PluggableImpressionsStorageAsync(PluggableImpressionsStorageBase): """Pluggable Impressions storage class.""" @@ -1473,6 +1473,7 @@ def __init__(self, pluggable_adapter, sdk_metadata, prefix=None): :type prefix: str """ PluggableEventsStorageBase.__init__(self, pluggable_adapter, sdk_metadata, prefix) + self._events = queue.Queue() # adding it to allow hook condition in recorder def put(self, events): """ diff --git a/tests/client/test_client.py b/tests/client/test_client.py index 5c061957..91f1db5a 100644 --- a/tests/client/test_client.py +++ b/tests/client/test_client.py @@ -42,7 +42,7 @@ def test_get_treatment(self, mocker): 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) + impression_storage = InMemoryImpressionStorage(100000, telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) destroyed_property = mocker.PropertyMock() @@ -52,7 +52,7 @@ def test_get_treatment(self, mocker): mocker.patch('splitio.client.client.get_latency_bucket_index', new=lambda x: 5) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) - recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer(), + recorder = StandardRecorder(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer(), unique_keys_tracker=UniqueKeysTracker(), imp_counter=ImpressionsCounter()) class TelemetrySubmitterMock(): @@ -122,14 +122,14 @@ def test_get_treatment_with_config(self, mocker): 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) + impression_storage = InMemoryImpressionStorage(10000, telemetry_runtime_producer) impmanager = ImpressionManager(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()) + recorder = StandardRecorder(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactory(mocker.Mock(), {'splits': split_storage, 'segments': segment_storage, @@ -202,7 +202,7 @@ def test_get_treatments(self, mocker): 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) + impression_storage = InMemoryImpressionStorage(10000, telemetry_runtime_producer) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0]), from_raw(splits_json['splitChange1_1']['ff']['d'][1])], [], -1) @@ -210,7 +210,7 @@ def test_get_treatments(self, mocker): 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()) + recorder = StandardRecorder(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactory(mocker.Mock(), {'splits': split_storage, 'segments': segment_storage, @@ -285,7 +285,7 @@ def test_get_treatments_by_flag_set(self, mocker): 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) + impression_storage = InMemoryImpressionStorage(10000, telemetry_runtime_producer) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0]), from_raw(splits_json['splitChange1_1']['ff']['d'][1])], [], -1) @@ -293,7 +293,7 @@ def test_get_treatments_by_flag_set(self, mocker): 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()) + recorder = StandardRecorder(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactory(mocker.Mock(), {'splits': split_storage, 'segments': segment_storage, @@ -367,7 +367,7 @@ def test_get_treatments_by_flag_sets(self, mocker): 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) + impression_storage = InMemoryImpressionStorage(10000, telemetry_runtime_producer) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0]), from_raw(splits_json['splitChange1_1']['ff']['d'][1])], [], -1) @@ -375,7 +375,7 @@ def test_get_treatments_by_flag_sets(self, mocker): 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()) + recorder = StandardRecorder(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactory(mocker.Mock(), {'splits': split_storage, 'segments': segment_storage, @@ -449,14 +449,14 @@ def test_get_treatments_with_config(self, mocker): 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) + impression_storage = InMemoryImpressionStorage(10000, telemetry_runtime_producer) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0]), from_raw(splits_json['splitChange1_1']['ff']['d'][1])], [], -1) 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()) + recorder = StandardRecorder(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactory(mocker.Mock(), {'splits': split_storage, 'segments': segment_storage, @@ -535,14 +535,14 @@ def test_get_treatments_with_config_by_flag_set(self, mocker): 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) + impression_storage = InMemoryImpressionStorage(10000, telemetry_runtime_producer) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0]), from_raw(splits_json['splitChange1_1']['ff']['d'][1])], [], -1) 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()) + recorder = StandardRecorder(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) events_queue = queue.Queue() factory = SplitFactory(mocker.Mock(), {'splits': split_storage, @@ -619,14 +619,14 @@ def test_get_treatments_with_config_by_flag_sets(self, mocker): 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) + impression_storage = InMemoryImpressionStorage(10000, telemetry_runtime_producer) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0]), from_raw(splits_json['splitChange1_1']['ff']['d'][1])], [], -1) 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()) + recorder = StandardRecorder(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactory(mocker.Mock(), {'splits': split_storage, 'segments': segment_storage, @@ -702,7 +702,7 @@ def test_impression_toggle_optimized(self, mocker): 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) + impression_storage = InMemoryImpressionStorage(10000, telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) destroyed_property = mocker.PropertyMock() @@ -712,7 +712,7 @@ def test_impression_toggle_optimized(self, mocker): mocker.patch('splitio.client.client.get_latency_bucket_index', new=lambda x: 5) impmanager = ImpressionManager(StrategyOptimizedMode(), StrategyNoneMode(), telemetry_runtime_producer) - recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorder(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) class TelemetrySubmitterMock(): def synchronize_config(*_): pass @@ -769,7 +769,7 @@ def test_impression_toggle_debug(self, mocker): 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) + impression_storage = InMemoryImpressionStorage(10000, telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) destroyed_property = mocker.PropertyMock() @@ -779,7 +779,7 @@ def test_impression_toggle_debug(self, mocker): mocker.patch('splitio.client.client.get_latency_bucket_index', new=lambda x: 5) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) - recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorder(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) class TelemetrySubmitterMock(): def synchronize_config(*_): pass @@ -836,7 +836,7 @@ def test_impression_toggle_none(self, mocker): 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) + impression_storage = InMemoryImpressionStorage(10000, telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) destroyed_property = mocker.PropertyMock() @@ -846,7 +846,7 @@ def test_impression_toggle_none(self, mocker): mocker.patch('splitio.client.client.get_latency_bucket_index', new=lambda x: 5) non_strategy = StrategyNoneMode() impmanager = ImpressionManager(non_strategy, non_strategy, telemetry_runtime_producer) - recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorder(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) class TelemetrySubmitterMock(): def synchronize_config(*_): pass @@ -896,7 +896,7 @@ def test_destroy(self, mocker): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) - recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorder(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) events_queue = queue.Queue() factory = SplitFactory(mocker.Mock(), {'splits': split_storage, @@ -932,11 +932,14 @@ def test_track(self, mocker): impression_storage = mocker.Mock(spec=ImpressionStorage) event_storage = mocker.Mock(spec=EventStorage) event_storage.put.return_value = True + event_storage._size = 10 + event_storage._queue_size = 1000 + event_storage._events = queue.Queue() impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) - recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorder(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) events_queue = queue.Queue() factory = SplitFactory(mocker.Mock(), {'splits': split_storage, @@ -978,7 +981,7 @@ def test_evaluations_before_running_post_fork(self, mocker): telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer() - impression_storage = InMemoryImpressionStorage(10, telemetry_runtime_producer) + impression_storage = InMemoryImpressionStorage(10000, telemetry_runtime_producer) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) events_queue = queue.Queue() split_storage = InMemorySplitStorage(events_queue) @@ -989,7 +992,7 @@ def test_evaluations_before_running_post_fork(self, mocker): destroyed_property.return_value = False impmanager = mocker.Mock(spec=ImpressionManager) - recorder = StandardRecorder(impmanager, mocker.Mock(), impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorder(None, None, impmanager, mocker.Mock(), impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) events_queue = queue.Queue() factory = SplitFactory(mocker.Mock(), {'splits': split_storage, @@ -1063,14 +1066,14 @@ def test_telemetry_not_ready(self, mocker): telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer() - impression_storage = InMemoryImpressionStorage(10, telemetry_runtime_producer) + impression_storage = InMemoryImpressionStorage(10000, telemetry_runtime_producer) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) events_queue = queue.Queue() split_storage = InMemorySplitStorage(events_queue) segment_storage = InMemorySegmentStorage(events_queue) rb_segment_storage = InMemoryRuleBasedSegmentStorage(events_queue) split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0])], [], -1) - recorder = StandardRecorder(impmanager, mocker.Mock(), mocker.Mock(), telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorder(None, None, impmanager, mocker.Mock(), mocker.Mock(), telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactory('localhost', {'splits': split_storage, 'segments': segment_storage, @@ -1117,7 +1120,7 @@ def test_telemetry_record_treatment_exception(self, mocker): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) - recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorder(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactory('localhost', {'splits': split_storage, 'segments': segment_storage, @@ -1206,7 +1209,7 @@ def test_telemetry_method_latency(self, mocker): telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer() - impression_storage = InMemoryImpressionStorage(10, telemetry_runtime_producer) + impression_storage = InMemoryImpressionStorage(10000, telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) events_queue = queue.Queue() @@ -1214,7 +1217,7 @@ def test_telemetry_method_latency(self, mocker): segment_storage = InMemorySegmentStorage(events_queue) rb_segment_storage = InMemoryRuleBasedSegmentStorage(events_queue) split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0])], [], -1) - recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorder(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) destroyed_property = mocker.PropertyMock() destroyed_property.return_value = False @@ -1292,7 +1295,7 @@ def test_telemetry_track_exception(self, mocker): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) - recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorder(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) events_queue = queue.Queue() factory = SplitFactory(mocker.Mock(), {'splits': split_storage, @@ -1332,7 +1335,7 @@ def test_impressions_properties(self, mocker): 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) + impression_storage = InMemoryImpressionStorage(10000, telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) destroyed_property = mocker.PropertyMock() @@ -1342,7 +1345,7 @@ def test_impressions_properties(self, mocker): mocker.patch('splitio.client.client.get_latency_bucket_index', new=lambda x: 5) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) - recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer(), + recorder = StandardRecorder(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer(), unique_keys_tracker=UniqueKeysTracker(), imp_counter=ImpressionsCounter()) class TelemetrySubmitterMock(): @@ -1441,8 +1444,16 @@ def synchronize_config(*_): def test_fallback_treatment_eval_exception(self, mocker): # using fallback when the evaluator has RuntimeError exception split_storage = mocker.Mock(spec=SplitStorage) + def fetch_many_splits(*_): + return {"some": from_raw(splits_json['splitChange1_1']['ff']['d'][0]), + "some2": from_raw(splits_json['splitChange1_1']['ff']['d'][1])} + split_storage.fetch_many = fetch_many_splits + segment_storage = mocker.Mock(spec=SegmentStorage) - rb_segment_storage = mocker.Mock(spec=RuleBasedSegmentsStorage) + rb_segment_storage = mocker.Mock(spec=RuleBasedSegmentsStorage) + def fetch_many_rbs(*_): + return {} + rb_segment_storage.fetch_many = fetch_many_rbs impression_storage = mocker.Mock(spec=ImpressionStorage) event_storage = mocker.Mock(spec=EventStorage) destroyed_property = mocker.PropertyMock() @@ -1454,7 +1465,7 @@ def test_fallback_treatment_eval_exception(self, mocker): telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) impmanager = ImpressionManager(StrategyOptimizedMode(), StrategyNoneMode(), telemetry_producer.get_telemetry_runtime_producer()) - recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorder(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) internal_events_queue = queue.Queue() factory = SplitFactory(mocker.Mock(), {'splits': split_storage, @@ -1473,9 +1484,9 @@ def test_fallback_treatment_eval_exception(self, mocker): mocker.Mock() ) - self.imps = None + self.imps = [] def put(impressions): - self.imps = impressions + self.imps.append(impressions[0]) impression_storage.put = put class TelemetrySubmitterMock(): @@ -1493,7 +1504,7 @@ def get_feature_flag_names_by_flag_sets(*_): assert(self.imps[0].treatment == "on-global") assert(self.imps[0].label == "fallback - exception") - self.imps = None + self.imps = [] treatment = client.get_treatments("key_m", ["some", "some2"]) assert(treatment == {"some": "on-global", "some2": "on-global"}) assert(self.imps[0].treatment == "on-global") @@ -1508,14 +1519,14 @@ def get_feature_flag_names_by_flag_sets(*_): assert(client.get_treatments_with_config_by_flag_set("key_m", "set") == {"some": ("on-global", '{"prop": "val"}'), "some2": ("on-global", '{"prop": "val"}')}) assert(client.get_treatments_with_config_by_flag_sets("key_m", ["set"]) == {"some": ("on-global", '{"prop": "val"}'), "some2": ("on-global", '{"prop": "val"}')}) - self.imps = None + self.imps = [] client._fallback_treatment_calculator = FallbackTreatmentCalculator(FallbackTreatmentsConfiguration(FallbackTreatment("on-global", '{"prop": "val"}'), {'some': FallbackTreatment("on-local")})) treatment = client.get_treatment("key2", "some") assert(treatment == "on-local") assert(self.imps[0].treatment == "on-local") assert(self.imps[0].label == "fallback - exception") - self.imps = None + self.imps = [] treatment = client.get_treatments("key2_m", ["some", "some2"]) assert(treatment == {"some": "on-local", "some2": "on-global"}) assert_both = 0 @@ -1537,14 +1548,14 @@ def get_feature_flag_names_by_flag_sets(*_): assert(client.get_treatments_with_config_by_flag_set("key_m", "set") == {"some": ("on-local", None), "some2": ("on-global", '{"prop": "val"}')}) assert(client.get_treatments_with_config_by_flag_sets("key_m", ["set"]) == {"some": ("on-local", None), "some2": ("on-global", '{"prop": "val"}')}) - self.imps = None + self.imps = [] client._fallback_treatment_calculator = FallbackTreatmentCalculator(FallbackTreatmentsConfiguration(None, {'some': FallbackTreatment("on-local", '{"prop": "val"}')})) treatment = client.get_treatment("key3", "some") assert(treatment == "on-local") assert(self.imps[0].treatment == "on-local") assert(self.imps[0].label == "fallback - exception") - self.imps = None + self.imps = [] treatment = client.get_treatments("key3_m", ["some", "some2"]) assert(treatment == {"some": "on-local", "some2": "control"}) assert_both = 0 @@ -1566,7 +1577,7 @@ def get_feature_flag_names_by_flag_sets(*_): assert(client.get_treatments_with_config_by_flag_set("key_m", "set") == {"some": ("on-local", '{"prop": "val"}'), "some2": ("control", None)}) assert(client.get_treatments_with_config_by_flag_sets("key_m", ["set"]) == {"some": ("on-local", '{"prop": "val"}'), "some2": ("control", None)}) - self.imps = None + self.imps = [] client._fallback_treatment_calculator = FallbackTreatmentCalculator(FallbackTreatmentsConfiguration(None, {'some2': FallbackTreatment("on-local")})) treatment = client.get_treatment("key4", "some") assert(treatment == "control") @@ -1595,7 +1606,7 @@ def test_fallback_treatment_exception(self, mocker): telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) impmanager = ImpressionManager(StrategyOptimizedMode(), StrategyNoneMode(), telemetry_producer.get_telemetry_runtime_producer()) - recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorder(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) internal_events_queue = queue.Queue() factory = SplitFactory(mocker.Mock(), {'splits': split_storage, @@ -1666,7 +1677,7 @@ def test_fallback_treatment_not_ready_impressions(self, mocker): telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) impmanager = ImpressionManager(StrategyOptimizedMode(), StrategyNoneMode(), telemetry_producer.get_telemetry_runtime_producer()) - recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorder(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) internal_events_queue = queue.Queue() factory = SplitFactory(mocker.Mock(), {'splits': split_storage, @@ -1761,7 +1772,7 @@ async def notify_internal_event(sdk_internal_event, event_metadata): impression_storage = InMemoryImpressionStorageAsync(10, telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) - recorder = StandardRecorderAsync(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorderAsync(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) await split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0])], [], -1) destroyed_property = mocker.PropertyMock() @@ -1841,7 +1852,7 @@ async def notify_internal_event(sdk_internal_event, event_metadata): impression_storage = InMemoryImpressionStorageAsync(10, telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) - recorder = StandardRecorderAsync(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorderAsync(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) await split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0])], [], -1) destroyed_property = mocker.PropertyMock() @@ -1926,7 +1937,7 @@ async def notify_internal_event(sdk_internal_event, event_metadata): impression_storage = InMemoryImpressionStorageAsync(10, telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) - recorder = StandardRecorderAsync(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorderAsync(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) await split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0]), from_raw(splits_json['splitChange1_1']['ff']['d'][1])], [], -1) destroyed_property = mocker.PropertyMock() @@ -2014,7 +2025,7 @@ async def notify_internal_event(sdk_internal_event, event_metadata): impression_storage = InMemoryImpressionStorageAsync(10, telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) - recorder = StandardRecorderAsync(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorderAsync(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) await split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0]), from_raw(splits_json['splitChange1_1']['ff']['d'][1])], [], -1) destroyed_property = mocker.PropertyMock() @@ -2102,7 +2113,7 @@ async def notify_internal_event(sdk_internal_event, event_metadata): impression_storage = InMemoryImpressionStorageAsync(10, telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) - recorder = StandardRecorderAsync(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorderAsync(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) await split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0]), from_raw(splits_json['splitChange1_1']['ff']['d'][1])], [], -1) destroyed_property = mocker.PropertyMock() @@ -2190,7 +2201,7 @@ async def notify_internal_event(sdk_internal_event, event_metadata): impression_storage = InMemoryImpressionStorageAsync(10, telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) - recorder = StandardRecorderAsync(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorderAsync(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) await split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0]), from_raw(splits_json['splitChange1_1']['ff']['d'][1])], [], -1) destroyed_property = mocker.PropertyMock() @@ -2282,7 +2293,7 @@ async def notify_internal_event(sdk_internal_event, event_metadata): impression_storage = InMemoryImpressionStorageAsync(10, telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) - recorder = StandardRecorderAsync(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorderAsync(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) await split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0]), from_raw(splits_json['splitChange1_1']['ff']['d'][1])], [], -1) destroyed_property = mocker.PropertyMock() @@ -2374,7 +2385,7 @@ async def notify_internal_event(sdk_internal_event, event_metadata): impression_storage = InMemoryImpressionStorageAsync(10, telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) - recorder = StandardRecorderAsync(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorderAsync(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) await split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0]), from_raw(splits_json['splitChange1_1']['ff']['d'][1])], [], -1) destroyed_property = mocker.PropertyMock() @@ -2473,7 +2484,7 @@ async def notify_internal_event(sdk_internal_event, event_metadata): mocker.patch('splitio.client.client.get_latency_bucket_index', new=lambda x: 5) impmanager = ImpressionManager(StrategyOptimizedMode(), StrategyNoneMode(), telemetry_runtime_producer) - recorder = StandardRecorderAsync(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorderAsync(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactoryAsync(mocker.Mock(), {'splits': split_storage, 'segments': segment_storage, @@ -2544,7 +2555,7 @@ async def notify_internal_event(sdk_internal_event, event_metadata): mocker.patch('splitio.client.client.get_latency_bucket_index', new=lambda x: 5) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) - recorder = StandardRecorderAsync(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorderAsync(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactoryAsync(mocker.Mock(), {'splits': split_storage, 'segments': segment_storage, @@ -2612,7 +2623,7 @@ async def notify_internal_event(sdk_internal_event, event_metadata): mocker.patch('splitio.client.client.get_latency_bucket_index', new=lambda x: 5) non_strategy = StrategyNoneMode() impmanager = ImpressionManager(non_strategy, non_strategy, telemetry_runtime_producer) - recorder = StandardRecorderAsync(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorderAsync(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactoryAsync(mocker.Mock(), {'splits': split_storage, 'segments': segment_storage, @@ -2664,11 +2675,14 @@ async def put(event): self.events.append(event) return True event_storage.put = put + event_storage._size = 10 + event_storage._queue_size = 1000 + event_storage._events = asyncio.Queue() impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = await InMemoryTelemetryStorageAsync.create() telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage) - recorder = StandardRecorderAsync(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorderAsync(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactoryAsync(mocker.Mock(), {'splits': split_storage, 'segments': segment_storage, @@ -2720,7 +2734,7 @@ async def notify_internal_event(sdk_internal_event, event_metadata): impression_storage = InMemoryImpressionStorageAsync(10, telemetry_runtime_producer) event_storage = InMemoryEventStorageAsync(10, telemetry_runtime_producer) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) - recorder = StandardRecorderAsync(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorderAsync(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) await split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0])], [], -1) factory = SplitFactoryAsync('localhost', {'splits': split_storage, @@ -2771,7 +2785,7 @@ async def notify_internal_event(sdk_internal_event, event_metadata): impression_storage = InMemoryImpressionStorageAsync(10, telemetry_runtime_producer) event_storage = InMemoryEventStorageAsync(10, telemetry_runtime_producer) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) - recorder = StandardRecorderAsync(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorderAsync(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) await split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0])], [], -1) destroyed_property = mocker.PropertyMock() destroyed_property.return_value = False @@ -2853,7 +2867,7 @@ async def notify_internal_event(sdk_internal_event, event_metadata): impression_storage = InMemoryImpressionStorageAsync(10, telemetry_runtime_producer) event_storage = InMemoryEventStorageAsync(10, telemetry_runtime_producer) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) - recorder = StandardRecorderAsync(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorderAsync(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) await split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0])], [], -1) destroyed_property = mocker.PropertyMock() destroyed_property.return_value = False @@ -2941,7 +2955,7 @@ async def notify_internal_event(sdk_internal_event, event_metadata): telemetry_storage = await InMemoryTelemetryStorageAsync.create() telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage) event_storage = InMemoryEventStorageAsync(10, telemetry_producer.get_telemetry_runtime_producer()) - recorder = StandardRecorderAsync(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorderAsync(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactoryAsync(mocker.Mock(), {'splits': split_storage, 'segments': segment_storage, @@ -2993,7 +3007,7 @@ async def notify_internal_event(sdk_internal_event, event_metadata): impression_storage = InMemoryImpressionStorageAsync(10, telemetry_runtime_producer) event_storage = mocker.Mock(spec=EventStorage) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) - recorder = StandardRecorderAsync(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer(), imp_counter=ImpressionsCounter()) + recorder = StandardRecorderAsync(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer(), imp_counter=ImpressionsCounter()) await split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0])], [], -1) destroyed_property = mocker.PropertyMock() @@ -3105,6 +3119,9 @@ async def notify_internal_event(sdk_internal_event, event_metadata): segment_storage = mocker.Mock(spec=SegmentStorage) rb_segment_storage = mocker.Mock(spec=RuleBasedSegmentsStorage) impression_storage = mocker.Mock(spec=ImpressionStorage) + impression_storage._size = 10 + impression_storage._queue_size = 1000 + impression_storage._impressions = asyncio.Queue() event_storage = mocker.Mock(spec=EventStorage) destroyed_property = mocker.PropertyMock() destroyed_property.return_value = False @@ -3116,7 +3133,7 @@ async def notify_internal_event(sdk_internal_event, event_metadata): telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage) telemetry_evaluation_producer = telemetry_producer.get_telemetry_evaluation_producer() impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_producer.get_telemetry_runtime_producer()) - recorder = StandardRecorderAsync(impmanager, event_storage, impression_storage, telemetry_evaluation_producer, telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorderAsync(None, None, impmanager, event_storage, impression_storage, telemetry_evaluation_producer, telemetry_producer.get_telemetry_runtime_producer()) class TelemetrySubmitterMock(): async def synchronize_config(*_): @@ -3143,9 +3160,9 @@ async def synchronize_config(*_): ready_property.return_value = True type(factory).ready = ready_property - self.imps = None + self.imps = [] async def put(impressions): - self.imps = impressions + self.imps.append(impressions[0]) impression_storage.put = put client = ClientAsync(factory, recorder, events_manager, True, FallbackTreatmentCalculator(FallbackTreatmentsConfiguration(FallbackTreatment("on-global", '{"prop": "val"}')))) @@ -3171,7 +3188,7 @@ async def fetch_many_rbs(*_): assert(self.imps[0].treatment == "on-global") assert(self.imps[0].label == "fallback - exception") - self.imps = None + self.imps = [] treatment = await client.get_treatments("key_m", ["some", "some2"]) assert(treatment == {"some": "on-global", "some2": "on-global"}) assert(self.imps[0].treatment == "on-global") @@ -3186,14 +3203,14 @@ async def fetch_many_rbs(*_): assert(await client.get_treatments_with_config_by_flag_set("key_m", "set") == {"some": ("on-global", '{"prop": "val"}'), "some2": ("on-global", '{"prop": "val"}')}) assert(await client.get_treatments_with_config_by_flag_sets("key_m", ["set"]) == {"some": ("on-global", '{"prop": "val"}'), "some2": ("on-global", '{"prop": "val"}')}) - self.imps = None + self.imps = [] client._fallback_treatment_calculator = FallbackTreatmentCalculator(FallbackTreatmentsConfiguration(FallbackTreatment("on-global", '{"prop": "val"}'), {'some': FallbackTreatment("on-local")})) treatment = await client.get_treatment("key2", "some") assert(treatment == "on-local") assert(self.imps[0].treatment == "on-local") assert(self.imps[0].label == "fallback - exception") - self.imps = None + self.imps = [] treatment = await client.get_treatments("key2_m", ["some", "some2"]) assert(treatment == {"some": "on-local", "some2": "on-global"}) assert_both = 0 @@ -3215,14 +3232,14 @@ async def fetch_many_rbs(*_): assert(await client.get_treatments_with_config_by_flag_set("key_m", "set") == {"some": ("on-local", None), "some2": ("on-global", '{"prop": "val"}')}) assert(await client.get_treatments_with_config_by_flag_sets("key_m", ["set"]) == {"some": ("on-local", None), "some2": ("on-global", '{"prop": "val"}')}) - self.imps = None + self.imps = [] client._fallback_treatment_calculator = FallbackTreatmentCalculator(FallbackTreatmentsConfiguration(None, {'some': FallbackTreatment("on-local", '{"prop": "val"}')})) treatment = await client.get_treatment("key3", "some") assert(treatment == "on-local") assert(self.imps[0].treatment == "on-local") assert(self.imps[0].label == "fallback - exception") - self.imps = None + self.imps = [] treatment = await client.get_treatments("key3_m", ["some", "some2"]) assert(treatment == {"some": "on-local", "some2": "control"}) assert_both = 0 @@ -3244,7 +3261,7 @@ async def fetch_many_rbs(*_): assert(await client.get_treatments_with_config_by_flag_set("key_m", "set") == {"some": ("on-local", '{"prop": "val"}'), "some2": ("control", None)}) assert(await client.get_treatments_with_config_by_flag_sets("key_m", ["set"]) == {"some": ("on-local", '{"prop": "val"}'), "some2": ("control", None)}) - self.imps = None + self.imps = [] client._fallback_treatment_calculator = FallbackTreatmentCalculator(FallbackTreatmentsConfiguration(None, {'some2': FallbackTreatment("on-local")})) treatment = await client.get_treatment("key4", "some") assert(treatment == "control") @@ -3269,6 +3286,10 @@ async def notify_internal_event(sdk_internal_event, event_metadata): segment_storage = mocker.Mock(spec=SegmentStorage) rb_segment_storage = mocker.Mock(spec=RuleBasedSegmentsStorage) impression_storage = mocker.Mock(spec=ImpressionStorage) + impression_storage._size = 10 + impression_storage._queue_size = 1000 + impression_storage._impressions = asyncio.Queue() + event_storage = mocker.Mock(spec=EventStorage) destroyed_property = mocker.PropertyMock() destroyed_property.return_value = False @@ -3279,7 +3300,7 @@ async def notify_internal_event(sdk_internal_event, event_metadata): telemetry_storage = await InMemoryTelemetryStorageAsync.create() telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage) impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_producer.get_telemetry_runtime_producer()) - recorder = StandardRecorderAsync(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorderAsync(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactoryAsync(mocker.Mock(), {'splits': split_storage, @@ -3369,6 +3390,10 @@ async def notify_internal_event(sdk_internal_event, event_metadata): segment_storage = mocker.Mock(spec=SegmentStorage) rb_segment_storage = mocker.Mock(spec=RuleBasedSegmentsStorage) impression_storage = mocker.Mock(spec=ImpressionStorage) + impression_storage._size = 10 + impression_storage._queue_size = 1000 + impression_storage._impressions = asyncio.Queue() + event_storage = mocker.Mock(spec=EventStorage) mocker.patch('splitio.client.client.utctime_ms', new=lambda: 1000) @@ -3377,7 +3402,7 @@ async def notify_internal_event(sdk_internal_event, event_metadata): telemetry_storage = await InMemoryTelemetryStorageAsync.create() telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage) impmanager = ImpressionManager(StrategyOptimizedMode(), StrategyNoneMode(), telemetry_producer.get_telemetry_runtime_producer()) - recorder = StandardRecorderAsync(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) + recorder = StandardRecorderAsync(None, None, impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) async def manager_start_task(): pass diff --git a/tests/client/test_factory.py b/tests/client/test_factory.py index 15d9ee12..054bd7ac 100644 --- a/tests/client/test_factory.py +++ b/tests/client/test_factory.py @@ -733,13 +733,13 @@ def _split_synchronizer(self, ready_flag, some, auth_api, streaming_enabled, sdk 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) + impression_storage = InMemoryImpressionStorage(10000, 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()) + recorder = StandardRecorder(None, None, 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, diff --git a/tests/client/test_input_validator.py b/tests/client/test_input_validator.py index 28aab806..97dd7e6f 100644 --- a/tests/client/test_input_validator.py +++ b/tests/client/test_input_validator.py @@ -2,6 +2,7 @@ import pytest import logging import asyncio +import queue from splitio.client.factory import SplitFactory, get_factory, SplitFactoryAsync, get_factory_async from splitio.client.client import CONTROL, Client, _LOGGER as _logger, ClientAsync @@ -41,7 +42,7 @@ def test_get_treatment(self, mocker): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) - recorder = StandardRecorder(impmanager, mocker.Mock(spec=EventStorage), ImpressionStorage, telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorder(None, None, impmanager, mocker.Mock(spec=EventStorage), ImpressionStorage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactory(mocker.Mock(), { @@ -284,7 +285,7 @@ def _configs(treatment): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) - recorder = StandardRecorder(impmanager, mocker.Mock(spec=EventStorage), ImpressionStorage, telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorder(None, None, impmanager, mocker.Mock(spec=EventStorage), ImpressionStorage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactory(mocker.Mock(), { @@ -552,16 +553,18 @@ class Test: def test_track(self, mocker): """Test track method().""" events_storage_mock = mocker.Mock(spec=EventStorage) + events_storage_mock._size = 10 + events_storage_mock._queue_size = 1000 + events_storage_mock._events = queue.Queue() events_storage_mock.put.return_value = True - event_storage = mocker.Mock(spec=EventStorage) - event_storage.put.return_value = True + split_storage_mock = mocker.Mock(spec=DefinitionStorage) split_storage_mock.is_valid_traffic_type.return_value = True impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) - recorder = StandardRecorder(impmanager, events_storage_mock, ImpressionStorage, telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorder(None, None, impmanager, events_storage_mock, ImpressionStorage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactory(mocker.Mock(), { @@ -584,7 +587,7 @@ def test_track(self, mocker): factory._sdk_key = 'some-test' client = Client(factory, recorder, mocker.Mock(), mocker.Mock(), FallbackTreatmentCalculator(None)) - client._event_storage = event_storage + client._event_storage = events_storage_mock _logger = mocker.Mock() mocker.patch('splitio.client.input_validator._LOGGER', new=_logger) @@ -843,7 +846,7 @@ def test_get_treatments(self, mocker): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) - recorder = StandardRecorder(impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorder(None, None, impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactory(mocker.Mock(), { @@ -993,7 +996,7 @@ def test_get_treatments_with_config(self, mocker): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) - recorder = StandardRecorder(impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorder(None, None, impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactory(mocker.Mock(), { @@ -1143,7 +1146,7 @@ def test_get_treatments_by_flag_set(self, mocker): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) - recorder = StandardRecorder(impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorder(None, None, impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactory(mocker.Mock(), { @@ -1264,7 +1267,7 @@ def test_get_treatments_by_flag_sets(self, mocker): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) - recorder = StandardRecorder(impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorder(None, None, impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactory(mocker.Mock(), { @@ -1396,7 +1399,7 @@ def _configs(treatment): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) - recorder = StandardRecorder(impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorder(None, None, impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactory(mocker.Mock(), { @@ -1522,7 +1525,7 @@ def _configs(treatment): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) - recorder = StandardRecorder(impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorder(None, None, impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactory(mocker.Mock(), { @@ -1717,7 +1720,7 @@ async def get_change_number(*_): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = await InMemoryTelemetryStorageAsync.create() telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage) - recorder = StandardRecorderAsync(impmanager, mocker.Mock(spec=EventStorage), ImpressionStorage, telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorderAsync(None, None, impmanager, mocker.Mock(spec=EventStorage), ImpressionStorage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactoryAsync(mocker.Mock(), { @@ -1988,7 +1991,7 @@ async def get_change_number(*_): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = await InMemoryTelemetryStorageAsync.create() telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage) - recorder = StandardRecorderAsync(impmanager, mocker.Mock(spec=EventStorage), ImpressionStorage, telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorderAsync(None, None, impmanager, mocker.Mock(spec=EventStorage), ImpressionStorage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactoryAsync(mocker.Mock(), { @@ -2232,16 +2235,17 @@ async def notify_internal_event(sdk_internal_event, event_metadata): async def put(*_): return True events_storage_mock.put = put - - event_storage = mocker.Mock(spec=EventStorage) - event_storage.put = put + events_storage_mock._size = 10 + events_storage_mock._queue_size = 1000 + events_storage_mock._events = asyncio.Queue() + split_storage_mock = mocker.Mock(spec=DefinitionStorage) split_storage_mock.is_valid_traffic_type = put impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = await InMemoryTelemetryStorageAsync.create() telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage) - recorder = StandardRecorderAsync(impmanager, events_storage_mock, ImpressionStorage, telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorderAsync(None, None, impmanager, events_storage_mock, ImpressionStorage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactoryAsync(mocker.Mock(), { @@ -2264,7 +2268,7 @@ async def put(*_): factory._sdk_key = 'some-test' client = ClientAsync(factory, recorder, events_manager, mocker.Mock(), FallbackTreatmentCalculator(None)) - client._event_storage = event_storage + client._event_storage = events_storage_mock _logger = mocker.Mock() mocker.patch('splitio.client.input_validator._LOGGER', new=_logger) @@ -2540,7 +2544,7 @@ async def fetch_many_rbs(*_): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = await InMemoryTelemetryStorageAsync.create() telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage) - recorder = StandardRecorderAsync(impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorderAsync(None, None, impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactoryAsync(mocker.Mock(), { @@ -2713,7 +2717,7 @@ async def fetch_many_rbs(*_): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = await InMemoryTelemetryStorageAsync.create() telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage) - recorder = StandardRecorderAsync(impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorderAsync(None, None, impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactoryAsync(mocker.Mock(), { @@ -2889,7 +2893,7 @@ async def fetch_many_rbs(*_): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = await InMemoryTelemetryStorageAsync.create() telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage) - recorder = StandardRecorderAsync(impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorderAsync(None, None, impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactoryAsync(mocker.Mock(), { @@ -3044,7 +3048,7 @@ async def get_feature_flags_by_sets(*_): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = await InMemoryTelemetryStorageAsync.create() telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage) - recorder = StandardRecorderAsync(impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorderAsync(None, None, impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactoryAsync(mocker.Mock(), { @@ -3208,7 +3212,7 @@ async def get_feature_flags_by_sets(*_): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = await InMemoryTelemetryStorageAsync.create() telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage) - recorder = StandardRecorderAsync(impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorderAsync(None, None, impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactoryAsync(mocker.Mock(), { @@ -3366,7 +3370,7 @@ async def get_feature_flags_by_sets(*_): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = await InMemoryTelemetryStorageAsync.create() telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage) - recorder = StandardRecorderAsync(impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorderAsync(None, None, impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactoryAsync(mocker.Mock(), { @@ -3527,7 +3531,7 @@ def test_split_(self, mocker): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) - recorder = StandardRecorder(impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorder(None, None, impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactory(mocker.Mock(), { @@ -3612,7 +3616,12 @@ async def get(*_): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = await InMemoryTelemetryStorageAsync.create() telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage) - recorder = StandardRecorderAsync(impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), + event_storage = mocker.Mock(spec=EventStorage) + event_storage._size = 10 + event_storage._queue_size = 1000 + event_storage._events = queue.Queue() + + recorder = StandardRecorderAsync(None, None, impmanager, mocker.Mock(spec=EventStorage), mocker.Mock(spec=ImpressionStorage), telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactoryAsync(mocker.Mock(), { @@ -3620,7 +3629,7 @@ async def get(*_): 'segments': mocker.Mock(spec=SegmentStorage), 'rule_based_segments': mocker.Mock(spec=RuleBasedSegmentsStorage), 'impressions': mocker.Mock(spec=ImpressionStorage), - 'events': mocker.Mock(spec=EventStorage), + 'events': event_storage, }, mocker.Mock(), recorder, diff --git a/tests/client/test_manager.py b/tests/client/test_manager.py index 6491901e..e0f4d6cc 100644 --- a/tests/client/test_manager.py +++ b/tests/client/test_manager.py @@ -47,7 +47,7 @@ def test_evaluations_before_running_post_fork(self, mocker): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) - recorder = StandardRecorder(impmanager, mocker.Mock(), mocker.Mock(), telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorder(None, None, impmanager, mocker.Mock(), mocker.Mock(), telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactory(mocker.Mock(), {'splits': mocker.Mock(), @@ -123,7 +123,7 @@ async def test_evaluations_before_running_post_fork(self, mocker): impmanager = mocker.Mock(spec=ImpressionManager) telemetry_storage = InMemoryTelemetryStorageAsync() telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage) - recorder = StandardRecorderAsync(impmanager, mocker.Mock(), mocker.Mock(), telemetry_producer.get_telemetry_evaluation_producer(), + recorder = StandardRecorderAsync(None, None, impmanager, mocker.Mock(), mocker.Mock(), telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer()) factory = SplitFactory(mocker.Mock(), {'splits': mocker.Mock(), diff --git a/tests/integration/test_client_e2e.py b/tests/integration/test_client_e2e.py index cab26578..8fd96a12 100644 --- a/tests/integration/test_client_e2e.py +++ b/tests/integration/test_client_e2e.py @@ -565,11 +565,11 @@ def setup_method(self): 'splits': split_storage, 'segments': segment_storage, 'rule_based_segments': rb_segment_storage, - 'impressions': InMemoryImpressionStorage(5000, telemetry_runtime_producer), - 'events': InMemoryEventStorage(5000, telemetry_runtime_producer), + 'impressions': InMemoryImpressionStorage(50000, telemetry_runtime_producer), + 'events': InMemoryEventStorage(50000, telemetry_runtime_producer), } impmanager = ImpressionsManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener - recorder = StandardRecorder(impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, imp_counter=ImpressionsCounter()) + recorder = StandardRecorder(None, None, impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, imp_counter=ImpressionsCounter()) events_manager = EventsManager(EventsManagerConfig(), EventsDelivery()) internal_events_task = EventsTask(events_manager.notify_internal_event, events_queue) @@ -733,11 +733,11 @@ def setup_method(self): 'splits': split_storage, 'segments': segment_storage, 'rule_based_segments': rb_segment_storage, - 'impressions': InMemoryImpressionStorage(5000, telemetry_runtime_producer), - 'events': InMemoryEventStorage(5000, telemetry_runtime_producer), + 'impressions': InMemoryImpressionStorage(50000, telemetry_runtime_producer), + 'events': InMemoryEventStorage(50000, telemetry_runtime_producer), } impmanager = ImpressionsManager(StrategyOptimizedMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener - recorder = StandardRecorder(impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, imp_counter=ImpressionsCounter()) + recorder = StandardRecorder(None, None, impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, imp_counter=ImpressionsCounter()) events_manager = EventsManager(EventsManagerConfig(), EventsDelivery()) internal_events_task = EventsTask(events_manager.notify_internal_event, events_queue) self.factory = SplitFactory('some_api_key', @@ -1478,7 +1478,7 @@ def setup_method(self): } impmanager = ImpressionsManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener - recorder = StandardRecorder(impmanager, storages['events'], + recorder = StandardRecorder(None, None, impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, imp_counter=ImpressionsCounter()) events_manager = EventsManager(EventsManagerConfig(), EventsDelivery()) @@ -1678,7 +1678,7 @@ def setup_method(self): } impmanager = ImpressionsManager(StrategyOptimizedMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener - recorder = StandardRecorder(impmanager, storages['events'], + recorder = StandardRecorder(None, None, impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, imp_counter=ImpressionsCounter()) events_manager = EventsManager(EventsManagerConfig(), EventsDelivery()) @@ -1859,7 +1859,7 @@ def setup_method(self): imp_strategy, none_strategy = set_classes('PLUGGABLE', ImpressionsMode.NONE, self.pluggable_storage_adapter, imp_counter, unique_keys_tracker) impmanager = ImpressionsManager(imp_strategy, none_strategy, telemetry_runtime_producer) # no listener - recorder = StandardRecorder(impmanager, storages['events'], + recorder = StandardRecorder(None, None, impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, unique_keys_tracker=unique_keys_tracker, imp_counter=imp_counter) synchronizers = HarnessSynchronizers(None, None, None, None, @@ -1925,7 +1925,7 @@ def setup_method(self): def test_get_treatment(self): """Test client.get_treatment().""" _get_treatment(self.factory) - assert self.pluggable_storage_adapter._keys['SPLITIO.impressions'] == [] + assert self.pluggable_storage_adapter._keys.get('SPLITIO.impressions') == None def test_get_treatments(self): """Test client.get_treatments().""" @@ -1942,7 +1942,7 @@ def test_get_treatments(self): assert result['invalid_feature'] == 'control' assert result['sample_feature'] == 'off' - assert self.pluggable_storage_adapter._keys['SPLITIO.impressions'] == [] + assert self.pluggable_storage_adapter._keys.get('SPLITIO.impressions') == None def test_get_treatments_with_config(self): """Test client.get_treatments_with_config().""" @@ -1958,12 +1958,12 @@ def test_get_treatments_with_config(self): assert result['killed_feature'] == ('defTreatment', '{"size":15,"defTreatment":true}') assert result['invalid_feature'] == ('control', None) assert result['sample_feature'] == ('off', None) - assert self.pluggable_storage_adapter._keys['SPLITIO.impressions'] == [] + assert self.pluggable_storage_adapter._keys.get('SPLITIO.impressions') == None def test_get_treatments_by_flag_set(self): """Test client.get_treatments_by_flag_set().""" _get_treatments_by_flag_set(self.factory) - assert self.pluggable_storage_adapter._keys['SPLITIO.impressions'] == [] + assert self.pluggable_storage_adapter._keys.get('SPLITIO.impressions') == None def test_get_treatments_by_flag_sets(self): """Test client.get_treatments_by_flag_sets().""" @@ -1974,12 +1974,12 @@ def test_get_treatments_by_flag_sets(self): 'whitelist_feature': 'off', 'all_feature': 'on' } - assert self.pluggable_storage_adapter._keys['SPLITIO.impressions'] == [] + assert self.pluggable_storage_adapter._keys.get('SPLITIO.impressions') == None def test_get_treatments_with_config_by_flag_set(self): """Test client.get_treatments_with_config_by_flag_set().""" _get_treatments_with_config_by_flag_set(self.factory) - assert self.pluggable_storage_adapter._keys['SPLITIO.impressions'] == [] + assert self.pluggable_storage_adapter._keys.get('SPLITIO.impressions') == None def test_get_treatments_with_config_by_flag_sets(self): """Test client.get_treatments_with_config_by_flag_sets().""" @@ -1990,7 +1990,7 @@ def test_get_treatments_with_config_by_flag_sets(self): 'whitelist_feature': ('off', None), 'all_feature': ('on', None) } - assert self.pluggable_storage_adapter._keys['SPLITIO.impressions'] == [] + assert self.pluggable_storage_adapter._keys.get('SPLITIO.impressions') == None def test_track(self): """Test client.track().""" @@ -2033,11 +2033,11 @@ def test_optimized(self): 'splits': split_storage, 'segments': segment_storage, 'rule_based_segments': InMemoryRuleBasedSegmentStorage(events_queue), - 'impressions': InMemoryImpressionStorage(5000, telemetry_runtime_producer), - 'events': InMemoryEventStorage(5000, telemetry_runtime_producer), + 'impressions': InMemoryImpressionStorage(50000, telemetry_runtime_producer), + 'events': InMemoryEventStorage(50000, telemetry_runtime_producer), } impmanager = ImpressionsManager(StrategyOptimizedMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener - recorder = StandardRecorder(impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, None, UniqueKeysTracker(), ImpressionsCounter()) + recorder = StandardRecorder(None, None, impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, None, UniqueKeysTracker(), ImpressionsCounter()) events_manager = EventsManager(EventsManagerConfig(), EventsDelivery()) # Since we are passing None as SDK_Ready event, the factory will use the Redis telemetry call, using try catch to ignore the exception. try: @@ -2095,11 +2095,11 @@ def test_debug(self): 'splits': split_storage, 'segments': segment_storage, 'rule_based_segments': InMemoryRuleBasedSegmentStorage(events_queue), - 'impressions': InMemoryImpressionStorage(5000, telemetry_runtime_producer), - 'events': InMemoryEventStorage(5000, telemetry_runtime_producer), + 'impressions': InMemoryImpressionStorage(50000, telemetry_runtime_producer), + 'events': InMemoryEventStorage(50000, telemetry_runtime_producer), } impmanager = ImpressionsManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener - recorder = StandardRecorder(impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, None, UniqueKeysTracker(), ImpressionsCounter()) + recorder = StandardRecorder(None, None, impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, None, UniqueKeysTracker(), ImpressionsCounter()) events_manager = EventsManager(EventsManagerConfig(), EventsDelivery()) internal_events_task = EventsTask(events_manager.notify_internal_event, events_queue) # Since we are passing None as SDK_Ready event, the factory will use the Redis telemetry call, using try catch to ignore the exception. @@ -2159,11 +2159,11 @@ def test_none(self): 'splits': split_storage, 'segments': segment_storage, 'rule_based_segments': InMemoryRuleBasedSegmentStorage(events_queue), - 'impressions': InMemoryImpressionStorage(5000, telemetry_runtime_producer), - 'events': InMemoryEventStorage(5000, telemetry_runtime_producer), + 'impressions': InMemoryImpressionStorage(50000, telemetry_runtime_producer), + 'events': InMemoryEventStorage(50000, telemetry_runtime_producer), } impmanager = ImpressionsManager(StrategyNoneMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener - recorder = StandardRecorder(impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, None, UniqueKeysTracker(), ImpressionsCounter()) + recorder = StandardRecorder(None, None, impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, None, UniqueKeysTracker(), ImpressionsCounter()) events_manager = EventsManager(EventsManagerConfig(), EventsDelivery()) internal_events_task = EventsTask(events_manager.notify_internal_event, events_queue) # Since we are passing None as SDK_Ready event, the factory will use the Redis telemetry call, using try catch to ignore the exception. @@ -2477,11 +2477,11 @@ def test_sdk_ready(self): 'splits': split_storage, 'segments': segment_storage, 'rule_based_segments': rb_segment_storage, - 'impressions': InMemoryImpressionStorage(5000, telemetry_runtime_producer), - 'events': InMemoryEventStorage(5000, telemetry_runtime_producer), + 'impressions': InMemoryImpressionStorage(50000, telemetry_runtime_producer), + 'events': InMemoryEventStorage(50000, telemetry_runtime_producer), } impmanager = ImpressionsManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener - recorder = StandardRecorder(impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, imp_counter=ImpressionsCounter()) + recorder = StandardRecorder(None, None, impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, imp_counter=ImpressionsCounter()) events_manager = EventsManager(EventsManagerConfig(), EventsDelivery()) internal_events_task = EventsTask(events_manager.notify_internal_event, events_queue) @@ -2547,11 +2547,11 @@ def test_sdk_ready_fire_later(self): 'splits': split_storage, 'segments': segment_storage, 'rule_based_segments': rb_segment_storage, - 'impressions': InMemoryImpressionStorage(5000, telemetry_runtime_producer), - 'events': InMemoryEventStorage(5000, telemetry_runtime_producer), + 'impressions': InMemoryImpressionStorage(50000, telemetry_runtime_producer), + 'events': InMemoryEventStorage(50000, telemetry_runtime_producer), } impmanager = ImpressionsManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener - recorder = StandardRecorder(impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, imp_counter=ImpressionsCounter()) + recorder = StandardRecorder(None, None, impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, imp_counter=ImpressionsCounter()) events_manager = EventsManager(EventsManagerConfig(), EventsDelivery()) internal_events_task = EventsTask(events_manager.notify_internal_event, events_queue) @@ -2635,7 +2635,7 @@ async def test_sdk_ready(self): 'events': InMemoryEventStorageAsync(5000, telemetry_runtime_producer), } impmanager = ImpressionsManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener - recorder = StandardRecorderAsync(impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, imp_counter=ImpressionsCounter()) + recorder = StandardRecorderAsync(None, None, impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, imp_counter=ImpressionsCounter()) events_manager = EventsManagerAsync(EventsManagerConfig(), EventsDelivery()) internal_events_task = EventsTaskAsync(events_manager.notify_internal_event, events_queue) @@ -2705,7 +2705,7 @@ async def test_sdk_ready_fire_later(self): 'events': InMemoryEventStorageAsync(5000, telemetry_runtime_producer), } impmanager = ImpressionsManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener - recorder = StandardRecorderAsync(impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, imp_counter=ImpressionsCounter()) + recorder = StandardRecorderAsync(None, None, impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, imp_counter=ImpressionsCounter()) events_manager = EventsManagerAsync(EventsManagerConfig(), EventsDelivery()) internal_events_task = EventsTaskAsync(events_manager.notify_internal_event, events_queue) @@ -2784,7 +2784,7 @@ async def _setup_method(self): 'events': InMemoryEventStorageAsync(5000, telemetry_runtime_producer), } impmanager = ImpressionsManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener - recorder = StandardRecorderAsync(impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, imp_counter=ImpressionsCounter()) + recorder = StandardRecorderAsync(None, None, impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, imp_counter=ImpressionsCounter()) # Since we are passing None as SDK_Ready event, the factory will use the Redis telemetry call, using try catch to ignore the exception. try: self.factory = SplitFactoryAsync('some_api_key', @@ -2960,7 +2960,7 @@ async def _setup_method(self): 'events': InMemoryEventStorageAsync(5000, telemetry_runtime_producer), } impmanager = ImpressionsManager(StrategyOptimizedMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener - recorder = StandardRecorderAsync(impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, + recorder = StandardRecorderAsync(None, None, impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, imp_counter = ImpressionsCounter()) # Since we are passing None as SDK_Ready event, the factory will use the Redis telemetry call, using try catch to ignore the exception. try: @@ -3802,7 +3802,7 @@ async def _setup_method(self): } impmanager = ImpressionsManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener - recorder = StandardRecorderAsync(impmanager, storages['events'], + recorder = StandardRecorderAsync(None, None, impmanager, storages['events'], storages['impressions'], telemetry_producer.get_telemetry_evaluation_producer(), telemetry_runtime_producer, imp_counter=ImpressionsCounter()) @@ -4036,7 +4036,7 @@ async def _setup_method(self): } impmanager = ImpressionsManager(StrategyOptimizedMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener - recorder = StandardRecorderAsync(impmanager, storages['events'], + recorder = StandardRecorderAsync(None, None, impmanager, storages['events'], storages['impressions'], telemetry_producer.get_telemetry_evaluation_producer(), telemetry_runtime_producer, @@ -4068,6 +4068,7 @@ async def _setup_method(self): for flag_set in split.get('sets'): await self.pluggable_storage_adapter.push_items(split_storage._flag_set_prefix.format(flag_set=flag_set), split['name']) await self.pluggable_storage_adapter.set(split_storage._feature_flag_till_prefix, data['ff']['t']) + await self.pluggable_storage_adapter.set('SPLITIO.trafficType.user', 1) for rbs in data['rbs']['d']: await self.pluggable_storage_adapter.set(rb_segment_storage._prefix.format(segment_name=rbs['name']), rbs) @@ -4111,7 +4112,7 @@ async def test_get_treatments(self): assert result['killed_feature'] == 'defTreatment' assert result['invalid_feature'] == 'control' assert result['sample_feature'] == 'off' - assert len(self.pluggable_storage_adapter._keys['SPLITIO.impressions']) == 0 + assert self.pluggable_storage_adapter._keys.get('SPLITIO.impressions') == None await self.factory.destroy() await self._teardown_method() @@ -4262,7 +4263,7 @@ async def _setup_method(self): imp_strategy, none_strategy = set_classes_async('PLUGGABLE', ImpressionsMode.NONE, self.pluggable_storage_adapter, imp_counter, unique_keys_tracker) impmanager = ImpressionsManager(imp_strategy, none_strategy, telemetry_runtime_producer) # no listener - recorder = StandardRecorderAsync(impmanager, storages['events'], + recorder = StandardRecorderAsync(None, None, impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, unique_keys_tracker=unique_keys_tracker, imp_counter=imp_counter) synchronizers = HarnessSynchronizers(None, None, None, None, @@ -4326,7 +4327,7 @@ async def test_get_treatment(self): """Test client.get_treatment().""" await self.setup_task await _get_treatment_async(self.factory) - assert self.pluggable_storage_adapter._keys['SPLITIO.impressions'] == [] + assert self.pluggable_storage_adapter._keys.get('SPLITIO.impressions') == None await self.factory.destroy() await self._teardown_method() @@ -4347,7 +4348,7 @@ async def test_get_treatments(self): assert result['killed_feature'] == 'defTreatment' assert result['invalid_feature'] == 'control' assert result['sample_feature'] == 'off' - assert self.pluggable_storage_adapter._keys['SPLITIO.impressions'] == [] + assert self.pluggable_storage_adapter._keys.get('SPLITIO.impressions') == None await self.factory.destroy() await self._teardown_method() @@ -4368,7 +4369,7 @@ async def test_get_treatments_with_config(self): assert result['killed_feature'] == ('defTreatment', '{"size":15,"defTreatment":true}') assert result['invalid_feature'] == ('control', None) assert result['sample_feature'] == ('off', None) - assert self.pluggable_storage_adapter._keys['SPLITIO.impressions'] == [] + assert self.pluggable_storage_adapter._keys.get('SPLITIO.impressions') == None await self.factory.destroy() await self._teardown_method() @@ -4377,7 +4378,7 @@ async def test_get_treatments_by_flag_set(self): """Test client.get_treatments_by_flag_set().""" await self.setup_task await _get_treatments_by_flag_set_async(self.factory) - assert self.pluggable_storage_adapter._keys['SPLITIO.impressions'] == [] + assert self.pluggable_storage_adapter._keys.get('SPLITIO.impressions') == None await self.factory.destroy() await self._teardown_method() @@ -4393,7 +4394,7 @@ async def test_get_treatments_by_flag_sets(self): 'whitelist_feature': 'off', 'all_feature': 'on' } - assert self.pluggable_storage_adapter._keys['SPLITIO.impressions'] == [] + assert self.pluggable_storage_adapter._keys.get('SPLITIO.impressions') == None await self.factory.destroy() await self._teardown_method() @@ -4402,7 +4403,7 @@ async def test_get_treatments_with_config_by_flag_set(self): """Test client.get_treatments_with_config_by_flag_set().""" await self.setup_task await _get_treatments_with_config_by_flag_set_async(self.factory) - assert self.pluggable_storage_adapter._keys['SPLITIO.impressions'] == [] + assert self.pluggable_storage_adapter._keys.get('SPLITIO.impressions') == None await self.factory.destroy() await self._teardown_method() @@ -4418,7 +4419,7 @@ async def test_get_treatments_with_config_by_flag_sets(self): 'whitelist_feature': ('off', None), 'all_feature': ('on', None) } - assert self.pluggable_storage_adapter._keys['SPLITIO.impressions'] == [] + assert self.pluggable_storage_adapter._keys.get('SPLITIO.impressions') == None await self.factory.destroy() await self._teardown_method() @@ -4499,7 +4500,7 @@ async def test_optimized(self): 'events': InMemoryEventStorageAsync(5000, telemetry_runtime_producer), } impmanager = ImpressionsManager(StrategyOptimizedMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener - recorder = StandardRecorderAsync(impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, None, UniqueKeysTrackerAsync(), ImpressionsCounter()) + recorder = StandardRecorderAsync(None, None, impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, None, UniqueKeysTrackerAsync(), ImpressionsCounter()) # Since we are passing None as SDK_Ready event, the factory will use the Redis telemetry call, using try catch to ignore the exception. try: factory = SplitFactoryAsync('some_api_key', @@ -4565,7 +4566,7 @@ async def test_debug(self): 'events': InMemoryEventStorageAsync(5000, telemetry_runtime_producer), } impmanager = ImpressionsManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener - recorder = StandardRecorderAsync(impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, None, UniqueKeysTrackerAsync(), ImpressionsCounter()) + recorder = StandardRecorderAsync(None, None, impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, None, UniqueKeysTrackerAsync(), ImpressionsCounter()) # Since we are passing None as SDK_Ready event, the factory will use the Redis telemetry call, using try catch to ignore the exception. try: factory = SplitFactoryAsync('some_api_key', @@ -4631,7 +4632,7 @@ async def test_none(self): 'events': InMemoryEventStorageAsync(5000, telemetry_runtime_producer), } impmanager = ImpressionsManager(StrategyNoneMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener - recorder = StandardRecorderAsync(impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, None, UniqueKeysTrackerAsync(), ImpressionsCounter()) + recorder = StandardRecorderAsync(None, None, impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, None, UniqueKeysTrackerAsync(), ImpressionsCounter()) # Since we are passing None as SDK_Ready event, the factory will use the Redis telemetry call, using try catch to ignore the exception. try: factory = SplitFactoryAsync('some_api_key', diff --git a/tests/integration/test_streaming_e2e.py b/tests/integration/test_streaming_e2e.py index 792616f6..1c73d0b8 100644 --- a/tests/integration/test_streaming_e2e.py +++ b/tests/integration/test_streaming_e2e.py @@ -176,12 +176,6 @@ def test_happiness(self): assert req.path == '/api/splitChanges?s=1.3&since=-1&rbSince=-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?s=1.3&since=1&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Auth req = split_backend_requests.get() assert req.method == 'GET' @@ -390,12 +384,6 @@ def test_occupancy_flicker(self): assert req.path == '/api/splitChanges?s=1.3&since=-1&rbSince=-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?s=1.3&since=1&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Auth req = split_backend_requests.get() assert req.method == 'GET' @@ -414,12 +402,6 @@ def test_occupancy_flicker(self): assert req.path == '/api/splitChanges?s=1.3&since=1&rbSince=-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?s=1.3&since=2&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Fetch after second notification req = split_backend_requests.get() assert req.method == 'GET' @@ -432,12 +414,6 @@ def test_occupancy_flicker(self): assert req.path == '/api/splitChanges?s=1.3&since=3&rbSince=-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?s=1.3&since=3&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Iteration until since == till req = split_backend_requests.get() assert req.method == 'GET' @@ -450,12 +426,6 @@ def test_occupancy_flicker(self): assert req.path == '/api/splitChanges?s=1.3&since=4&rbSince=-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?s=1.3&since=5&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Cleanup destroy_event = threading.Event() factory.destroy(destroy_event) @@ -568,12 +538,6 @@ def test_start_without_occupancy(self): assert req.path == '/api/splitChanges?s=1.3&since=-1&rbSince=-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?s=1.3&since=1&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Auth req = split_backend_requests.get() assert req.method == 'GET' @@ -598,12 +562,6 @@ def test_start_without_occupancy(self): assert req.path == '/api/splitChanges?s=1.3&since=1&rbSince=-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?s=1.3&since=2&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Cleanup destroy_event = threading.Event() factory.destroy(destroy_event) @@ -760,12 +718,6 @@ def test_streaming_status_changes(self): assert req.path == '/api/splitChanges?s=1.3&since=-1&rbSince=-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?s=1.3&since=1&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Auth req = split_backend_requests.get() assert req.method == 'GET' @@ -784,24 +736,12 @@ def test_streaming_status_changes(self): assert req.path == '/api/splitChanges?s=1.3&since=1&rbSince=-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?s=1.3&since=2&rbSince=-1' - 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?s=1.3&since=2&rbSince=-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?s=1.3&since=3&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Fetch after notification req = split_backend_requests.get() assert req.method == 'GET' @@ -820,12 +760,6 @@ def test_streaming_status_changes(self): assert req.path == '/api/splitChanges?s=1.3&since=4&rbSince=-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?s=1.3&since=5&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Cleanup destroy_event = threading.Event() factory.destroy(destroy_event) @@ -987,12 +921,6 @@ def test_server_closes_connection(self): assert req.path == '/api/splitChanges?s=1.3&since=-1&rbSince=-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?s=1.3&since=1&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Auth req = split_backend_requests.get() assert req.method == 'GET' @@ -1224,12 +1152,6 @@ def test_ably_errors_handling(self): assert req.path == '/api/splitChanges?s=1.3&since=-1&rbSince=-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?s=1.3&since=1&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Auth req = split_backend_requests.get() assert req.method == 'GET' @@ -1248,12 +1170,6 @@ def test_ably_errors_handling(self): assert req.path == '/api/splitChanges?s=1.3&since=1&rbSince=-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?s=1.3&since=2&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Auth again req = split_backend_requests.get() assert req.method == 'GET' @@ -1506,12 +1422,6 @@ async def test_happiness(self): assert req.path == '/api/splitChanges?s=1.3&since=-1&rbSince=-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?s=1.3&since=1&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Auth req = split_backend_requests.get() assert req.method == 'GET' @@ -1714,12 +1624,6 @@ async def test_occupancy_flicker(self): assert req.path == '/api/splitChanges?s=1.3&since=-1&rbSince=-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?s=1.3&since=1&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Auth req = split_backend_requests.get() assert req.method == 'GET' @@ -1738,12 +1642,6 @@ async def test_occupancy_flicker(self): assert req.path == '/api/splitChanges?s=1.3&since=1&rbSince=-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?s=1.3&since=2&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Fetch after second notification req = split_backend_requests.get() assert req.method == 'GET' @@ -1756,12 +1654,6 @@ async def test_occupancy_flicker(self): assert req.path == '/api/splitChanges?s=1.3&since=3&rbSince=-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?s=1.3&since=3&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Iteration until since == till req = split_backend_requests.get() assert req.method == 'GET' @@ -1774,12 +1666,6 @@ async def test_occupancy_flicker(self): assert req.path == '/api/splitChanges?s=1.3&since=4&rbSince=-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?s=1.3&since=5&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Cleanup await factory.destroy() sse_server.publish(sse_server.GRACEFUL_REQUEST_END) @@ -1892,12 +1778,6 @@ async def test_start_without_occupancy(self): assert req.path == '/api/splitChanges?s=1.3&since=-1&rbSince=-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?s=1.3&since=1&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Auth req = split_backend_requests.get() assert req.method == 'GET' @@ -1922,12 +1802,6 @@ async def test_start_without_occupancy(self): assert req.path == '/api/splitChanges?s=1.3&since=1&rbSince=-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?s=1.3&since=2&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Cleanup await factory.destroy() sse_server.publish(sse_server.GRACEFUL_REQUEST_END) @@ -2086,12 +1960,6 @@ async def test_streaming_status_changes(self): assert req.path == '/api/splitChanges?s=1.3&since=-1&rbSince=-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?s=1.3&since=1&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Auth req = split_backend_requests.get() assert req.method == 'GET' @@ -2110,48 +1978,24 @@ async def test_streaming_status_changes(self): assert req.path == '/api/splitChanges?s=1.3&since=1&rbSince=-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?s=1.3&since=2&rbSince=-1' - 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?s=1.3&since=2&rbSince=-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?s=1.3&since=3&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Fetch after notification req = split_backend_requests.get() assert req.method == 'GET' assert req.path == '/api/splitChanges?s=1.3&since=3&rbSince=-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?s=1.3&since=4&rbSince=-1' - 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?s=1.3&since=4&rbSince=-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?s=1.3&since=5&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Cleanup await factory.destroy() sse_server.publish(sse_server.GRACEFUL_REQUEST_END) @@ -2306,12 +2150,6 @@ async def test_server_closes_connection(self): assert req.path == '/api/splitChanges?s=1.3&since=-1&rbSince=-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?s=1.3&since=1&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Auth req = split_backend_requests.get() assert req.method == 'GET' @@ -2542,12 +2380,6 @@ async def test_ably_errors_handling(self): assert req.path == '/api/splitChanges?s=1.3&since=-1&rbSince=-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?s=1.3&since=1&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Auth req = split_backend_requests.get() assert req.method == 'GET' @@ -2566,12 +2398,6 @@ async def test_ably_errors_handling(self): assert req.path == '/api/splitChanges?s=1.3&since=1&rbSince=-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?s=1.3&since=2&rbSince=-1' - assert req.headers['authorization'] == 'Bearer some_apikey' - # Auth again req = split_backend_requests.get() assert req.method == 'GET' diff --git a/tests/sync/test_splits_synchronizer.py b/tests/sync/test_splits_synchronizer.py index 4edf8eb3..a8755e68 100644 --- a/tests/sync/test_splits_synchronizer.py +++ b/tests/sync/test_splits_synchronizer.py @@ -257,9 +257,6 @@ def get_changes(*args, **kwargs): assert api.fetch_definitions.mock_calls[0][1][0] == -1 assert api.fetch_definitions.mock_calls[0][1][2].cache_control_headers == True - assert api.fetch_definitions.mock_calls[1][1][0] == 123 - assert api.fetch_definitions.mock_calls[1][1][1] == 123 - assert api.fetch_definitions.mock_calls[1][1][2].cache_control_headers == True inserted_split = storage.update.mock_calls[0][1][0][0] assert isinstance(inserted_split, Split) @@ -381,8 +378,6 @@ def intersect(sets): assert api.fetch_definitions.mock_calls[0][1][0] == -1 assert api.fetch_definitions.mock_calls[0][1][2].cache_control_headers == True - assert api.fetch_definitions.mock_calls[1][1][0] == 123 - assert api.fetch_definitions.mock_calls[1][1][2].cache_control_headers == True split_synchronizer._backoff = Backoff(1, 0.1) split_synchronizer.synchronize_definitions(12345) @@ -620,7 +615,6 @@ async def get_changes(change_number, rbs_change_number, fetch_options): await split_synchronizer.synchronize_definitions() assert (-1, FetchOptions(True)._cache_control_headers) == (self.change_number_1, self.fetch_options_1._cache_control_headers) - assert (123, FetchOptions(True)._cache_control_headers) == (self.change_number_2, self.fetch_options_2._cache_control_headers) inserted_split = self.parsed_split[0] assert isinstance(inserted_split, Split) assert inserted_split.name == 'some_name' @@ -772,7 +766,6 @@ async def clear(): await split_synchronizer.synchronize_definitions() assert (-1, FetchOptions(True).cache_control_headers) == (self.change_number_1, self.fetch_options_1.cache_control_headers) - assert (123, FetchOptions(True).cache_control_headers) == (self.change_number_2, self.fetch_options_2.cache_control_headers) split_synchronizer._backoff = Backoff(1, 0.1) await split_synchronizer.synchronize_definitions(12345) diff --git a/tests/tasks/test_split_sync.py b/tests/tasks/test_split_sync.py index 99dcb771..b4d38ca3 100644 --- a/tests/tasks/test_split_sync.py +++ b/tests/tasks/test_split_sync.py @@ -108,8 +108,6 @@ def get_changes(*args, **kwargs): assert not task.is_running() assert api.fetch_definitions.mock_calls[0][1][0] == -1 assert api.fetch_definitions.mock_calls[0][1][2].cache_control_headers == True - assert api.fetch_definitions.mock_calls[1][1][0] == 123 - assert api.fetch_definitions.mock_calls[1][1][2].cache_control_headers == True inserted_split = storage.update.mock_calls[0][1][0][0] assert isinstance(inserted_split, Split)