From 963c3c754a9cd6d3a1041bd9cf1c6c71a4ae48e8 Mon Sep 17 00:00:00 2001 From: petruki <31597636+petruki@users.noreply.github.com> Date: Sun, 28 Dec 2025 15:36:01 -0800 Subject: [PATCH 1/2] feat: integrated snapshot processors to Switcher resolver --- .coveragerc | 1 + switcher_client/client.py | 27 ++++++--- switcher_client/errors/__init__.py | 9 ++- switcher_client/lib/globals/global_auth.py | 2 - switcher_client/lib/globals/global_context.py | 1 - .../lib/globals/global_snapshot.py | 2 +- switcher_client/lib/remote.py | 31 +++------- switcher_client/lib/remote_auth.py | 8 +-- switcher_client/lib/resolver.py | 48 ++++++++++++++- switcher_client/lib/snapshot.py | 16 ++--- switcher_client/lib/snapshot_auto_updater.py | 1 + switcher_client/lib/snapshot_loader.py | 8 +-- switcher_client/lib/types.py | 11 ++++ switcher_client/lib/utils/__init__.py | 12 ++++ switcher_client/lib/utils/execution_logger.py | 3 +- switcher_client/lib/utils/payload_reader.py | 2 +- .../lib/utils/timed_match/timed_match.py | 4 +- switcher_client/switcher.py | 18 +++--- switcher_client/switcher_data.py | 13 +++- tests/strategy-operations/test_date.py | 17 +++--- tests/strategy-operations/test_network.py | 17 +++--- tests/strategy-operations/test_numeric.py | 18 +++--- tests/strategy-operations/test_payload.py | 18 +++--- tests/strategy-operations/test_regex.py | 18 +++--- tests/strategy-operations/test_time.py | 18 +++--- tests/strategy-operations/test_value.py | 18 +++--- tests/test_switcher_local.py | 59 +++++++++++++++++++ 27 files changed, 272 insertions(+), 128 deletions(-) diff --git a/.coveragerc b/.coveragerc index 05d3abb..153576a 100644 --- a/.coveragerc +++ b/.coveragerc @@ -1,6 +1,7 @@ [run] omit = switcher_client/lib/utils/timed_match/worker.py + switcher_client/version.py */tests/* */test_* */__init__.py diff --git a/switcher_client/client.py b/switcher_client/client.py index 9ac6921..b07703c 100644 --- a/switcher_client/client.py +++ b/switcher_client/client.py @@ -1,14 +1,15 @@ from typing import Optional, Callable -from switcher_client.lib.globals.global_snapshot import GlobalSnapshot, LoadSnapshotOptions -from switcher_client.lib.remote_auth import RemoteAuth -from switcher_client.lib.globals.global_context import Context, ContextOptions -from switcher_client.lib.globals.global_context import DEFAULT_ENVIRONMENT -from switcher_client.lib.snapshot_auto_updater import SnapshotAutoUpdater -from switcher_client.lib.snapshot_loader import load_domain, validate_snapshot, save_snapshot -from switcher_client.lib.utils.execution_logger import ExecutionLogger -from switcher_client.lib.utils import get -from switcher_client.switcher import Switcher +from .lib.globals.global_snapshot import GlobalSnapshot, LoadSnapshotOptions +from .lib.remote_auth import RemoteAuth +from .lib.globals.global_context import Context, ContextOptions +from .lib.globals.global_context import DEFAULT_ENVIRONMENT +from .lib.snapshot_auto_updater import SnapshotAutoUpdater +from .lib.snapshot_loader import 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 +from .switcher import Switcher class SwitcherOptions: SNAPSHOT_AUTO_UPDATE_INTERVAL = 'snapshot_auto_update_interval' @@ -146,6 +147,14 @@ def get_execution(switcher: Switcher) -> ExecutionLogger: def clear_logger() -> None: """Clear all logged executions""" ExecutionLogger.clear_logger() + + @staticmethod + def clear_resources() -> None: + """ Clear all resources used by the Client """ + Client.terminate_snapshot_auto_update() + ExecutionLogger.clear_logger() + GlobalSnapshot.clear() + TimedMatch.terminate_worker() @staticmethod def _is_check_snapshot_available(fetch_remote = False) -> bool: diff --git a/switcher_client/errors/__init__.py b/switcher_client/errors/__init__.py index 68bfe7b..2f70b77 100644 --- a/switcher_client/errors/__init__.py +++ b/switcher_client/errors/__init__.py @@ -14,4 +14,11 @@ def __init__(self, message): class LocalCriteriaError(Exception): def __init__(self, message): self.message = message - super().__init__(self.message) \ No newline at end of file + super().__init__(self.message) + +__all__ = [ + 'RemoteError', + 'RemoteAuthError', + 'RemoteCriteriaError', + 'LocalCriteriaError', +] \ No newline at end of file diff --git a/switcher_client/lib/globals/global_auth.py b/switcher_client/lib/globals/global_auth.py index 5bcd4e8..393a6de 100644 --- a/switcher_client/lib/globals/global_auth.py +++ b/switcher_client/lib/globals/global_auth.py @@ -1,5 +1,3 @@ -from typing import Optional - class GlobalAuth: __token = None __exp = None diff --git a/switcher_client/lib/globals/global_context.py b/switcher_client/lib/globals/global_context.py index 830e7e2..7274d78 100644 --- a/switcher_client/lib/globals/global_context.py +++ b/switcher_client/lib/globals/global_context.py @@ -1,6 +1,5 @@ from typing import Optional - DEFAULT_ENVIRONMENT = 'default' DEFAULT_LOCAL = False diff --git a/switcher_client/lib/globals/global_snapshot.py b/switcher_client/lib/globals/global_snapshot.py index df93c59..88aed97 100644 --- a/switcher_client/lib/globals/global_snapshot.py +++ b/switcher_client/lib/globals/global_snapshot.py @@ -1,4 +1,4 @@ -from switcher_client.lib.types import Snapshot +from ...lib.types import Snapshot class GlobalSnapshot: diff --git a/switcher_client/lib/remote.py b/switcher_client/lib/remote.py index a968fb4..f04b012 100644 --- a/switcher_client/lib/remote.py +++ b/switcher_client/lib/remote.py @@ -3,12 +3,11 @@ from typing import Optional -from switcher_client.errors import RemoteAuthError, RemoteError -from switcher_client.errors import RemoteCriteriaError -from switcher_client.lib.globals.global_context import DEFAULT_ENVIRONMENT, Context -from switcher_client.lib.types import ResultDetail -from switcher_client.lib.utils import get -from switcher_client.switcher_data import SwitcherData +from ..errors import RemoteAuthError, RemoteError, RemoteCriteriaError +from ..lib.globals.global_context import DEFAULT_ENVIRONMENT, Context +from ..lib.types import ResultDetail +from ..lib.utils import get, get_entry +from ..switcher_data import SwitcherData class Remote: _client: Optional[httpx.Client] = None @@ -33,8 +32,8 @@ def auth(context: Context): @staticmethod def check_criteria(token: Optional[str], context: Context, switcher: SwitcherData) -> ResultDetail: url = f'{context.url}/criteria?showReason={str(switcher._show_details).lower()}&key={switcher._key}' - entry = Remote._get_entry(switcher._input) - response = Remote._do_post(url, entry, Remote._get_header(token)) + entry = get_entry(switcher._input) + response = Remote._do_post(url, { 'entry': [e.to_dict() for e in entry] }, Remote._get_header(token)) if response.status_code == 200: json_response = response.json() @@ -115,18 +114,4 @@ def _get_header(token: Optional[str]): return { 'Authorization': f'Bearer {token}', 'Content-Type': 'application/json', - } - - @staticmethod - def _get_entry(input: list): - entry = [] - for strategy_type, input_value in input: - entry.append({ - 'strategy': strategy_type, - 'input': input_value - }) - - if not entry: - return None - - return {'entry': entry} \ No newline at end of file + } \ No newline at end of file diff --git a/switcher_client/lib/remote_auth.py b/switcher_client/lib/remote_auth.py index c4800c4..70d3cee 100644 --- a/switcher_client/lib/remote_auth.py +++ b/switcher_client/lib/remote_auth.py @@ -1,8 +1,8 @@ from time import time -from switcher_client.lib.remote import Remote -from switcher_client.lib.globals.global_context import Context -from switcher_client.lib.globals import GlobalAuth +from .remote import Remote +from .globals.global_context import Context +from .globals import GlobalAuth class RemoteAuth: __context: Context = Context.empty() @@ -42,5 +42,3 @@ def is_valid(): errors = [name for name, value in required_fields if not value] if errors: raise ValueError(f"Something went wrong: Missing or empty required fields ({', '.join(errors)})") - - diff --git a/switcher_client/lib/resolver.py b/switcher_client/lib/resolver.py index 60bd85a..60fb354 100644 --- a/switcher_client/lib/resolver.py +++ b/switcher_client/lib/resolver.py @@ -1,11 +1,17 @@ +from typing import Optional + +from switcher_client.lib.types import Config, Domain, Entry, Group, ResultDetail, Snapshot, StrategyConfig +from switcher_client.lib.snapshot import process_operation +from switcher_client.lib.utils import get, get_entry +from switcher_client.lib.utils import get_entry from switcher_client.errors import LocalCriteriaError -from switcher_client.lib.types import Config, Domain, Group, ResultDetail, Snapshot from switcher_client.switcher_data import SwitcherData class Resolver: @staticmethod def check_criteria(snapshot: Snapshot | None, switcher: SwitcherData) -> ResultDetail: + """ Resolves the criteria for a given switcher request against the snapshot domain. """ if not snapshot: raise LocalCriteriaError("Snapshot not loaded. Try to use 'Client.load_snapshot()'") @@ -13,6 +19,7 @@ def check_criteria(snapshot: Snapshot | None, switcher: SwitcherData) -> ResultD @staticmethod def _check_domain(domain: Domain, switcher: SwitcherData) -> ResultDetail: + """ Checks if the domain is activated and proceeds to check groups. """ if domain.activated is False: return ResultDetail.disabled("Domain is disabled") @@ -20,6 +27,7 @@ def _check_domain(domain: Domain, switcher: SwitcherData) -> ResultDetail: @staticmethod def _check_group(groups: list[Group], switcher: SwitcherData) -> ResultDetail: + """ Finds the correct config in the groups and checks it. """ key = switcher._key for group in groups: @@ -35,7 +43,43 @@ def _check_group(groups: list[Group], switcher: SwitcherData) -> ResultDetail: @staticmethod def _check_config(config: Config, switcher: SwitcherData) -> ResultDetail: + """ Checks if the config is activated and proceeds to check strategies. """ if config.activated is False: return ResultDetail.disabled("Config disabled") - return ResultDetail.success() \ No newline at end of file + if config.strategies is not None and len(config.strategies) > 0: + return Resolver._check_strategy(config.strategies, switcher._input) + + return ResultDetail.success() + + @staticmethod + def _check_strategy(strategy_configs: list[StrategyConfig], inputs: list[list[str]]) -> ResultDetail: + """ Checks each strategy configuration against the provided inputs. """ + entry = get_entry(get(inputs, [])) + + for strategy_config in strategy_configs: + if not strategy_config.activated: + continue + + strategy_result = Resolver._check_strategy_config(strategy_config, entry) + if strategy_result is not None: + return strategy_result + + return ResultDetail.success() + + @staticmethod + def _check_strategy_config(strategy_config: StrategyConfig, entry: list[Entry]) -> Optional[ResultDetail]: + """ Checks a single strategy configuration against the provided entries. """ + if len(entry) == 0: + return ResultDetail.disabled(f"Strategy '{strategy_config.strategy}' did not receive any input") + + strategy_entry = [e for e in entry if e.strategy == strategy_config.strategy] + if not Resolver._is_strategy_fulfilled(strategy_entry, strategy_config): + return ResultDetail.disabled(f"Strategy '{strategy_config.strategy}' does not agree") + + return None + + @staticmethod + def _is_strategy_fulfilled(strategy_entry: list[Entry], strategy_config: StrategyConfig) -> bool: + """ Determines if the strategy conditions are fulfilled based on the entries and configuration. """ + return len(strategy_entry) > 0 and process_operation(strategy_config, strategy_entry[0].input) is True \ No newline at end of file diff --git a/switcher_client/lib/snapshot.py b/switcher_client/lib/snapshot.py index 8bee3f9..b0fff1c 100644 --- a/switcher_client/lib/snapshot.py +++ b/switcher_client/lib/snapshot.py @@ -4,6 +4,8 @@ from typing import Optional from datetime import datetime +from switcher_client.lib.types import StrategyConfig + from .utils.payload_reader import parse_json, payload_reader from .utils.ipcidr import IPCIDR from .utils.timed_match import TimedMatch @@ -13,9 +15,9 @@ class StrategiesType(Enum): NUMERIC = "NUMERIC_VALIDATION" DATE = "DATE_VALIDATION" TIME = "TIME_VALIDATION" - PAYLOAD = "PAYLOAD" - NETWORK = "NETWORK" - REGEX = "REGEX" + PAYLOAD = "PAYLOAD_VALIDATION" + NETWORK = "NETWORK_VALIDATION" + REGEX = "REGEX_VALIDATION" class OperationsType(Enum): EXIST = "EXIST" @@ -28,12 +30,12 @@ class OperationsType(Enum): HAS_ONE = "HAS_ONE" HAS_ALL = "HAS_ALL" -def process_operation(strategy_config: dict, input_value: str) -> Optional[bool]: +def process_operation(strategy_config: StrategyConfig, input_value: str) -> Optional[bool]: """Process the operation based on strategy configuration and input value.""" - strategy = strategy_config.get('strategy') - operation = strategy_config.get('operation', '') - values = strategy_config.get('values', []) + strategy = strategy_config.strategy + operation = strategy_config.operation + values = strategy_config.values match strategy: case StrategiesType.VALUE.value: diff --git a/switcher_client/lib/snapshot_auto_updater.py b/switcher_client/lib/snapshot_auto_updater.py index ce6d064..df6aa7d 100644 --- a/switcher_client/lib/snapshot_auto_updater.py +++ b/switcher_client/lib/snapshot_auto_updater.py @@ -1,5 +1,6 @@ import threading import time + from typing import Callable, Optional class SnapshotAutoUpdater: diff --git a/switcher_client/lib/snapshot_loader.py b/switcher_client/lib/snapshot_loader.py index d607d1b..f4cf72a 100644 --- a/switcher_client/lib/snapshot_loader.py +++ b/switcher_client/lib/snapshot_loader.py @@ -1,10 +1,10 @@ import json import os -from switcher_client.lib.globals.global_auth import GlobalAuth -from switcher_client.lib.globals.global_context import Context -from switcher_client.lib.remote import Remote -from switcher_client.lib.types import Snapshot +from .globals.global_auth import GlobalAuth +from .globals.global_context import Context +from .remote import Remote +from .types import Snapshot def load_domain(snapshot_location: str, environment: str): """ Load Domain from snapshot file """ diff --git a/switcher_client/lib/types.py b/switcher_client/lib/types.py index cbb1f93..8c8733c 100644 --- a/switcher_client/lib/types.py +++ b/switcher_client/lib/types.py @@ -41,6 +41,17 @@ def __init__(self): self.operation: str self.values: list[str] +class Entry: + def __init__(self, strategy: str, input: str): + self.strategy = strategy + self.input = input + + def to_dict(self) -> dict: + return { + 'strategy': self.strategy, + 'input': self.input + } + class Relay: def __init__(self): self.type: str diff --git a/switcher_client/lib/utils/__init__.py b/switcher_client/lib/utils/__init__.py index 22d8ec1..4828e3a 100644 --- a/switcher_client/lib/utils/__init__.py +++ b/switcher_client/lib/utils/__init__.py @@ -1,10 +1,22 @@ +from typing import Optional + +from switcher_client.lib.types import Entry from .execution_logger import ExecutionLogger def get(value, default_value): """ Return value if not None, otherwise return default_value """ return value if value is not None else default_value +def get_entry(input: list) -> list[Entry]: + """ Prepare entry dictionary from input strategy handling """ + entry: list[Entry] = [] + for strategy_type, input_value in input: + entry.append(Entry(strategy_type, input_value)) + + return entry + __all__ = [ 'ExecutionLogger', + 'get_entry', 'get', ] diff --git a/switcher_client/lib/utils/execution_logger.py b/switcher_client/lib/utils/execution_logger.py index 2ec952f..da7ed43 100644 --- a/switcher_client/lib/utils/execution_logger.py +++ b/switcher_client/lib/utils/execution_logger.py @@ -1,5 +1,6 @@ from typing import Optional, Callable, List -from switcher_client.lib.types import ResultDetail + +from ...lib.types import ResultDetail # Global logger storage _logger: List['ExecutionLogger'] = [] diff --git a/switcher_client/lib/utils/payload_reader.py b/switcher_client/lib/utils/payload_reader.py index 1f770a8..c61f07e 100644 --- a/switcher_client/lib/utils/payload_reader.py +++ b/switcher_client/lib/utils/payload_reader.py @@ -1,6 +1,6 @@ import json -from typing import Any, List, Union +from typing import Any, List, Union def payload_reader(payload: Any) -> List[str]: """Extract all field keys from a JSON payload structure. diff --git a/switcher_client/lib/utils/timed_match/timed_match.py b/switcher_client/lib/utils/timed_match/timed_match.py index 2a2818c..be164b0 100644 --- a/switcher_client/lib/utils/timed_match/timed_match.py +++ b/switcher_client/lib/utils/timed_match/timed_match.py @@ -1,12 +1,10 @@ import multiprocessing -import signal -import os import time from typing import List, Optional, Any from dataclasses import dataclass -from switcher_client.lib.utils.timed_match.worker import TaskType, WorkerResult, WorkerTask, persistent_regex_worker +from .worker import TaskType, WorkerResult, WorkerTask, persistent_regex_worker # Default constants DEFAULT_REGEX_MAX_TIME_LIMIT = 3000 # 3 seconds in milliseconds diff --git a/switcher_client/switcher.py b/switcher_client/switcher.py index 1a93361..6b3c5ea 100644 --- a/switcher_client/switcher.py +++ b/switcher_client/switcher.py @@ -1,14 +1,14 @@ 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.lib.utils.execution_logger import ExecutionLogger -from switcher_client.switcher_data import SwitcherData +from .lib.globals.global_context import Context +from .lib.globals.global_snapshot import GlobalSnapshot +from .lib.remote_auth import RemoteAuth +from .lib.globals import GlobalAuth +from .lib.remote import Remote +from .lib.resolver import Resolver +from .lib.types import ResultDetail +from .lib.utils.execution_logger import ExecutionLogger +from .switcher_data import SwitcherData class Switcher(SwitcherData): def __init__(self, context: Context, key: Optional[str] = None): diff --git a/switcher_client/switcher_data.py b/switcher_client/switcher_data.py index 31fa75a..b7460e8 100644 --- a/switcher_client/switcher_data.py +++ b/switcher_client/switcher_data.py @@ -2,8 +2,7 @@ from abc import ABCMeta from typing import Optional, Self -# Strategy types -VALUE_VALIDATION = 'VALUE_VALIDATION' +from .lib.snapshot import StrategiesType class SwitcherData(metaclass=ABCMeta): def __init__(self, key: Optional[str] = None): @@ -21,7 +20,15 @@ def check(self, strategy_type: str, input: str)-> Self: def check_value(self, input: str) -> Self: """ Adds VALUE_VALIDATION input for strategy validation """ - return self.check(VALUE_VALIDATION, input) + return self.check(StrategiesType.VALUE.value, input) + + def check_network(self, input: str) -> Self: + """ Adds NETWORK_VALIDATION input for strategy validation """ + return self.check(StrategiesType.NETWORK.value, input) + + def check_regex(self, input: str) -> Self: + """ Adds REGEX_VALIDATION input for strategy validation """ + return self.check(StrategiesType.REGEX.value, input) def throttle(self, period: int) -> Self: """ Sets throttle period in milliseconds """ diff --git a/tests/strategy-operations/test_date.py b/tests/strategy-operations/test_date.py index de4f762..ef7c625 100644 --- a/tests/strategy-operations/test_date.py +++ b/tests/strategy-operations/test_date.py @@ -1,7 +1,8 @@ import pytest -from typing import Dict, List, Any +from typing import List from switcher_client.lib.snapshot import OperationsType, StrategiesType, process_operation +from switcher_client.lib.types import StrategyConfig class TestDateStrategy: """Test suite for Strategy [DATE] tests.""" @@ -21,13 +22,13 @@ def mock_values3(self) -> List[str]: """Date with time component mock data.""" return ['2019-12-01T08:30'] - def given_strategy_config(self, operation: str, values: List[str]) -> Dict[str, Any]: - return { - 'strategy': StrategiesType.DATE.value, - 'operation': operation, - 'values': values, - 'activated': True - } + def given_strategy_config(self, operation: str, values: List[str]) -> StrategyConfig: + strategy_config = StrategyConfig() + strategy_config.strategy = StrategiesType.DATE.value + strategy_config.operation = operation + strategy_config.values = values + strategy_config.activated = True + return strategy_config def test_should_agree_when_input_is_lower(self, mock_values1): """Should agree when input is LOWER.""" diff --git a/tests/strategy-operations/test_network.py b/tests/strategy-operations/test_network.py index 0cda9fd..acfb87f 100644 --- a/tests/strategy-operations/test_network.py +++ b/tests/strategy-operations/test_network.py @@ -1,7 +1,8 @@ import pytest -from typing import Dict, List, Any +from typing import List from switcher_client.lib.snapshot import OperationsType, StrategiesType, process_operation +from switcher_client.lib.types import StrategyConfig class TestNetworkStrategy: """Test suite for Strategy [NETWORK] tests.""" @@ -21,14 +22,14 @@ def mock_values3(self) -> List[str]: """Multiple IP addresses mock data.""" return ['192.168.56.56', '192.168.56.57', '192.168.56.58'] - def given_strategy_config(self, operation: str, values: List[str]) -> Dict[str, Any]: + def given_strategy_config(self, operation: str, values: List[str]) -> StrategyConfig: """Create a strategy configuration for NETWORK strategy.""" - return { - 'strategy': StrategiesType.NETWORK.value, - 'operation': operation, - 'values': values, - 'activated': True - } + strategy_config = StrategyConfig() + strategy_config.strategy = StrategiesType.NETWORK.value + strategy_config.operation = operation + strategy_config.values = values + strategy_config.activated = True + return strategy_config def test_should_agree_when_input_range_exist(self, mock_values1): """Should agree when input range EXIST.""" diff --git a/tests/strategy-operations/test_numeric.py b/tests/strategy-operations/test_numeric.py index 33e3f50..fe6040a 100644 --- a/tests/strategy-operations/test_numeric.py +++ b/tests/strategy-operations/test_numeric.py @@ -1,7 +1,8 @@ import pytest -from typing import Dict, List, Any +from typing import List from switcher_client.lib.snapshot import OperationsType, StrategiesType, process_operation +from switcher_client.lib.types import StrategyConfig class TestNumericStrategy: """Test suite for Strategy [NUMERIC] tests.""" @@ -21,13 +22,14 @@ def mock_values3(self) -> List[str]: """Decimal numeric value mock data.""" return ['1.5'] - def given_strategy_config(self, operation: str, values: List[str]) -> Dict[str, Any]: - return { - 'strategy': StrategiesType.NUMERIC.value, - 'operation': operation, - 'values': values, - 'activated': True - } + def given_strategy_config(self, operation: str, values: List[str]) -> StrategyConfig: + """Create a strategy configuration for NUMERIC strategy.""" + strategy_config = StrategyConfig() + strategy_config.strategy = StrategiesType.NUMERIC.value + strategy_config.operation = operation + strategy_config.values = values + strategy_config.activated = True + return strategy_config def test_should_agree_when_input_exist_in_values_string_type(self, mock_values2): """Should agree when input EXIST in values - String type.""" diff --git a/tests/strategy-operations/test_payload.py b/tests/strategy-operations/test_payload.py index 9ed637f..f501a04 100644 --- a/tests/strategy-operations/test_payload.py +++ b/tests/strategy-operations/test_payload.py @@ -1,7 +1,8 @@ import json import pytest -from typing import Dict, List, Any +from typing import List +from switcher_client.lib.types import StrategyConfig from switcher_client.lib.utils.payload_reader import payload_reader from switcher_client.lib.snapshot import OperationsType, StrategiesType, process_operation @@ -51,13 +52,14 @@ def fixture_values3(self) -> str: 'env': 'default' }) - def given_strategy_config(self, operation: str, values: List[str]) -> Dict[str, Any]: - return { - 'strategy': StrategiesType.PAYLOAD.value, - 'operation': operation, - 'values': values, - 'activated': True - } + def given_strategy_config(self, operation: str, values: List[str]) -> StrategyConfig: + """Create a strategy configuration for PAYLOAD strategy.""" + strategy_config = StrategyConfig() + strategy_config.strategy = StrategiesType.PAYLOAD.value + strategy_config.operation = operation + strategy_config.values = values + strategy_config.activated = True + return strategy_config def test_should_read_keys_from_payload_1(self, fixture_values2): """Should read keys from payload #1.""" diff --git a/tests/strategy-operations/test_regex.py b/tests/strategy-operations/test_regex.py index 6209eb6..1121668 100644 --- a/tests/strategy-operations/test_regex.py +++ b/tests/strategy-operations/test_regex.py @@ -1,7 +1,8 @@ import pytest -from typing import Dict, List, Any +from typing import List from switcher_client.lib.snapshot import OperationsType, StrategiesType, process_operation +from switcher_client.lib.types import StrategyConfig from switcher_client.lib.utils.timed_match import TimedMatch class TestRegexStrategy: @@ -37,13 +38,14 @@ def mock_values3(self) -> List[str]: """Simple regex pattern without word boundaries.""" return ['USER_[0-9]{1,2}'] - def given_strategy_config(self, operation: str, values: List[str]) -> Dict[str, Any]: - return { - 'strategy': StrategiesType.REGEX.value, - 'operation': operation, - 'values': values, - 'activated': True - } + def given_strategy_config(self, operation: str, values: List[str]) -> StrategyConfig: + """Create a strategy configuration for REGEX strategy.""" + strategy_config = StrategyConfig() + strategy_config.strategy = StrategiesType.REGEX.value + strategy_config.operation = operation + strategy_config.values = values + strategy_config.activated = True + return strategy_config def test_should_agree_when_expect_to_exist_using_exist_operation(self, mock_values1, mock_values2): """Should agree when expect to exist using EXIST operation.""" diff --git a/tests/strategy-operations/test_time.py b/tests/strategy-operations/test_time.py index be4c4d2..b69d95d 100644 --- a/tests/strategy-operations/test_time.py +++ b/tests/strategy-operations/test_time.py @@ -1,7 +1,8 @@ import pytest -from typing import Dict, List, Any +from typing import List from switcher_client.lib.snapshot import OperationsType, StrategiesType, process_operation +from switcher_client.lib.types import StrategyConfig class TestTimeStrategy: """Test suite for Strategy [TIME] tests.""" @@ -16,13 +17,14 @@ def mock_values2(self) -> List[str]: """Multiple time values for BETWEEN operations.""" return ['08:00', '10:00'] - def given_strategy_config(self, operation: str, values: List[str]) -> Dict[str, Any]: - return { - 'strategy': StrategiesType.TIME.value, - 'operation': operation, - 'values': values, - 'activated': True - } + def given_strategy_config(self, operation: str, values: List[str]) -> StrategyConfig: + """Create a strategy configuration for TIME strategy.""" + strategy_config = StrategyConfig() + strategy_config.strategy = StrategiesType.TIME.value + strategy_config.operation = operation + strategy_config.values = values + strategy_config.activated = True + return strategy_config def test_should_agree_when_input_is_lower(self, mock_values1): """Should agree when input is LOWER.""" diff --git a/tests/strategy-operations/test_value.py b/tests/strategy-operations/test_value.py index 0a2c39a..c8e8d5b 100644 --- a/tests/strategy-operations/test_value.py +++ b/tests/strategy-operations/test_value.py @@ -1,7 +1,8 @@ import pytest -from typing import Dict, List, Any +from typing import List from switcher_client.lib.snapshot import OperationsType, StrategiesType, process_operation +from switcher_client.lib.types import StrategyConfig class TestValueStrategy: """Test suite for Strategy [VALUE] tests.""" @@ -16,13 +17,14 @@ def mock_values2(self) -> List[str]: """Multiple users mock data.""" return ['USER_1', 'USER_2'] - def given_strategy_config(self, operation: str, values: List[str]) -> Dict[str, Any]: - return { - 'strategy': StrategiesType.VALUE.value, - 'operation': operation, - 'values': values, - 'activated': True - } + def given_strategy_config(self, operation: str, values: List[str]) -> StrategyConfig: + """Create a strategy configuration for VALUE strategy.""" + strategy_config = StrategyConfig() + strategy_config.strategy = StrategiesType.VALUE.value + strategy_config.operation = operation + strategy_config.values = values + strategy_config.activated = True + return strategy_config def test_should_agree_when_input_exist(self, mock_values1): """Should agree when input EXIST.""" diff --git a/tests/test_switcher_local.py b/tests/test_switcher_local.py index 34d8f8a..a24ebae 100644 --- a/tests/test_switcher_local.py +++ b/tests/test_switcher_local.py @@ -1,5 +1,6 @@ from switcher_client.client import Client, ContextOptions from switcher_client.errors import LocalCriteriaError +from switcher_client.lib.utils.timed_match.timed_match import TimedMatch def test_local(): """ Should use local Snapshot to evaluate the switcher """ @@ -14,6 +15,51 @@ def test_local(): assert snapshot_version == 1 assert switcher.is_on('FF2FOR2022') +def test_local_with_strategy(): + """ Should use local Snapshot to evaluate the switcher with strategy """ + + # given + given_context('tests/snapshots') + Client.load_snapshot() + + switcher = Client.get_switcher() + + # test + assert switcher \ + .check_value('Japan') \ + .check_network('10.0.0.3') \ + .is_on('FF2FOR2020') + +def test_local_with_strategy_no_input(): + """ Should return disabled when no input is provided for the strategy """ + + # given + given_context('tests/snapshots') + Client.load_snapshot() + + switcher = Client.get_switcher() + + # test + assert switcher.is_on('FF2FOR2020') is False + +def test_local_with_strategy_no_matching_input(): + """ Should return disabled when no matching input is provided for the strategy """ + + # given + TimedMatch.set_max_time_limit(100) + given_context('tests/snapshots') + Client.load_snapshot() + + switcher = Client.get_switcher() + + # test + assert switcher \ + .check_regex('123') \ + .is_on('FF2FOR2024') is False + + # teardown + Client.clear_resources() + def test_local_domain_disabled(): """ Should return disabled when domain is deactivated """ @@ -53,6 +99,19 @@ def test_local_config_disabled(): assert snapshot_version == 1 assert switcher.is_on('FF2FOR2031') is False +def test_local_strategy_disabled(): + """ Should return disabled when strategy is deactivated """ + + # given + given_context('tests/snapshots') + snapshot_version = Client.load_snapshot() + + switcher = Client.get_switcher() + + # test + assert snapshot_version == 1 + assert switcher.check_network('10.0.0.3').is_on('FF2FOR2021') + def test_local_no_key_found(): """ Should raise an error when no key is found in the snapshot """ From 9ce9f6d4ea7858d4b7f3cdd6b88b13a126a519cc Mon Sep 17 00:00:00 2001 From: petruki <31597636+petruki@users.noreply.github.com> Date: Sun, 28 Dec 2025 15:41:25 -0800 Subject: [PATCH 2/2] chore: removed init from cov omit --- .coveragerc | 1 - 1 file changed, 1 deletion(-) diff --git a/.coveragerc b/.coveragerc index 153576a..1944e48 100644 --- a/.coveragerc +++ b/.coveragerc @@ -4,7 +4,6 @@ omit = switcher_client/version.py */tests/* */test_* - */__init__.py [report] exclude_lines =