844 lines
26 KiB
Python
844 lines
26 KiB
Python
#!/usr/bin/env python
|
|
# SPDX-License-Identifier: ISC
|
|
|
|
#
|
|
# test_isis_topo1.py
|
|
# Part of NetDEF Topology Tests
|
|
#
|
|
# Copyright (c) 2017 by
|
|
# Network Device Education Foundation, Inc. ("NetDEF")
|
|
#
|
|
|
|
"""
|
|
test_isis_topo1.py: Test ISIS topology.
|
|
"""
|
|
import datetime
|
|
import functools
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import pytest
|
|
|
|
CWD = os.path.dirname(os.path.realpath(__file__))
|
|
sys.path.append(os.path.join(CWD, "../"))
|
|
|
|
# pylint: disable=C0413
|
|
from lib import topotest
|
|
from lib.common_config import (
|
|
retry,
|
|
stop_router,
|
|
start_router,
|
|
)
|
|
from lib.topogen import Topogen, TopoRouter, get_topogen
|
|
from lib.topolog import logger
|
|
|
|
|
|
pytestmark = [pytest.mark.isisd]
|
|
|
|
VERTEX_TYPE_LIST = [
|
|
"pseudo_IS",
|
|
"pseudo_TE-IS",
|
|
"IS",
|
|
"TE-IS",
|
|
"ES",
|
|
"IP internal",
|
|
"IP external",
|
|
"IP TE",
|
|
"IP6 internal",
|
|
"IP6 external",
|
|
"UNKNOWN",
|
|
]
|
|
|
|
|
|
def build_topo(tgen):
|
|
"Build function"
|
|
|
|
# Add ISIS routers:
|
|
# r1 r2
|
|
# | sw1 | sw2
|
|
# r3 r4
|
|
# | |
|
|
# sw3 sw4
|
|
# \ /
|
|
# r5
|
|
for routern in range(1, 6):
|
|
tgen.add_router("r{}".format(routern))
|
|
|
|
# r1 <- sw1 -> r3
|
|
sw = tgen.add_switch("sw1")
|
|
sw.add_link(tgen.gears["r1"])
|
|
sw.add_link(tgen.gears["r3"])
|
|
|
|
# r2 <- sw2 -> r4
|
|
sw = tgen.add_switch("sw2")
|
|
sw.add_link(tgen.gears["r2"])
|
|
sw.add_link(tgen.gears["r4"])
|
|
|
|
# r3 <- sw3 -> r5
|
|
sw = tgen.add_switch("sw3")
|
|
sw.add_link(tgen.gears["r3"])
|
|
sw.add_link(tgen.gears["r5"])
|
|
|
|
# r4 <- sw4 -> r5
|
|
sw = tgen.add_switch("sw4")
|
|
sw.add_link(tgen.gears["r4"])
|
|
sw.add_link(tgen.gears["r5"])
|
|
|
|
|
|
def setup_module(mod):
|
|
"Sets up the pytest environment"
|
|
tgen = Topogen(build_topo, mod.__name__)
|
|
tgen.start_topology()
|
|
|
|
# For all registered routers, load the zebra configuration file
|
|
for rname, router in tgen.routers().items():
|
|
router.load_config(
|
|
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
|
|
)
|
|
router.load_config(
|
|
TopoRouter.RD_ISIS, os.path.join(CWD, "{}/isisd.conf".format(rname))
|
|
)
|
|
|
|
# After loading the configurations, this function loads configured daemons.
|
|
tgen.start_router()
|
|
|
|
|
|
def teardown_module():
|
|
"Teardown the pytest environment"
|
|
tgen = get_topogen()
|
|
|
|
# This function tears down the whole topology.
|
|
tgen.stop_topology()
|
|
|
|
|
|
def test_isis_convergence():
|
|
"Wait for the protocol to converge before starting to test"
|
|
tgen = get_topogen()
|
|
# Don't run this test if we have any failure.
|
|
if tgen.routers_have_failure():
|
|
pytest.skip(tgen.errors)
|
|
|
|
logger.info("waiting for ISIS protocol to converge")
|
|
for rname, router in tgen.routers().items():
|
|
filename = "{0}/{1}/{1}_topology.json".format(CWD, rname)
|
|
expected = json.loads(open(filename).read())
|
|
|
|
def compare_isis_topology(router, expected):
|
|
"Helper function to test ISIS topology convergence."
|
|
actual = json.loads(router.vtysh_cmd("show isis topology json"))
|
|
return topotest.json_cmp(actual, expected)
|
|
|
|
test_func = functools.partial(compare_isis_topology, router, expected)
|
|
(result, diff) = topotest.run_and_expect(test_func, None, wait=0.5, count=120)
|
|
assert result, "ISIS did not converge on {}:\n{}".format(rname, diff)
|
|
|
|
|
|
def test_isis_route_installation():
|
|
"Check whether all expected routes are present"
|
|
tgen = get_topogen()
|
|
# Don't run this test if we have any failure.
|
|
if tgen.routers_have_failure():
|
|
pytest.skip(tgen.errors)
|
|
|
|
logger.info("Checking routers for installed ISIS routes")
|
|
|
|
# Check for routes in 'show ip route json'
|
|
for rname, router in tgen.routers().items():
|
|
filename = "{0}/{1}/{1}_route.json".format(CWD, rname)
|
|
expected = json.loads(open(filename, "r").read())
|
|
|
|
def compare_isis_installed_routes(router, expected):
|
|
"Helper function to test ISIS routes installed in rib."
|
|
actual = router.vtysh_cmd("show ip route json", isjson=True)
|
|
return topotest.json_cmp(actual, expected)
|
|
|
|
test_func = functools.partial(compare_isis_installed_routes, router, expected)
|
|
(result, _) = topotest.run_and_expect(test_func, None, wait=1, count=10)
|
|
assertmsg = "Router '{}' routes mismatch".format(rname)
|
|
assert result, assertmsg
|
|
|
|
|
|
def test_isis_linux_route_installation():
|
|
"Check whether all expected routes are present and installed in the OS"
|
|
tgen = get_topogen()
|
|
# Don't run this test if we have any failure.
|
|
if tgen.routers_have_failure():
|
|
pytest.skip(tgen.errors)
|
|
|
|
logger.info("Checking routers for installed ISIS routes in OS")
|
|
|
|
# Check for routes in `ip route`
|
|
for rname, router in tgen.routers().items():
|
|
filename = "{0}/{1}/{1}_route_linux.json".format(CWD, rname)
|
|
expected = json.loads(open(filename, "r").read())
|
|
actual = topotest.ip4_route(router)
|
|
assertmsg = "Router '{}' OS routes mismatch".format(rname)
|
|
assert topotest.json_cmp(actual, expected) is None, assertmsg
|
|
|
|
|
|
def test_isis_route6_installation():
|
|
"Check whether all expected routes are present"
|
|
tgen = get_topogen()
|
|
# Don't run this test if we have any failure.
|
|
if tgen.routers_have_failure():
|
|
pytest.skip(tgen.errors)
|
|
|
|
logger.info("Checking routers for installed ISIS IPv6 routes")
|
|
|
|
# Check for routes in 'show ip route json'
|
|
for rname, router in tgen.routers().items():
|
|
filename = "{0}/{1}/{1}_route6.json".format(CWD, rname)
|
|
expected = json.loads(open(filename, "r").read())
|
|
|
|
def compare_isis_v6_installed_routes(router, expected):
|
|
"Helper function to test ISIS v6 routes installed in rib."
|
|
actual = router.vtysh_cmd("show ipv6 route json", isjson=True)
|
|
return topotest.json_cmp(actual, expected)
|
|
|
|
test_func = functools.partial(
|
|
compare_isis_v6_installed_routes, router, expected
|
|
)
|
|
(result, _) = topotest.run_and_expect(test_func, None, wait=1, count=10)
|
|
assertmsg = "Router '{}' routes mismatch".format(rname)
|
|
assert result, assertmsg
|
|
|
|
|
|
def test_isis_linux_route6_installation():
|
|
"Check whether all expected routes are present and installed in the OS"
|
|
tgen = get_topogen()
|
|
# Don't run this test if we have any failure.
|
|
if tgen.routers_have_failure():
|
|
pytest.skip(tgen.errors)
|
|
|
|
logger.info("Checking routers for installed ISIS IPv6 routes in OS")
|
|
|
|
# Check for routes in `ip route`
|
|
for rname, router in tgen.routers().items():
|
|
filename = "{0}/{1}/{1}_route6_linux.json".format(CWD, rname)
|
|
expected = json.loads(open(filename, "r").read())
|
|
actual = topotest.ip6_route(router)
|
|
assertmsg = "Router '{}' OS routes mismatch".format(rname)
|
|
assert topotest.json_cmp(actual, expected) is None, assertmsg
|
|
|
|
|
|
def test_isis_summary_json():
|
|
"Check json struct in show isis summary json"
|
|
|
|
tgen = get_topogen()
|
|
# Don't run this test if we have any failure.
|
|
if tgen.routers_have_failure():
|
|
pytest.skip(tgen.errors)
|
|
|
|
logger.info("Checking 'show isis summary json'")
|
|
for rname, _ in tgen.routers().items():
|
|
logger.info("Checking router %s", rname)
|
|
json_output = tgen.gears[rname].vtysh_cmd("show isis summary json", isjson=True)
|
|
assertmsg = "Test isis summary json failed in '{}' data '{}'".format(
|
|
rname, json_output
|
|
)
|
|
assert json_output["vrf"] == "default", assertmsg
|
|
assert json_output["areas"][0]["area"] == "1", assertmsg
|
|
assert json_output["areas"][0]["levels"][0]["id"] != "3", assertmsg
|
|
|
|
|
|
def test_isis_interface_json():
|
|
"Check json struct in show isis interface json"
|
|
|
|
tgen = get_topogen()
|
|
# Don't run this test if we have any failure.
|
|
if tgen.routers_have_failure():
|
|
pytest.skip(tgen.errors)
|
|
|
|
logger.info("Checking 'show isis interface json'")
|
|
for rname, _ in tgen.routers().items():
|
|
logger.info("Checking router %s", rname)
|
|
json_output = tgen.gears[rname].vtysh_cmd(
|
|
"show isis interface json", isjson=True
|
|
)
|
|
assertmsg = "Test isis interface json failed in '{}' data '{}'".format(
|
|
rname, json_output
|
|
)
|
|
assert (
|
|
json_output["areas"][0]["circuits"][0]["interface"]["name"]
|
|
== rname + "-eth0"
|
|
), assertmsg
|
|
|
|
for rname, router in tgen.routers().items():
|
|
logger.info("Checking router %s", rname)
|
|
json_output = tgen.gears[rname].vtysh_cmd(
|
|
"show isis interface detail json", isjson=True
|
|
)
|
|
assertmsg = "Test isis interface json failed in '{}' data '{}'".format(
|
|
rname, json_output
|
|
)
|
|
assert (
|
|
json_output["areas"][0]["circuits"][0]["interface"]["name"]
|
|
== rname + "-eth0"
|
|
), assertmsg
|
|
|
|
|
|
def test_isis_neighbor_json():
|
|
"Check json struct in show isis neighbor json"
|
|
|
|
tgen = get_topogen()
|
|
# Don't run this test if we have any failure.
|
|
if tgen.routers_have_failure():
|
|
pytest.skip(tgen.errors)
|
|
|
|
# tgen.mininet_cli()
|
|
logger.info("Checking 'show isis neighbor json'")
|
|
for rname, _ in tgen.routers().items():
|
|
logger.info("Checking router %s", rname)
|
|
json_output = tgen.gears[rname].vtysh_cmd(
|
|
"show isis neighbor json", isjson=True
|
|
)
|
|
assertmsg = "Test isis neighbor json failed in '{}' data '{}'".format(
|
|
rname, json_output
|
|
)
|
|
assert (
|
|
json_output["areas"][0]["circuits"][0]["interface"] == rname + "-eth0"
|
|
), assertmsg
|
|
|
|
for rname, router in tgen.routers().items():
|
|
logger.info("Checking router %s", rname)
|
|
json_output = tgen.gears[rname].vtysh_cmd(
|
|
"show isis neighbor detail json", isjson=True
|
|
)
|
|
assertmsg = "Test isis neighbor json failed in '{}' data '{}'".format(
|
|
rname, json_output
|
|
)
|
|
assert (
|
|
json_output["areas"][0]["circuits"][0]["interface"]["name"]
|
|
== rname + "-eth0"
|
|
), assertmsg
|
|
|
|
|
|
def test_isis_database_json():
|
|
"Check json struct in show isis database json"
|
|
|
|
tgen = get_topogen()
|
|
# Don't run this test if we have any failure.
|
|
if tgen.routers_have_failure():
|
|
pytest.skip(tgen.errors)
|
|
|
|
# tgen.mininet_cli()
|
|
logger.info("Checking 'show isis database json'")
|
|
for rname, _ in tgen.routers().items():
|
|
logger.info("Checking router %s", rname)
|
|
json_output = tgen.gears[rname].vtysh_cmd(
|
|
"show isis database json", isjson=True
|
|
)
|
|
assertmsg = "Test isis database json failed in '{}' data '{}'".format(
|
|
rname, json_output
|
|
)
|
|
assert json_output["areas"][0]["area"]["name"] == "1", assertmsg
|
|
assert json_output["areas"][0]["levels"][0]["id"] != "3", assertmsg
|
|
|
|
for rname, router in tgen.routers().items():
|
|
logger.info("Checking router %s", rname)
|
|
json_output = tgen.gears[rname].vtysh_cmd(
|
|
"show isis database detail json", isjson=True
|
|
)
|
|
assertmsg = "Test isis database json failed in '{}' data '{}'".format(
|
|
rname, json_output
|
|
)
|
|
assert json_output["areas"][0]["area"]["name"] == "1", assertmsg
|
|
assert json_output["areas"][0]["levels"][0]["id"] != "3", assertmsg
|
|
|
|
|
|
def test_isis_overload_on_startup():
|
|
"Check that overload on startup behaves as expected"
|
|
|
|
tgen = get_topogen()
|
|
net = get_topogen().net
|
|
overload_time = 120
|
|
|
|
# Don't run this test if we have any failure.
|
|
if tgen.routers_have_failure():
|
|
pytest.skip(tgen.errors)
|
|
|
|
logger.info("Testing overload on startup behavior")
|
|
|
|
# Configure set-overload-bit on-startup on r3
|
|
r3 = tgen.gears["r3"]
|
|
r3.vtysh_cmd(
|
|
f"""
|
|
configure
|
|
router isis 1
|
|
set-overload-bit on-startup {overload_time}
|
|
"""
|
|
)
|
|
# Restart r3
|
|
logger.info("Stop router")
|
|
stop_router(tgen, "r3")
|
|
logger.info("Start router")
|
|
|
|
tstamp_before_start_router = datetime.datetime.now()
|
|
start_router(tgen, "r3")
|
|
tstamp_after_start_router = datetime.datetime.now()
|
|
startup_router_time = (
|
|
tstamp_after_start_router - tstamp_before_start_router
|
|
).total_seconds()
|
|
|
|
# Check that the overload bit is set in r3's LSP
|
|
check_lsp_overload_bit("r3", "r3.00-00", "0/0/1")
|
|
check_lsp_overload_bit("r1", "r3.00-00", "0/0/1")
|
|
|
|
# Attempt to unset overload bit while timer is still running
|
|
r3.vtysh_cmd(
|
|
"""
|
|
configure
|
|
router isis 1
|
|
no set-overload-bit on-startup
|
|
no set-overload-bit
|
|
"""
|
|
)
|
|
|
|
# Check overload bit is still set
|
|
check_lsp_overload_bit("r1", "r3.00-00", "0/0/1")
|
|
|
|
# Check that overload bit is unset after timer completes
|
|
check_lsp_overload_bit("r3", "r3.00-00", "0/0/0")
|
|
tstamp_after_bit_unset = datetime.datetime.now()
|
|
check_lsp_overload_bit("r1", "r3.00-00", "0/0/0")
|
|
|
|
# Collect time overloaded
|
|
time_overloaded = (
|
|
tstamp_after_bit_unset - tstamp_after_start_router
|
|
).total_seconds()
|
|
logger.info(f"Time Overloaded: {time_overloaded}")
|
|
|
|
# Use time it took to startup router as lower bound
|
|
logger.info(
|
|
f"Assert that overload time falls in range: {overload_time - startup_router_time} < {time_overloaded} <= {overload_time}"
|
|
)
|
|
result = overload_time - startup_router_time < time_overloaded <= overload_time
|
|
assert result
|
|
|
|
|
|
def test_isis_overload_on_startup_cancel_timer():
|
|
"Check that overload on startup timer is cancelled when overload bit is set/unset"
|
|
|
|
tgen = get_topogen()
|
|
net = get_topogen().net
|
|
overload_time = 90
|
|
|
|
# Don't run this test if we have any failure.
|
|
if tgen.routers_have_failure():
|
|
pytest.skip(tgen.errors)
|
|
|
|
logger.info(
|
|
"Testing overload on startup behavior with set overload bit: cancel timer"
|
|
)
|
|
|
|
# Configure set-overload-bit on-startup on r3
|
|
r3 = tgen.gears["r3"]
|
|
r3.vtysh_cmd(
|
|
f"""
|
|
configure
|
|
router isis 1
|
|
set-overload-bit on-startup {overload_time}
|
|
set-overload-bit
|
|
"""
|
|
)
|
|
# Restart r3
|
|
logger.info("Stop router")
|
|
stop_router(tgen, "r3")
|
|
logger.info("Start router")
|
|
start_router(tgen, "r3")
|
|
|
|
# Check that the overload bit is set in r3's LSP
|
|
check_lsp_overload_bit("r3", "r3.00-00", "0/0/1")
|
|
|
|
# Check that overload timer is running
|
|
check_overload_timer("r3", True)
|
|
|
|
# Unset overload bit while timer is running
|
|
r3.vtysh_cmd(
|
|
"""
|
|
configure
|
|
router isis 1
|
|
no set-overload-bit
|
|
"""
|
|
)
|
|
|
|
# Check that overload timer is cancelled
|
|
check_overload_timer("r3", False)
|
|
|
|
# Check overload bit is unset
|
|
check_lsp_overload_bit("r3", "r3.00-00", "0/0/0")
|
|
|
|
|
|
def test_isis_overload_on_startup_override_timer():
|
|
"Check that overload bit remains set after overload timer expires if overload bit is configured"
|
|
|
|
tgen = get_topogen()
|
|
net = get_topogen().net
|
|
overload_time = 60
|
|
|
|
# Don't run this test if we have any failure.
|
|
if tgen.routers_have_failure():
|
|
pytest.skip(tgen.errors)
|
|
|
|
logger.info(
|
|
"Testing overload on startup behavior with set overload bit: override timer"
|
|
)
|
|
|
|
# Configure set-overload-bit on-startup on r3
|
|
r3 = tgen.gears["r3"]
|
|
r3.vtysh_cmd(
|
|
f"""
|
|
configure
|
|
router isis 1
|
|
set-overload-bit on-startup {overload_time}
|
|
set-overload-bit
|
|
"""
|
|
)
|
|
# Restart r3
|
|
logger.info("Stop router")
|
|
stop_router(tgen, "r3")
|
|
logger.info("Start router")
|
|
start_router(tgen, "r3")
|
|
|
|
# Check that the overload bit is set in r3's LSP
|
|
check_lsp_overload_bit("r3", "r3.00-00", "0/0/1")
|
|
|
|
# Check that overload timer is running
|
|
check_overload_timer("r3", True)
|
|
|
|
# Check that overload timer expired
|
|
check_overload_timer("r3", False)
|
|
|
|
# Check overload bit is still set
|
|
check_lsp_overload_bit("r3", "r3.00-00", "0/0/1")
|
|
|
|
|
|
def test_isis_advertise_passive_only():
|
|
"""Check that we only advertise prefixes of passive interfaces when advertise-passive-only is enabled."""
|
|
tgen = get_topogen()
|
|
net = get_topogen().net
|
|
# Don't run this test if we have any failure.
|
|
if tgen.routers_have_failure():
|
|
pytest.skip(tgen.errors)
|
|
|
|
logger.info("Testing isis advertise-passive-only behavior")
|
|
expected_prefixes_no_advertise_passive_only = set(
|
|
["10.0.20.0/24", "10.254.0.1/32", "2001:db8:f::1/128", "2001:db8:1:1::/64"]
|
|
)
|
|
expected_prefixes_advertise_passive_only = set(
|
|
["10.254.0.1/32", "2001:db8:f::1/128"]
|
|
)
|
|
lsp_id = "r1.00-00"
|
|
|
|
r1 = tgen.gears["r1"]
|
|
r1.vtysh_cmd(
|
|
"""
|
|
configure
|
|
router isis 1
|
|
no redistribute ipv4 connected level-2
|
|
no redistribute ipv6 connected level-2
|
|
interface lo
|
|
ip router isis 1
|
|
ipv6 router isis 1
|
|
isis passive
|
|
end
|
|
"""
|
|
)
|
|
|
|
result = check_advertised_prefixes(
|
|
r1, lsp_id, expected_prefixes_no_advertise_passive_only
|
|
)
|
|
assert result is True, result
|
|
|
|
r1.vtysh_cmd(
|
|
"""
|
|
configure
|
|
router isis 1
|
|
advertise-passive-only
|
|
end
|
|
"""
|
|
)
|
|
|
|
result = check_advertised_prefixes(
|
|
r1, lsp_id, expected_prefixes_advertise_passive_only
|
|
)
|
|
assert result is True, result
|
|
|
|
|
|
def test_isis_hello_padding_during_adjacency_formation():
|
|
"""Check that IIH packets is only padded when adjacency is still being formed
|
|
when isis hello padding during-adjacency-formation is configured
|
|
"""
|
|
tgen = get_topogen()
|
|
net = get_topogen().net
|
|
# Don't run this test if we have any failure.
|
|
if tgen.routers_have_failure():
|
|
pytest.skip(tgen.errors)
|
|
|
|
logger.info("Testing isis hello padding during-adjacency-formation behavior")
|
|
r3 = tgen.gears["r3"]
|
|
|
|
# Reduce hello-multiplier to make the adjacency go down faster.
|
|
r3.vtysh_cmd(
|
|
"""
|
|
configure
|
|
interface r3-eth0
|
|
isis hello-multiplier 2
|
|
"""
|
|
)
|
|
|
|
r1 = tgen.gears["r1"]
|
|
cmd_output = r1.vtysh_cmd(
|
|
"""
|
|
configure
|
|
interface r1-eth0
|
|
isis hello padding during-adjacency-formation
|
|
end
|
|
debug isis adj-packets
|
|
"""
|
|
)
|
|
result = check_last_iih_packet_for_padding(r1, expect_padding=False)
|
|
assert result is True, result
|
|
|
|
r3.vtysh_cmd(
|
|
"""
|
|
configure
|
|
interface r3-eth0
|
|
shutdown
|
|
"""
|
|
)
|
|
result = check_last_iih_packet_for_padding(r1, expect_padding=True)
|
|
assert result is True, result
|
|
|
|
r3 = tgen.gears["r3"]
|
|
r3.vtysh_cmd(
|
|
"""
|
|
configure
|
|
interface r3-eth0
|
|
no shutdown
|
|
"""
|
|
)
|
|
result = check_last_iih_packet_for_padding(r1, expect_padding=False)
|
|
assert result is True, result
|
|
|
|
|
|
@retry(retry_timeout=10)
|
|
def check_last_iih_packet_for_padding(router, expect_padding):
|
|
logfilename = "{}/{}".format(router.gearlogdir, "isisd.log")
|
|
last_hello_packet_line = None
|
|
with open(logfilename, "r") as f:
|
|
lines = f.readlines()
|
|
for line in lines:
|
|
if re.search("Sending .+? IIH", line):
|
|
last_hello_packet_line = line
|
|
|
|
if last_hello_packet_line is None:
|
|
return "Expected IIH packet in {}, but no packet found".format(logfilename)
|
|
|
|
interface_name, packet_length = re.search(
|
|
r"Sending .+ IIH on (.+), length (\d+)", last_hello_packet_line
|
|
).group(1, 2)
|
|
packet_length = int(packet_length)
|
|
interface_output = router.vtysh_cmd("show interface {} json".format(interface_name))
|
|
interface_json = json.loads(interface_output)
|
|
padded_packet_length = interface_json[interface_name]["mtu"] - 3
|
|
if expect_padding:
|
|
if packet_length == padded_packet_length:
|
|
return True
|
|
return (
|
|
"Expected padded packet with length {}, got packet with length {}".format(
|
|
padded_packet_length, packet_length
|
|
)
|
|
)
|
|
if packet_length < padded_packet_length:
|
|
return True
|
|
return "Expected unpadded packet with length less than {}, got packet with length {}".format(
|
|
padded_packet_length, packet_length
|
|
)
|
|
|
|
|
|
@retry(retry_timeout=5)
|
|
def check_advertised_prefixes(router, lsp_id, expected_prefixes):
|
|
output = router.vtysh_cmd("show isis database detail {}".format(lsp_id))
|
|
prefixes = set(re.findall(r"IP(?:v6)? Reachability: (.*) \(Metric: 10\)", output))
|
|
if prefixes == expected_prefixes:
|
|
return True
|
|
return str({"expected_prefixes:": expected_prefixes, "prefixes": prefixes})
|
|
|
|
|
|
@retry(retry_timeout=200)
|
|
def _check_lsp_overload_bit(router, overloaded_router_lsp, att_p_ol_expected):
|
|
"Verfiy overload bit in router's LSP"
|
|
|
|
tgen = get_topogen()
|
|
router = tgen.gears[router]
|
|
logger.info(f"check_overload_bit {router}")
|
|
isis_database_output = router.vtysh_cmd(
|
|
"show isis database {} json".format(overloaded_router_lsp)
|
|
)
|
|
|
|
database_json = json.loads(isis_database_output)
|
|
if "lsps" not in database_json["areas"][0]["levels"][1]:
|
|
return "The LSP of {} has not been synchronized yet ".format(router.name)
|
|
|
|
att_p_ol = database_json["areas"][0]["levels"][1]["lsps"][0]["attPOl"]
|
|
if att_p_ol == att_p_ol_expected:
|
|
return True
|
|
return "{} peer with expected att_p_ol {} got {} ".format(
|
|
router.name, att_p_ol_expected, att_p_ol
|
|
)
|
|
|
|
|
|
def check_lsp_overload_bit(router, overloaded_router_lsp, att_p_ol_expected):
|
|
"Verfiy overload bit in router's LSP"
|
|
|
|
assertmsg = _check_lsp_overload_bit(
|
|
router, overloaded_router_lsp, att_p_ol_expected
|
|
)
|
|
assert assertmsg is True, assertmsg
|
|
|
|
|
|
@retry(retry_timeout=200)
|
|
def _check_overload_timer(router, timer_expected):
|
|
"Verfiy overload bit in router's LSP"
|
|
|
|
tgen = get_topogen()
|
|
router = tgen.gears[router]
|
|
output = router.vtysh_cmd("show event timers")
|
|
|
|
timer_running = "set_overload_on_start_timer" in output
|
|
if timer_running == timer_expected:
|
|
return True
|
|
return "Expected timer running status: {}".format(timer_expected)
|
|
|
|
|
|
def check_overload_timer(router, timer_expected):
|
|
"Verfiy overload bit in router's LSP"
|
|
|
|
assertmsg = _check_overload_timer(router, timer_expected)
|
|
assert assertmsg is True, assertmsg
|
|
|
|
|
|
def test_memory_leak():
|
|
"Run the memory leak test and report results."
|
|
tgen = get_topogen()
|
|
if not tgen.is_memleak_enabled():
|
|
pytest.skip("Memory leak test/report is disabled")
|
|
|
|
tgen.report_memory_leaks()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
args = ["-s"] + sys.argv[1:]
|
|
sys.exit(pytest.main(args))
|
|
|
|
|
|
#
|
|
# Auxiliary functions
|
|
#
|
|
|
|
|
|
def dict_merge(dct, merge_dct):
|
|
"""
|
|
Recursive dict merge. Inspired by :meth:``dict.update()``, instead of
|
|
updating only top-level keys, dict_merge recurses down into dicts nested
|
|
to an arbitrary depth, updating keys. The ``merge_dct`` is merged into
|
|
``dct``.
|
|
:param dct: dict onto which the merge is executed
|
|
:param merge_dct: dct merged into dct
|
|
:return: None
|
|
|
|
Source:
|
|
https://gist.github.com/angstwad/bf22d1822c38a92ec0a9
|
|
"""
|
|
for k, _ in merge_dct.items():
|
|
if k in dct and isinstance(dct[k], dict) and topotest.is_mapping(merge_dct[k]):
|
|
dict_merge(dct[k], merge_dct[k])
|
|
else:
|
|
dct[k] = merge_dct[k]
|
|
|
|
|
|
def parse_topology(lines, level):
|
|
"""
|
|
Parse the output of 'show isis topology level-X' into a Python dict.
|
|
"""
|
|
areas = {}
|
|
area = None
|
|
ipv = None
|
|
vertex_type_regex = "|".join(VERTEX_TYPE_LIST)
|
|
|
|
for line in lines:
|
|
area_match = re.match(r"Area (.+):", line)
|
|
if area_match:
|
|
area = area_match.group(1)
|
|
if area not in areas:
|
|
areas[area] = {level: {"ipv4": [], "ipv6": []}}
|
|
ipv = None
|
|
continue
|
|
elif area is None:
|
|
continue
|
|
|
|
if re.match(r"IS\-IS paths to level-. routers that speak IPv6", line):
|
|
ipv = "ipv6"
|
|
continue
|
|
if re.match(r"IS\-IS paths to level-. routers that speak IP", line):
|
|
ipv = "ipv4"
|
|
continue
|
|
|
|
item_match = re.match(
|
|
r"([^\s]+) ([^\s]+) ([^\s]+) ([^\s]+) ([^\s]+) ([^\s]+)", line
|
|
)
|
|
if (
|
|
item_match is not None
|
|
and item_match.group(1) == "Vertex"
|
|
and item_match.group(2) == "Type"
|
|
and item_match.group(3) == "Metric"
|
|
and item_match.group(4) == "Next-Hop"
|
|
and item_match.group(5) == "Interface"
|
|
and item_match.group(6) == "Parent"
|
|
):
|
|
# Skip header
|
|
continue
|
|
|
|
item_match = re.match(
|
|
r"([^\s]+) ({}) ([0]|([1-9][0-9]*)) ([^\s]+) ([^\s]+) ([^\s]+)".format(
|
|
vertex_type_regex
|
|
),
|
|
line,
|
|
)
|
|
if item_match is not None:
|
|
areas[area][level][ipv].append(
|
|
{
|
|
"vertex": item_match.group(1),
|
|
"type": item_match.group(2),
|
|
"metric": item_match.group(3),
|
|
"next-hop": item_match.group(5),
|
|
"interface": item_match.group(6),
|
|
"parent": item_match.group(7),
|
|
}
|
|
)
|
|
continue
|
|
|
|
item_match = re.match(
|
|
r"([^\s]+) ({}) ([0]|([1-9][0-9]*)) ([^\s]+)".format(vertex_type_regex),
|
|
line,
|
|
)
|
|
|
|
if item_match is not None:
|
|
areas[area][level][ipv].append(
|
|
{
|
|
"vertex": item_match.group(1),
|
|
"type": item_match.group(2),
|
|
"metric": item_match.group(3),
|
|
"parent": item_match.group(5),
|
|
}
|
|
)
|
|
continue
|
|
|
|
item_match = re.match(r"([^\s]+)", line)
|
|
if item_match is not None:
|
|
areas[area][level][ipv].append({"vertex": item_match.group(1)})
|
|
continue
|
|
|
|
return areas
|