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:

Nornir 3.0 and beyond
Cerberus module for CerberusTest function

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 arguments/keys each test dictionary may contain:

test - mandatory, name of test function to run
name - optional, name of the test, if not provided derived from task, test and criteria arguments
task - optional, name of the task to check results for or list of task names to use with custom test function, task parameter might be omitted if use_all_tasks is set to true
err_msg - optional, error message string to use for exception in case of test failure
path - optional, dot separated string representing path to data to test within results
report_all - optional, boolean, default is False, if path evaluates to a list of items and report_all set to True, reports all tests, even successful ones
use_all_tasks - optional, boolean to indicate if need to supply all task results to the test function

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.plugins.processors import TestsProcessor
from nornir_salt.plugins.functions import ResultSerializer
from nornir_salt.plugins.tasks import 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 unicast 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 unicast 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.plugins.processors import TestsProcessor
from nornir_salt.plugins.functions import ResultSerializer
from nornir_salt.plugins.tasks import 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.

Using Tests Suite Templates

Starting with Nornir-Salt version 0.16.0 support added to dynamically render hosts’ tests suites from hosts’ data using Jinja2 templates YAML formatted strings.

Warning

Nornir-Salt tasks coded to make use of per-host commands - netmiko_send_commands, scrapli_send_commands, pyats_send_commands, napalm_send_commands - custom task plugins can make use of host.data["__task__"]["commands"] commands list to produce per host commands output. This is required for TestsProcessor to work, as sub-task should be named after cli commands they containing results for.

Given hosts’ Nornir inventory data content:

hosts:
  IOL1:
    data:
      interfaces_test:
      - admin_status: is up
        description: Description
        line_status: line protocol is up
        mtu: IP MTU 9200
        name: Ethernet1
      - admin_status: is up
        description: Description
        line_status: line protocol is up
        mtu: IP MTU 65535
        name: Loopback1
      software_version: cEOS
  IOL2:
    data:
      software_version: cEOS

Sample code to run test suite Jinja2 template:

import pprint

from nornir import InitNornir
from nornir_salt.plugins.processors import TestsProcessor
from nornir_salt.plugins.functions import ResultSerializer
from nornir_salt.plugins.tasks import netmiko_send_commands

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

tests = [
    '''
- task: "show version"
  test: contains
  pattern: "{{ host.software_version }}"
  name: check ceos version

{% for interface in host.interfaces_test %}
- task: "show interface {{ interface.name }}"
  test: contains_lines
  pattern:
    - {{ interface.admin_status }}
    - {{ interface.line_status }}
    - {{ interface.mtu }}
    - {{ interface.description }}
  name: check interface {{ interface.name }} status
{% endfor %}
''',
    {
        "name": "Test NTP config",
        "task": "show run | inc ntp",
        "test": "contains",
        "pattern": "ntp server 7.7.7.8",
    }
]

nr_with_tests = nr.with_processors([
    TestsProcessor(tests)
])

results = nr_with_tests.run(
    task=netmiko_send_commands
)

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

pprint.pprint(results_dictionary)

# should print something like:

# [{'host': 'IOL1', 'name': 'show version', 'result': 'PASS'},
# {'host': 'IOL2', 'name': 'show version', 'result': 'PASS'},
# {'host': 'IOL1', 'name': 'show interface Ethernet1', 'result': 'PASS'},
# {'host': 'IOL1', 'name': 'show interface Ethernet2', 'result': 'PASS'},
# {'host': 'IOL1', 'name': 'show run | inc ntp', 'result': 'PASS'},
# {'host': 'IOL2', 'name': 'show run | inc ntp', 'result': 'PASS'}]

Test suite template rendered using individual host’s data forming per-host test suite. CLI show commands to collect from host device automatically extracted from per-host test suite. For example, for above data these are commands collected from devices:

ceos1 - “show version”, “show interface Ethernet1”, “show interface Ethernet2”, “show run | inc ntp”
ceos2 - “show version”, “show run | inc ntp”

Collected show commands output tested using rendered test suite on a per-host basis.

Tests Reference

class nornir_salt.plugins.processors.TestsProcessor.TestsProcessor(tests=None, remove_tasks=True, failed_only=False, jinja_kwargs=None, tests_data=None, build_per_host_tests=False, render_tests=True, subset=None, **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
build_per_host_tests – (bool) if True, renders and forms per host tests and show commands
render_tests – (bool) if True will use Jinja2 to render tests, rendering skipped otherwise
jinja_kwargs – (dict) Dictionary of arguments for jinja2.Template object, default is {"trim_blocks": True, "lstrip_blocks": True}
subset – (list or str) list or string with comma separated glob patterns to match tests’ names to execute. Patterns are not case-sensitive. Uses fnmatch Python built-in function to do glob patterns matching
tests_data – (dict) dictionary of parameters to supply for test suite templates rendering

nornir_salt.plugins.processors.TestsProcessor.ContainsTest(host, result, pattern, use_re=False, count=None, count_ge=None, count_le=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
count_ge – (int) check number of pattern occurrences in the output is greater or equal to given value
count_le – (int) check number of pattern occurrences in the output is lower or equal to given value
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
result – nornir.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=None, **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=None, globals_dictionary=None, add_host=False, **kwargs)

Wrapper around calling custom function to perform results checks.

Parameters

host – (obj) Nornir host object
result – nornir.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
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
add_host – (bool) default is False, if True adds host argument to function_kwargs as a reference to Nornir Host object that this function executing for
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

If add_host set to True, custom function must accept host argument as a reference to Nornir host object this task results are being tested for.

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