mirror of
https://github.com/kernelkit/infix.git
synced 2026-07-31 13:03:02 +02:00
Fix problems when running test on real hardware not qemu
* sniffer did not work correctly * links did not come upp as quick as in qemu * must_reach in netns was to narrow, increase timeout * add wait function to handle poll
This commit is contained in:
@@ -3,7 +3,7 @@
|
||||
import copy
|
||||
import infamy
|
||||
import infamy.iface as iface
|
||||
|
||||
from infamy.wait import wait_mac_address
|
||||
|
||||
with infamy.Test() as test:
|
||||
with test.step("Initialize"):
|
||||
@@ -37,8 +37,6 @@ with infamy.Test() as test:
|
||||
del i["phys-address"]
|
||||
break
|
||||
target.put_diff_dicts("ietf-interfaces", running, new)
|
||||
mac = iface.get_phys_address(target, tport)
|
||||
print(f"Target iface {tport} current mac: {mac}")
|
||||
assert mac == pmac
|
||||
wait_mac_address(target, tport, pmac)
|
||||
|
||||
test.succeed()
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
import infamy
|
||||
import copy
|
||||
from infamy.wait import wait_links
|
||||
def test_ping(hport, should_pass):
|
||||
with infamy.IsolatedMacVlan(hport) as ns:
|
||||
pingtest = ns.runsh("""
|
||||
@@ -10,13 +11,11 @@ def test_ping(hport, should_pass):
|
||||
ip link set iface up
|
||||
ip link add dev vlan10 link iface up type vlan id 10
|
||||
ip addr add 10.0.0.1/24 dev vlan10
|
||||
|
||||
ping -c1 -w5 10.0.0.2 || exit 1
|
||||
""")
|
||||
|
||||
if (pingtest.returncode and should_pass) or (not pingtest.returncode and not should_pass):
|
||||
print(pingtest.stdout)
|
||||
test.fail()
|
||||
if(should_pass):
|
||||
ns.must_reach("10.0.0.2")
|
||||
else:
|
||||
ns.must_not_reach("10.0.0.2")
|
||||
|
||||
with infamy.Test() as test:
|
||||
with test.step("Initialize"):
|
||||
@@ -55,6 +54,9 @@ with infamy.Test() as test:
|
||||
}
|
||||
})
|
||||
|
||||
with test.step("Waiting for links to come up"):
|
||||
wait_links(target, [tport])
|
||||
|
||||
with test.step("Ping 10.0.0.2 from VLAN 10 on host:data with IP 10.0.0.1"):
|
||||
_, hport = env.ltop.xlate("host", "data")
|
||||
test_ping(hport,True)
|
||||
|
||||
@@ -46,6 +46,14 @@ def get_phys_address(target, iface):
|
||||
"""Fetch interface MAC address (operational status)"""
|
||||
return _iface_get_param(target, iface, "phys-address")
|
||||
|
||||
def is_phys_address(target,iface,addr):
|
||||
"""Check if MAC address is a specific value"""
|
||||
return get_phys_address(target,iface) == addr
|
||||
|
||||
def get_oper_up(target,iface):
|
||||
state=get_oper_status(target,iface)
|
||||
return state == "up"
|
||||
|
||||
def print_iface_status(target):
|
||||
"""Print status parameters for all target interfaces"""
|
||||
try:
|
||||
|
||||
@@ -85,8 +85,9 @@ class IsolatedMacVlan:
|
||||
ip addr add {addr}/{prefix_length} dev iface
|
||||
""", check=True)
|
||||
|
||||
def ping(self, daddr, count=1, timeout=2, check=False):
|
||||
return self.runsh(f"""set -ex; ping -c {count} -w {timeout} {daddr}""", check=check)
|
||||
|
||||
def ping(self, daddr, count=1, timeout=5, interval=2, check=False):
|
||||
return self.runsh(f"""set -ex; ping -c {count} -w {timeout} -i {interval} {daddr}""", check=check)
|
||||
|
||||
def must_reach(self, daddr):
|
||||
res = self.ping(daddr)
|
||||
|
||||
@@ -0,0 +1,18 @@
|
||||
def _get_routes(target,protocol):
|
||||
xpath=f"/ietf-routing:routing"
|
||||
rib = target.get_data(xpath)["routing"]["ribs"]["rib"]
|
||||
for r in rib:
|
||||
if r["name"] != protocol:
|
||||
continue
|
||||
return r.get("routes", {}).get("route",{})
|
||||
|
||||
def _exist_route(target,destination_prefix, protocol):
|
||||
routes=_get_routes(target,protocol)
|
||||
for r in routes:
|
||||
if(r["destination-prefix"] == destination_prefix):
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def ipv4_route_exist(target, destination_prefix):
|
||||
return _exist_route(target,destination_prefix, "ipv4")
|
||||
@@ -18,7 +18,7 @@ class Sniffer:
|
||||
os.unlink(self.pcap.name)
|
||||
|
||||
def __enter__(self):
|
||||
cmd = f"tcpdump -lni iface -w {self.pcap.name} {self.expr}"
|
||||
cmd = f"tshark -lni iface -w {self.pcap.name} {self.expr}"
|
||||
arg = cmd.split(" ")
|
||||
self.proc = self.netns.popen(arg, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
|
||||
|
||||
@@ -34,8 +34,7 @@ class Sniffer:
|
||||
except OSError:
|
||||
pass
|
||||
self.proc.wait()
|
||||
return True
|
||||
|
||||
def output(self):
|
||||
"""Return PCAP output"""
|
||||
return self.netns.runsh(f"tcpdump -n -r {self.pcap.name}")
|
||||
return self.netns.runsh(f"tshark -n -r {self.pcap.name}")
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
import time
|
||||
import infamy.iface
|
||||
def wait(func, *args):
|
||||
timeout = 10
|
||||
while(timeout>0):
|
||||
if len(args) == 0:
|
||||
f=func()
|
||||
else:
|
||||
f=func(*args)
|
||||
|
||||
if(f):
|
||||
return True
|
||||
timeout-=1
|
||||
time.sleep(1)
|
||||
|
||||
return False
|
||||
|
||||
def wait_links(target, ifaces):
|
||||
for i in ifaces:
|
||||
if not wait(infamy.iface.get_oper_up, target, i):
|
||||
raise Exception("Interface did not come up in time.")
|
||||
|
||||
|
||||
def wait_mac_address(target, iface, mac):
|
||||
if not wait(infamy.iface.is_phys_address, target, iface, mac):
|
||||
raise Exception("Failed waiting for MAC address")
|
||||
Reference in New Issue
Block a user