Updated integration tests

Author: chillaq

splitio/api/impressions.py

@@ -37,7 +37,8 @@ def _build_bulk(impressions):
                         'c': impression.change_number,
                         'r': impression.label,
                         'b': impression.bucketing_key,
-                        'pt': impression.previous_time
+                        'pt': impression.previous_time,
+                        'properties': impression.properties
                     }
                     for impression in imps
                 ]

splitio/client/client.py

@@ -137,7 +137,7 @@ def _build_impression(self, key, bucketing, feature, result, properties=None):
                 bucketing_key=bucketing,
                 time=utctime_ms(),
                 previous_time=None,
-                properties=json.dumps(properties)),
+                properties=json.dumps(properties) if properties is not None else None),
                 disabled=result['impressions_disabled'])
 
     def _build_impressions(self, key, bucketing, results, properties=None):

splitio/storage/pluggable.py

@@ -1231,6 +1231,7 @@ def _wrap_impressions(self, impressions):
                         'r': impression.label,
                         'c': impression.change_number,
                         'm': impression.time,
+                        'properties': impression.properties
                     }
                 }
             bulk_impressions.append(json.dumps(to_store))

splitio/storage/redis.py

@@ -1100,6 +1100,7 @@ def _wrap_impressions(self, impressions):
                         'r': impression.label,
                         'c': impression.change_number,
                         'm': impression.time,
+                        'properties': impression.properties
                     }
                 }
             bulk_impressions.append(json.dumps(to_store))

tests/api/test_impressions_api.py

@@ -14,20 +14,20 @@
 from splitio.storage.inmemmory import InMemoryTelemetryStorage, InMemoryTelemetryStorageAsync
 
 impressions_mock = [
-    Impression('k1', 'f1', 'on', 'l1', 123456, 'b1', 321654, {}),
-    Impression('k2', 'f2', 'off', 'l1', 123456, 'b1', 321654, {}),
-    Impression('k3', 'f1', 'on', 'l1', 123456, 'b1', 321654, {})
+    Impression('k1', 'f1', 'on', 'l1', 123456, 'b1', 321654, None, {'prop': 'val'}),
+    Impression('k2', 'f2', 'off', 'l1', 123456, 'b1', 321654, None, None),
+    Impression('k3', 'f1', 'on', 'l1', 123456, 'b1', 321654, None, None)
 ]
 expectedImpressions = [{
     'f': 'f1',
     'i': [
-        {'k': 'k1', 'b': 'b1', 't': 'on', 'r': 'l1', 'm': 321654, 'c': 123456, 'pt': None},
-        {'k': 'k3', 'b': 'b1', 't': 'on', 'r': 'l1', 'm': 321654, 'c': 123456, 'pt': None},
+        {'k': 'k1', 'b': 'b1', 't': 'on', 'r': 'l1', 'm': 321654, 'c': 123456, 'pt': None, 'properties': {"prop": "val"}},
+        {'k': 'k3', 'b': 'b1', 't': 'on', 'r': 'l1', 'm': 321654, 'c': 123456, 'pt': None, 'properties': None},
     ],
 }, {
     'f': 'f2',
     'i': [
-        {'k': 'k2', 'b': 'b1', 't': 'off', 'r': 'l1', 'm': 321654, 'c': 123456, 'pt': None},
+        {'k': 'k2', 'b': 'b1', 't': 'off', 'r': 'l1', 'm': 321654, 'c': 123456, 'pt': None, 'properties': None},
     ]
 }]
 

tests/client/test_client.py

@@ -766,14 +766,14 @@ def test_track(self, mocker):
         _logger.reset_mock()
         assert client.track("some_key", "traffic_type", "event_type", 1, []) is False
         assert _logger.error.mock_calls == [
-            mocker.call("track: properties must be of type dictionary.")
+            mocker.call("%s: properties must be of type dictionary.", "track")
         ]
 
         # Test track with invalid properties
         _logger.reset_mock()
         assert client.track("some_key", "traffic_type", "event_type", 1, True) is False
         assert _logger.error.mock_calls == [
-            mocker.call("track: properties must be of type dictionary.")
+            mocker.call("%s: properties must be of type dictionary.", "track")
         ]
 
         # Test track with properties
@@ -788,7 +788,7 @@ def test_track(self, mocker):
         _logger.reset_mock()
         assert client.track("some_key", "traffic_type", "event_type", 1, props1) is True
         assert _logger.warning.mock_calls == [
-            mocker.call("Property %s is of invalid type. Setting value to None", [])
+            mocker.call("%s: Property %s is of invalid type. Setting value to None", "track", [])
         ]
 
         # Test track with more than 300 properties
@@ -798,7 +798,7 @@ def test_track(self, mocker):
         _logger.reset_mock()
         assert client.track("some_key", "traffic_type", "event_type", 1, props2) is True
         assert _logger.warning.mock_calls == [
-            mocker.call("Event has more than 300 properties. Some of them will be trimmed when processed")
+            mocker.call("%s: Event has more than 300 properties. Some of them will be trimmed when processed", "track")
         ]
 
         # Test track with properties higher than 32kb
@@ -808,7 +808,7 @@ def test_track(self, mocker):
             props3["prop" + str(i)] = "a" * 300
         assert client.track("some_key", "traffic_type", "event_type", 1, props3) is False
         assert _logger.error.mock_calls == [
-            mocker.call("The maximum size allowed for the properties is 32768 bytes. Current one is 32952 bytes. Event not queued")
+            mocker.call("%s: The maximum size allowed for the properties is 32768 bytes. Current one is 32952 bytes. Event not queued", "track")
         ]
         factory.destroy
 
@@ -2378,14 +2378,14 @@ async def is_valid_traffic_type(*_):
         _logger.reset_mock()
         assert await client.track("some_key", "traffic_type", "event_type", 1, []) is False
         assert _logger.error.mock_calls == [
-            mocker.call("track: properties must be of type dictionary.")
+            mocker.call("%s: properties must be of type dictionary.", "track")
         ]
 
         # Test track with invalid properties
         _logger.reset_mock()
         assert await client.track("some_key", "traffic_type", "event_type", 1, True) is False
         assert _logger.error.mock_calls == [
-            mocker.call("track: properties must be of type dictionary.")
+            mocker.call("%s: properties must be of type dictionary.", "track")
         ]
 
         # Test track with properties
@@ -2400,7 +2400,7 @@ async def is_valid_traffic_type(*_):
         _logger.reset_mock()
         assert await client.track("some_key", "traffic_type", "event_type", 1, props1) is True
         assert _logger.warning.mock_calls == [
-            mocker.call("Property %s is of invalid type. Setting value to None", [])
+            mocker.call("%s: Property %s is of invalid type. Setting value to None", "track", [])
         ]
 
         # Test track with more than 300 properties
@@ -2410,7 +2410,7 @@ async def is_valid_traffic_type(*_):
         _logger.reset_mock()
         assert await client.track("some_key", "traffic_type", "event_type", 1, props2) is True
         assert _logger.warning.mock_calls == [
-            mocker.call("Event has more than 300 properties. Some of them will be trimmed when processed")
+            mocker.call("%s: Event has more than 300 properties. Some of them will be trimmed when processed", "track")
         ]
 
         # Test track with properties higher than 32kb
@@ -2420,7 +2420,7 @@ async def is_valid_traffic_type(*_):
             props3["prop" + str(i)] = "a" * 300
         assert await client.track("some_key", "traffic_type", "event_type", 1, props3) is False
         assert _logger.error.mock_calls == [
-            mocker.call("The maximum size allowed for the properties is 32768 bytes. Current one is 32952 bytes. Event not queued")
+            mocker.call("%s: The maximum size allowed for the properties is 32768 bytes. Current one is 32952 bytes. Event not queued", "track")
         ]
         await factory.destroy()
 

tests/client/test_input_validator.py

@@ -49,6 +49,7 @@
 def _validate_last_impressions(client, *to_validate):
     """Validate the last N impressions are present disregarding the order."""
     imp_storage = client._factory._get_storage('impressions')
+    as_tup_set = set()
     if isinstance(client._factory._get_storage('splits'), RedisSplitStorage) or isinstance(client._factory._get_storage('splits'), PluggableSplitStorage):
         if isinstance(client._factory._get_storage('splits'), RedisSplitStorage):
             redis_client = imp_storage._redis
@@ -64,15 +65,28 @@ def _validate_last_impressions(client, *to_validate):
                 json.loads(i)
                 for i in results
             ]
-        as_tup_set = set(
-            (i['i']['f'], i['i']['k'], i['i']['t'])
-            for i in impressions_raw
-        )
+        if to_validate != ():
+            if len(to_validate[0]) == 3:
+                as_tup_set = set(
+                    (i['i']['f'], i['i']['k'], i['i']['t'])
+                    for i in impressions_raw
+                )
+            else:
+                as_tup_set = set(
+                    (i['i']['f'], i['i']['k'], i['i']['t'], i['i']['properties'])
+                    for i in impressions_raw
+                )
+            
         assert as_tup_set == set(to_validate)
         time.sleep(0.2) # delay for redis to sync
     else:
         impressions = imp_storage.pop_many(len(to_validate))
-        as_tup_set = set((i.feature_name, i.matching_key, i.treatment) for i in impressions)
+        if to_validate != ():
+            if len(to_validate[0]) == 3:
+                as_tup_set = set((i.feature_name, i.matching_key, i.treatment) for i in impressions)
+            else:
+                as_tup_set = set((i.feature_name, i.matching_key, i.treatment, i.properties) for i in impressions)
+                
         assert as_tup_set == set(to_validate)
 
 def _validate_last_events(client, *to_validate):
@@ -108,9 +122,9 @@ def _get_treatment(factory, skip_rbs=False):
     except:
         pass
 
-    assert client.get_treatment('user1', 'sample_feature') == 'on'
+    assert client.get_treatment('user1', 'sample_feature', impressions_properties={'prop':'value'}) == 'on'
     if not isinstance(factory._recorder._impressions_manager._strategy, StrategyNoneMode):
-        _validate_last_impressions(client, ('sample_feature', 'user1', 'on'))
+        _validate_last_impressions(client, ('sample_feature', 'user1', 'on', '{"prop": "value"}'))
 
     assert client.get_treatment('invalidKey', 'sample_feature') == 'off'
     if not isinstance(factory._recorder._impressions_manager._strategy, StrategyNoneMode):
@@ -514,7 +528,7 @@ def setup_method(self):
             'events': InMemoryEventStorage(5000, 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)
+        recorder = StandardRecorder(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 = SplitFactory('some_api_key',
@@ -674,7 +688,7 @@ def setup_method(self):
             'events': InMemoryEventStorage(5000, 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)
+        recorder = StandardRecorder(impmanager, storages['events'], storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, imp_counter=ImpressionsCounter())
         self.factory = SplitFactory('some_api_key',
                                     storages,
                                     True,
@@ -967,7 +981,7 @@ def setup_method(self):
         }
         impmanager = ImpressionsManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener
         recorder = PipelinedRecorder(redis_client.pipeline, impmanager, storages['events'],
-                                    storages['impressions'], telemetry_redis_storage)
+                                    storages['impressions'], telemetry_redis_storage, imp_counter=ImpressionsCounter())
         self.factory = SplitFactory('some_api_key',
                                     storages,
                                     True,
@@ -1155,7 +1169,7 @@ def setup_method(self):
         }
         impmanager = ImpressionsManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener
         recorder = PipelinedRecorder(redis_client.pipeline, impmanager,
-                                     storages['events'], storages['impressions'], telemetry_redis_storage)
+                                     storages['events'], storages['impressions'], telemetry_redis_storage, imp_counter=ImpressionsCounter())
         self.factory = SplitFactory('some_api_key',
                                     storages,
                                     True,
@@ -1375,7 +1389,7 @@ def setup_method(self):
 
         impmanager = ImpressionsManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener
         recorder = StandardRecorder(impmanager, storages['events'],
-                                    storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer)
+                                    storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, imp_counter=ImpressionsCounter())
 
         self.factory = SplitFactory('some_api_key',
                                     storages,
@@ -1570,7 +1584,7 @@ def setup_method(self):
 
         impmanager = ImpressionsManager(StrategyOptimizedMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener
         recorder = StandardRecorder(impmanager, storages['events'],
-                                    storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer)
+                                    storages['impressions'], telemetry_evaluation_producer, telemetry_runtime_producer, imp_counter=ImpressionsCounter())
 
         self.factory = SplitFactory('some_api_key',
                                     storages,
@@ -1617,7 +1631,7 @@ def test_get_treatment(self):
         client.get_treatment('user1', 'sample_feature')
         client.get_treatment('user1', 'sample_feature')
         client.get_treatment('user1', 'sample_feature')
-        assert self.pluggable_storage_adapter._keys['SPLITIO.impressions'] == []
+        assert len(self.pluggable_storage_adapter._keys['SPLITIO.impressions']) == 1
 
     def test_get_treatment_with_config(self):
         """Test client.get_treatment_with_config()."""
@@ -2317,7 +2331,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)
+        recorder = StandardRecorderAsync(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',
@@ -2839,7 +2853,7 @@ async def _setup_method(self):
         }
         impmanager = ImpressionsManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener
         recorder = PipelinedRecorderAsync(redis_client.pipeline, impmanager, storages['events'],
-                                    storages['impressions'], telemetry_redis_storage)
+                                    storages['impressions'], telemetry_redis_storage, imp_counter=ImpressionsCounter())
         self.factory = SplitFactoryAsync('some_api_key',
                                     storages,
                                     True,
@@ -3061,7 +3075,7 @@ async def _setup_method(self):
         }
         impmanager = ImpressionsManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer) # no listener
         recorder = PipelinedRecorderAsync(redis_client.pipeline, impmanager, storages['events'],
-                                    storages['impressions'], telemetry_redis_storage)
+                                    storages['impressions'], telemetry_redis_storage, imp_counter=ImpressionsCounter())
         self.factory = SplitFactoryAsync('some_api_key',
                                     storages,
                                     True,
@@ -3293,7 +3307,7 @@ async def _setup_method(self):
         recorder = StandardRecorderAsync(impmanager, storages['events'],
                                     storages['impressions'],
                                     telemetry_producer.get_telemetry_evaluation_producer(),
-                                    telemetry_runtime_producer)
+                                    telemetry_runtime_producer, imp_counter=ImpressionsCounter())
 
         self.factory = SplitFactoryAsync('some_api_key',
                                     storages,

tests/integration/test_client_e2e.py

@@ -158,9 +158,9 @@ class PluggableImpressionsStorageIntegrationTests(object):
     def _put_impressions(self, adapter, metadata):
         storage = PluggableImpressionsStorage(adapter, metadata)
         storage.put([
-            impressions.Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654),
-            impressions.Impression('key2', 'feature1', 'on', 'l1', 123456, 'b1', 321654),
-            impressions.Impression('key3', 'feature1', 'on', 'l1', 123456, 'b1', 321654)
+            impressions.Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654, None, None),
+            impressions.Impression('key2', 'feature1', 'on', 'l1', 123456, 'b1', 321654, None, None),
+            impressions.Impression('key3', 'feature1', 'on', 'l1', 123456, 'b1', 321654, None, None)
         ])
 
 

tests/integration/test_pluggable_integration.py

@@ -161,9 +161,9 @@ class RedisImpressionsStorageTests(object):
     def _put_impressions(self, adapter, metadata):
         storage = RedisImpressionsStorage(adapter, metadata)
         storage.put([
-            impressions.Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654),
-            impressions.Impression('key2', 'feature1', 'on', 'l1', 123456, 'b1', 321654),
-            impressions.Impression('key3', 'feature1', 'on', 'l1', 123456, 'b1', 321654)
+            impressions.Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654, None, None),
+            impressions.Impression('key2', 'feature1', 'on', 'l1', 123456, 'b1', 321654, None, None),
+            impressions.Impression('key3', 'feature1', 'on', 'l1', 123456, 'b1', 321654, None, None)
         ])
 
 
@@ -394,9 +394,9 @@ class RedisImpressionsStorageAsyncTests(object):
     async def _put_impressions(self, adapter, metadata):
         storage = RedisImpressionsStorageAsync(adapter, metadata)
         await storage.put([
-            impressions.Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654),
-            impressions.Impression('key2', 'feature1', 'on', 'l1', 123456, 'b1', 321654),
-            impressions.Impression('key3', 'feature1', 'on', 'l1', 123456, 'b1', 321654)
+            impressions.Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654, None, None),
+            impressions.Impression('key2', 'feature1', 'on', 'l1', 123456, 'b1', 321654, None, None),
+            impressions.Impression('key3', 'feature1', 'on', 'l1', 123456, 'b1', 321654, None, None)
         ])