feat: added Client.test_mode() to prevent subprocess locking

README.md

@@ -344,9 +344,11 @@ assert response.metadata['message'] == 'Feature is disabled'
 
 ### Test Mode Configuration
 
+Convenient functionality to prevent subprocess locking of snapshot files during testing.
+
 ```python
-# 🚧 TODO: Enable test mode to prevent file locking
-Client.enable_test_mode()
+# Enable test mode to prevent snapshot file locking when running tests
+Client.test_mode()
 ```
 
 ### Configuration Validation

switcher_client/client.py

@@ -3,7 +3,7 @@
 from switcher_client.lib.bypasser import Bypasser, Key
 from switcher_client.lib.globals.global_auth import GlobalAuth
 from switcher_client.lib.globals.global_snapshot import GlobalSnapshot, LoadSnapshotOptions
-from switcher_client.lib.globals.global_context import Context, ContextOptions, DEFAULT_ENVIRONMENT
+from switcher_client.lib.globals.global_context import Context, ContextOptions, DEFAULT_ENVIRONMENT, DEFAULT_TEST_MODE
 from switcher_client.lib.remote_auth import RemoteAuth
 from switcher_client.lib.remote import Remote
 from switcher_client.lib.snapshot_auto_updater import SnapshotAutoUpdater
@@ -12,7 +12,7 @@
 from switcher_client.lib.utils.execution_logger import ExecutionLogger
 from switcher_client.lib.utils.timed_match.timed_match import TimedMatch
 from switcher_client.lib.utils import get
-from switcher_client.errors import SnapshotNotFoundError
+from switcher_client.errors import SnapshotNotFoundError, TestModeError
 from switcher_client.switcher import Switcher
 
 REGEX_MAX_BLACK_LIST = 'regex_max_black_list'
@@ -44,6 +44,7 @@ class Client:
     _switcher: dict[str, Switcher] = {}
     _snapshot_auto_updater: SnapshotAutoUpdater = SnapshotAutoUpdater()
     _snapshot_watcher: SnapshotWatcher = SnapshotWatcher()
+    _test_mode: bool = DEFAULT_TEST_MODE
 
     # pylint: disable=too-many-arguments
     @staticmethod
@@ -73,6 +74,7 @@ def build_context(*,
             options=options)
 
         # Default values
+        Client._test_mode = DEFAULT_TEST_MODE
         GlobalSnapshot.clear()
 
         # Build Options
@@ -193,6 +195,10 @@ def watch_snapshot(callback: Optional[WatchSnapshotCallback] = None) -> None:
         callback = get(callback, WatchSnapshotCallback())
         snapshot_location = Client._context.options.snapshot_location
 
+        if Client._test_mode:
+            callback.reject(TestModeError("Snapshot watcher is not available in test mode"))
+            return
+
         if snapshot_location is None:
             callback.reject(SnapshotNotFoundError("Snapshot location is not defined in the context options"))
             return
@@ -260,6 +266,11 @@ def forget(key: str) -> None:
         """ Remove forced value from a switcher """
         Bypasser.forget(key)
 
+    @staticmethod
+    def test_mode() -> None:
+        """ It prevents subprocess to run during tests such as snapshot watcher """
+        Client._test_mode = True
+
     @staticmethod
     def _is_check_snapshot_available(fetch_remote = False) -> bool:
         return Client.snapshot_version() == 0 and (fetch_remote or not Client._context.options.local)

switcher_client/errors/__init__.py

@@ -33,12 +33,19 @@ def __init__(self, message):
         self.message = message
         super().__init__(self.message)
 
+class TestModeError(Exception):
+    """ Raised when an operation is not allowed in test mode """
+    def __init__(self, message):
+        self.message = message
+        super().__init__(self.message)
+
 __all__ = [
     'RemoteError',
     'RemoteAuthError',
     'RemoteCriteriaError',
     'RemoteSwitcherError',
     'LocalSwitcherError',
     'LocalCriteriaError',
-    'SnapshotNotFoundError'
+    'SnapshotNotFoundError',
+    'TestModeError'
 ]

switcher_client/lib/globals/global_context.py

@@ -8,6 +8,7 @@
 DEFAULT_RESTRICT_RELAY = True
 DEFAULT_REGEX_MAX_BLACKLISTED = 100
 DEFAULT_REGEX_MAX_TIME_LIMIT = 3000
+DEFAULT_TEST_MODE = False
 
 @dataclass
 class ContextOptions:

tests/test_client_watch_snapshot.py

@@ -8,6 +8,7 @@
 from switcher_client.lib.snapshot_watcher import SnapshotWatcher, WatchSnapshotCallback
 
 context_options_local = ContextOptions(local=True, snapshot_location='tests/snapshots/temp')
+async_error = None
 
 class TestClientWatchSnapshot:
     """ Test suite for Client.watch_snapshot """
@@ -98,6 +99,26 @@ def test_watch_snapshot_err_malformed_snapshot(self):
         else:
             print("Warning: Snapshot watcher did not detect the change within the time limit")
 
+    def test_watch_snapshot_blocked_for_test(self):
+        """ Should not watch the snapshot file when Client is configured for test mode """
+
+        globals().update(async_error=None)
+        fixture_env = 'default_load_1'
+        fixture_location = 'tests/snapshots/temp'
+
+        # given
+        modify_fixture_snapshot(fixture_location, fixture_env, f'tests/snapshots/{fixture_env}.json')
+        given_context(options=context_options_local, environment=fixture_env)
+        Client.load_snapshot()
+
+        # test
+        Client.test_mode()
+        Client.watch_snapshot(WatchSnapshotCallback(
+            reject=lambda error: globals().update(async_error=str(error))
+        ))
+
+        assert async_error == 'Snapshot watcher is not available in test mode'
+
 # Helpers
 
 def modify_fixture_snapshot(location: str, environment: str, fixture_modified_location: str):