Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
Fail if `var` doesn't contain element with `k` == `a`
::
check:
any:
of: '{{ var }}'
equals: {the: '{{ ITEM.k }}', is: 'a'}
"""
def operator(self, data):
return any(data)
class Check(Step):
"""
Run check and fail if it was not successful.
There are two types of checks: terminators and nodes. Terminators like `equals` or `contains`
just perform checks while nodes contain like `all`, `any`, `or` and others contain other checks.
Check has a short form
::
check: '{{ variable }}'
which equals
::
check:
equals: {the: '{{ variable }}', is: true}
import json
from catcher.steps.step import Step, update_variables
from catcher.utils.misc import fill_template_str
from catcher.utils import external_utils
class External(Step):
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
@update_variables
def action(self, includes: dict, variables: dict) -> tuple:
"""
Call external script.
:param includes: testcase's includes
:param variables: variables
:return: script's output
"""
json_args = fill_template_str(json.dumps(self.data), variables)
from catcher.steps.step import Step, update_variables
from catcher.utils.misc import fill_template
from catcher.utils.logger import debug
from catcher.utils import external_utils
class Sh(Step):
"""
Run shell command and return output.
:Input:
- command: Command to run.
- path: Path to be used as a root for the command. *Optional*.
- return_code: expected return code. *Optional*. 0 is default.
:Examples:
List current directory
::
- sh:
command: 'ls -la'
import os
from os.path import join
from catcher.steps.step import Step, update_variables
from catcher.utils.file_utils import read_file
from catcher.utils.logger import info
from catcher.utils.misc import fill_template, fill_template_str
class Echo(Step):
"""
:Input:
:from: data source. Can be variable or constant string
:from_file: file in resources.
:to: output to file. *Optional* If not set - stdout will be used.
Has short from which just prints variable to stdout.
:Examples:
Use short form to print variable to stdout
::
echo: '{{ var }}'
import json
from abc import abstractmethod
from catcher.steps.step import Step
from catcher.utils.misc import fill_template_str, try_get_objects
class ExternalStep(Step):
"""
Implement this step in case you are adding external python module to
catcher-modules project
"""
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
method = Step.filter_predefined_keys(kwargs)
self.data = {method: kwargs[method]}
def simple_input(self, variables):
"""
Use this method to get simple input as python object, with all
templates filled in
:param variables:
import json
import os
import re
from typing import Union, Optional
import requests
from catcher.steps.step import Step, update_variables
from catcher.utils.logger import debug, warning
from catcher.utils.misc import fill_template, fill_template_str
from catcher.utils import file_utils
class Http(Step):
"""
:Input:
:: http method. See https://www.w3.org/Protocols/rfc2616/rfc2616-sec9.html for details
- headers: Dictionary with custom headers. *Optional*
- url: url to call
- response_code: Code to await. Use 'x' for a wildcard or '-' to set a range between 2 codes.
*Optional* default is 200.
- body: body to send. *Optional*.
- body_from_file: File can be used as data source. *Optional*.
- files: send file from resources (only for methods which support it). *Optional*
- verify: Verify SSL Certificate in case of https. *Optional*. Default is true.
- should_fail: true, if this request should fail, f.e. to test connection refused. Will fail the test if no errors.
- session: http session name. Cookies are saved between sessions. *Optional*. Default session is 'default'.
If set to null - there would be no session.
import collections
import itertools
import json
from catcher.steps.check import Operator
from catcher.steps.step import Step, update_variables
from catcher.utils.logger import debug
from catcher.utils.misc import fill_template_str, try_get_objects
class Loop(Step):
"""
:Input:
:while: perform action while the condition is true
- if: your condition. It can be in short format: `if: '{{ counter < 10 }}'` and
long one: `if: {equals: {the: '{{ counter }}', is_not: 10000}}`. The clause
format is the same as in [checks](checks.md)
- do: the aciton to be performed. Can be a list of actions or single one.
- max_cycle: the limit of reductions. *Optional* default is no limit.
:foreach: iterate data structure
- in: variable or static list. `ITEM` variable can be used to access each element of the data structure.
Data structure can be list, dict or any other python data structure which supports iteration.
- do: the aciton to be performed. Can be a list of actions or single one.
from time import sleep
from pykafka import KafkaClient, SimpleConsumer
from pykafka.common import OffsetType
from catcher.steps.check import Operator
from catcher.steps.step import Step
from catcher.utils.file_utils import read_file
from catcher.utils.misc import try_get_object, fill_template_str
from catcher.utils.time_utils import to_seconds
from catcher.utils.logger import debug
class Kafka(Step):
"""
:Input:
:consume: Consume message from kafka.
- server: is the kafka host. Can be multiple, comma-separated.
- group_id: is the consumer group id. If not specified - `catcher` will be used. *Optional*
- topic: the name of the topic
- timeout: is the consumer timeout. *Optional* (default is 1 sec)
- where: search for specific message clause. *Optional*
:produce: Produce message to kafka.
- server: is the kafka host. Can be multiple, comma-separated.
- topic: the name of the topic
- data: data to be produced.