TestsProcessor Plugin

It is often required to validate network devices state against certain criteria, test functions designed to address most common use cases in that area.

Majority of available checks help to implement simple approach of “if this then that” logic, for instance, if output contains this pattern then test failed.

Dependencies:

Test functions returns Nornir Result object and make use of these attributes:

  • name - name of the test

  • task - name of the task

  • result - test result PASS, FAIL or ERROR

  • success - test success status True (PASS) or False (FAIL or ERROR)

  • exception - description of failure reason

  • test - test type to perform e.g. contains, custom, cerberus etc.

  • criteria - criteria that failed the test e.g. pattern or string

  • failed - this attribute not used by test functions to signal any status and should always be False

Running tests

Running tests as simple as defining a list of dictionaries - test suite - each dictionary represents single test definition. Reference to a particular test function API for description of test function specific arguments it supports.

These are mandatory arguments/keys each test dictionary must contain:

  • name - name of the test

  • task - name of the task to check results for or list of task names to use with custom test function

  • test - name of test function to run

Additional arguments/keys that test dictionary can contain:

  • err_msg - string, error message to use for exception in case of test failure

  • path - string, dot separated path to data to test within results

  • report_all - boolean, default is False, if path evaluates to a list of items and report_all set to True, reports all tests, even successful ones

To simplify test functions calls, TestsProcessor implements these set of aliases for test argument:

  • contains calls ContainsTest

  • !contains or ncontains calls ContainsTest with kwargs: {"revert": True}

  • contains_lines calls ContainsLinesTest

  • !contains_lines or ncontains_lines calls ContainsLinesTest with kwargs: {"revert": True}

  • contains_re calls ContainsTest with kwargs: {"use_re": True}

  • !contains_re or ncontains_re calls ContainsTest with kwargs: {"revert": True, "use_re": True}

  • equal calls EqualTest

  • !equal calls or nequal calls EqualTest with kwargs: {"revert": True}

  • cerberus calls CerberusTest

  • custom calls CustomFunctionTest

  • eval calls EvalTest

In addition to aliases, test argument can reference actual test functions names:

  • ContainsTest calls ContainsTest

  • ContainsLinesTest calls ContainsLinesTest

  • EqualTest calls EqualTest

  • CerberusTest calls CerberusTest

  • CustomFunctionTest calls CustomFunctionTest

  • EvalTest calls EvalTest

Sample code to run tests:

import pprint

from nornir import InitNornir
from nornir_salt import TestsProcessor, ResultSerializer, netmiko_send_commands

nr = InitNornir(config_file="nornir.yaml")

tests = [
    {
        "name": "Test NTP config",
        "task": "show run | inc ntp",
        "test": "contains",
        "pattern": "ntp server 7.7.7.8",
    },
    {
        "name": "Test Logging config",
        "task": "show run | inc logging",
        "test": "contains_lines",
        "pattern": ["logging host 1.1.1.1", "logging host 1.1.1.2"]
    },
    {
        "name": "Test BGP peers state",
        "task": "show bgp ipv4 un summary",
        "test": "!contains_lines",
        "pattern": ["Idle", "Active", "Connect"]
    },
    {
        "task": "show run | inc ntp",
        "name": "Test NTP config",
        "expr": "assert '7.7.7.8' in result, 'NTP server 7.7.7.8 not in config'",
        "test": "eval",
    }
]

nr_with_tests = nr.with_processors([
    TestsProcessor(tests, remove_tasks=True)
])

# netmiko_send_commands maps commands to sub-task names
results = nr_with_tests.run(
    task=netmiko_send_commands,
    commands=[
        "show run | inc ntp",
        "show run | inc logging",
        "show bgp ipv4 un summary"
    ]
)

results_dictionary = ResultSerializer(results, to_dict=False, add_details=False)

pprint.pprint(results_dictionary)

# should print something like:
#
# [{'host': 'IOL1', 'name': 'Test NTP config', 'result': 'PASS'},
# {'host': 'IOL1', 'name': 'Test Logging config', 'result': 'PASS'},
# {'host': 'IOL1', 'name': 'Test BGP peers state', 'result': 'FAIL'},
# {'host': 'IOL2', 'name': 'Test NTP config', 'result': 'PASS'},
# {'host': 'IOL2', 'name': 'Test Logging config', 'result': 'PASS'},
# {'host': 'IOL2', 'name': 'Test BGP peers state', 'result': 'PASS'}]

Notes on path attribute. path attribute allows to run tests against portions of overall results, but works only if results are structured data, e.g. nested dictionary or list of dictionaries. For example:

import pprint
from nornir import InitNornir
from nornir_salt import TestsProcessor, ResultSerializer, nr_test

nr = InitNornir(config_file="nornir.yaml")

tests = [
    {
        "test": "eval",
        "task": "show run interface",
        "name": "Test MTU config",
        "path": "interfaces.*",
        "expr": "assert result['mtu'] > 9000, '{} MTU less then 9000'.format(result['interface'])"
    }
]

nr_with_tests = nr.with_processors([
    TestsProcessor(tests, remove_tasks=True)
])

# nr_test function echoes back ret_data_per_host as task results
output = nr_with_tests.run(
    task=nr_test,
    ret_data_per_host={
        "IOL1": {
            "interfaces":  [
                {"interface": "Gi1", "mtu": 1500},
                {"interface": "Gi2", "mtu": 9200},
            ]
        },
        "IOL2": {
            "interfaces":  [
                {"interface": "Eth1/9", "mtu": 9600}
            ]
        }
    },
    name="show run interface"
)

check_result = ResultSerializer(output, add_details=True, to_dict=False)
# pprint.pprint(check_result)
# [{'changed': False,
#   'criteria': '',
#   'diff': '',
#   'exception': 'Gi1 MTU less then 9000',
#   'failed': True,
#   'host': 'IOL1',
#   'name': 'Test MTU config',
#   'result': 'FAIL',
#   'success': False,
#   'task': 'show run interface',
#   'test': 'eval'},
#  {'changed': False,
#   'criteria': '',
#   'diff': '',
#   'exception': None,
#   'failed': False,
#   'host': 'IOL2',
#   'name': 'Test MTU config',
#   'result': 'PASS',
#   'success': True,
#   'task': 'show run interface',
#   'test': 'eval'}]

In above example path interfaces.* tells TestsProcessor to retrieve data from results under interfaces key, single star * symbol tells to iterate over list items, instead of star, list item index can be given as well, e.g. interfaces.0.

Tests Reference

class nornir_salt.plugins.processors.TestsProcessor.TestsProcessor(tests=None, remove_tasks=True, failed_only=False, **kwargs)

TestsProcessor designed to run a series of tests for Nornir tasks results.

Parameters
  • tests – (list of dictionaries) list of tests to run

  • remove_tasks – (bool) if True (default) removes tasks output from results

  • kwargs – (any) if provided, **kwargs will form a single test item

  • failed_only – (bool) if True, includes only failed tests in results, default is False

nornir_salt.plugins.processors.TestsProcessor.ContainsTest(host, result, pattern, use_re=False, count=None, revert=False, err_msg=None, **kwargs)

Function to check if pattern contained in output of given result.

Parameters
  • host – (obj) Nornir host object

  • result – (obj) nornir.core.task.Result object

  • pattern – (str) pattern to check containment for

  • use_re – (bool) if True uses re.search to check for pattern in output

  • count – (int) check exact number of pattern occurrences in the output

  • revert – (bool) if True, changes results to opposite - check lack of containment

  • err_msg – (str) exception message to use on test failure

  • kwargs – (dict) any additional **kwargs keyword arguments to include in return Result object

Return result

nornir.core.task.Result object with test results

nornir_salt.plugins.processors.TestsProcessor.ContainsLinesTest(host, result, pattern, use_re=False, count=None, revert=False, err_msg=None, **kwargs)

Function to check that all lines contained in result output.

Tests each line one by one, this is the key difference compared to ContainsTest function, where whole pattern checked for presence in output from device.

Parameters
  • host – (obj) Nornir host object

  • result – (obj) nornir.core.task.Result object

  • pattern – (str or list) multiline string or list of lines to check

  • use_re – (bool) if True uses re.search to check for line pattern in output

  • count – (int) check exact number of line pattern occurrences in the output

  • revert – (bool) if True, changes results to opposite - check lack of lines in output

  • err_msg – (str) exception message to use on test failure

  • kwargs – (dict) any additional **kwargs keyword arguments to include in return Result object

Return result

nornir.core.task.Result object with test results

nornir_salt.plugins.processors.TestsProcessor.EqualTest(host, result, pattern, revert=False, err_msg=None, **kwargs)

Function to check result is equal to the pattern.

Parameters
  • host – (obj) Nornir host object

  • result – (obj) nornir.core.task.Result object

  • pattern – (any) string, dict, list or any other object to check for equality

  • revert – (bool) if True, changes results to opposite - check for inequality

  • err_msg – (str) exception message to use on test failure

  • kwargs – (dict) any additional **kwargs keyword arguments to include in return Result object

Return result

nornir.core.task.Result object with test results

nornir_salt.plugins.processors.TestsProcessor.CerberusTest(host, result, schema, allow_unknown=True, **kwargs)

Function to check results using Cerberus module schema. Results must be a structured data - dictionary, list - strings and other types of data not supported.

Parameters
  • host – (obj) Nornir host object

  • resultnornir.core.task.Result object

  • schema – (dictionary) Cerberus schema definition to us for validation

  • allow_uncknown – (bool) Cerberus allow unknown parameter, default is True

  • kwargs – (dict) any additional **kwargs keyword arguments to include in return Result object

Warning

Cerberus library only supports validation of dictionary structures, while nested elements could be lists, as a result, CerberusTest function was coded to support validation of dictionary or list of dictionaries results.

Note

kwargs name key value formatted using python format function supplying dictionary being validated as arguments

nornir_salt.plugins.processors.TestsProcessor.EvalTest(host, result, expr, revert=False, err_msg=None, globs={}, **kwargs)

Function to check result running python built-in Eval or Exec function against provided python expression.

This function in its use cases sits in between pre-built test function such as ContainsTest or EqualTest and running custom Python test function using CustomFunctionTest function. Eval allows to use any python expressions that evaluates to True or False without the need to write custom functions.

If expression string starts with assert, will use Exec function, uses Eval for everything else.

Eval and Exec functions’ globals dictionary populated with result and host variables, result contains nornir.core.task.Result result attribute while host references Nornir host object. This allows to use expressions like this:

"'7.7.7.7' in result"
"assert '7.7.7.8' in result, 'NTP server 7.7.7.8 not in config'"
"len(result.splitlines()) == 3"

Eval and Exec functions’ globals dictionary attribute merged with **globs supplied to EvalTest function call, that allows to use any additional variables or functions. For example, below is equivalent to running contains_lines test:

tests = [
    {
        "test": "eval",
        "task": "show run | inc logging",
        "name": "Test Syslog config",
        "expr": "all(map(lambda line: line in result, lines))",
        "globs": {
            "lines": ["logging host 1.1.1.1", "logging host 2.2.2.2"]
        },
        "err_msg": "Syslog config is wrong"
    }
]

lines variable shared with eval globals space, allowing to reference it as part of expression.

Parameters
  • host – (obj) Nornir host object

  • result – (obj) nornir.core.task.Result object

  • expr – (str) Python expression to evaluate

  • revert – (bool) if True, changes results to opposite - check for inequality

  • err_msg – (str) exception message to use on test failure

  • globs – (dict) dictionary to use as eval/exec globals space

  • kwargs – (dict) any additional **kwargs keyword arguments to include in return Result object

Return result

nornir.core.task.Result object with test results

nornir_salt.plugins.processors.TestsProcessor.CustomFunctionTest(host, result, function_file=None, function_text=None, function_call=None, function_name='run', function_kwargs={}, use_all_tasks=False, globals_dictionary={}, **kwargs)

Wrapper around calling custom function to perform results checks.

Parameters
  • host – (obj) Nornir host object

  • resultnornir.core.task.Result object

  • function_name – (str) function name, default is run

  • function_file – (str) OS path to file with function_name function

  • function_text – (str) Python code text for function_name function

  • function_call – (callable) reference to callable python function

  • use_all_tasks – (bool) if True passes all host results to custom function, default False

  • globals_dictionary – (dict) dictionary to merge with global space of the custom function, used only if function_file or function_text arguments provided.

  • function_kwargs – (dict) **function_kwargs to pass on to custom function

  • kwargs – (dict) any additional key word arguments to include in results

Warning

function_file and function_text use exec function to compile python code, using test functions from untrusted sources can be dangerous.

Custom functions should accept one positional argument for results following these rules:

  • if task is a string result is nornir.core.task.Result

  • if task is a list of task names result is a list of nornir.core.task.Result objects of corresponding tasks

  • if use_all_tasks set to True result is nornir.core.task.MultiResult object

Any additional parameters can be passed to custom test function using function_kwargs arguments.

Custom function can return a dictionary or a list of dictionaries to include in results. Each dictionary can have any keys, but it is recommended to have at least these keys:

  • exception - error description if any

  • result - “PASS”, “FAIL” or “ERROR” string

  • success - boolean True or False

If a list returned by custom function, each list item forms individual result item.

If custom test function returns empty list, empty dictionary or None or True test considered successful and dictionary added to overall results with result key set to PASS.

If custom test function returns False test outcome considered unsuccessful and dictionary added to overall results with result key set to FAIL.

Sample custom test function to accept Result object when use_all_tasks set to False and task is a sting representing name of the task:

def custom_test_function(result):
    # result is nornir.core.task.Result object
    if "7.7.7.8" not in result.result:
        return {
            "exception": "Server 7.7.7.8 not in config",
            "result": "FAIL",
            "success": False
        }

Sample custom test function to accept MultiResult object when use_all_tasks set to True:

def custom_test_function(result):
    # result is nornir.core.task.MultiResult object - list of nornir.core.task.Result objects
    ret = []
    for item in result:
        if item.result == None: # skip empty results
            continue
        elif item.name == "show run | inc ntp":
            if "7.7.7.8" not in item.result:
                ret.append({
                    "exception": "NTP Server 7.7.7.8 not in config",
                    "result": "FAIL",
                    "success": False
                })
        elif item.name == "show run | inc logging":
            if "1.1.1.1" not in item.result:
                ret.append({
                    "exception": "Logging Server 1.1.1.1 not in config",
                    "result": "FAIL",
                    "success": False
                })
    return ret