feat: added scoped mock for improved concurrent test exection

README.md

@@ -47,7 +47,7 @@ The **Switcher Client SDK for Python** provides seamless integration with [Switc
 - **Clean & Maintainable**: Flexible and robust functions that keep your code organized
 - **Local Mode**: Work offline using snapshot files from your Switcher-API Domain
 - **Silent Mode**: Hybrid configuration with automatic fallback for connectivity issues
-- **Built-in Mocking**: Easy implementation of automated testing with mock support
+- **Built-in Mocking**: Easy implementation of automated testing with scoped mock isolation for concurrent execution
 - **Zero Latency**: Local snapshot execution for high-performance scenarios
 - **Secure**: Built-in protection against ReDoS attacks with regex safety features
 - **Monitoring**: Comprehensive logging and error handling capabilities
@@ -312,7 +312,7 @@ Client.watch_snapshot(WatchSnapshotCallback(
 
 ### Built-in Mocking
 
-The SDK will include powerful mocking capabilities for testing:
+The SDK includes powerful mocking capabilities for testing. Forced values are scoped to the current execution context, which adds a safe-net for concurrent test runs in the same process by reducing mock leakage between overlapping test executions while keeping the mocking API unchanged.
 
 ```python
 # Mock feature states for testing

switcher_client/client.py

@@ -75,6 +75,7 @@ def build_context(*,
 
         # Default values
         Client._test_mode = DEFAULT_TEST_MODE
+        Bypasser.clear()
         GlobalSnapshot.clear()
 
         # Build Options
@@ -244,6 +245,7 @@ def clear_resources() -> None:
         """ Clear all resources used by the Client """
         Client.terminate_snapshot_auto_update()
         Client.unwatch_snapshot()
+        Bypasser.clear()
         ExecutionLogger.clear_logger()
         GlobalSnapshot.clear()
         TimedMatch.terminate_worker()

switcher_client/lib/bypasser/bypasser.py

@@ -1,33 +1,42 @@
+from contextvars import ContextVar
+
 from switcher_client.lib.bypasser.key import Key
 
 class Bypasser:
     """
     Bypasser allows to force a switcher value to return a given value by calling one of both methods - true() false()
     """
 
-    _bypassed_keys = []
+    _bypassed_keys: ContextVar[dict[str, Key] | None] = ContextVar('bypassed_keys', default=None)
 
     @staticmethod
     def assume(key: str) -> Key:
         # Remove previous forced value if exists to avoid conflicts
-        Bypasser.forget(key)
-
         new_key = Key(key)
-        Bypasser._bypassed_keys.append(new_key)
+        bypassed_keys = dict(Bypasser._current_keys())
+        bypassed_keys[key] = new_key
+        Bypasser._bypassed_keys.set(bypassed_keys)
         return new_key
 
     @staticmethod
     def forget(key: str) -> None:
         """ Remove forced value from a switcher """
-        key_stored = Bypasser.search_key(key)
-        if key_stored is not None:
-            Bypasser._bypassed_keys.remove(key_stored)
+        bypassed_keys = dict(Bypasser._current_keys())
+        if key in bypassed_keys:
+            del bypassed_keys[key]
+            Bypasser._bypassed_keys.set(bypassed_keys)
 
     @staticmethod
     def search_key(key: str) -> Key | None:
         """ Search for key registered via 'assume' """
-        for bypassed_key in Bypasser._bypassed_keys:
-            if bypassed_key.key == key:
-                return bypassed_key
+        return Bypasser._current_keys().get(key)
 
-        return None
+    @staticmethod
+    def clear() -> None:
+        """ Remove all forced values from the current execution context """
+        Bypasser._bypassed_keys.set({})
+
+    @staticmethod
+    def _current_keys() -> dict[str, Key]:
+        bypassed_keys = Bypasser._bypassed_keys.get()
+        return bypassed_keys if bypassed_keys is not None else {}

switcher_client/lib/bypasser/key.py

@@ -43,11 +43,6 @@ def get_response(self, input_list: list[str] | None) -> ResultDetail:
 
         return ResultDetail.create(result=result, reason=self._reason, metadata=self._metadata)
 
-    @property
-    def key(self):
-        """ Return selected switcher name """
-        return self._key
-
     def _get_result_based_on_when(self, input_list: list[str]) -> bool:
         """ Evaluate the when conditions to determine the result """
         for strategy_when, input_when in self._when.items():

tests/test_switcher_stub.py

@@ -1,3 +1,5 @@
+from contextvars import Context
+
 from tests.test_switcher_integration import given_context
 
 from switcher_client.lib.snapshot import StrategiesType
@@ -104,4 +106,36 @@ def test_switcher_stub_with_multiple_criteria(self):
 
         assert switcher.check_value("Canada").is_on()
         assert switcher.check_value("Brazil").is_on()
-        assert not switcher.check_value("USA").is_on()
+        assert not switcher.check_value("USA").is_on()
+
+    def test_switcher_stub_is_isolated_per_context(self):
+        """ Should keep bypassed keys scoped to the current execution context """
+
+        # given
+        switcher = Client.get_switcher(self.key)
+
+        # test
+        Client.assume(self.key).true()
+
+        isolated_result = Context().run(lambda: switcher.reset_inputs().is_on())
+
+        assert switcher.reset_inputs().is_on()
+        assert isolated_result is False
+
+    def test_switcher_stub_can_override_per_context(self):
+        """ Should allow different bypass values in different execution contexts """
+
+        # given
+        switcher = Client.get_switcher(self.key)
+
+        # test
+        Client.assume(self.key).true()
+
+        def isolated_context_result():
+            Client.assume(self.key).false()
+            return switcher.reset_inputs().is_on()
+
+        isolated_result = Context().run(isolated_context_result)
+
+        assert switcher.reset_inputs().is_on()
+        assert isolated_result is False