router.vtysh_cmd("no debug northbound callbacks configuration")
-@retry(retry_timeout=30, initial_wait=1)
+@retry(retry_timeout=30, initial_wait=0.1)
def _do_oper_test(tgen, qr, seconds_left=None):
r1 = tgen.gears["r1"].net
"-------DIFF---------\n%s\n---------DIFF----------",
pprint.pformat(cmpout),
)
+ cmpout = str(cmpout)
else:
cmpout = tt_json_cmp(ojson, ejson, exact=True)
if cmpout and ejson_alt is not None:
a += step
+@retry(retry_timeout=30, initial_wait=0.1)
+def check_kernel_net(r1, net, vrf):
+ addr = ipaddress.ip_network(net)
+ vrfstr = f" vrf {vrf}" if vrf else ""
+ if addr.version == 6:
+ kernel = r1.cmd_raises(f"ip -6 route show{vrfstr}")
+ else:
+ kernel = r1.cmd_raises(f"ip -4 route show{vrfstr}")
+
+ nentries = len(re.findall("\n", kernel))
+ logging.info("checking kernel routing table%s: (%s entries)", vrfstr, nentries)
+ assert str(net) in kernel, f"Failed to find '{net}' in {nentries} entries"
+
+
@retry(retry_timeout=30, initial_wait=0.1)
def check_kernel_32(r1, start_addr, count, vrf, step=1):
start = ipaddress.ip_address(start_addr)
import pytest
from lib.topogen import Topogen
-from oper import check_kernel_32, do_oper_test
+from oper import check_kernel_32, check_kernel_net, do_oper_test
pytestmark = [pytest.mark.staticd, pytest.mark.mgmtd]
]
r1 = tgen.gears["r1"].net
+
check_kernel_32(r1, "11.11.11.11", 1, "")
check_kernel_32(r1, "12.12.12.12", 1, "")
check_kernel_32(r1, "13.13.13.13", 1, "red")
check_kernel_32(r1, "14.14.14.14", 1, "red")
+
+ check_kernel_net(r1, "2001:1111::/64", "")
+ check_kernel_net(r1, "2002:2222::/64", "")
+ check_kernel_net(r1, "2003:333::/64", "red")
+ check_kernel_net(r1, "2004:4444::/64", "red")
+
do_oper_test(tgen, query_results)