feat: added check_payload to Switcher payload strategy check

Author: petruki

switcher_client/lib/resolver.py

@@ -1,11 +1,10 @@
 from typing import Optional
 
-from switcher_client.lib.types import Config, Domain, Entry, Group, ResultDetail, Snapshot, StrategyConfig
-from switcher_client.lib.snapshot import process_operation
-from switcher_client.lib.utils import get, get_entry
-from switcher_client.lib.utils import get_entry
-from switcher_client.errors import LocalCriteriaError
-from switcher_client.switcher_data import SwitcherData
+from ..lib.types import Config, Domain, Entry, Group, ResultDetail, Snapshot, StrategyConfig
+from ..lib.snapshot import process_operation
+from ..lib.utils import get, get_entry
+from ..errors import LocalCriteriaError
+from ..switcher_data import SwitcherData
 
 class Resolver:
 

switcher_client/lib/snapshot.py

@@ -4,8 +4,7 @@
 from typing import Optional
 from datetime import datetime
 
-from switcher_client.lib.types import StrategyConfig
-
+from ..lib.types import StrategyConfig
 from .utils.payload_reader import parse_json, payload_reader
 from .utils.ipcidr import IPCIDR
 from .utils.timed_match import TimedMatch

switcher_client/switcher_data.py

@@ -1,6 +1,6 @@
 from datetime import datetime
 from abc import ABCMeta
-from typing import Optional, Self
+from typing import Optional, Self, Union
 
 from .lib.snapshot import StrategiesType
 
@@ -30,6 +30,16 @@ def check_regex(self, input: str) -> Self:
         """ Adds REGEX_VALIDATION input for strategy validation """
         return self.check(StrategiesType.REGEX.value, input)
 
+    def check_payload(self, input: Union[str, dict]) -> Self:
+        """ Adds PAYLOAD_VALIDATION input for strategy validation """
+        if isinstance(input, dict):
+            import json
+            payload_str = json.dumps(input)
+        else:
+            payload_str = input
+            
+        return self.check(StrategiesType.PAYLOAD.value, payload_str)
+    
     def throttle(self, period: int) -> Self:
         """ Sets throttle period in milliseconds """
         self._throttle_period = period

tests/test_switcher_local.py

@@ -1,3 +1,5 @@
+import json
+
 from switcher_client.client import Client, ContextOptions
 from switcher_client.errors import LocalCriteriaError
 from switcher_client.lib.utils.timed_match.timed_match import TimedMatch
@@ -30,6 +32,32 @@ def test_local_with_strategy():
         .check_network('10.0.0.3') \
         .is_on('FF2FOR2020')
 
+def test_local_with_strategy_payload():
+    """ Should use local Snapshot to evaluate the switcher with payload validation strategy """
+
+    # given
+    given_context('tests/snapshots')
+    Client.load_snapshot()
+
+    switcher = Client.get_switcher()
+    payload = {
+        'id': 12345, 
+        'user': {
+            'login': 'test_user',
+            'role': 'admin'
+        }
+    }
+
+    # test (using stringified JSON)
+    assert switcher \
+        .check_payload(json.dumps(payload)) \
+        .is_on('FF2FOR2023')
+    
+    # test (using dict)
+    assert switcher \
+        .check_payload(payload) \
+        .is_on('FF2FOR2023')
+
 def test_local_with_strategy_no_input():
     """ Should return disabled when no input is provided for the strategy """
 

tests/utils/test_timed_match.py

@@ -45,8 +45,8 @@ def test_should_return_true(self):
         assert result is True
 
     def test_should_return_false_and_abort_processing(self):
-
         """Should return false and abort processing for ReDoS pattern."""
+        
         result = TimedMatch.try_match([NOK_RE], NOK_INPUT)
         assert result is False