mirror of
https://github.com/kernelkit/infix.git
synced 2026-07-22 01:13:00 +02:00
test/case: Add interface status tests
This commit is contained in:
committed by
Joachim Wiberg
parent
a1a4491d37
commit
b0d3976ae0
@@ -1,3 +1,4 @@
|
||||
---
|
||||
- case: vlan_ping.py
|
||||
- case: ipv4_address.py
|
||||
- case: iface_status.py
|
||||
|
||||
Executable
+50
@@ -0,0 +1,50 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import infamy
|
||||
import infamy.iface as iface
|
||||
|
||||
|
||||
def print_err_msg(interface, param, val):
|
||||
f"{param} failure for interface <{interface}> [{param}: {val}]"
|
||||
|
||||
def assert_if_index(target, interface):
|
||||
if_index = iface.get_if_index(target, interface)
|
||||
if interface == "lo":
|
||||
assert (if_index == 1), print_err_msg(interface, "if-index", if_index)
|
||||
else:
|
||||
assert (if_index > 1 and if_index < 65535), print_err_msg(interface, "if-index", if_index)
|
||||
|
||||
def assert_oper_status(target, interface):
|
||||
oper_status = iface.get_oper_status(target, interface)
|
||||
if interface == "lo":
|
||||
assert (oper_status == "up" or oper_status == "unknown"), print_err_msg(interface, "oper-status", oper_status)
|
||||
else:
|
||||
assert (oper_status == "up"), print_err_msg(interface, "oper-status", oper_status)
|
||||
|
||||
def asser_iface_exists(target, interface):
|
||||
assert iface.interface_exist(target, interface), f"Interface <{interface}> does not exist."
|
||||
|
||||
|
||||
with infamy.Test() as test:
|
||||
with test.step("Initialize"):
|
||||
env = infamy.Env(infamy.std_topology("1x1"))
|
||||
target = env.attach("target", "mgmt")
|
||||
|
||||
iface.print_iface_status(target)
|
||||
|
||||
loopback_iface = "lo"
|
||||
_, mgmt_iface = env.ltop.xlate("target", "mgmt")
|
||||
ifaces_under_test = [loopback_iface, mgmt_iface]
|
||||
print(f"Interfaces under test: {ifaces_under_test}")
|
||||
|
||||
for interface in ifaces_under_test:
|
||||
with test.step(f"Verifying <{interface}> interface"):
|
||||
asser_iface_exists(target, interface)
|
||||
|
||||
with test.step(f"Verifying <if-index> for <{interface}> interface"):
|
||||
assert_if_index(target, interface)
|
||||
|
||||
with test.step(f"Verifying <oper-status> for <{interface}> interface"):
|
||||
assert_oper_status(target, interface)
|
||||
|
||||
test.succeed()
|
||||
@@ -0,0 +1,61 @@
|
||||
def extract_iface_param(json_content, param):
|
||||
"""Returns (extracted) value for parameter 'param'"""
|
||||
interfaces = json_content.get('interfaces')
|
||||
if not interfaces:
|
||||
return None
|
||||
|
||||
for interface in interfaces.get('interface'):
|
||||
if param not in interface:
|
||||
continue
|
||||
return interface[param]
|
||||
|
||||
return None
|
||||
|
||||
def interface_exist(target, iface):
|
||||
"""Verify that the target interface exists"""
|
||||
try:
|
||||
content = target.get_dict(f"/ietf-interfaces:interfaces/ietf-interfaces:interface[name='{iface}']")
|
||||
name = extract_iface_param(content, 'name')
|
||||
return (name != None)
|
||||
except:
|
||||
return False
|
||||
|
||||
def get_if_index(target, iface):
|
||||
"""Return value of 'if-index' (interface index) parameter for the target interface"""
|
||||
try:
|
||||
content = target.get_dict(f"/ietf-interfaces:interfaces/ietf-interfaces:interface[name='{iface}']/ietf-interfaces:if-index")
|
||||
if_index = extract_iface_param(content, 'if-index')
|
||||
return if_index
|
||||
except:
|
||||
return None
|
||||
|
||||
def get_oper_status(target, iface):
|
||||
"""Return value of 'oper-status' (operational status) parameter for the target interface"""
|
||||
try:
|
||||
content = target.get_dict(f"/ietf-interfaces:interfaces/ietf-interfaces:interface[name='{iface}']/ietf-interfaces:oper-status")
|
||||
oper_status = extract_iface_param(content, 'oper-status')
|
||||
return oper_status
|
||||
except:
|
||||
return None
|
||||
|
||||
def print_iface_status(target):
|
||||
"""Print status parameters for all target interfaces"""
|
||||
try:
|
||||
content = target.get_dict(f"/ietf-interfaces:interfaces")
|
||||
interfaces = content.get('interfaces')
|
||||
if interfaces:
|
||||
interface_list = interfaces.get('interface')
|
||||
if interface_list and isinstance(interface_list, list):
|
||||
col1 = "name"
|
||||
col2 = "if-index"
|
||||
col3 = "oper-status"
|
||||
print('-'*36)
|
||||
print(f"{col1: <12}{col2: <12}{col3: <12}")
|
||||
print('-'*36)
|
||||
for interface in interface_list:
|
||||
print(f"{interface['name']: <12}"
|
||||
f"{interface['if-index']: <12}"
|
||||
f"{interface['oper-status']: <12}")
|
||||
print('-'*36)
|
||||
except:
|
||||
print(f"Failed to get interfaces' status from target {target}")
|
||||
Reference in New Issue
Block a user