Merge pull request #332 from splitio/pluggable-optimized-factory

Factory and test integration for pluggable optimized mode

Author: chillaq

splitio/client/config.py

@@ -111,10 +111,6 @@ def _sanitize_impressions_mode(storage_type, mode, refresh_rate=None):
                             'one of the following values: `debug`, `none` or `optimized`. '
                             ' Defaulting to `optimized` mode.')
 
-    if storage_type == 'pluggable' and mode != ImpressionsMode.DEBUG:
-        mode = ImpressionsMode.DEBUG
-        _LOGGER.warning('`pluggable` storageMode only support `debug` impressionMode, adjusting impressionsMode to `debug`. ')
-
     if mode == ImpressionsMode.DEBUG:
         refresh_rate = max(1, refresh_rate) if refresh_rate is not None else 60
     else:

splitio/client/factory.py

@@ -537,15 +537,30 @@ def _build_pluggable_factory(api_key, cfg):
 
     unique_keys_synchronizer, clear_filter_sync, unique_keys_task, \
     clear_filter_task, impressions_count_sync, impressions_count_task, \
-    imp_strategy = set_classes('PLUGGABLE', cfg['impressionsMode'], pluggable_adapter)
+    imp_strategy = set_classes('PLUGGABLE', cfg['impressionsMode'], pluggable_adapter, storage_prefix)
 
     imp_manager = ImpressionsManager(
         imp_strategy,
         telemetry_runtime_producer,
         _wrap_impression_listener(cfg['impressionListener'], sdk_metadata),
         )
 
-    synchronizer = PluggableSynchronizer()
+    synchronizers = SplitSynchronizers(None, None, None, None,
+        impressions_count_sync,
+        None,
+        unique_keys_synchronizer,
+        clear_filter_sync
+    )
+
+    tasks = SplitTasks(None, None, None, None,
+        impressions_count_task,
+        None,
+        unique_keys_task,
+        clear_filter_task
+    )
+
+    # Using same class as redis for consumer mode only
+    synchronizer = RedisSynchronizer(synchronizers, tasks)
     recorder = StandardRecorder(
         imp_manager,
         storages['events'],

splitio/engine/impressions/__init__.py

@@ -1,13 +1,13 @@
 from splitio.engine.impressions.impressions import ImpressionsMode
 from splitio.engine.impressions.manager import Counter as ImpressionsCounter
 from splitio.engine.impressions.strategies import StrategyNoneMode, StrategyDebugMode, StrategyOptimizedMode
-from splitio.engine.impressions.adapters import InMemorySenderAdapter, RedisSenderAdapter
+from splitio.engine.impressions.adapters import InMemorySenderAdapter, RedisSenderAdapter, PluggableSenderAdapter
 from splitio.tasks.unique_keys_sync import UniqueKeysSyncTask, ClearFilterSyncTask
 from splitio.sync.unique_keys import UniqueKeysSynchronizer, ClearFilterSynchronizer
 from splitio.sync.impression import ImpressionsCountSynchronizer
 from splitio.tasks.impressions_sync import ImpressionsCountSyncTask
 
-def set_classes(storage_mode, impressions_mode, api_adapter):
+def set_classes(storage_mode, impressions_mode, api_adapter, prefix=None):
     unique_keys_synchronizer = None
     clear_filter_sync = None
     unique_keys_task = None
@@ -16,6 +16,7 @@ def set_classes(storage_mode, impressions_mode, api_adapter):
     impressions_count_task = None
     sender_adapter = None
     if storage_mode == 'PLUGGABLE':
+        sender_adapter = PluggableSenderAdapter(api_adapter, prefix)
         api_telemetry_adapter = sender_adapter
         api_impressions_adapter = sender_adapter
     elif storage_mode == 'REDIS':

tests/client/test_config.py

@@ -39,15 +39,15 @@ def test_sanitize_imp_mode(self):
         assert rate == 200
 
         mode, rate = config._sanitize_impressions_mode('pluggable', 'ANYTHING', 200)
-        assert mode == ImpressionsMode.DEBUG
+        assert mode == ImpressionsMode.OPTIMIZED
         assert rate == 200
 
         mode, rate = config._sanitize_impressions_mode('pluggable', 'NONE', 200)
-        assert mode == ImpressionsMode.DEBUG
+        assert mode == ImpressionsMode.NONE
         assert rate == 200
 
         mode, rate = config._sanitize_impressions_mode('pluggable', 'OPTIMIZED', 200)
-        assert mode == ImpressionsMode.DEBUG
+        assert mode == ImpressionsMode.OPTIMIZED
         assert rate == 200
 
         mode, rate = config._sanitize_impressions_mode('memory', 43, -1)

tests/integration/test_client_e2e.py

@@ -1452,3 +1452,272 @@ def teardown_method(self):
 
         for key in keys_to_delete:
             self.pluggable_storage_adapter.delete(key)
+
+class PluggableOptimizedIntegrationTests(object):
+    """Pluggable storage-based integration tests."""
+
+    def setup_method(self):
+        """Prepare storages with test data."""
+        metadata = SdkMetadata('python-1.2.3', 'some_ip', 'some_name')
+        self.pluggable_storage_adapter = StorageMockAdapter()
+        split_storage = PluggableSplitStorage(self.pluggable_storage_adapter, 'myprefix')
+        segment_storage = PluggableSegmentStorage(self.pluggable_storage_adapter, 'myprefix')
+
+        telemetry_pluggable_storage = PluggableTelemetryStorage(self.pluggable_storage_adapter, metadata, 'myprefix')
+        telemetry_producer = TelemetryStorageProducer(telemetry_pluggable_storage)
+        telemetry_consumer = TelemetryStorageConsumer(telemetry_pluggable_storage)
+        telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
+
+        storages = {
+            'splits': split_storage,
+            'segments': segment_storage,
+            'impressions': PluggableImpressionsStorage(self.pluggable_storage_adapter, metadata, 'myprefix'),
+            'events': PluggableEventsStorage(self.pluggable_storage_adapter, metadata, 'myprefix'),
+            'telemetry': telemetry_pluggable_storage
+        }
+
+        impmanager = ImpressionsManager(StrategyOptimizedMode(ImpressionsCounter()), telemetry_runtime_producer) # no listener
+        recorder = StandardRecorder(impmanager, storages['events'],
+                                    storages['impressions'], storages['telemetry'])
+
+        self.factory = SplitFactory('some_api_key',
+                                    storages,
+                                    True,
+                                    recorder,
+                                    RedisManager(PluggableSynchronizer()),
+                                    sdk_ready_flag=None,
+                                    telemetry_producer=telemetry_producer,
+                                    telemetry_init_producer=telemetry_producer.get_telemetry_init_producer(),
+                                    )  # pylint:disable=attribute-defined-outside-init
+
+        # Adding data to storage
+        split_fn = os.path.join(os.path.dirname(__file__), 'files', 'splitChanges.json')
+        with open(split_fn, 'r') as flo:
+            data = json.loads(flo.read())
+        for split in data['splits']:
+            self.pluggable_storage_adapter.set(split_storage._prefix.format(split_name=split['name']), split)
+        self.pluggable_storage_adapter.set(split_storage._split_till_prefix, data['till'])
+
+        segment_fn = os.path.join(os.path.dirname(__file__), 'files', 'segmentEmployeesChanges.json')
+        with open(segment_fn, 'r') as flo:
+            data = json.loads(flo.read())
+        self.pluggable_storage_adapter.set(segment_storage._prefix.format(segment_name=data['name']), set(data['added']))
+        self.pluggable_storage_adapter.set(segment_storage._segment_till_prefix.format(segment_name=data['name']), data['till'])
+
+        segment_fn = os.path.join(os.path.dirname(__file__), 'files', 'segmentHumanBeignsChanges.json')
+        with open(segment_fn, 'r') as flo:
+            data = json.loads(flo.read())
+        self.pluggable_storage_adapter.set(segment_storage._prefix.format(segment_name=data['name']), set(data['added']))
+        self.pluggable_storage_adapter.set(segment_storage._segment_till_prefix.format(segment_name=data['name']), data['till'])
+
+    def _validate_last_events(self, client, *to_validate):
+        """Validate the last N impressions are present disregarding the order."""
+        event_storage = client._factory._get_storage('events')
+        events_raw = []
+        stored_events = self.pluggable_storage_adapter.pop_items(event_storage._events_queue_key)
+        if stored_events is not None:
+            events_raw = [json.loads(im) for im in stored_events]
+
+        as_tup_set = set(
+            (i['e']['key'], i['e']['trafficTypeName'], i['e']['eventTypeId'], i['e']['value'], str(i['e']['properties']))
+            for i in events_raw
+        )
+        assert as_tup_set == set(to_validate)
+
+    def _validate_last_impressions(self, client, *to_validate):
+        """Validate the last N impressions are present disregarding the order."""
+        imp_storage = client._factory._get_storage('impressions')
+        impressions_raw = []
+        stored_impressions = self.pluggable_storage_adapter.pop_items(imp_storage._impressions_queue_key)
+        if stored_impressions is not None:
+            impressions_raw = [json.loads(im) for im in stored_impressions]
+        as_tup_set = set(
+            (i['i']['f'], i['i']['k'], i['i']['t'])
+            for i in impressions_raw
+        )
+
+        assert as_tup_set == set(to_validate)
+
+    def test_get_treatment(self):
+        """Test client.get_treatment()."""
+        client = self.factory.client()
+
+        assert client.get_treatment('user1', 'sample_feature') == 'on'
+        self._validate_last_impressions(client, ('sample_feature', 'user1', 'on'))
+        client.get_treatment('user1', 'sample_feature')
+        client.get_treatment('user1', 'sample_feature')
+        client.get_treatment('user1', 'sample_feature')
+
+        # Only one impression was added, and popped when validating, the rest were ignored
+#        pytest.set_trace()
+        assert self.pluggable_storage_adapter._keys['myprefix.SPLITIO.impressions'] == []
+
+        assert client.get_treatment('invalidKey', 'sample_feature') == 'off'
+        self._validate_last_impressions(client, ('sample_feature', 'invalidKey', 'off'))
+
+        assert client.get_treatment('invalidKey', 'invalid_feature') == 'control'
+        self._validate_last_impressions(client)  # No impressions should be present
+
+        # testing a killed feature. No matter what the key, must return default treatment
+        assert client.get_treatment('invalidKey', 'killed_feature') == 'defTreatment'
+        self._validate_last_impressions(client, ('killed_feature', 'invalidKey', 'defTreatment'))
+
+        # testing ALL matcher
+        assert client.get_treatment('invalidKey', 'all_feature') == 'on'
+        self._validate_last_impressions(client, ('all_feature', 'invalidKey', 'on'))
+
+        # testing WHITELIST matcher
+        assert client.get_treatment('whitelisted_user', 'whitelist_feature') == 'on'
+        self._validate_last_impressions(client, ('whitelist_feature', 'whitelisted_user', 'on'))
+        assert client.get_treatment('unwhitelisted_user', 'whitelist_feature') == 'off'
+        self._validate_last_impressions(client, ('whitelist_feature', 'unwhitelisted_user', 'off'))
+
+        #  testing INVALID matcher
+        assert client.get_treatment('some_user_key', 'invalid_matcher_feature') == 'control'
+        self._validate_last_impressions(client)  # No impressions should be present
+
+        #  testing Dependency matcher
+        assert client.get_treatment('somekey', 'dependency_test') == 'off'
+        self._validate_last_impressions(client, ('dependency_test', 'somekey', 'off'))
+
+        #  testing boolean matcher
+        assert client.get_treatment('True', 'boolean_test') == 'on'
+        self._validate_last_impressions(client, ('boolean_test', 'True', 'on'))
+
+        #  testing regex matcher
+        assert client.get_treatment('abc4', 'regex_test') == 'on'
+        self._validate_last_impressions(client, ('regex_test', 'abc4', 'on'))
+
+    def test_get_treatments(self):
+        """Test client.get_treatments()."""
+        client = self.factory.client()
+
+        result = client.get_treatments('user1', ['sample_feature'])
+        assert len(result) == 1
+        assert result['sample_feature'] == 'on'
+        self._validate_last_impressions(client, ('sample_feature', 'user1', 'on'))
+
+        result = client.get_treatments('invalidKey', ['sample_feature'])
+        assert len(result) == 1
+        assert result['sample_feature'] == 'off'
+        self._validate_last_impressions(client, ('sample_feature', 'invalidKey', 'off'))
+
+        result = client.get_treatments('invalidKey', ['invalid_feature'])
+        assert len(result) == 1
+        assert result['invalid_feature'] == 'control'
+        self._validate_last_impressions(client)
+
+        # testing a killed feature. No matter what the key, must return default treatment
+        result = client.get_treatments('invalidKey', ['killed_feature'])
+        assert len(result) == 1
+        assert result['killed_feature'] == 'defTreatment'
+        self._validate_last_impressions(client, ('killed_feature', 'invalidKey', 'defTreatment'))
+
+        # testing ALL matcher
+        result = client.get_treatments('invalidKey', ['all_feature'])
+        assert len(result) == 1
+        assert result['all_feature'] == 'on'
+        self._validate_last_impressions(client, ('all_feature', 'invalidKey', 'on'))
+
+        # testing multiple splitNames
+        result = client.get_treatments('invalidKey', [
+            'all_feature',
+            'killed_feature',
+            'invalid_feature',
+            'sample_feature'
+        ])
+        assert len(result) == 4
+        assert result['all_feature'] == 'on'
+        assert result['killed_feature'] == 'defTreatment'
+        assert result['invalid_feature'] == 'control'
+        assert result['sample_feature'] == 'off'
+        assert self.pluggable_storage_adapter._keys['myprefix.SPLITIO.impressions'] == []
+
+    def test_get_treatments_with_config(self):
+        """Test client.get_treatments_with_config()."""
+        client = self.factory.client()
+
+        result = client.get_treatments_with_config('user1', ['sample_feature'])
+        assert len(result) == 1
+        assert result['sample_feature'] == ('on', '{"size":15,"test":20}')
+        self._validate_last_impressions(client, ('sample_feature', 'user1', 'on'))
+
+        result = client.get_treatments_with_config('invalidKey', ['sample_feature'])
+        assert len(result) == 1
+        assert result['sample_feature'] == ('off', None)
+        self._validate_last_impressions(client, ('sample_feature', 'invalidKey', 'off'))
+
+        result = client.get_treatments_with_config('invalidKey', ['invalid_feature'])
+        assert len(result) == 1
+        assert result['invalid_feature'] == ('control', None)
+        self._validate_last_impressions(client)
+
+        # testing a killed feature. No matter what the key, must return default treatment
+        result = client.get_treatments_with_config('invalidKey', ['killed_feature'])
+        assert len(result) == 1
+        assert result['killed_feature'] == ('defTreatment', '{"size":15,"defTreatment":true}')
+        self._validate_last_impressions(client, ('killed_feature', 'invalidKey', 'defTreatment'))
+
+        # testing ALL matcher
+        result = client.get_treatments_with_config('invalidKey', ['all_feature'])
+        assert len(result) == 1
+        assert result['all_feature'] == ('on', None)
+        self._validate_last_impressions(client, ('all_feature', 'invalidKey', 'on'))
+
+        # testing multiple splitNames
+        result = client.get_treatments_with_config('invalidKey', [
+            'all_feature',
+            'killed_feature',
+            'invalid_feature',
+            'sample_feature'
+        ])
+        assert len(result) == 4
+
+        assert result['all_feature'] == ('on', None)
+        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['myprefix.SPLITIO.impressions'] == []
+
+    def test_manager_methods(self):
+        """Test manager.split/splits."""
+        manager = self.factory.manager()
+        result = manager.split('all_feature')
+        assert result.name == 'all_feature'
+        assert result.traffic_type is None
+        assert result.killed is False
+        assert len(result.treatments) == 2
+        assert result.change_number == 123
+        assert result.configs == {}
+
+        result = manager.split('killed_feature')
+        assert result.name == 'killed_feature'
+        assert result.traffic_type is None
+        assert result.killed is True
+        assert len(result.treatments) == 2
+        assert result.change_number == 123
+        assert result.configs['defTreatment'] == '{"size":15,"defTreatment":true}'
+        assert result.configs['off'] == '{"size":15,"test":20}'
+
+        result = manager.split('sample_feature')
+        assert result.name == 'sample_feature'
+        assert result.traffic_type is None
+        assert result.killed is False
+        assert len(result.treatments) == 2
+        assert result.change_number == 123
+        assert result.configs['on'] == '{"size":15,"test":20}'
+
+        assert len(manager.split_names()) == 7
+        assert len(manager.splits()) == 7
+
+    def test_track(self):
+        """Test client.track()."""
+        client = self.factory.client()
+        assert(client.track('user1', 'user', 'conversion', 1, {"prop1": "value1"}))
+        assert(not client.track(None, 'user', 'conversion'))
+        assert(not client.track('user1', None, 'conversion'))
+        assert(not client.track('user1', 'user', None))
+        self._validate_last_events(
+            client,
+            ('user1', 'user', 'conversion', 1, "{'prop1': 'value1'}")
+        )