Added __execute_local_criteria() API for local evaluation

switcher_client/errors/__init__.py

@@ -9,4 +9,9 @@ def __init__(self, message):
 
 class RemoteCriteriaError(RemoteError):
     def __init__(self, message):
-        super().__init__(message)
+        super().__init__(message)
+
+class LocalCriteriaError(Exception):
+    def __init__(self, message):
+        self.message = message
+        super().__init__(self.message)

switcher_client/lib/resolver.py

@@ -0,0 +1,41 @@
+from switcher_client.errors import LocalCriteriaError
+from switcher_client.lib.types import Config, Group, ResultDetail, Snapshot, SnapshotData
+from switcher_client.switcher_data import SwitcherData
+
+class Resolver:
+
+    @staticmethod
+    def check_criteria(snapshot: Snapshot | None, switcher: SwitcherData) -> ResultDetail:
+        if not snapshot:
+            raise LocalCriteriaError("Snapshot not loaded. Try to use 'Client.load_snapshot()'")
+
+        return Resolver.__check_domain(snapshot.data, switcher)
+
+    @staticmethod
+    def __check_domain(data: SnapshotData, switcher: SwitcherData) -> ResultDetail:
+        if data.domain.activated is False:
+            return ResultDetail.disabled("Domain is disabled")
+
+        return Resolver.__check_group(data.domain.group, switcher)
+
+    @staticmethod
+    def __check_group(groups: list[Group], switcher: SwitcherData) -> ResultDetail:
+        key = switcher._key
+
+        for group in groups:
+            config_found = next((c for c in group.config if c.key == key), None)
+
+            if config_found is not None:
+                if group.activated is False:
+                    return ResultDetail.disabled("Group disabled")
+
+                return Resolver.__check_config(config_found, switcher)
+
+        raise LocalCriteriaError(f"Config with key '{key}' not found in the snapshot")
+
+    @staticmethod
+    def __check_config(config: Config, switcher: SwitcherData) -> ResultDetail:
+        if config.activated is False:
+            return ResultDetail.disabled("Config disabled")
+
+        return ResultDetail.success()

switcher_client/lib/types.py

@@ -1,46 +1,54 @@
-from typing import Optional, List
+from typing import Optional
 
 class ResultDetail:
-    def __init__(self, result: bool, reason: str, metadata: dict):
+    def __init__(self, result: bool, reason: Optional[str], metadata: Optional[dict] = None):
         self.result = result
         self.reason = reason
         self.metadata = metadata
 
+    @staticmethod
+    def disabled(reason: str, metadata: Optional[dict] = None) -> 'ResultDetail':
+        return ResultDetail(result=False, reason=reason, metadata=metadata)
+
+    @staticmethod
+    def success(reason: str = "Success", metadata: Optional[dict] = None) -> 'ResultDetail':
+        return ResultDetail(result=True, reason=reason, metadata=metadata)
+
 class SnapshotData:
     def __init__(self):
         self.domain: Domain
 
 class Domain:
     def __init__(self):
-        self.name: Optional[str] = None
+        self.name: str
         self.version: int = 0
-        self.activated: Optional[bool] = None
-        self.group: Optional[List[Group]] = None
+        self.activated: bool
+        self.group: list[Group]
 
 class Group:
     def __init__(self):
-        self.name: Optional[str] = None
-        self.activated: Optional[bool] = None
-        self.config: Optional[List[Config]] = None
+        self.name: str
+        self.activated: bool
+        self.config: list[Config]
 
 class Config:
     def __init__(self):
-        self.key: Optional[str] = None
-        self.activated: Optional[bool] = None
-        self.strategies: Optional[List[StrategyConfig]] = None
+        self.key: str
+        self.activated: bool
+        self.strategies: Optional[list[StrategyConfig]] = None
         self.relay: Optional[Relay] = None
 
 class StrategyConfig:
     def __init__(self):
-        self.strategy: Optional[str] = None
-        self.activated: Optional[bool] = None
-        self.operation: Optional[str] = None
-        self.values: Optional[List[str]] = None
+        self.strategy: str
+        self.activated: bool
+        self.operation: str
+        self.values: list[str]
 
 class Relay:
     def __init__(self):
-        self.type: Optional[str] = None
-        self.activated: Optional[bool] = None
+        self.type: str
+        self.activated: bool
 
 class Snapshot:
     def __init__(self, json_data: dict):
@@ -53,8 +61,8 @@ def _parse_domain(self, domain_data: dict) -> Domain:
         """ Parse domain data from JSON """
 
         domain = Domain()
-        domain.name = domain_data.get('name')
-        domain.activated = domain_data.get('activated')
+        domain.name = domain_data.get('name', '')
+        domain.activated = domain_data.get('activated', False)
         domain.version = domain_data.get('version', 0)
 
         if 'group' in domain_data and domain_data['group']:
@@ -68,9 +76,9 @@ def _parse_group(self, group_data: dict) -> Group:
         """ Parse group data from JSON """
 
         group = Group()
-        group.name = group_data.get('name')
-        group.activated = group_data.get('activated')
-        
+        group.name = group_data.get('name', '')
+        group.activated = group_data.get('activated', False)
+
         if 'config' in group_data and group_data['config']:
             group.config = []
             for config_data in group_data['config']:
@@ -82,9 +90,9 @@ def _parse_config(self, config_data: dict) -> Config:
         """ Parse config data from JSON """
 
         config = Config()
-        config.key = config_data.get('key')
-        config.activated = config_data.get('activated')
-        
+        config.key = config_data.get('key', '')
+        config.activated = config_data.get('activated', False)
+
         if 'strategies' in config_data and config_data['strategies']:
             config.strategies = []
             for strategy_data in config_data['strategies']:
@@ -99,20 +107,20 @@ def _parse_strategy(self, strategy_data: dict) -> StrategyConfig:
         """ Parse strategy data from JSON """
 
         strategy = StrategyConfig()
-        strategy.strategy = strategy_data.get('strategy')
-        strategy.activated = strategy_data.get('activated')
-        strategy.operation = strategy_data.get('operation')
-        strategy.values = strategy_data.get('values')
+        strategy.strategy = strategy_data.get('strategy', '')
+        strategy.activated = strategy_data.get('activated', False)
+        strategy.operation = strategy_data.get('operation', '')
+        strategy.values = strategy_data.get('values', [])
 
         return strategy
 
     def _parse_relay(self, relay_data: dict) -> Relay:
         """ Parse relay data from JSON """
 
         relay = Relay()
-        relay.type = relay_data.get('type')
-        relay.activated = relay_data.get('activated')
-        
+        relay.type = relay_data.get('type', '')
+        relay.activated = relay_data.get('activated', False)
+
         return relay
 
     def to_dict(self) -> dict:

switcher_client/switcher.py

@@ -1,16 +1,19 @@
 from typing import Optional
 
 from switcher_client.lib.globals.global_context import Context
+from switcher_client.lib.globals.global_snapshot import GlobalSnapshot
 from switcher_client.lib.remote_auth import RemoteAuth
 from switcher_client.lib.globals import GlobalAuth
 from switcher_client.lib.remote import Remote
+from switcher_client.lib.resolver import Resolver
 from switcher_client.lib.types import ResultDetail
 from switcher_client.switcher_data import SwitcherData
 
 class Switcher(SwitcherData):
     def __init__(self, context: Context, key: Optional[str] = None):
         super().__init__(key) 
         self._context = context
+        self.__validate_args(key)
 
     def prepare(self, key: Optional[str] = None):
         """ Checks API credentials and connectivity """
@@ -30,6 +33,9 @@ def is_on_with_details(self, key: Optional[str] = None) -> ResultDetail:
 
     def __submit(self) -> ResultDetail:
         """ Submit criteria for execution (local or remote) """
+        if (self._context.options.local):
+            return self.__execute_local_criteria()
+
         self.validate()
         response = self.__execute_remote_criteria()
 
@@ -69,4 +75,8 @@ def __execute_api_checks(self):
     def __execute_remote_criteria(self):
         """ Execute remote criteria """
         token = GlobalAuth.get_token()
-        return Remote.check_criteria(token, self._context, self)
+        return Remote.check_criteria(token, self._context, self)
+
+    def __execute_local_criteria(self):
+        """ Execute local criteria """
+        return Resolver.check_criteria(GlobalSnapshot.snapshot(), self)

tests/snapshots/default_disabled.json

@@ -0,0 +1,11 @@
+{
+    "data": {
+        "domain": {
+            "name": "Business",
+            "description": "Business description",
+            "activated": false,
+            "version": 1,
+            "group": []
+        }
+    }
+}

tests/test_switcher_local.py

@@ -0,0 +1,95 @@
+from switcher_client.client import Client, ContextOptions
+from switcher_client.errors import LocalCriteriaError
+
+def test_local():
+    """ Should use local Snapshot to evaluate the switcher """
+
+    # given
+    given_context('tests/snapshots')
+    snapshot_version = Client.load_snapshot()
+
+    switcher = Client.get_switcher()
+
+    # test
+    assert snapshot_version == 1
+    assert switcher.is_on('FF2FOR2022')
+
+def test_local_domain_disabled():
+    """ Should return disabled when domain is deactivated """
+
+    # given
+    given_context('tests/snapshots', environment='default_disabled')
+    snapshot_version = Client.load_snapshot()
+
+    switcher = Client.get_switcher()
+
+    # test
+    assert snapshot_version == 1
+    assert switcher.is_on('FEATURE') is False
+
+def test_local_group_disabled():
+    """ Should return disabled when group is deactivated """
+
+    # given
+    given_context('tests/snapshots')
+    snapshot_version = Client.load_snapshot()
+
+    switcher = Client.get_switcher()
+
+    # test
+    assert snapshot_version == 1
+    assert switcher.is_on('FF2FOR2040') is False
+
+def test_local_config_disabled():
+    """ Should return disabled when config is deactivated """
+
+    # given
+    given_context('tests/snapshots')
+    snapshot_version = Client.load_snapshot()
+
+    switcher = Client.get_switcher()
+
+    # test
+    assert snapshot_version == 1
+    assert switcher.is_on('FF2FOR2031') is False
+
+def test_local_no_key_found():
+    """ Should raise an error when no key is found in the snapshot """
+
+    # given
+    given_context('tests/snapshots')
+    snapshot_version = Client.load_snapshot()
+
+    switcher = Client.get_switcher()
+
+    # test
+    assert snapshot_version == 1
+    try:
+        switcher.is_on('INVALID_KEY')
+    except LocalCriteriaError as e:
+        assert str(e) == "Config with key 'INVALID_KEY' not found in the snapshot"
+
+def test_local_no_snapshot():
+    """ Should raise an error when no snapshot is loaded """
+
+    # given
+    given_context('tests/invalid_location')
+    switcher = Client.get_switcher()
+
+    # test
+    try:
+        switcher.is_on('FF2FOR2022')
+    except LocalCriteriaError as e:
+        assert str(e) == "Snapshot not loaded. Try to use 'Client.load_snapshot()'"
+
+# Helpers
+
+def given_context(snapshot_location: str, environment: str = 'default') -> None:
+    Client.build_context(
+        domain='Playground',
+        environment=environment,
+        options=ContextOptions(
+            local=True,
+            snapshot_location=snapshot_location
+        )
+    )