feat: added Client.check_switchers for smoke testing (local)

switcher_client/client.py

@@ -6,7 +6,7 @@
 from .lib.remote_auth import RemoteAuth
 from .lib.remote import Remote
 from .lib.snapshot_auto_updater import SnapshotAutoUpdater
-from .lib.snapshot_loader import load_domain, validate_snapshot, save_snapshot
+from .lib.snapshot_loader import check_switchers, load_domain, validate_snapshot, save_snapshot
 from .lib.utils.execution_logger import ExecutionLogger
 from .lib.utils.timed_match.timed_match import TimedMatch
 from .lib.utils import get
@@ -171,7 +171,10 @@ def snapshot_version() -> int:
     @staticmethod
     def check_switchers(switcher_keys: list[str]) -> None:
         """ Verifies if switchers are properly configured """
-        Client._check_switchers_remote(switcher_keys)
+        if Client._context.options.local:
+            check_switchers(GlobalSnapshot.snapshot(), switcher_keys)
+        else:
+            Client._check_switchers_remote(switcher_keys)
 
     @staticmethod
     def get_execution(switcher: Switcher) -> ExecutionLogger:

switcher_client/errors/__init__.py

@@ -13,7 +13,12 @@ def __init__(self, message):
 
 class RemoteSwitcherError(RemoteError):
     def __init__(self, not_found: list):
-        super().__init__(f'{", ".join(not_found)} not found')
+        super().__init__(f'{', '.join(not_found)} not found')
+
+class LocalSwitcherError(Exception):
+    def __init__(self, not_found: list):
+        self.message = f'{', '.join(not_found)} not found'
+        super().__init__(self.message)
 
 class LocalCriteriaError(Exception):
     def __init__(self, message):
@@ -25,5 +30,6 @@ def __init__(self, message):
     'RemoteAuthError',
     'RemoteCriteriaError',
     'RemoteSwitcherError',
+    'LocalSwitcherError',
     'LocalCriteriaError',
 ]

switcher_client/lib/snapshot_loader.py

@@ -5,6 +5,7 @@
 from .globals.global_context import Context
 from .remote import Remote
 from .types import Snapshot
+from ..errors import LocalSwitcherError
 
 def load_domain(snapshot_location: str, environment: str):
     """ Load Domain from snapshot file """
@@ -56,4 +57,21 @@ def save_snapshot(snapshot: Snapshot, snapshot_location: str, environment: str):
     os.makedirs(snapshot_location, exist_ok=True)
     snapshot_file = f"{snapshot_location}/{environment}.json"
     with open(snapshot_file, 'w') as file:
-        json.dump(snapshot.to_dict(), file, indent=4)
+        json.dump(snapshot.to_dict(), file, indent=4)
+
+def check_switchers(snapshot: Snapshot | None, switcher_keys: list[str]) -> None:
+    """ Check if switchers are properly configured in snapshot """
+    groups = getattr(snapshot.domain, 'group', []) if snapshot else []
+    not_found = []
+
+    for switcher in switcher_keys:
+        found = False
+        for group in groups:
+            if any(c.key == switcher for c in getattr(group, 'config', [])):
+                found = True
+                break
+        if not found:
+            not_found.append(switcher)
+
+    if not_found:
+        raise LocalSwitcherError(not_found)

tests/test_client_check_switchers.py

@@ -4,7 +4,7 @@
 from pytest_httpx import HTTPXMock
 
 from switcher_client.client import Client
-from switcher_client.errors import RemoteSwitcherError
+from switcher_client.errors import LocalSwitcherError, RemoteSwitcherError
 from switcher_client.lib.globals.global_context import ContextOptions
 
 
@@ -49,6 +49,59 @@ def test_check_remote_switchers_api_error(httpx_mock):
     except Exception as e:
         assert str(e) == '[check_switchers] failed with status: 500'
 
+def test_chek_remote_switchers_auth_error(httpx_mock):
+    """ Should check remote switchers and raise RemoteError when API returns an auth error """
+
+    # given
+    given_auth(httpx_mock, status=401)
+    given_context()
+
+    # test
+    try:
+        Client.check_switchers(['MY_SWITCHER', 'ANOTHER_SWITCHER'])
+        assert False, 'Expected RemoteError to be raised'
+    except Exception as e:
+        assert str(e) == 'Invalid API key'
+
+def test_check_local_switchers():
+    """ Should check local switchers with success """
+
+    # given
+    given_context(options=ContextOptions(snapshot_location='tests/snapshots', local=True))
+    snapshot_version = Client.load_snapshot()
+
+    # test
+    assert snapshot_version == 1
+    Client.check_switchers(['FF2FOR2020', 'FF2FOR2021'])
+
+def test_check_local_switchers_not_found():
+    """ Should check local switchers and raise LocalSwitcherError with not found switchers """
+
+    # given
+    given_context(options=ContextOptions(snapshot_location='tests/snapshots', local=True))
+    snapshot_version = Client.load_snapshot()
+
+    # test
+    assert snapshot_version == 1
+    try:
+        Client.check_switchers(['FF2FOR2020', 'NON_EXISTENT_SWITCHER'])
+        assert False, 'Expected LocalSwitcherError to be raised'
+    except LocalSwitcherError as e:
+        assert str(e) == 'NON_EXISTENT_SWITCHER not found'
+
+def test_check_local_switchers_no_snapshot():
+    """ Should check local switchers and raise LocalSwitcherError when no snapshot is loaded """
+
+    # given
+    given_context(options=ContextOptions(local=True))
+
+    # test
+    try:
+        Client.check_switchers(['FF2FOR2020', 'NON_EXISTENT_SWITCHER'])
+        assert False, 'Expected LocalSwitcherError to be raised'
+    except LocalSwitcherError as e:
+        assert str(e) == 'FF2FOR2020, NON_EXISTENT_SWITCHER not found'
+
 # Helpers
 
 def given_context(url='https://api.switcherapi.com', api_key='[API_KEY]', options = ContextOptions()):