Merge branch 'development' into feature/telemetryv2

Author: chillaq

.github/workflows/ci.yml

@@ -10,20 +10,20 @@ on:
 
 jobs:
   test:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-20.04
     services:
       redis:
         image: redis
         ports:
           - 6379:6379
     steps:
       - name: Checkout code
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
         with:
           fetch-depth: 0
 
       - name: Set up Python
-        uses: actions/setup-python@v2
+        uses: actions/setup-python@v3
         with:
           python-version: '3.6'
 

splitio/client/factory.py

@@ -68,6 +68,7 @@
 _MIN_DEFAULT_DATA_SAMPLING_ALLOWED = 0.1  # 10%
 _MAX_RETRY_SYNC_ALL = 3
 
+
 class Status(Enum):
     """Factory Status."""
 

splitio/engine/impressions/adapters.py

@@ -88,8 +88,8 @@ def flush_counters(self, to_send):
         """
         post the impression counters to redis.
 
-        :param uniques: unique keys disctionary
-        :type uniques: Dictionary {'feature1': set(), 'feature2': set(), .. }
+        :param to_send: unique keys disctionary
+        :type to_send: Dictionary {'feature1': set(), 'feature2': set(), .. }
         """
         try:
             resulted = 0

splitio/engine/impressions/strategies.py

@@ -99,6 +99,6 @@ def process_impressions(self, impressions):
         :rtype: list[tuple[splitio.models.impression.Impression, dict]]
         """
         imps = [(self._observer.test_and_set(imp), attrs) for imp, attrs in impressions]
-        self._counter.track([imp for imp, _ in imps])
-        this_hour = truncate_time(utctime_ms())
+        self._counter.track([imp for imp, _ in imps if imp.previous_time != None])
+        this_hour = truncate_time(util.utctime_ms())
         return [i for i, _ in imps if i.previous_time is None or i.previous_time < this_hour], imps

splitio/sync/synchronizer.py

@@ -8,6 +8,7 @@
 from splitio.api import APIException
 from splitio.util.backoff import Backoff
 
+
 _LOGGER = logging.getLogger(__name__)
 _SYNC_ALL_NO_RETRIES = -1
 
@@ -306,6 +307,7 @@ def synchronize_splits(self, till, sync_segments=True):
     def sync_all(self, max_retry_attempts=_SYNC_ALL_NO_RETRIES):
         """
         Synchronize all splits.
+        
         :param max_retry_attempts: apply max attempts if it set to absilute integer.
         :type max_retry_attempts: int
         """
@@ -332,7 +334,10 @@ def sync_all(self, max_retry_attempts=_SYNC_ALL_NO_RETRIES):
                 how_long = self._backoff.get()
                 time.sleep(how_long)
 
-            _LOGGER.error("Could not correctly synchronize splits and segments after %d attempts.", retry_attempts)
+        _LOGGER.error("Could not correctly synchronize splits and segments after %d attempts.", retry_attempts)
+
+    def _retry_block(self, max_retry_attempts, retry_attempts):
+        return retry_attempts
 
     def shutdown(self, blocking):
         """
@@ -504,6 +509,7 @@ def __init__(self, split_synchronizers, split_tasks):
     def sync_all(self, max_retry_attempts=-1):
         """
         Synchronize all splits.
+        
         :param max_retry_attempts: Not used, added for compatibility
         """
         try:

tests/client/test_factory.py

@@ -5,6 +5,7 @@
 import os
 import time
 import threading
+import pytest
 from splitio.client.factory import get_factory, SplitFactory, _INSTANTIATED_FACTORIES, Status,\
     _LOGGER as _logger
 from splitio.client.config import DEFAULT_CONFIG
@@ -126,7 +127,7 @@ def test_redis_client_creation(self, mocker):
             ssl_certfile='some_cert_file',
             ssl_cert_reqs='some_cert_req',
             ssl_ca_certs='some_ca_cert',
-            max_connections=999
+            max_connections=999,
         )
         assert factory._labels_enabled is False
         assert isinstance(factory._recorder, PipelinedRecorder)
@@ -238,6 +239,7 @@ def _split_synchronizer(self, ready_flag, some, auth_api, streaming_enabled, sdk
         mocker.patch('splitio.sync.manager.Manager.__init__', new=_split_synchronizer)
 
         # Start factory and make assertions
+        # Using invalid key should result in a timeout exception
         factory = get_factory('some_api_key')
         try:
             factory.block_until_ready(1)
@@ -338,6 +340,7 @@ def _split_synchronizer(self, ready_flag, some, auth_api, streaming_enabled, sdk
             factory.block_until_ready(1)
         except:
             pass
+            
         assert factory.ready is True
         assert factory.destroyed is False
 

tests/engine/test_impressions.py

@@ -121,11 +121,6 @@ def test_standalone_optimized(self, mocker):
         assert imps == [Impression('k1', 'f1', 'on', 'l1', 123, None, utc_now-3),
                         Impression('k1', 'f2', 'on', 'l1', 123, None, utc_now-3)]
 
-        assert [Counter.CountPerFeature(k.feature, k.timeframe, v)
-                for (k, v) in manager._strategy._counter._data.items()] == [
-            Counter.CountPerFeature('f1', truncate_time(utc_now-3), 1),
-            Counter.CountPerFeature('f2', truncate_time(utc_now-3), 1)]
-
         # Tracking the same impression a ms later should be empty
         imps = manager.process_impressions([
             (Impression('k1', 'f1', 'on', 'l1', 123, None, utc_now-2), None)
@@ -154,14 +149,26 @@ def test_standalone_optimized(self, mocker):
                         Impression('k2', 'f1', 'on', 'l1', 123, None, utc_now-2, old_utc-1)]
 
         assert len(manager._strategy._observer._cache._data) == 3  # distinct impressions seen
-        assert len(manager._strategy._counter._data) == 3  # 2 distinct features. 1 seen in 2 different timeframes
+        assert len(manager._strategy._counter._data) == 2  # 2 distinct features. 1 seen in 2 different timeframes
 
         assert set(manager._strategy._counter.pop_all()) == set([
-            Counter.CountPerFeature('f1', truncate_time(old_utc), 3),
-            Counter.CountPerFeature('f2', truncate_time(old_utc), 1),
+            Counter.CountPerFeature('f1', truncate_time(old_utc), 1),
             Counter.CountPerFeature('f1', truncate_time(utc_now), 2)
         ])
 
+        # Test counting only from the second impression
+        imps = manager.process_impressions([
+            (Impression('k3', 'f3', 'on', 'l1', 123, None, utc_now-1), None)
+        ])
+        assert set(manager._strategy._counter.pop_all()) == set([])
+
+        imps = manager.process_impressions([
+            (Impression('k3', 'f3', 'on', 'l1', 123, None, utc_now-1), None)
+        ])
+        assert set(manager._strategy._counter.pop_all()) == set([
+            Counter.CountPerFeature('f3', truncate_time(utc_now), 1)
+        ])
+
     def test_standalone_debug(self, mocker):
         """Test impressions manager in debug mode with sdk in standalone mode."""
 
@@ -304,10 +311,6 @@ def test_standalone_optimized_listener(self, mocker):
         ])
         assert imps == [Impression('k1', 'f1', 'on', 'l1', 123, None, utc_now-3),
                         Impression('k1', 'f2', 'on', 'l1', 123, None, utc_now-3)]
-        assert [Counter.CountPerFeature(k.feature, k.timeframe, v)
-                for (k, v) in manager._strategy._counter._data.items()] == [
-            Counter.CountPerFeature('f1', truncate_time(utc_now-3), 1),
-            Counter.CountPerFeature('f2', truncate_time(utc_now-3), 1)]
 
         # Tracking the same impression a ms later should return empty
         imps = manager.process_impressions([
@@ -336,11 +339,10 @@ def test_standalone_optimized_listener(self, mocker):
                         Impression('k2', 'f1', 'on', 'l1', 123, None, utc_now-2, old_utc-1)]
 
         assert len(manager._strategy._observer._cache._data) == 3  # distinct impressions seen
-        assert len(manager._strategy._counter._data) == 3  # 2 distinct features. 1 seen in 2 different timeframes
+        assert len(manager._strategy._counter._data) == 2  # 2 distinct features. 1 seen in 2 different timeframes
 
         assert set(manager._strategy._counter.pop_all()) == set([
-            Counter.CountPerFeature('f1', truncate_time(old_utc), 3),
-            Counter.CountPerFeature('f2', truncate_time(old_utc), 1),
+            Counter.CountPerFeature('f1', truncate_time(old_utc), 1),
             Counter.CountPerFeature('f1', truncate_time(utc_now), 2)
         ])
 
@@ -353,6 +355,19 @@ def test_standalone_optimized_listener(self, mocker):
             mocker.call(Impression('k2', 'f1', 'on', 'l1', 123, None, utc_now-2, old_utc-1), None)
         ]
 
+        # Test counting only from the second impression
+        imps = manager.process_impressions([
+            (Impression('k3', 'f3', 'on', 'l1', 123, None, utc_now-1), None)
+        ])
+        assert set(manager._strategy._counter.pop_all()) == set([])
+
+        imps = manager.process_impressions([
+            (Impression('k3', 'f3', 'on', 'l1', 123, None, utc_now-1), None)
+        ])
+        assert set(manager._strategy._counter.pop_all()) == set([
+            Counter.CountPerFeature('f3', truncate_time(utc_now), 1)
+        ])
+
     def test_standalone_debug_listener(self, mocker):
         """Test impressions manager in optimized mode with sdk in standalone mode."""
 

tests/sync/test_manager.py

@@ -56,7 +56,7 @@ def run(x):
         manager = Manager(threading.Event(), synchronizer,  mocker.Mock(), False, SdkMetadata('1.0', 'some', '1.2.3.4'), mocker.Mock())
 
         manager._SYNC_ALL_ATTEMPTS = 1
-        manager.start(2) # should not throw!
+        manager.start(2)  # should not throw!
 
     def test_start_streaming_false(self, mocker):
         splits_ready_event = threading.Event()