From 11b2e0db7ac443fb58ddf030a572b792a28c622e Mon Sep 17 00:00:00 2001 From: Alexandre Catarino Date: Fri, 14 Aug 2026 23:50:01 +0100 Subject: [PATCH] feature: validate configuration values with a regex from the modules json Adds an optional "input-regex" and "input-regex-message" to the module configurations, so a module describes the values it accepts instead of the CLI hard coding them. The regex is checked wherever a value enters the CLI: the command line options, the interactive prompts and the values read from the Lean config, which don't go through click and are validated by config_build. In interactive mode an unsupported value read from the Lean config is prompted for again, instead of aborting the deployment. Co-Authored-By: Claude Opus 5 (1M context) --- lean/click.py | 22 +++++++ lean/models/click_options.py | 3 + lean/models/configuration.py | 43 +++++++++++++- lean/models/json_module.py | 48 +++++++++++++-- tests/models/test_configuration.py | 90 +++++++++++++++++++++++++++++ tests/models/test_json_module.py | 93 ++++++++++++++++++++++++++++++ tests/test_click.py | 50 +++++++++++++++- 7 files changed, 340 insertions(+), 9 deletions(-) create mode 100644 tests/models/test_configuration.py create mode 100644 tests/models/test_json_module.py diff --git a/lean/click.py b/lean/click.py index abef2ee5..06346adb 100644 --- a/lean/click.py +++ b/lean/click.py @@ -383,6 +383,28 @@ def convert(self, value: str, param: Parameter, ctx: Context): self.fail(f"'{value}' does not match the yyyyMMdd format.", param, ctx) +class RegexParameter(ParamType): + """A click parameter which requires the value to match a regular expression from the modules json.""" + + name = "text" + + def __init__(self, pattern: str, error_message: str = None): + """Creates a new RegexParameter instance. + + :param pattern: the regular expression the value must match completely + :param error_message: the message describing the expected value, or None to describe the pattern itself + """ + self._pattern = pattern + self._error_message = error_message if error_message else f"must match the '{pattern}' format" + + def convert(self, value: str, param: Parameter, ctx: Context) -> str: + from re import fullmatch + if fullmatch(self._pattern, str(value)) is None: + self.fail(f"'{value}' is not supported, it {self._error_message}.", param, ctx) + + return value + + def ensure_options(options: List[str]) -> None: """Ensures certain options have values, raises an error if not. diff --git a/lean/models/click_options.py b/lean/models/click_options.py index 5531395c..f7b8042b 100644 --- a/lean/models/click_options.py +++ b/lean/models/click_options.py @@ -53,6 +53,9 @@ def get_click_option_type(configuration: Configuration): # TODO: handle input can inherit type prompt. if configuration._config_type == "internal-input": return str + regex_type = configuration.get_regex_type() + if regex_type is not None: + return regex_type if configuration._input_method == "confirm": return bool elif configuration._input_method == "choice": diff --git a/lean/models/configuration.py b/lean/models/configuration.py index e7a61e27..7ffc697e 100644 --- a/lean/models/configuration.py +++ b/lean/models/configuration.py @@ -13,11 +13,11 @@ from pathlib import Path from typing import Any, Dict, List -from click import prompt +from click import prompt, ClickException from lean.click import CaseInsensitiveChoice from abc import ABC, abstractmethod from lean.components.util.logger import Logger -from lean.click import PathParameter +from lean.click import PathParameter, RegexParameter class BaseCondition(ABC): @@ -107,6 +107,7 @@ def __init__(self, config_json_object): self._filter = Filter([]) self._input_default = config_json_object["input-default"] if "input-default" in config_json_object else None self._optional = config_json_object["optional"] if "optional" in config_json_object else False + self._regex_type = self._create_regex_type(config_json_object) def factory(config_json_object) -> 'Configuration': """Creates an instance of the child classes. @@ -128,6 +129,41 @@ def factory(config_json_object) -> 'Configuration': raise ValueError( f'Undefined input method type {config_json_object["type"]}') + @staticmethod + def _create_regex_type(config_json_object): + """Creates the click type validating the values of a configuration, as described by the modules json. + + :param config_json_object: the json object dict with configuration info + :return: a RegexParameter instance, or None when the modules json doesn't describe a regex + """ + if "input-regex" not in config_json_object.keys(): + return None + message_key = "input-regex-message" + message = config_json_object[message_key] if message_key in config_json_object.keys() else None + return RegexParameter(config_json_object["input-regex"], message) + + def get_regex_type(self): + """Returns the click type validating this configuration against the regex given by the modules json. + + :return: a RegexParameter instance, or None when the modules json doesn't describe a regex + """ + return self._regex_type + + def validate(self, value): + """Validates a value which didn't go through a click prompt or option, like the ones read from the Lean config. + + :param value: the value to validate + :raises RuntimeError: when the value doesn't match the regex given by the modules json + :return: the validated value + """ + regex_type = self.get_regex_type() + if value is None or regex_type is None: + return value + try: + return regex_type.convert(value, None, None) + except ClickException as e: + raise RuntimeError(f"Invalid value for '{self._id}': {e.message}") + def __repr__(self): return f'{self._id}: {self._value}' @@ -273,6 +309,9 @@ def ask_user_for_input(self, default_value, logger: Logger, hide_input: bool = F return prompt(self._prompt_info, default_value, type=self.get_input_type()) def get_input_type(self): + regex_type = self.get_regex_type() + if regex_type is not None: + return regex_type return self.map_to_types.get(self._input_type, self._input_type) diff --git a/lean/models/json_module.py b/lean/models/json_module.py index e1001406..3110da27 100644 --- a/lean/models/json_module.py +++ b/lean/models/json_module.py @@ -224,6 +224,45 @@ def get_project_id(self, default_project_id: int, require_project_id: bool) -> i -1, show_default=False) return project_id + def _ask_user_for_input(self, configuration: Configuration, logger: Logger, hide_input: bool): + """Prompts the user for the value of a configuration and saves it in the Lean config. + + :param configuration: the configuration to prompt the user for + :param logger: the logger to use + :param hide_input: whether to hide secrets inputs + :return: the value provided by the user + """ + user_choice = configuration.ask_user_for_input(configuration._input_default, logger, hide_input=hide_input) + + if not isinstance(configuration, BrokerageEnvConfiguration): + self._save_property({f"{configuration._id}": user_choice}) + + return user_choice + + def _validate(self, configuration: Configuration, user_choice, logger: Logger, interactive: bool, + hide_input: bool): + """Validates the value of a configuration, prompting the user for a new one when it isn't supported. + + Values which come from the Lean config or from the modules json didn't go through click, + so they are validated here instead. + + :param configuration: the configuration the value belongs to + :param user_choice: the value to validate + :param logger: the logger to use + :param interactive: true if running in interactive mode + :param hide_input: whether to hide secrets inputs + :raises RuntimeError: when the value isn't supported and the user cannot be prompted for a new one + :return: the validated value + """ + while True: + try: + return configuration.validate(user_choice) + except RuntimeError as e: + if not interactive: + raise + logger.info(str(e)) + user_choice = self._ask_user_for_input(configuration, logger, hide_input) + def config_build(self, lean_config: Dict[str, Any], logger: Logger, @@ -318,11 +357,7 @@ def config_build(self, # in which case we still want to prompt the user. if not user_choice: if interactive: - default_value = configuration._input_default - user_choice = configuration.ask_user_for_input(default_value, logger, hide_input=hide_input) - - if not isinstance(configuration, BrokerageEnvConfiguration): - self._save_property({f"{configuration._id}": user_choice}) + user_choice = self._ask_user_for_input(configuration, logger, hide_input) else: if configuration._input_default != None and configuration._optional: # if optional and we have a default input value and the user didn't provider it we use it @@ -330,7 +365,8 @@ def config_build(self, else: missing_options.append(f"--{configuration._id}") - configuration._value = user_choice + # the values that come from the Lean config didn't go through click, so they are validated here + configuration._value = self._validate(configuration, user_choice, logger, interactive, hide_input) if len(missing_options) > 0: raise RuntimeError(f"""You are missing the following option{"s" if len(missing_options) > 1 else ""}: {', ' diff --git a/tests/models/test_configuration.py b/tests/models/test_configuration.py new file mode 100644 index 00000000..6bfae83d --- /dev/null +++ b/tests/models/test_configuration.py @@ -0,0 +1,90 @@ +# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. +# Lean CLI v1.0. Copyright 2021 QuantConnect Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import pytest + +from lean.click import RegexParameter +from lean.models.click_options import get_click_option_type +from lean.models.configuration import Configuration + +# The regex the modules json uses for ib-weekly-restart-utc-time, +# Interactive Brokers doesn't support weekly restart times later than 23:30 UTC +TIME_REGEX = r"^(?:(?:[01][0-9]|2[0-2]):[0-5][0-9]:[0-5][0-9]|23:(?:[0-2][0-9]:[0-5][0-9]|30:00))$" +TIME_REGEX_MESSAGE = "must be a UTC time in hh:mm:ss format, no later than 23:30:00" + + +def create_configuration(**properties) -> Configuration: + return Configuration.factory({ + "id": "my-time", + "type": "input", + "input-method": "prompt", + "prompt-info": "My time", + **properties + }) + + +def test_validate_returns_value_when_modules_json_has_no_regex() -> None: + configuration = create_configuration() + + assert configuration.validate("this is not a time") == "this is not a time" + + +@pytest.mark.parametrize("value", ["00:00:00", "21:00:00", "23:29:59", "23:30:00"]) +def test_validate_returns_value_when_it_matches_the_regex(value: str) -> None: + configuration = create_configuration(**{"input-regex": TIME_REGEX, "input-regex-message": TIME_REGEX_MESSAGE}) + + assert configuration.validate(value) == value + + +@pytest.mark.parametrize("value", ["23:30:01", "23:50:00", "21:00", "25:00:00", "invalid"]) +def test_validate_raises_when_value_does_not_match_the_regex(value: str) -> None: + configuration = create_configuration(**{"input-regex": TIME_REGEX, "input-regex-message": TIME_REGEX_MESSAGE}) + + with pytest.raises(RuntimeError) as error: + configuration.validate(value) + + assert f"Invalid value for 'my-time'" in str(error.value) + assert value in str(error.value) + assert TIME_REGEX_MESSAGE in str(error.value) + + +def test_validate_returns_none_when_there_is_no_value_to_validate() -> None: + configuration = create_configuration(**{"input-regex": TIME_REGEX}) + + assert configuration.validate(None) is None + + +def test_get_input_type_returns_the_regex_type_when_modules_json_has_a_regex() -> None: + configuration = create_configuration(**{"input-type": "string", "input-regex": TIME_REGEX}) + + assert isinstance(configuration.get_input_type(), RegexParameter) + + +def test_get_input_type_returns_the_mapped_type_when_modules_json_has_no_regex() -> None: + configuration = create_configuration(**{"input-type": "integer"}) + + assert configuration.get_input_type() is int + + +@pytest.mark.parametrize("input_method", ["prompt", "prompt-password", "path-parameter", "choice"]) +def test_get_click_option_type_returns_the_regex_type_for_all_input_methods(input_method: str) -> None: + configuration = create_configuration(**{"input-method": input_method, "input-regex": TIME_REGEX}) + + assert get_click_option_type(configuration) is configuration.get_regex_type() + + +def test_regex_is_not_validated_for_configurations_without_one() -> None: + configuration = create_configuration(**{"input-method": "prompt-password"}) + + assert configuration.get_regex_type() is None + assert get_click_option_type(configuration) is str diff --git a/tests/models/test_json_module.py b/tests/models/test_json_module.py new file mode 100644 index 00000000..742667b9 --- /dev/null +++ b/tests/models/test_json_module.py @@ -0,0 +1,93 @@ +# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. +# Lean CLI v1.0. Copyright 2021 QuantConnect Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from json import loads +from pathlib import Path +from typing import Any, Dict + +import pytest +from click import Command, Context, command, echo +from click.testing import CliRunner + +from lean.constants import MODULE_BROKERAGE, MODULE_CLI_PLATFORM +from lean.container import container +from lean.models.json_module import JsonModule +from tests.models.test_configuration import TIME_REGEX, TIME_REGEX_MESSAGE +from tests.test_helpers import create_fake_lean_cli_directory + + +def create_module(regex: bool = True) -> JsonModule: + configuration = { + "id": "my-time", + "type": "input", + "input-method": "prompt", + "prompt-info": "My time" + } + if regex: + configuration["input-regex"] = TIME_REGEX + configuration["input-regex-message"] = TIME_REGEX_MESSAGE + + return JsonModule({ + "id": "MyBrokerage", + "display-id": "My Brokerage", + "configurations": [configuration] + }, MODULE_BROKERAGE, MODULE_CLI_PLATFORM) + + +def build_config(lean_config: Dict[str, Any], interactive: bool = False, regex: bool = True) -> JsonModule: + # config_build() reads the options the user passed from the click context + with Context(Command("test")): + return create_module(regex).config_build(lean_config, container.logger, interactive=interactive) + + +def test_config_build_accepts_lean_config_value_matching_the_regex() -> None: + module = build_config({"my-time": "21:00:00"}) + + assert module.get_config_value_from_name("my-time") == "21:00:00" + + +@pytest.mark.parametrize("value", ["23:30:01", "23:50:00", "21:00", "invalid"]) +def test_config_build_raises_when_lean_config_value_does_not_match_the_regex(value: str) -> None: + # values read from the Lean config don't go through click, they are validated by config_build() + with pytest.raises(RuntimeError) as error: + build_config({"my-time": value}) + + assert "Invalid value for 'my-time'" in str(error.value) + assert TIME_REGEX_MESSAGE in str(error.value) + + +def test_config_build_prompts_again_when_lean_config_value_does_not_match_the_regex() -> None: + create_fake_lean_cli_directory() + + @command() + def test_command(): + module = build_config({"my-time": "23:50:00"}, interactive=True) + echo(f"value: {module.get_config_value_from_name('my-time')}") + + # the first answer isn't supported either, so the user is asked once more + result = CliRunner().invoke(test_command, input="21:00\n22:00:00\n") + + assert result.exit_code == 0 + assert TIME_REGEX_MESSAGE in result.output + assert "value: 22:00:00" in result.output + + # the value the user provided replaces the unsupported one in the Lean config + assert loads((Path.cwd() / "lean.json").read_text(encoding="utf-8"))["my-time"] == "22:00:00" + + +@pytest.mark.parametrize("value", ["23:50:00", "21:00", "invalid"]) +def test_config_build_accepts_any_lean_config_value_when_the_module_has_no_regex(value: str) -> None: + # the regex is optional, modules json files which don't describe one behave like they did before + module = build_config({"my-time": value}, regex=False) + + assert module.get_config_value_from_name("my-time") == value diff --git a/tests/test_click.py b/tests/test_click.py index 2b9c6f61..3cd72c6c 100644 --- a/tests/test_click.py +++ b/tests/test_click.py @@ -22,7 +22,7 @@ import pytest from click.testing import CliRunner -from lean.click import DateParameter, LeanCommand, PathParameter +from lean.click import DateParameter, LeanCommand, PathParameter, RegexParameter from lean.container import container from tests.test_helpers import create_fake_lean_cli_directory @@ -209,3 +209,51 @@ def command(arg: datetime) -> None: result = CliRunner().invoke(command, [input]) assert result.exit_code != 0 + + +# The regex the modules json uses for ib-weekly-restart-utc-time, +# Interactive Brokers doesn't support weekly restart times later than 23:30 UTC +TIME_REGEX = r"^(?:(?:[01][0-9]|2[0-2]):[0-5][0-9]:[0-5][0-9]|23:(?:[0-2][0-9]:[0-5][0-9]|30:00))$" +TIME_REGEX_MESSAGE = "must be a UTC time in hh:mm:ss format, no later than 23:30:00" + + +@pytest.mark.parametrize("input", ["21:00:00", "00:00:00", "23:29:59", "23:30:00"]) +def test_regex_parameter_returns_input_when_it_matches(input: str) -> None: + given_arg: Optional[str] = None + + @click.command() + @click.argument("arg", type=RegexParameter(TIME_REGEX, TIME_REGEX_MESSAGE)) + def command(arg: str) -> None: + nonlocal given_arg + given_arg = arg + + result = CliRunner().invoke(command, [input]) + + assert result.exit_code == 0 + + assert given_arg == input + + +@pytest.mark.parametrize("input", ["23:30:01", "23:50:00", "21:00", "25:00:00", "21:60:00", "210000", "invalid"]) +def test_regex_parameter_fails_when_input_does_not_match(input: str) -> None: + @click.command() + @click.argument("arg", type=RegexParameter(TIME_REGEX, TIME_REGEX_MESSAGE)) + def command(arg: str) -> None: + pass + + result = CliRunner().invoke(command, [input]) + + assert result.exit_code != 0 + assert f"'{input}' is not supported, it {TIME_REGEX_MESSAGE}" in result.output + + +def test_regex_parameter_falls_back_to_the_pattern_when_no_message_is_given() -> None: + @click.command() + @click.argument("arg", type=RegexParameter(TIME_REGEX)) + def command(arg: str) -> None: + pass + + result = CliRunner().invoke(command, ["invalid"]) + + assert result.exit_code != 0 + assert f"must match the '{TIME_REGEX}' format" in result.output