Implemented Impressions strategy classes for Impression Manager

Author: chillaq

splitio/engine/impressions.py

@@ -1,4 +1,5 @@
 """Split evaluator module."""
+import abc
 import threading
 from collections import defaultdict, namedtuple
 from enum import Enum
@@ -9,6 +10,8 @@
 from splitio.client.listener import ImpressionListenerException
 from splitio import util
 
+import logging
+_LOGGER = logging.getLogger(__name__)
 
 _TIME_INTERVAL_MS = 3600 * 1000  # one hour
 _IMPRESSION_OBSERVER_CACHE_SIZE = 500000
@@ -150,6 +153,130 @@ def pop_all(self):
         return [Counter.CountPerFeature(k.feature, k.timeframe, v)
                 for (k, v) in old.items()]
 
+class BaseStrategy(object, metaclass=abc.ABCMeta):
+    """Strategy interface."""
+
+    @abc.abstractmethod
+    def process_impressions(self):
+        """
+        Return a list(impressions) object
+
+        """
+        pass
+
+    @abc.abstractmethod
+    def truncate_impressions_time(self):
+        """
+        Return list(impressions) object
+
+        """
+        pass
+
+    def get_counts(self):
+        """
+        Return A list of counter objects.
+
+        """
+        pass
+
+class StrategyOptimizedMode(BaseStrategy):
+    """Optimized mode strategy."""
+
+    def __init__(self, standalone=True):
+        """
+        Construct a strategy instance for optimized mode.
+
+        """
+        self._standalone = standalone
+        self._counter = Counter() if self._standalone else None
+        self._observer =  Observer(_IMPRESSION_OBSERVER_CACHE_SIZE) if self._standalone else None
+
+    def process_impressions(self, impressions):
+        """
+        Process impressions.
+
+        Impressions are analyzed to see if they've been seen before and counted.
+
+        :param impressions: List of impression objects with attributes
+        :type impressions: list[tuple[splitio.models.impression.Impression, dict]]
+
+        :returns: Observed list of impressions
+        :rtype: list[tuple[splitio.models.impression.Impression, dict]]
+        """
+        imps = [(self._observer.test_and_set(imp), attrs) for imp, attrs in impressions] \
+        if self._observer else impressions
+        if self._counter is not None:
+            self._counter.track([imp for imp, _ in imps])
+        return imps
+
+    def truncate_impressions_time(self, imps):
+        """
+        Process impressions.
+
+        Impressions are truncated based on time
+
+        :param impressions: List of impression objects with attributes
+        :type impressions: list[tuple[splitio.models.impression.Impression, dict]]
+
+        :returns: truncated list of impressions
+        :rtype: list[splitio.models.impression.Impression]
+        """
+        this_hour = truncate_time(util.utctime_ms())
+        return [imp for imp, _ in imps] if self._counter is None \
+            else [i for i, _ in imps if i.previous_time is None or i.previous_time < this_hour]
+
+    def get_counts(self):
+        """
+        Return counts of impressions per features.
+
+        :returns: A list of counter objects.
+        :rtype: list[Counter.CountPerFeature]
+        """
+        return self._counter.pop_all() if self._counter is not None else []
+
+class StrategyDebugMode(BaseStrategy):
+    """Debug mode strategy."""
+
+    def __init__(self, standalone=True):
+        """
+        Construct a strategy instance for debug mode.
+
+        """
+        self._standalone = standalone
+        self._observer =  Observer(_IMPRESSION_OBSERVER_CACHE_SIZE) if self._standalone else None
+
+    def process_impressions(self, impressions):
+        """
+        Process impressions.
+
+        Impressions are analyzed to see if they've been seen before.
+
+        :param impressions: List of impression objects with attributes
+        :type impressions: list[tuple[splitio.models.impression.Impression, dict]]
+
+        :returns: Observed list of impressions
+        :rtype: list[tuple[splitio.models.impression.Impression, dict]]
+        """
+        imps = [(self._observer.test_and_set(imp), attrs) for imp, attrs in impressions] if self._observer is not None else impressions
+        return imps
+
+    def truncate_impressions_time(self, imps):
+        """
+        No counter implemented, return same impresisons passed.
+
+        :returns: list of impressions
+        :rtype: list[splitio.models.impression.Impression]
+        """
+        return [imp for imp, _ in imps]
+
+    def get_counts(self):
+        """
+        No counter implemented, return empty array
+
+        :returns: empty list
+        :rtype: list[]
+        """
+        return []
 
 class Manager(object):  # pylint:disable=too-few-public-methods
     """Impression manager."""
@@ -167,10 +294,18 @@ def __init__(self, mode=ImpressionsMode.OPTIMIZED, standalone=True, listener=Non
         :param listener: Optional impressions listener that will capture all seen impressions.
         :type listener: splitio.client.listener.ImpressionListenerWrapper
         """
-        self._observer = Observer(_IMPRESSION_OBSERVER_CACHE_SIZE) if standalone else None
-        self._counter = Counter() if standalone and mode == ImpressionsMode.OPTIMIZED else None
+        self._strategy = self.get_strategy(mode, standalone)
         self._listener = listener
 
+    def get_strategy(self, mode, standalone):
+        """
+        Return a strategy object based on mode value.
+
+        :returns: A strategy object
+        :rtype: (BaseStrategy)
+        """
+        return StrategyOptimizedMode(standalone) if mode == ImpressionsMode.OPTIMIZED else StrategyDebugMode(standalone)
+
     def process_impressions(self, impressions):
         """
         Process impressions.
@@ -180,17 +315,9 @@ def process_impressions(self, impressions):
         :param impressions: List of impression objects with attributes
         :type impressions: list[tuple[splitio.models.impression.Impression, dict]]
         """
-        imps = [(self._observer.test_and_set(imp), attrs) for imp, attrs in impressions] \
-            if self._observer else impressions
-
-        if self._counter:
-            self._counter.track([imp for imp, _ in imps])
-
+        imps = self._strategy.process_impressions(impressions)
         self._send_impressions_to_listener(imps)
-
-        this_hour = truncate_time(util.utctime_ms())
-        return [imp for imp, _ in imps] if self._counter is None \
-            else [i for i, _ in imps if i.previous_time is None or i.previous_time < this_hour]
+        return self._strategy.truncate_impressions_time(imps)
 
     def get_counts(self):
         """
@@ -199,7 +326,7 @@ def get_counts(self):
         :returns: A list of counter objects.
         :rtype: list[Counter.CountPerFeature]
         """
-        return self._counter.pop_all() if self._counter is not None else []
+        return self._strategy.get_counts()
 
     def _send_impressions_to_listener(self, impressions):
         """

tests/engine/test_impressions.py

@@ -1,11 +1,10 @@
 """Impression manager, observer & hasher tests."""
 from datetime import datetime
 from splitio.engine.impressions import Hasher, Observer, Counter, Manager, \
-    ImpressionsMode, truncate_time
+    ImpressionsMode, truncate_time, StrategyDebugMode, StrategyOptimizedMode
 from splitio.models.impressions import Impression
 from splitio.client.listener import ImpressionListenerWrapper
 
-
 def utctime_ms_reimplement():
     """Re-implementation of utctime_ms to avoid conflicts with mock/patching."""
     return int((datetime.utcnow() - datetime(1970, 1, 1)).total_seconds() * 1000)
@@ -99,18 +98,25 @@ def test_standalone_optimized(self, mocker):
         mocker.patch('splitio.util.utctime_ms', new=utc_time_mock)
 
         manager = Manager()  # no listener
-        assert manager._counter is not None
-        assert manager._observer is not None
+        assert manager._strategy._counter is not None
+        assert manager._strategy._observer is not None
         assert manager._listener is None
+        assert isinstance(manager._strategy, StrategyOptimizedMode)
 
         # An impression that hasn't happened in the last hour (pt = None) should be tracked
         imps = manager.process_impressions([
             (Impression('k1', 'f1', 'on', 'l1', 123, None, utc_now-3), None),
             (Impression('k1', 'f2', 'on', 'l1', 123, None, utc_now-3), None)
         ])
+
         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)
@@ -136,10 +142,10 @@ def test_standalone_optimized(self, mocker):
         assert imps == [Impression('k1', 'f1', 'on', 'l1', 123, None, utc_now-1, old_utc-3),
                         Impression('k2', 'f1', 'on', 'l1', 123, None, utc_now-2, old_utc-1)]
 
-        assert len(manager._observer._cache._data) == 3  # distinct impressions seen
-        assert len(manager._counter._data) == 3  # 2 distinct features. 1 seen in 2 different timeframes
+        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 set(manager._counter.pop_all()) == set([
+        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(utc_now), 2)
@@ -155,9 +161,10 @@ def test_standalone_debug(self, mocker):
         mocker.patch('splitio.util.utctime_ms', new=utc_time_mock)
 
         manager = Manager(ImpressionsMode.DEBUG)  # no listener
-        assert manager._counter is None
-        assert manager._observer is not None
+        assert manager._strategy.get_counts() == []
+        assert manager._strategy._observer is not None
         assert manager._listener is None
+        assert isinstance(manager._strategy, StrategyDebugMode)
 
         # An impression that hasn't happened in the last hour (pt = None) should be tracked
         imps = manager.process_impressions([
@@ -192,7 +199,7 @@ def test_standalone_debug(self, mocker):
         assert imps == [Impression('k1', 'f1', 'on', 'l1', 123, None, utc_now-1, old_utc-3),
                         Impression('k2', 'f1', 'on', 'l1', 123, None, utc_now-2, old_utc-1)]
 
-        assert len(manager._observer._cache._data) == 3  # distinct impressions seen
+        assert len(manager._strategy._observer._cache._data) == 3  # distinct impressions seen
 
     def test_non_standalone_optimized(self, mocker):
         """Test impressions manager in optimized mode with sdk in standalone mode."""
@@ -204,9 +211,10 @@ def test_non_standalone_optimized(self, mocker):
         mocker.patch('splitio.util.utctime_ms', new=utc_time_mock)
 
         manager = Manager(ImpressionsMode.OPTIMIZED, False)  # no listener
-        assert manager._counter is None
-        assert manager._observer is None
+        assert manager._strategy._counter is None
+        assert manager._strategy._observer is None
         assert manager._listener is None
+        assert isinstance(manager._strategy, StrategyOptimizedMode)
 
         # An impression that hasn't happened in the last hour (pt = None) should be tracked
         imps = manager.process_impressions([
@@ -250,9 +258,10 @@ def test_non_standalone_debug(self, mocker):
         mocker.patch('splitio.util.utctime_ms', new=utc_time_mock)
 
         manager = Manager(ImpressionsMode.DEBUG, False)  # no listener
-        assert manager._counter is None
-        assert manager._observer is None
+        assert manager._strategy.get_counts() == []
+        assert manager._strategy._observer is None
         assert manager._listener is None
+        assert isinstance(manager._strategy, StrategyDebugMode)
 
         # An impression that hasn't happened in the last hour (pt = None) should be tracked
         imps = manager.process_impressions([
@@ -298,9 +307,10 @@ def test_standalone_optimized_listener(self, mocker):
         listener = mocker.Mock(spec=ImpressionListenerWrapper)
 
         manager = Manager(listener=listener)  # no listener
-        assert manager._counter is not None
-        assert manager._observer is not None
+        assert manager._strategy._counter is not None
+        assert manager._strategy._observer is not None
         assert manager._listener is not None
+        assert isinstance(manager._strategy, StrategyOptimizedMode)
 
         # An impression that hasn't happened in the last hour (pt = None) should be tracked
         imps = manager.process_impressions([
@@ -309,6 +319,10 @@ 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([
@@ -335,10 +349,10 @@ def test_standalone_optimized_listener(self, mocker):
         assert imps == [Impression('k1', 'f1', 'on', 'l1', 123, None, utc_now-1, old_utc-3),
                         Impression('k2', 'f1', 'on', 'l1', 123, None, utc_now-2, old_utc-1)]
 
-        assert len(manager._observer._cache._data) == 3  # distinct impressions seen
-        assert len(manager._counter._data) == 3  # 2 distinct features. 1 seen in 2 different timeframes
+        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 set(manager._counter.pop_all()) == set([
+        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(utc_now), 2)
@@ -365,9 +379,10 @@ def test_standalone_debug_listener(self, mocker):
         imps = []
         listener = mocker.Mock(spec=ImpressionListenerWrapper)
         manager = Manager(ImpressionsMode.DEBUG, listener=listener)
-        assert manager._counter is None
-        assert manager._observer is not None
+        assert manager._strategy.get_counts() == []
+        assert manager._strategy._observer is not None
         assert manager._listener is not None
+        assert isinstance(manager._strategy, StrategyDebugMode)
 
         # An impression that hasn't happened in the last hour (pt = None) should be tracked
         imps = manager.process_impressions([
@@ -402,7 +417,7 @@ def test_standalone_debug_listener(self, mocker):
         assert imps == [Impression('k1', 'f1', 'on', 'l1', 123, None, utc_now-1, old_utc-3),
                         Impression('k2', 'f1', 'on', 'l1', 123, None, utc_now-2, old_utc-1)]
 
-        assert len(manager._observer._cache._data) == 3  # distinct impressions seen
+        assert len(manager._strategy._observer._cache._data) == 3  # distinct impressions seen
 
         assert listener.log_impression.mock_calls == [
             mocker.call(Impression('k1', 'f1', 'on', 'l1', 123, None, old_utc-3), None),
@@ -425,9 +440,10 @@ def test_non_standalone_optimized_listener(self, mocker):
         imps = []
         listener = mocker.Mock(spec=ImpressionListenerWrapper)
         manager = Manager(ImpressionsMode.OPTIMIZED, False, listener)  # no listener
-        assert manager._counter is None
-        assert manager._observer is None
+        assert manager._strategy._counter is None
+        assert manager._strategy._observer is None
         assert manager._listener is not None
+        assert isinstance(manager._strategy, StrategyOptimizedMode)
 
         # An impression that hasn't happened in the last hour (pt = None) should be tracked
         imps = manager.process_impressions([
@@ -482,9 +498,10 @@ def test_non_standalone_debug_listener(self, mocker):
 
         listener = mocker.Mock(spec=ImpressionListenerWrapper)
         manager = Manager(ImpressionsMode.DEBUG, False, listener)  # no listener
-        assert manager._counter is None
-        assert manager._observer is None
+        assert manager._strategy.get_counts() == []
+        assert manager._strategy._observer is None
         assert manager._listener is not None
+        assert isinstance(manager._strategy, StrategyDebugMode)
 
         # An impression that hasn't happened in the last hour (pt = None) should be tracked
         imps = manager.process_impressions([