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 testtask
- name of the taskresult
- test resultPASS
,FAIL
orERROR
success
- test success status True (PASS) or False (FAIL or ERROR)exception
- description of failure reasontest
- test type to perform e.g.contains
,custom
,cerberus
etc.criteria
- criteria that failed the test e.g. pattern or stringfailed
- 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 runname
- optional, name of the test, if not provided derived from task, test and criteria argumentstask
- 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 ifuse_all_tasks
is set to trueerr_msg
- optional, error message string to use for exception in case of test failurepath
- optional, dot separated string representing path to data to test within resultsreport_all
- optional, boolean, default is False, ifpath
evaluates to a list of items andreport_all
set to True, reports all tests, even successful onesuse_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
callsContainsTest
!contains
orncontains
callsContainsTest
with kwargs:{"revert": True}
contains_lines
callsContainsLinesTest
!contains_lines
orncontains_lines
callsContainsLinesTest
with kwargs:{"revert": True}
contains_re
callsContainsTest
with kwargs:{"use_re": True}
!contains_re
orncontains_re
callsContainsTest
with kwargs:{"revert": True, "use_re": True}
equal
callsEqualTest
!equal
calls ornequal
callsEqualTest
with kwargs:{"revert": True}
cerberus
callsCerberusTest
custom
callsCustomFunctionTest
eval
callsEvalTest
In addition to aliases, test
argument can reference actual test functions names:
ContainsTest
callsContainsTest
ContainsLinesTest
callsContainsLinesTest
EqualTest
callsEqualTest
CerberusTest
callsCerberusTest
CustomFunctionTest
callsCustomFunctionTest
EvalTest
callsEvalTest
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
.
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 itemfailed_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, 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
objectpattern – (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
objectpattern – (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
objectpattern – (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
objectschema – (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
orExec
function against provided python expression.This function in its use cases sits in between pre-built test function such as
ContainsTest
orEqualTest
and running custom Python test function usingCustomFunctionTest
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 useExec
function, usesEval
for everything else.Eval and Exec functions’
globals
dictionary populated withresult
andhost
variables,result
containsnornir.core.task.Result
result attribute whilehost
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 toEvalTest
function call, that allows to use any additional variables or functions. For example, below is equivalent to runningcontains_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 witheval
globals space, allowing to reference it as part of expression.- Parameters
host – (obj) Nornir host object
result – (obj)
nornir.core.task.Result
objectexpr – (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
spacekwargs – (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
objectfunction_name – (str) function name, default is
run
function_file – (str) OS path to file with
function_name
functionfunction_text – (str) Python code text for
function_name
functionfunction_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
orfunction_text
arguments provided.function_kwargs – (dict)
**function_kwargs
to pass on to custom functionadd_host – (bool) default is False, if True adds
host
argument tofunction_kwargs
as a reference to Nornir Host object that this function executing forkwargs – (dict) any additional key word arguments to include in results
Warning
function_file
andfunction_text
useexec
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 isnornir.core.task.Result
if
task
is a list of task names result is a list ofnornir.core.task.Result
objects of corresponding tasksif
use_all_tasks
set to True result isnornir.core.task.MultiResult
object
If
add_host
set to True, custom function must accepthost
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 anyresult
- “PASS”, “FAIL” or “ERROR” stringsuccess
- 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 toPASS
.If custom test function returns False test outcome considered unsuccessful and dictionary added to overall results with
result
key set toFAIL
.Sample custom test function to accept
Result
object whenuse_all_tasks
set to False andtask
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 whenuse_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