Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
method = Step.filter_predefined_keys(kwargs) # get/post/put...
self.method = method.lower()
conf = kwargs[method]
self.url = conf['url']
self.headers = conf.get('headers', {})
self.verify = conf.get('verify', True)
self._should_fail = conf.get('should_fail', False)
if not self.verify:
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
self.code = conf.get('response_code', 200)
self.body = conf.get('body')
self.file = conf.get('body_from_file')
self.files = conf.get('files')
self.session = conf.get('session', 'default')
self.fix_cookies = conf.get('fix_cookies', True)
self.timeout = conf.get('timeout')
def __init__(self, _module: str = None, **kwargs) -> None:
super().__init__(**kwargs)
method = Step.filter_predefined_keys(kwargs)
self.data = {method: kwargs[method]}
self.module = _module
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
method = Step.filter_predefined_keys(kwargs)
self.data = {method: kwargs[method]}
def __init__(self, _get_action=None, _get_actions=None, **kwargs):
super().__init__(**kwargs)
self.type = Step.filter_predefined_keys(kwargs) # while/foreach
do = kwargs[self.type]['do']
if len(do) == 1: # just action
if isinstance(do, list): # list with single action
do = do[0]
[loop_action] = do.keys()
self.do_action = [_get_action((loop_action, do[loop_action]))]
else:
self.do_action = list(itertools.chain.from_iterable([_get_actions(act) for act in do]))
self.max_cycle = kwargs[self.type].get('max_cycle')
if self.type == 'while':
if_clause = kwargs['while']['if']
if isinstance(if_clause, str):
self.if_clause = {'equals': if_clause}
else:
self.if_clause = if_clause
elif self.type == 'foreach':
def __init__(self, body: dict) -> None:
super().__init__(body)
method = Step.filter_predefined_keys(body) # produce/consume
self._method = method.lower()
conf = body[method]
self._group_id = conf.get('group_id', 'catcher')
self._server = conf['server']
self._topic = conf['topic']
timeout = conf.get('timeout', {'seconds': 1})
self._timeout = to_seconds(timeout)
self._where = conf.get('where', None)
self._data = None
if self.method != 'consume':
self._data = conf.get('data', None)
if self.data is None:
self._file = conf['data_from_file']