All methods of using tests¶
There are some things you can and some things you can’t do with tests. This tutorial will show all of them.
[1]:
from nornir_tests.plugins.tests import *
from nornir_tests.plugins.tasks import wrap_task
from nornir_napalm.plugins.tasks import napalm_get, napalm_ping
from nornir_tests.plugins.functions import print_result
from nornir import InitNornir
nr = InitNornir(
inventory={
"plugin": "SimpleInventory",
"options": {
"host_file": "data/hosts.yaml",
"group_file": "data/groups.yaml",
"defaults_file": "data/defaults.yaml",
},
},
)
vyos = nr.filter(name="vyos")
@ style decorator use¶
This will only work as designed if not using task.run inside the function. If there were a bunch of task.run statements in the function the decorators applied using @ syntax would not get applied.
[2]:
@until(retries=5, delay=5)
@timing(max_run_time=5, fail_task=True)
@jpath(path='interfaces.eth0.is_enabled', assertion='is_true', fail_task=True)
def at_syntax_test(task):
return napalm_get(task, getters=['interfaces'])
print_result(vyos.run(task=at_syntax_test), vars=['result', 'tests'])
at_syntax_test******************************************************************
* vyos ** changed : False ******************************************************
vvvv at_syntax_test ** changed : False vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv INFO
{'interfaces':{'eth0':{'description':'',
'is_enabled':True,
'is_up':True,
'last_flapped':-1.0,
'mac_address':'08:00:27:e0:28:63',
'mtu':-1,
'speed':0},
'lo':{'description':'',
'is_enabled':True,
'is_up':True,
'last_flapped':-1.0,
'mac_address':'00:00:00:00:00:00',
'mtu':-1,
'speed':0}}}
P JpathRecord - {'assertion': 'is_true',
'fail_task': True,
'path': 'interfaces.eth0.is_enabled',
'result_attr': 'result'}
{'matches': ['interfaces.eth0.is_enabled']}
P TimingRecord - {'fail_task': True, 'max_run_time': 5}
{'run_time': 2.0251331329345703,
't0': 1600801296.0013554,
't1': 1600801298.0264885}
P UntilRecord - {'delay': 5, 'retries': 5}
{'run_time': 13.532038927078247,
't0': 1600801284.494459,
't1': 1600801298.0264978}
^^^^ END at_syntax_test ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Using nr.run with tasks that return results¶
[3]:
nr.data.reset_failed_hosts()
print_result(vyos.run(
task=wrap_task(napalm_get),
getters=['interfaces'],
tests=[
timing(),
jpath(path='interfaces.eth0.is_up', assertion='is_true')
]
), vars=['result', 'tests'])
napalm_get**********************************************************************
* vyos ** changed : False ******************************************************
vvvv napalm_get ** changed : False vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv INFO
{'interfaces':{'eth0':{'description':'',
'is_enabled':True,
'is_up':True,
'last_flapped':-1.0,
'mac_address':'08:00:27:e0:28:63',
'mtu':-1,
'speed':0},
'lo':{'description':'',
'is_enabled':True,
'is_up':True,
'last_flapped':-1.0,
'mac_address':'00:00:00:00:00:00',
'mtu':-1,
'speed':0}}}
P TimingRecord - {'max_run_time': 9223372036854775807}
{'run_time': 2.011794090270996,
't0': 1600801298.441358,
't1': 1600801300.4531522}
P JpathRecord - {'assertion': 'is_true',
'path': 'interfaces.eth0.is_up',
'result_attr': 'result'}
{'matches': ['interfaces.eth0.is_up']}
^^^^ END napalm_get ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Using task.run inside grouped task¶
[4]:
def grouped_task(task):
task.run(
wrap_task(napalm_ping),
dest='192.168.99.1',
tests=[
timing()
]
)
task.run(
wrap_task(napalm_get),
getters=['interfaces'],
tests=[
jpath(path='interfaces.eth0.is_up', assertion='is_true')
]
)
print_result(vyos.run(task=grouped_task), vars=['result', 'tests'])
grouped_task********************************************************************
* vyos ** changed : False ******************************************************
vvvv grouped_task ** changed : False vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv INFO
---- napalm_ping ** changed : False -------------------------------------------- INFO
{'success':{'packet_loss':0,
'probes_sent':5,
'results':[{'ip_address': '192.168.99.1', 'rtt': 3.588}],
'rtt_avg':3.588,
'rtt_max':4.07,
'rtt_min':3.105,
'rtt_stddev':0.35}}
P TimingRecord - {'max_run_time': 9223372036854775807}
{'run_time': 4.9137420654296875,
't0': 1600801300.7780375,
't1': 1600801305.6917796}
---- napalm_get ** changed : False --------------------------------------------- INFO
{'interfaces':{'eth0':{'description':'',
'is_enabled':True,
'is_up':True,
'last_flapped':-1.0,
'mac_address':'08:00:27:e0:28:63',
'mtu':-1,
'speed':0},
'lo':{'description':'',
'is_enabled':True,
'is_up':True,
'last_flapped':-1.0,
'mac_address':'00:00:00:00:00:00',
'mtu':-1,
'speed':0}}}
P JpathRecord - {'assertion': 'is_true',
'path': 'interfaces.eth0.is_up',
'result_attr': 'result'}
{'matches': ['interfaces.eth0.is_up']}
^^^^ END grouped_task ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Not supported¶
Testing a grouped_task at the level of running nr.run will not work. There is not much you would be able to test without really complex logic anyhow considering this always returns an AggregatedResult.