How to use add_secrets_from_pairs method in lisa

Best Python code snippet using lisa_python

variable.py

Source:variable.py

...72        higher_level_variables = {}73    current_variables: Dict[str, VariableEntry] = {}74    if isinstance(higher_level_variables, list):75        env_variables = _load_from_env()76        cmd_variables = add_secrets_from_pairs(higher_level_variables)77    else:78        merge_variables(current_variables, higher_level_variables)79        env_variables = {}80        cmd_variables = {}81    # current_variables uses to support variable in variable file path82    merge_variables(current_variables, env_variables)83    merge_variables(current_variables, cmd_variables)84    final_variables: Dict[str, VariableEntry] = {}85    merge_variables(86        final_variables,87        _load_from_runbook(runbook_data, higher_level_variables=current_variables),88    )89    if isinstance(higher_level_variables, dict):90        merge_variables(final_variables, higher_level_variables)91    else:92        merge_variables(final_variables, env_variables)93        merge_variables(final_variables, cmd_variables)94    return final_variables95def merge_variables(96    variables: Dict[str, VariableEntry], new_variables: Dict[str, VariableEntry]97) -> None:98    """99    in place update variables. If variables don't exist, will create them100    """101    for name, new_variable in new_variables.items():102        variable = variables.get(name, None)103        if variable:104            variable.update(new_variable)105        else:106            variables[name] = new_variable.copy()107def _get_undefined_variables(108    value: str, variables: Dict[str, VariableEntry]109) -> List[str]:110    undefined_variables: List[str] = []111    # check if there is variable or not in a value112    matches = _VARIABLE_PATTERN.findall(value)113    for variable_name in matches:114        lower_variable_name = variable_name[2:-1].lower()115        if lower_variable_name not in variables:116            undefined_variables.append(variable_name)117    return undefined_variables118def _load_from_env() -> Dict[str, VariableEntry]:119    results: Dict[str, VariableEntry] = {}120    for env_name in os.environ:121        is_lisa_variable = True122        is_secret = False123        name = ""124        if env_name.startswith(_ENV_START):125            name = env_name[len(_ENV_START) :]126            value = os.environ[env_name]127        elif env_name.startswith(_SECRET_ENV_START):128            name = env_name[len(_SECRET_ENV_START) :]129            is_secret = True130        else:131            is_lisa_variable = False132        if is_lisa_variable:133            value = os.environ[env_name]134            _add_variable(name, value, results, is_secret=is_secret)135    return results136def _load_from_runbook(137    runbook_data: Any, higher_level_variables: Dict[str, VariableEntry]138) -> Dict[str, VariableEntry]:139    # make a copy to prevent modifying existing dict140    current_variables = higher_level_variables.copy()141    if constants.VARIABLE in runbook_data:142        variable_entries: List[schema.Variable] = schema.load_by_type_many(143            schema.Variable, runbook_data[constants.VARIABLE]144        )145        left_variables = variable_entries.copy()146        undefined_variables: List[str] = []147        is_current_updated = True148        # when is_current_updated, it means one of variable is processed, and149        #  it's ok to loop again. If it's false, there are some variables cannot150        #  be resolved.151        while left_variables and is_current_updated:152            is_current_updated = False153            undefined_variables = []154            # solved variable will be removed later, so use a copy here to prevent155            #  operate enumerating collection.156            for entry in left_variables.copy():157                # this value is used to detect whether value used undefined variables.158                # in final merge, the referred variable may be defined later than used.159                # So it should to load referred variables firstly.160                checked_value = f"{entry.file}{entry.value}"161                current_undefined_variables = _get_undefined_variables(162                    checked_value, current_variables163                )164                if current_undefined_variables:165                    undefined_variables.extend(current_undefined_variables)166                    continue167                if entry.file:168                    path = replace_variables(entry.file, current_variables)169                    loaded_variables = _load_from_file(path, is_secret=entry.is_secret)170                else:171                    value = replace_variables(entry.value, current_variables)172                    loaded_variables = load_from_variable_entry(173                        entry.name,174                        value,175                        is_case_visible=entry.is_case_visible,176                        is_secret=entry.is_secret,177                        mask=entry.mask,178                    )179                merge_variables(current_variables, loaded_variables)180                merge_variables(current_variables, higher_level_variables)181                is_current_updated = True182                left_variables.remove(entry)183        if undefined_variables:184            raise LisaException(f"variables are undefined: {undefined_variables}")185    return current_variables186def _load_from_file(187    file_name: str,188    is_secret: bool = False,189) -> Dict[str, VariableEntry]:190    results: Dict[str, VariableEntry] = {}191    if is_secret:192        secret.add_secret(file_name, secret.PATTERN_FILENAME)193    path = constants.RUNBOOK_PATH.joinpath(file_name)194    if path.suffix.lower() not in [".yaml", ".yml"]:195        raise LisaException("variable support only yaml and yml")196    try:197        with open(path, "r") as fp:198            raw_variables = yaml.safe_load(fp)199    except FileNotFoundError:200        raise FileNotFoundError(f"cannot find variable file: {path}")201    if not isinstance(raw_variables, Dict):202        raise LisaException("variable file must be dict")203    for key, raw_value in raw_variables.items():204        try:205            entry = convert_to_variable_entry(raw_value, is_secret=is_secret)206            raw_value = entry.value207            is_secret = is_secret or entry.is_secret208            is_case_visible = entry.is_case_visible209            mask = entry.mask210        except AssertionError:211            # if parsing is failed, regard the raw_value as a value.212            is_case_visible = False213            mask = ""214        results.update(215            load_from_variable_entry(216                key,217                raw_value=raw_value,218                is_secret=is_secret,219                is_case_visible=is_case_visible,220                mask=mask,221            )222        )223    return results224def get_case_variables(variables: Dict[str, VariableEntry]) -> Dict[str, Any]:225    return {226        name: value.data for name, value in variables.items() if value.is_case_visible227    }228def add_secrets_from_pairs(229    raw_pairs: Optional[List[str]],230) -> Dict[str, VariableEntry]:231    """232    Given a list of command line style pairs of [s]:key:value tuples,233    take the ones prefixed with "s:" and make them recorded234    secrets. At the end, also return a dictionary of those tuples235    (still with raw values).236    """237    results: Dict[str, VariableEntry] = {}238    if raw_pairs is None:239        return results240    for raw_pair in raw_pairs:241        is_secret = False242        if raw_pair.lower().startswith("s:"):...

test_variable.py

Source:test_variable.py

...20    def test_in_pair(self) -> None:21        pair1 = "normal_value:nv_from_pair"22        pair2 = "S:normal_entry:s_value_from_env"23        variables = self._get_default_variables()24        variables.update(variable.add_secrets_from_pairs([pair1, pair2]))25        data = self._replace_and_validate(variables, {"normal_entry": "******"})26        self.assertEqual("nv_from_pair", data["nested"]["normal_value"])27        self.assertEqual("s_value_from_env", data["normal_entry"])28    def test_in_normal_file_outside_secret(self) -> None:29        self._test_files(30            "variable_normal.yml",31            True,32            {33                "normal_value": "******",34                "normal_entry": "******",35                "secret_guid": "12345678-****-****-****-********90ab",36                "secret_int": "1****0",37                "secret_head_tail": "a****h",38            },...

main.py

Source:main.py

...101        log.info(f"local time: {datetime.now().astimezone()}")102        _dump_code_information(log)103        # We don't want command line args logging to leak any provided104        # secrets, if any ("s:key:value" syntax)105        add_secrets_from_pairs(args.variables)106        log.debug(f"command line args: {sys.argv}")107        log.info(108            f"run log path: {constants.RUN_LOCAL_LOG_PATH}, "109            f"working path: {constants.RUN_LOCAL_WORKING_PATH}"110        )111        exit_code = args.func(args)112        assert isinstance(exit_code, int), f"actual: {type(exit_code)}"113    finally:114        log.info(f"completed in {total_timer}")115        if file_handler:116            remove_handler(log_handler=file_handler, logger=log)117        uninit_logger()118    return exit_code119if __name__ == "__main__":...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.