catcher.steps package

Submodules

catcher.steps.check module

class catcher.steps.check.All(body: dict, negative=False)[source]

Bases: catcher.steps.check.Operator

Fail if any check on the iterable fail.

Input:
Of:The source to check. Can be list or dictionary.
<check>:Check to perform on each element of the iterable.
Examples:

Pass if all elements of var has k == a

check:
    all:
        of: '{{ var }}'
        equals: {the: '{{ ITEM.k }}', is: 'a'}
operation(variables) → bool[source]
operator(data)[source]
class catcher.steps.check.And(body: dict, negative=False)[source]

Bases: catcher.steps.check.Operator

Fail if any of the conditions fails.

Input:The list of other checks.
Examples:

This is the same as 1 in list and list[1] != ‘b’ and list[2] > 2

check:
    and:
        - contains: {the: 1, in: '{{ list }}'}
        - equals: {the: '{{ list[1] }}', is_not: 'b'}
        - equals: {the: '{{ list[2] > 2 }}', is_not: true}
end
operation(variables) → bool[source]
class catcher.steps.check.Any(body: dict, negative=False)[source]

Bases: catcher.steps.check.All

Fail if all checks on the iterable fail.

Input:
Of:The source to check. Can be list or dictionary.
<check>:Check to perform on each element of the iterable.
Examples:

Fail if var doesn’t contain element with k == a

check:
    any:
        of: '{{ var }}'
        equals: {the: '{{ ITEM.k }}', is: 'a'}
operator(data)[source]
class catcher.steps.check.Check(_body=None, **kwargs)[source]

Bases: catcher.steps.step.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}
action(includes: dict, variables: dict) → dict[source]

Perform an action.

Parameters:
  • includes – Script includes.
  • variables – Script variables.
Returns:

step output

For code above

postgres:
    request:
        conf:
            dbname: test
            user: test
            password: test
            host: localhost
            port: 5433
        query: 'select count(*) from test'
register: {documents: '{{ OUTPUT }}'}

in_data will be

{'request' : {'conf': {'dbname': 'test', 'user': 'test', 'password': 'test', 'host': 'localhost', 'port': 5433},
              'query': 'select count(*) from test'}
}
class catcher.steps.check.Contains(body: dict, negative=False)[source]

Bases: catcher.steps.check.Operator

Fail if list of dictionary doesn’t contain the value

Input:
The:value to contain
In:variable to check
Not_in:inverted in. Only one can be used at a time.
Examples:

Check ‘a’ not in variable ‘list’

check:
    contains: {the: 'a', not_in: '{{ list }}'}

Check variable ‘dict’ has key a.

check:
    contains: {the: 'a', in: '{{ dict }}'}
determine_source(body: dict)[source]
operation(variables: dict)[source]
static to_long_form(source: any, value: any)[source]
class catcher.steps.check.Equals(body: dict, negative=False)[source]

Bases: catcher.steps.check.Operator

Fail if elements are not equal

Input:
The:value
Is:variable to compare
Is_not:inverted is. Only one can be used at a time.
Examples:

Check ‘bar’ equals variable ‘foo’

check: {equals: {the: 'bar', is: '{{ foo }}'}}

Check list’s third element is not greater than 2.

check: {equals: {the: '{{ list[2] > 2 }}', is_not: true}}
determine_source(body: dict)[source]
operation(variables: dict) → bool[source]
static to_long_form(source: any, value: any)[source]
class catcher.steps.check.Operator(body: dict, negative=False)[source]

Bases: object

body
static find_operator(source: dict) → catcher.steps.check.Operator[source]
operation(variables: dict) → bool[source]
class catcher.steps.check.Or(body: dict, negative=False)[source]

Bases: catcher.steps.check.And

Fail if all conditions fail.

Input:The list of other checks.
Examples:

This is the same as 1 in list or list[1] != ‘b’ or list[2] > 2

check:
    or:
        - contains: {the: 1, in: '{{ list }}'}
        - equals: {the: '{{ list[1] }}', is_not: 'b'}
        - equals: {the: '{{ list[2] > 2 }}', is_not: true}
end

catcher.steps.echo module

class catcher.steps.echo.Echo(_path: str = None, _body=None, to=None, **kwargs)[source]

Bases: catcher.steps.step.Step

Input:
From:data source. Can be variable or constant string
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 }}'

Print constant + variable to file

echo: {from: 'constant and {{ var }}', to: debug.output}

Use echo to register new variable

echo: {from: '{{ RANDOM_STR }}@test.com', register: {user_email: '{{ OUTPUT }}'}}
action(includes: dict, variables: dict) → tuple[source]

Perform an action.

Parameters:
  • includes – Script includes.
  • variables – Script variables.
Returns:

step output

For code above

postgres:
    request:
        conf:
            dbname: test
            user: test
            password: test
            host: localhost
            port: 5433
        query: 'select count(*) from test'
register: {documents: '{{ OUTPUT }}'}

in_data will be

{'request' : {'conf': {'dbname': 'test', 'user': 'test', 'password': 'test', 'host': 'localhost', 'port': 5433},
              'query': 'select count(*) from test'}
}

catcher.steps.external module

class catcher.steps.external.External(_module: str = None, **kwargs)[source]

Bases: catcher.steps.step.Step

action(includes: dict, variables: dict) → tuple[source]

Call external script.

Parameters:
  • includes – testcase’s includes
  • variables – variables
Returns:

script’s output

catcher.steps.external_step module

class catcher.steps.external_step.ExternalStep(**kwargs)[source]

Bases: catcher.steps.step.Step

Implement this step in case you are adding external python module to catcher-modules project

action(includes: dict, variables: dict) → dict[source]

Perform an action.

Parameters:
  • includes – Script includes.
  • variables – Script variables.
Returns:

variables and step’s output. Output is optional.

For code above

postgres:
    request:
        conf:
            dbname: test
            user: test
            password: test
            host: localhost
            port: 5433
        query: 'select count(*) from test'
register: {documents: '{{ OUTPUT }}'}

in_data will be

{'request' : {'conf': {'dbname': 'test', 'user': 'test', 'password': 'test', 'host': 'localhost', 'port': 5433},
              'query': 'select count(*) from test'}
}
simple_input(variables)[source]

Use this method to get simple input as python object, with all templates filled in

Parameters:variables
Returns:python object

catcher.steps.loop module

class catcher.steps.loop.Loop(_get_action=None, _get_actions=None, **kwargs)[source]

Bases: catcher.steps.step.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.
Examples:

Perform a single echo wile counter is less than 10

loop:
    while:
        if: '{{ counter < 10 }}'
        do:
            echo: {from: '{{ counter + 1 }}', register: {counter: '{{ OUTPUT }}'}}
        max_cycle: 100000

Perform to actions: consume message from kafka and send token via POST http. Do it until server returns passed true in http response.

loop:
    while:
        if:
            equals: {the: '{{ passed }}', is_not: True}
        do:
            - kafka:
                  consume:
                      server: '127.0.0.1:9092'
                      group_id: 'test'
                      topic: 'test_consume_with_timestamp'
                      timeout: {seconds: 5}
                      where:
                          equals: '{{ MESSAGE.timestamp > 1000 }}'
                  register: {token: '{{ OUTPUT.data.token }}'}
            - http:
                post:
                  headers: {Content-Type: 'application/json'}
                  url: 'http://test.com/check_my_token'
                  body: {'token': '{{ token }}'}
                register: {passed: '{{ OUTPUT.passed }}'}

Iterate over iterator variable, produce each element to kafka as json and debug it to file.

loop:
    foreach:
        in: '{{ iterator }}'
        do:
            - kafka:
                  produce:
                      server: '127.0.0.1:9092'
                      topic: 'test_produce_json'
                      data: '{{ ITEM|tojson }}'
            - echo: {from: '{{ ITEM }}', to: '{{ ITEM["filename"] }}.output'}
action(includes: dict, variables: dict) → dict[source]

Perform an action.

Parameters:
  • includes – Script includes.
  • variables – Script variables.
Returns:

step output

For code above

postgres:
    request:
        conf:
            dbname: test
            user: test
            password: test
            host: localhost
            port: 5433
        query: 'select count(*) from test'
register: {documents: '{{ OUTPUT }}'}

in_data will be

{'request' : {'conf': {'dbname': 'test', 'user': 'test', 'password': 'test', 'host': 'localhost', 'port': 5433},
              'query': 'select count(*) from test'}
}

catcher.steps.http module

class catcher.steps.http.Http(response_code=200, **kwargs)[source]

Bases: catcher.steps.step.Step

Input:
<method>: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. Optional default is 200.
  • body: body to send (only for methods which support it).
  • body_from_file: File can be used as data source. Optional Either body or body_from_file should present.
  • 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.
Examples:

Post data to server with custom header

http:
  post:
    headers: {Content-Type: 'application/json', Authorization: '{{ token }}'}
    url: 'http://test.com?user_id={{ user_id }}'
    body: {'foo': bar}

Post file to remote server

http:
  post:
    url: 'http://test.com'
    body_from_file: "data/answers.json"

SSL without verification

http:
  post:
    url: 'https://my_server.de'
    body: {'user':'test'}
    verify: false

Json body from a variable:

http:
  post:
    url: 'http://test.com?user_id={{ user_id }}'
    body: '{{ var |tojson }}'

Test disconnected service:

steps:
- docker:
    disconnect:
        hash: '{{ my_container }}'
- http:
    get:
        url: '{{ my_container_url }}'
        should_fail: true
action(includes: dict, variables: dict) → Union[tuple, dict][source]

Perform an action.

Parameters:
  • includes – Script includes.
  • variables – Script variables.
Returns:

step output

For code above

postgres:
    request:
        conf:
            dbname: test
            user: test
            password: test
            host: localhost
            port: 5433
        query: 'select count(*) from test'
register: {documents: '{{ OUTPUT }}'}

in_data will be

{'request' : {'conf': {'dbname': 'test', 'user': 'test', 'password': 'test', 'host': 'localhost', 'port': 5433},
              'query': 'select count(*) from test'}
}

catcher.steps.run module

class catcher.steps.run.Run(ignore_errors=False, _body=None, run=None, include=None, tag=None, variables=None, **keywords)[source]

Bases: catcher.steps.step.Step

Run include on demand

Input:
Include:include name
Tag:include tag. Optional If specified - only steps with this tag will be run. Can also be set up via dot notation: include: test.tag.
Variables:Variables to override. Optional
Examples:

Use short form to run sign_up

include:
    file: register_user.yaml
    as: sign_up
steps:
    # .... some steps
    - run: sign_up
    # .... some steps

Run sign_up with username overridden

include:
    file: register_user.yaml
    as: sign_up
steps:
    # .... some steps
    - run:
        include: sign_up
        variables:
            username: test
    # .... some steps

Include sign_up and run all steps with tag register from it. Use dot notation.

include:
    file: register_and_login.yaml
    as: sign_up
steps:
    - run:
        include: sign_up.register
action(includes: dict, variables: dict) → dict[source]

Perform an action.

Parameters:
  • includes – Script includes.
  • variables – Script variables.
Returns:

step output

For code above

postgres:
    request:
        conf:
            dbname: test
            user: test
            password: test
            host: localhost
            port: 5433
        query: 'select count(*) from test'
register: {documents: '{{ OUTPUT }}'}

in_data will be

{'request' : {'conf': {'dbname': 'test', 'user': 'test', 'password': 'test', 'host': 'localhost', 'port': 5433},
              'query': 'select count(*) from test'}
}
catcher.steps.run.get_tag(include: str, tag: str) → str[source]

catcher.steps.step module

class catcher.steps.step.MetaStep[source]

Bases: type

class catcher.steps.step.Step(register=None, name=None, ignore_errors=False, **kwargs)[source]

Bases: object

Abstract class for all Steps. Operates with common properties, available for all steps.

Register:register a new variable at the end of this step. Optional
Name:Set name for this step. Is used for passed steps output. Optional
Ignore_errors:Do not stop the test if this step fails. Can be useful with running includes. Optional
Tag:Tag this step to be called via run with tag. Optional
Actions:Each step can have one ore multiple actions. In case of one action actions list is not necessary and you can use short form. Also - in case of several actions each should have its own properties like register, tag etc…

One action - short form.

http:
    post:  # register client and get id
         url: '{{ user_service_url }}/sign_up'
         body: {email: '{{ email }}', name: '{{ user }}'}

Multiple actions.

http:
    actions:
      - post:  # register client and get id
          url: '{{ user_service_url }}/sign_up'
          body: {email: '{{ email }}', name: 'TestUser'}
      - post:  # fill some personal data
          url: '{{ user_service_url }}/data'
          body: {gender: 'M', age: 22, firstName: 'John', lastName: 'Doe'}
Examples:

Register new variable after echo step finishes.

echo: {from: '{{ RANDOM_STR }}@test.com', register: {user_email: '{{ OUTPUT }}'}}

Tag both http actions with sign_up

http:
    actions:
      - post:  # register client and get id
          url: '{{ user_service_url }}/sign_up'
          headers: {Content-Type: 'application/json'}
          body: {email: '{{ email }}', name: 'TestUser'}
          response_code: 201
        register: {token: '{{ OUTPUT.data.token }}'}
        tag: sign_up
      - post:  # fill some personal data
          url: '{{ user_service_url }}/data'
          headers: {Content-Type: 'application/json', Authorization: '{{ token }}'}
          body: {gender: 'M', age: 22, firstName: 'John', lastName: 'Doe'}
        register: {uuid: '{{ OUTPUT.data.uuid }}'}
        tag: sign_up

Use custom name for the step

http:
    post:  # register client and get id
         url: '{{ user_service_url }}/sign_up'
         headers: {Content-Type: 'application/json'}
         body: {email: '{{ email }}', name: '{{ user }}'}
         response_code: 201
    name: 'Register {{ user }} on remote server'

Ignore errors and continue to another step

http: {get: {url: 'http://test.com', response_code: 200}, ignore_errors: true}
action(includes: dict, variables: dict) → dict[source]

Perform an action.

Parameters:
  • includes – Script includes.
  • variables – Script variables.
Returns:

step output

For code above

postgres:
    request:
        conf:
            dbname: test
            user: test
            password: test
            host: localhost
            port: 5433
        query: 'select count(*) from test'
register: {documents: '{{ OUTPUT }}'}

in_data will be

{'request' : {'conf': {'dbname': 'test', 'user': 'test', 'password': 'test', 'host': 'localhost', 'port': 5433},
              'query': 'select count(*) from test'}
}
static filter_predefined_keys(kwargs: dict)[source]
process_register(variables, output: dict = None) → dict[source]
catcher.steps.step.register_class(target_class)[source]
catcher.steps.step.update_variables(func)[source]

Use this decorator on Step.action implementation.

Your action method should always return variables, or both variables and output.

This decorator will update variables with output.

catcher.steps.stop module

class catcher.steps.stop.Stop(**kwargs)[source]

Bases: catcher.steps.step.Step

Stop a test without error

Input:
If:condition
Examples:

Stop execution if migration was applied.

steps:
    - postgres:
        request:
            conf: '{{ migrations_postgres }}'
            query: "select count(*) from migration where hash = '{{ TEST_NAME }}';"
        register: {result: '{{ OUTPUT }}'}
        tag: check
        name: 'check_migration_{{ TEST_NAME }}'
    - stop:
        if:
            equals: {the: '{{ result }}', is: 1}
    - postgres:
        request:
            conf: '{{ migrations_postgres }}'
            query: "insert into migration(id, hash) values(1, '{{ TEST_NAME }}');"
action(includes: dict, variables: dict) → dict[source]

Perform an action.

Parameters:
  • includes – Script includes.
  • variables – Script variables.
Returns:

step output

For code above

postgres:
    request:
        conf:
            dbname: test
            user: test
            password: test
            host: localhost
            port: 5433
        query: 'select count(*) from test'
register: {documents: '{{ OUTPUT }}'}

in_data will be

{'request' : {'conf': {'dbname': 'test', 'user': 'test', 'password': 'test', 'host': 'localhost', 'port': 5433},
              'query': 'select count(*) from test'}
}
exception catcher.steps.stop.StopException(message)[source]

Bases: Exception

catcher.steps.wait module

class catcher.steps.wait.Wait(_get_action=None, _get_actions=None, **kwargs)[source]

Bases: catcher.steps.step.Step

Wait for some time before the next step

Input:
Days:several days
Hours:several hours
Minutes:several minutes
Seconds:several seconds
Microseconds:several microseconds
Milliseconds:several milliseconds
Nanoseconds:several nanoseconds
For:(list of actions) will repeat them till they all finishes successfully or time ends.
Examples:

Wait for 1 minute 30 seconds

wait: {minutes: 1, seconds: 30}

Wait for http to be ready. Will run the http for 5 seconds or till it finishes successfully

wait:
    seconds: 5
    for:
        http:
            put:
                url: 'http://localhost:8000/mockserver/expectation'
                body:
                    httpRequest: {'path': '/some/path'}
                    httpResponse: {'body': 'hello world'}
                response_code: 201

Wait for postgres to be populated

wait:
    seconds: 30
    for:
        - postgres:
              request:
                  conf: '{{ pg_conf }}'
                  query: 'select count(*) from users'
              register: {documents: '{{ OUTPUT }}'}
        - check: {equals: {the: '{{ documents }}', is_not: 0}}
action(includes: dict, variables: dict) → dict[source]

Perform an action.

Parameters:
  • includes – Script includes.
  • variables – Script variables.
Returns:

step output

For code above

postgres:
    request:
        conf:
            dbname: test
            user: test
            password: test
            host: localhost
            port: 5433
        query: 'select count(*) from test'
register: {documents: '{{ OUTPUT }}'}

in_data will be

{'request' : {'conf': {'dbname': 'test', 'user': 'test', 'password': 'test', 'host': 'localhost', 'port': 5433},
              'query': 'select count(*) from test'}
}
run_loop(includes, variables)[source]

Module contents