feat: added testing decorator for switcher mocks

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 scoped mock isolation for concurrent execution
+- **Built-in Mocking**: Manual and decorator-based test mocking 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
@@ -318,7 +318,9 @@ The SDK includes powerful mocking capabilities for testing. Forced values are sc
 # Mock feature states for testing
 Client.assume('FEATURE01').true()
 assert switcher.is_on('FEATURE01') == True
+```
 
+```python
 # Conditional mocking based on input criteria
 Client.assume('FEATURE01').true() \
     # value can be either 'guest' or 'admin'
@@ -329,10 +331,14 @@ assert switcher \
     .check_value('guest') \
     .check_network('10.0.0.3') \
     .is_on('FEATURE01') == True
+```
 
+```python
 # Reset to normal behavior
 Client.forget('FEATURE01')
+```
 
+```python
 # Mock with metadata
 Client.assume('FEATURE01').false().with_metadata({
     'message': 'Feature is disabled'
@@ -342,6 +348,48 @@ assert response.result == False
 assert response.metadata['message'] == 'Feature is disabled'
 ```
 
+### Decorator-Based Testing
+
+Decorator-based tests can use the same fluent mocking rules while automatically cleaning up mocked flags after the test finishes:
+
+```python
+@switcher_test(assume_test('FEATURE01').true())
+def test_feature_flag():
+    assert switcher.is_on('FEATURE01') == True
+```
+
+```python
+@switcher_test(
+    assume_test('FEATURE01').true()
+        .when(StrategiesType.VALUE.value, 'guest')
+        .with_metadata({'message': 'Decorated mock'})
+)
+def test_feature_flag_with_rules():
+    response = switcher.check_value('guest').is_on_with_details('FEATURE01')
+    assert response.result == True
+    assert response.metadata['message'] == 'Decorated mock'
+````
+
+```python
+@switcher_test(
+    assume_test('FEATURE01').true(),
+    assume_test('FEATURE02').false()
+)
+def test_multiple_flags():
+    assert switcher.is_on('FEATURE01') == True
+    assert switcher.is_on('FEATURE02') == False
+```
+
+Decorator behavior:
+
+- `assume_test('FEATURE')` builds a single mocked flag assumption
+- `switcher_test(...)` accepts one or more assumptions
+- fluent mock rules are preserved, including `.true()`, `.false()`, `.when(...)`, and `.with_metadata(...)`
+- mocked flags are always cleaned up with `Client.forget(...)` after the test finishes, even when the test raises an error
+- both regular `def` tests and `async def` tests are supported
+
+This decorator API is intended as a test convenience and mock-isolation improvement. It improves the safety of mocked flags in concurrent execution within the same process, but should not be treated as full SDK-wide parallel execution support.
+
 ### Test Mode Configuration
 
 Convenient functionality to prevent subprocess locking of snapshot files during testing.
@@ -381,4 +429,4 @@ Thank you for helping us improve the Switcher Client SDK!
 - Virtualenv - `pip install virtualenv`
 - Create a virtual environment - `python3 -m venv .venv`
 - Install Pipenv - `pip install pipenv`
-- Check Makefile for all available commands
+- Check Makefile for all available commands

switcher_client/__init__.py

@@ -5,6 +5,7 @@
 from switcher_client.lib.globals.global_retry import RetryOptions
 from switcher_client.lib.snapshot_watcher import WatchSnapshotCallback
 from switcher_client.lib.snapshot import StrategiesType
+from switcher_client.testing import assume_test, switcher_test
 
 __all__ = [
     'Client',
@@ -14,4 +15,6 @@
     'RetryOptions',
     'WatchSnapshotCallback',
     'StrategiesType',
+    'assume_test',
+    'switcher_test',
 ]

switcher_client/testing.py

@@ -0,0 +1,109 @@
+import inspect
+
+from copy import deepcopy
+from functools import wraps
+from typing import Any, Callable
+
+from switcher_client.client import Client
+from switcher_client.lib.compat import Self
+
+class TestAssumption:
+    """Builder for a test-scoped feature flag assumption."""
+
+    def __init__(self, key: str, operations: list[tuple[str, tuple[Any, ...]]] | None = None):
+        self._key = key
+        self._operations = operations or []
+
+    def true(self) -> Self:
+        self._operations.append(('true', ()))
+        return self
+
+    def false(self) -> Self:
+        self._operations.append(('false', ()))
+        return self
+
+    def when(self, strategy: str, input_strategy: str | list[str]) -> Self:
+        self._operations.append(('when', (strategy, input_strategy)))
+        return self
+
+    def with_metadata(self, metadata: dict) -> Self:
+        self._operations.append(('with_metadata', (metadata,)))
+        return self
+
+    @property
+    def key(self) -> str:
+        return self._key
+
+    def apply(self) -> None:
+        assumed_key = Client.assume(self._key)
+
+        for operation_name, args in self._operations:
+            getattr(assumed_key, operation_name)(*args)
+
+    def clone(self) -> 'TestAssumption':
+        return TestAssumption(
+            self._key,
+            [
+                (operation_name, deepcopy(args))
+                for operation_name, args in self._operations
+            ]
+        )
+
+def assume_test(key: str) -> TestAssumption:
+    """Build a test-scoped feature flag assumption."""
+    return TestAssumption(key)
+
+def switcher_test(*assumptions: TestAssumption) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
+    """Wrap a test function with one or more temporary feature flag assumptions."""
+    frozen_assumptions = _freeze_assumptions(assumptions)
+
+    def decorator(test_function: Callable[..., Any]) -> Callable[..., Any]:
+        if inspect.iscoroutinefunction(test_function):
+            @wraps(test_function)
+            async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
+                assumption_keys = _apply_assumptions(frozen_assumptions)
+                try:
+                    return await test_function(*args, **kwargs)
+                finally:
+                    _forget_assumptions(assumption_keys)
+
+            return async_wrapper
+
+        @wraps(test_function)
+        def wrapper(*args: Any, **kwargs: Any) -> Any:
+            assumption_keys = _apply_assumptions(frozen_assumptions)
+            try:
+                return test_function(*args, **kwargs)
+            finally:
+                _forget_assumptions(assumption_keys)
+
+        return wrapper
+
+    return decorator
+
+def _freeze_assumptions(assumptions: tuple[TestAssumption, ...]) -> tuple[TestAssumption, ...]:
+    if not assumptions:
+        raise ValueError('switcher_test requires at least one test assumption')
+
+    for assumption in assumptions:
+        if not isinstance(assumption, TestAssumption):
+            raise TypeError('switcher_test expects values created with assume_test(...)')
+
+    return tuple(assumption.clone() for assumption in assumptions)
+
+def _apply_assumptions(assumptions: tuple[TestAssumption, ...]) -> list[str]:
+    assumption_keys: list[str] = []
+
+    try:
+        for assumption in assumptions:
+            assumption.apply()
+            assumption_keys.append(assumption.key)
+    except Exception:
+        _forget_assumptions(assumption_keys)
+        raise
+
+    return assumption_keys
+
+def _forget_assumptions(assumption_keys: list[str]) -> None:
+    for key in reversed(assumption_keys):
+        Client.forget(key)

tests/playground/index.py

@@ -6,8 +6,7 @@
 
 from util import monitor_run
 from switcher_client.lib.globals.global_context import DEFAULT_ENVIRONMENT
-from switcher_client.lib.globals.global_snapshot import LoadSnapshotOptions
-from switcher_client import Client, ContextOptions, WatchSnapshotCallback
+from switcher_client import Client, ContextOptions, WatchSnapshotCallback, LoadSnapshotOptions
 
 load_dotenv()
 API_KEY = os.getenv('SWITCHER_API_KEY')

tests/test_switcher_stub_decorator.py

@@ -0,0 +1,151 @@
+import asyncio
+import pytest
+
+from contextvars import Context
+
+from tests.test_switcher_integration import given_context
+from switcher_client import Client, ContextOptions, StrategiesType, assume_test, switcher_test
+
+context_options_local = ContextOptions(snapshot_location='tests/snapshots', local=True, logger=True)
+PRIMARY_KEY = 'FF2FOR2020'
+SECONDARY_KEY = 'FF2FOR2022'
+
+@pytest.fixture(autouse=True)
+def local_context():
+    given_context(options=context_options_local)
+    Client.load_snapshot()
+    yield
+    Client.clear_resources()
+
+def test_switcher_test_applies_and_cleans_up_mocked_value():
+    """ Should apply the mocked value during the test and clean up after the test, even if the test fails. """
+    @switcher_test(assume_test(PRIMARY_KEY).true())
+    def decorated_test():
+        return Client.get_switcher(PRIMARY_KEY).reset_inputs().is_on()
+
+    assert decorated_test() is True
+    assert Client.get_switcher(PRIMARY_KEY).reset_inputs().is_on() is False
+
+def test_switcher_test_requires_at_least_one_assumption():
+    """ Should raise an error when no assumptions are provided. """
+    with pytest.raises(ValueError, match='requires at least one test assumption'):
+        @switcher_test()
+        def decorated_test():
+            """ This test should never run because the decorator should raise an error due to missing assumptions. """
+            pass
+
+def test_switcher_test_requires_assume_test_builders():
+    """ Should raise an error when assumptions are not created with assume_test. """
+    with pytest.raises(TypeError, match='expects values created with assume_test'):
+        @switcher_test(PRIMARY_KEY) # type: ignore
+        def decorated_test():
+            """ This test should never run because the decorator should raise an error due to invalid assumptions. """
+            pass
+
+def test_switcher_test_cleans_up_when_test_fails():
+    """ Should clean up mocked value even if the test raises an exception. """
+    @switcher_test(assume_test(PRIMARY_KEY).true())
+    def decorated_test():
+        assert Client.get_switcher(PRIMARY_KEY).reset_inputs().is_on() is True
+        raise RuntimeError('boom')
+
+    with pytest.raises(RuntimeError, match='boom'):
+        decorated_test()
+
+    assert Client.get_switcher(PRIMARY_KEY).reset_inputs().is_on() is False
+
+def test_switcher_test_rolls_back_applied_mocks_when_setup_fails(monkeypatch: pytest.MonkeyPatch):
+    """ Should forget already-applied assumptions when a later assumption fails during setup. """
+    original_assume = Client.assume
+
+    def flaky_assume(key: str):
+        if key == SECONDARY_KEY:
+            raise RuntimeError('setup failed')
+        return original_assume(key)
+
+    monkeypatch.setattr(Client, 'assume', staticmethod(flaky_assume))
+
+    @switcher_test(
+        assume_test(PRIMARY_KEY).true(),
+        assume_test(SECONDARY_KEY).false()
+    )
+    def decorated_test():
+        """ This test should never run because setup fails before execution. """
+        assert False
+
+    with pytest.raises(RuntimeError, match='setup failed'):
+        decorated_test()
+
+    assert Client.get_switcher(PRIMARY_KEY).reset_inputs().is_on() is False
+
+def test_switcher_test_supports_conditional_mocks():
+    """ Should support conditional mocks based on strategies. """
+    @switcher_test(
+        assume_test(PRIMARY_KEY)
+            .true()
+            .when(StrategiesType.VALUE.value, 'Canada')
+            .when(StrategiesType.NETWORK.value, '10.0.0.3')
+    )
+    def decorated_test():
+        result_detail = Client.get_switcher(PRIMARY_KEY) \
+            .check_value('Canada') \
+            .check_network('10.0.0.3') \
+            .is_on_with_details()
+
+        assert result_detail.result is True
+        assert result_detail.reason == 'Forced to True'
+
+    decorated_test()
+
+def test_switcher_test_supports_metadata():
+    """ Should support metadata in mocked values. """
+    @switcher_test(
+        assume_test(PRIMARY_KEY)
+            .false()
+            .with_metadata({'message': 'Feature is disabled'})
+    )
+    def decorated_test():
+        result_detail = Client.get_switcher(PRIMARY_KEY).reset_inputs().is_on_with_details()
+
+        assert result_detail.result is False
+        assert result_detail.metadata == {'message': 'Feature is disabled'}
+
+    decorated_test()
+
+def test_switcher_test_supports_multiple_mocked_flags():
+    """ Should support mocking multiple flags in the same test. """
+    @switcher_test(
+        assume_test(PRIMARY_KEY).true(),
+        assume_test(SECONDARY_KEY).false()
+    )
+    def decorated_test():
+        assert Client.get_switcher(PRIMARY_KEY).reset_inputs().is_on() is True
+        assert Client.get_switcher(SECONDARY_KEY).reset_inputs().is_on() is False
+
+    decorated_test()
+
+    assert Client.get_switcher(PRIMARY_KEY).reset_inputs().is_on() is False
+    assert Client.get_switcher(SECONDARY_KEY).reset_inputs().is_on() is True
+
+def test_switcher_test_keeps_mocks_scoped_to_the_current_execution_context():
+    """ Should keep mocks scoped to the current execution context. """
+    @switcher_test(assume_test(PRIMARY_KEY).true())
+    def decorated_test():
+        isolated_result = Context().run(
+            lambda: Client.get_switcher(PRIMARY_KEY).reset_inputs().is_on()
+        )
+
+        assert Client.get_switcher(PRIMARY_KEY).reset_inputs().is_on() is True
+        assert isolated_result is False
+
+    decorated_test()
+
+def test_switcher_test_supports_async_tests():
+    """ Should support async tests. """
+    @switcher_test(assume_test(PRIMARY_KEY).true())
+    async def decorated_test():
+        await asyncio.sleep(0)
+        return Client.get_switcher(PRIMARY_KEY).reset_inputs().is_on()
+
+    assert asyncio.run(decorated_test()) is True
+    assert Client.get_switcher(PRIMARY_KEY).reset_inputs().is_on() is False