How to use is_truthy method in Robotframework

Best Python code snippet using robotframework

Run Robotframework automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

config.py

Source: config.py Github

copy
1import os
2
3
4def is_truthy(v):
5    """
6    Returns True if v is a true-like value, or False otherwise.
7
8    >>> is_truthy(True)
9    True
10    >>> is_truthy("True")
11    True
12    >>> is_truthy("true")
13    True
14    >>> is_truthy("Yes")
15    True
16    >>> is_truthy("y")
17    True
18    >>> is_truthy(None)
19    False
20    >>> is_truthy(False)
21    False
22    >>> is_truthy("False")
23    False
24    >>> is_truthy("n")
25    False
26    >>> is_truthy("literally anything else")
27    False
28    """
29    return (isinstance(v, bool) and bool(v)) or (
30        isinstance(v, str) and v.lower() in ("1", "yes", "y", "true", "t")
31    )
32
33
34# when DEBUG is true, uvicorn will run in debug mode with autoreload enabled
35DEBUG = is_truthy(os.environ.get("DEBUG", False))
36
37# the set of corpora in the data folder
38CORPORA_SET = ('abstracts', 'fulltexts')
39
40# if env var USE_MEMMAP is truthy, loads word2vec models using the mmap='r' flag,
41# which largely keeps them on disk and keeps a single copy when sharing between
42# processes
43USE_MEMMAP = is_truthy(os.environ.get("USE_MEMMAP", True))
44
45# if env var MATERIALIZE_MODELS is truthy, load the models into memory once
46# (this requires a pretty big instance, as we're looking at ~20GB of RAM)
47MATERIALIZE_MODELS = is_truthy(os.environ.get("MATERIALIZE_MODELS", True))
48
49# if WARM_CACHE is truthy, warms the model cache when this module is first imported
50# (requires that MATERIALIZE_MODELS is true, since otherwise there's no cache to warm)
51WARM_CACHE = MATERIALIZE_MODELS and is_truthy(os.environ.get("WARM_CACHE", True))
52
53# if PARALLELIZE_QUERY is truthy or unspecified, queries year models in parallel
54PARALLELIZE_QUERY = is_truthy(os.environ.get("PARALLELIZE_QUERY", False))
55# integer number of pools to use for parallel year queries, default 4
56PARALLEL_POOLS = int(os.environ.get("PARALLEL_POOLS", 4))
57
58# controls what backend joblib uses to start parallel jobs
59# options are listed here: https://joblib.readthedocs.io/en/latest/generated/joblib.Parallel.html
60PARALLEL_BACKEND = os.environ.get("PARALLEL_BACKEND", "loky")
61
62# number of rq workers available, read from the environment
63RQ_CONCURRENCY = int(os.environ.get("RQ_CONCURRENCY", -1))
64
65
66def get_config_values():
67    return {
68        "DEBUG": DEBUG,
69        "CORPORA_SET": CORPORA_SET,
70        "USE_MEMMAP": USE_MEMMAP,
71        "MATERIALIZE_MODELS": MATERIALIZE_MODELS,
72        "WARM_CACHE": WARM_CACHE,
73        "PARALLELIZE_QUERY": PARALLELIZE_QUERY,
74        "PARALLEL_POOLS": PARALLEL_POOLS,
75        "PARALLEL_BACKEND": PARALLEL_BACKEND,
76        "RQ_CONCURRENCY": RQ_CONCURRENCY,
77    }
78
79
80def emit_config():
81    print("Debug enabled (DEBUG)?: %s" % DEBUG, flush=True)
82    print("Corpora set: %s" % CORPORA_SET, flush=True)
83    print("Memory-mapped models (USE_MEMMAP)?: %s" % USE_MEMMAP, flush=True)
84    print(
85        "Materialized models (MATERIALIZE_MODELS)?: %s" % MATERIALIZE_MODELS, flush=True
86    )
87    print("Pre-warmed model cache (WARM_CACHE)?: %s" % WARM_CACHE, flush=True)
88    print(
89        "Parallel year querying (PARALLELIZE_QUERY)?: %s" % PARALLELIZE_QUERY,
90        flush=True,
91    )
92    print("Parallel pools (PARALLEL_POOLS)?: %s" % PARALLEL_POOLS, flush=True)
93    print(
94        "Joblib.Parallel backend (PARALLEL_BACKEND)?: %s" % PARALLEL_BACKEND, flush=True
95    )
96
97
98if __name__ == "__main__":
99    import doctest
100
101    doctest.testmod()
102
Full Screen

ssh_runner.py

Source: ssh_runner.py Github

copy
1import uuid
2from abc import ABCMeta
3from contextlib import contextmanager
4from datetime import datetime, timedelta
5from enum import Enum
6from threading import Event, Thread, RLock
7from time import sleep
8from typing import Callable, Any
9
10import paramiko
11from SSHLibrary import SSHLibrary
12from SSHLibrary.pythonclient import Shell
13from robot.utils import DotDict, is_truthy, timestr_to_secs
14
15from RemoteMonitorLibrary.utils.logger_helper import logger
16
17from RemoteMonitorLibrary.api.tools import GlobalErrors
18from RemoteMonitorLibrary.model.errors import PlugInError, EmptyCommandSet, RunnerError
19from RemoteMonitorLibrary.model.runner_model import plugin_runner_abstract, _ExecutionResult, Parser, FlowCommands, \
20    Variable
21from RemoteMonitorLibrary.utils import evaluate_duration
22
23
24#
25# Solution for handling OSSocket error
26#
27def __shell_init__(self, client, term_type, term_width, term_height):
28    self._shell = client.invoke_shell(term_type, term_width, term_height)
29    # add use to solve socket.error: Socket is closed
30    self._shell.keep_this = client
31
32
33Shell.__init__ = __shell_init__
34
35SSHLibraryArgsMapping = {
36    SSHLibrary.execute_command.__name__: {'return_stdout': (is_truthy, True),
37                                          'return_stderr': (is_truthy, False),
38                                          'return_rc': (is_truthy, False),
39                                          'sudo': (is_truthy, False),
40                                          'sudo_password': (str, None),
41                                          'timeout': (timestr_to_secs, None),
42                                          'output_during_execution': (is_truthy, False),
43                                          'output_if_timeout': (is_truthy, False),
44                                          'invoke_subsystem': (is_truthy, False),
45                                          'forward_agent': (is_truthy, False)},
46    SSHLibrary.start_command.__name__: {'sudo': (is_truthy, False),
47                                        'sudo_password': (str, None),
48                                        'invoke_subsystem': (is_truthy, False),
49                                        'forward_agent': (is_truthy, False)},
50    SSHLibrary.write.__name__: {'text': (str, None),
51                                'loglevel': (str, 'INFO')},
52    SSHLibrary.read_command_output.__name__: {'return_stdout': (is_truthy, True),
53                                              'return_stderr': (is_truthy, False),
54                                              'return_rc': (is_truthy, False), 'sudo': (is_truthy, False),
55                                              'timeout': (timestr_to_secs, None)}
56}
57
58
59def _normalize_method_arguments(method_name, **kwargs):
60    assert method_name in SSHLibraryArgsMapping.keys(), f"Method {method_name} not supported"
61    for name, value in kwargs.items():
62        assert name in SSHLibraryArgsMapping.get(method_name, []).keys(), \
63            f"Argument '{name}' not supported for '{method_name}'"
64        arg_type, arg_default = SSHLibraryArgsMapping.get(method_name).get(name)
65        new_value = arg_type(value) if value else arg_default
66        yield name, new_value
67
68
69def extract_method_arguments(method_name, **kwargs):
70    assert method_name in SSHLibraryArgsMapping.keys(), f"Method {method_name} not supported"
71    return {name: value for name, value in kwargs.items() if name in SSHLibraryArgsMapping.get(method_name, []).keys()}
72
73
74class SSHLibraryCommand:
75    def __init__(self, method: Callable, command=None, **user_options):
76        self.variable_setter = user_options.pop('variable_setter', None)
77        if self.variable_setter:
78            assert isinstance(self.variable_setter, Variable), "Variable setter type error"
79        self.variable_getter = user_options.pop('variable_getter', None)
80        if self.variable_getter:
81            assert isinstance(self.variable_getter, Variable), "Variable getter vtype error"
82        self.parser: Parser = user_options.pop('parser', None)
83        self._sudo_expected = is_truthy(user_options.pop('sudo', False))
84        self._sudo_password_expected = is_truthy(user_options.pop('sudo_password', False))
85        self._start_in_folder = user_options.pop('start_in_folder', None)
86        # self._alias = user_options.pop('alias', None)
87        self._ssh_options = dict(_normalize_method_arguments(method.__name__, **user_options))
88        self._result_template = _ExecutionResult(**self._ssh_options)
89        if self.parser:
90            assert isinstance(self.parser, Parser), f"Parser type error [Error type: {type(self.parser).__name__}]"
91        self._method = method
92        self._command = command
93
94    @property
95    def command_template(self):
96        _command_res = f'cd {self._start_in_folder}; ' if self._start_in_folder else ''
97
98        _command = self._command.format(**self.variable_getter.result) if self.variable_getter else self._command
99
100        if self._sudo_password_expected:
101            _command_res += f'echo {{password}} | sudo --stdin --prompt "" {_command}'
102        elif self._sudo_expected:
103            _command_res += f'sudo {_command}'
104        else:
105            _command_res += _command
106        return _command_res
107
108    def __str__(self):
109        return f"{self._method.__name__}: " \
110               f"{', '.join([f'{a}' for a in [self._command] + [f'{k}={v}' for k, v in self._ssh_options.items()]])}" \
111               f"{'; Parser: '.format(self.parser) if self.parser else ''}"
112
113    def __call__(self, ssh_client: SSHLibrary, **runtime_options) -> Any:
114        # ssh_client.switch_connection(str(ssh_client))
115        if self._command is not None:
116            command = self.command_template.format(**runtime_options)
117            logger.debug(
118                f"Executing: {self._method.__name__}({command}, "
119                f"{', '.join([f'{k}={v}' for k, v in self._ssh_options.items()])})")
120            output = self._method(ssh_client, command, **self._ssh_options)
121        else:
122            logger.debug(f"Executing: {self._method.__name__}"
123                         f"({', '.join([f'{k}={v}' for k, v in self._ssh_options.items()])})")
124            output = self._method(ssh_client, **self._ssh_options)
125        if self.parser:
126            return self.parser(dict(self._result_template(output)))
127        if self.variable_setter:
128            self.variable_setter(output)
129        return output
130
131
132class SSHLibraryPlugInWrapper(plugin_runner_abstract, metaclass=ABCMeta):
133    def __init__(self, parameters: DotDict, data_handler, *user_args, **user_options):
134        self._sudo_expected = is_truthy(user_options.pop('sudo', False))
135        self._sudo_password_expected = is_truthy(user_options.pop('sudo_password', False))
136        super().__init__(parameters, data_handler, *user_args, **user_options)
137        self._execution_counter = 0
138        self._ssh = SSHLibrary()
139
140    @property
141    def content_object(self):
142        return self._ssh
143
144    @property
145    def sudo_expected(self):
146        return self._sudo_expected
147
148    @property
149    def sudo_password_expected(self):
150        return self._sudo_password_expected
151
152    def _close_ssh_library_connection_from_thread(self):
153        try:
154            with self._lock:
155                self._ssh.close_connection()
156        except RuntimeError:
157            pass
158        except Exception as e:
159            if 'Logging background messages is only allowed from the main thread' in str(e):
160                logger.warn(f"Ignore SSHLibrary error: '{e}'")
161                return True
162            raise
163
164    def _evaluate_tolerance(self):
165        if len(self._session_errors) == self._fault_tolerance:
166            e = PlugInError(f"{self}",
167                            "PlugIn stop invoked; Errors count arrived to limit ({})".format(
168                                self.host_alias,
169                                self._fault_tolerance,
170                            ), *self._session_errors)
171            logger.error(f"{e}")
172            GlobalErrors().append(e)
173            return False
174        return True
175
176    def login(self):
177        host = self.parameters.host
178        port = self.parameters.port
179        username = self.parameters.username
180        password = self.parameters.password
181        certificate = self.parameters.certificate
182
183        if len(self._session_errors) == 0:
184            logger.info(f"Host '{self.host_alias}': Connecting")
185        else:
186            logger.warn(f"Host '{self.host_alias}': Restoring at {len(self._session_errors)} time")
187
188        self._ssh.open_connection(host, repr(self), port)
189
190        start_ts = datetime.now()
191        while True:
192            try:
193                if certificate:
194                    logger.debug(f"Host '{self.host_alias}': Login with user/certificate")
195                    self._ssh.login_with_public_key(username, certificate, '')
196                else:
197                    logger.debug(f"Host '{self.host_alias}': Login with user/password")
198                    self._ssh.login(username, password)
199            except paramiko.AuthenticationException:
200                raise
201            except Exception as e:
202                logger.warn(f"Host '{self.host_alias}': Connection failed; Reason: {e}")
203            else:
204                self._is_logged_in = True
205                logger.info(f"Host '{self.host_alias}': Connection established")
206                break
207            finally:
208                duration = (datetime.now() - start_ts).total_seconds()
209                if duration >= self.parameters.timeout:
210                    raise TimeoutError(
211                        f"Cannot connect to '{self.host_alias}' during {self.parameters.timeout}s")
212
213    def exit(self):
214        if self._is_logged_in:
215            self._ssh.switch_connection(repr(self))
216            self._close_ssh_library_connection_from_thread()
217            self._is_logged_in = False
218            logger.info(f"Host '{self.id}::{self.host_alias}': Connection closed")
219        else:
220            logger.info(f"Host '{self.id}::{self.host_alias}': Connection close not required (not opened)")
221
222
Full Screen

helper_functions.py

Source: helper_functions.py Github

copy
1from code_page import *
2from check_type import *
3from collections import namedtuple
4
5Argument = namedtuple("Argument", ["char", "code"])
6
7loop_types = set("↑↓→←∟↔▲▼↨")
8block_creators = set("ÄÅÉæÆ{ôöò")
9string_creators = set("ûùÿ╢╖╕╣║╗")
10string_terminators = set("\"«»")
11compressed_letters_0 = "etaoinsrdluczbfp"
12compressed_letters_1 = "gwymvkxjqh ?*#.,"
13
14
15def while_true_no_pop(stack):
16	i = 0
17	while stack and is_truthy(stack[-1]):
18		yield i
19		i += 1
20
21def while_false_no_pop(stack):
22	i = 0
23	while stack and not is_truthy(stack[-1]):
24		yield i
25		i += 1
26
27def while_true_pop(stack):
28	i = 0
29	while stack and is_truthy(stack.pop("→")):
30		yield i
31		i += 1
32
33def while_false_pop(stack):
34	i = 0
35	while stack and not is_truthy(stack.pop("←")):
36		yield i
37		i += 1
38
39def do_while_true_no_pop(stack):
40	i = 0
41	while True:
42		yield i
43		i += 1
44		if not (stack and is_truthy(stack[-1])):
45			break
46
47def do_while_false_no_pop(stack):
48	i = 0
49	while True:
50		yield i
51		i += 1
52		if not (stack and not is_truthy(stack[-1])):
53			break
54
55def do_while_true_pop(stack):
56	i = 0
57	while True:
58		yield i
59		i += 1
60		if not (stack and is_truthy(stack.pop("▲"))):
61			break
62
63def do_while_false_pop(stack):
64	i = 0
65	while True:
66		yield i
67		i += 1
68		if not (stack and not is_truthy(stack.pop("▼"))):
69			break
70
71def decompress(string, compressed):
72	decompressed = ""
73	for i in range(len(string)):
74		letter_idx = code_page.index(string[i])
75		string_idx = (letter_idx >> 4) & 0xf
76		decompressed += compressed[string_idx]
77		string_idx = (letter_idx >> 0) & 0xf
78		decompressed += compressed[string_idx]
79	return decompressed
80
81def create_block(arg, code):
82	if arg.char == "Ä":
83		c = code[-1:]
84		code = code[:-1]
85	elif arg.char == "Å":
86		c = code[-2:]
87		code = code[:-2]
88	elif arg.char == "É":
89		c = code[-3:]
90		code = code[:-3]
91	elif arg.char == "æ":
92		c = code[-4:]
93		code = code[:-4]
94	elif arg.char == "Æ":
95		c = code[-5:]
96		code = code[:-5]
97	elif arg.char == "ô":
98		c = code[-6:]
99		code = code[:-6]
100	elif arg.char == "ö":
101		c = code[-7:]
102		code = code[:-7]
103	elif arg.char == "ò":
104		c = code[-8:]
105		code = code[:-8]
106	elif arg.char == "{":
107		c = []
108		temp = [] if not code else code.pop()
109		bracket_counter = 0
110		while bracket_counter > 0 or temp.char != "}" and code:
111			# print(temp)
112			if temp.char == "{":
113				bracket_counter += 1
114			if temp.char == "}":
115				bracket_counter -= 1
116			c.append(temp)
117			temp = code.pop()
118		if not code and temp.char != "}":
119			# print("temp:", temp)
120			c.append(temp)
121		# else:
122			# code.append(temp)
123
124		c = c[::-1]
125	return c, code
126
127def find_blocks(code):
128	level = 0
129	for i, arg in enumerate(code[::-1]):
130		if arg.char in block_creators:
131			level += 1
132		elif arg.char in loop_types or arg.char == "}":
133			level -= 1
134	while level < 0:
135		code.append(Argument("{", 0))
136		code.insert(1, Argument("}", 0))
137		level += 1
138	# print(''.join(c.char for c in code[::-1]))
139
140def for_looping(n):
141	for i in range(n):
142		yield i
143
144def get_char(c):
145	return code_page[c]
146def get_ord(c):
147	return code_page.index(c)
Full Screen

Accelerate Your Automation Test Cycles With LambdaTest

Leverage LambdaTest’s cloud-based platform to execute your automation tests in parallel and trim down your test execution time significantly. Your first 100 automation testing minutes are on us.

Try LambdaTest

Run Python Tests on LambdaTest Cloud Grid

Execute automation tests with Robotframework on a cloud-based Grid of 3000+ real browsers and operating systems for both web and mobile applications.

Test now for Free
LambdaTestX

We use cookies to give you the best experience. Cookies help to provide a more personalized experience and relevant advertising for you, and web analytics for us. Learn More in our Cookies policy, Privacy & Terms of service

Allow Cookie
Sarah

I hope you find the best code examples for your project.

If you want to accelerate automated browser testing, try LambdaTest. Your first 100 automation testing minutes are FREE.

Sarah Elson (Product & Growth Lead)