#!/usr/bin/env python
+# SPDX-License-Identifier: ISC
#
# test_isis_topo1.py
# Copyright (c) 2017 by
# Network Device Education Foundation, Inc. ("NetDEF")
#
-# Permission to use, copy, modify, and/or distribute this software
-# for any purpose with or without fee is hereby granted, provided
-# that the above copyright notice and this permission notice appear
-# in all copies.
-#
-# THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
-# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
-# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
-# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
-# DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
-# WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
-# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
-# OF THIS SOFTWARE.
-#
"""
test_isis_topo1.py: Test ISIS topology.
"""
-
+import datetime
import functools
import json
import os
# 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
for rname, router in tgen.routers().items():
filename = "{0}/{1}/{1}_route.json".format(CWD, rname)
expected = json.loads(open(filename, "r").read())
- actual = router.vtysh_cmd("show ip route json", isjson=True)
+
+ 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, diff) = topotest.run_and_expect(test_func, None, wait=1, count=10)
assertmsg = "Router '{}' routes mismatch".format(rname)
- assert topotest.json_cmp(actual, expected) is None, assertmsg
+ assert result, assertmsg
def test_isis_linux_route_installation():
for rname, router in tgen.routers().items():
filename = "{0}/{1}/{1}_route6.json".format(CWD, rname)
expected = json.loads(open(filename, "r").read())
- actual = router.vtysh_cmd("show ipv6 route json", isjson=True)
+
+ 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, diff) = topotest.run_and_expect(test_func, None, wait=1, count=10)
assertmsg = "Router '{}' routes mismatch".format(rname)
- assert topotest.json_cmp(actual, expected) is None, assertmsg
+ assert result, assertmsg
def test_isis_linux_route6_installation():
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, router 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, router 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, router 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, router 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")
+
+
+@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)
+ att_p_ol = database_json["areas"][0]["levels"][1]["att-p-ol"]
+ 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]
+ thread_output = router.vtysh_cmd("show thread timers")
+
+ timer_running = "set_overload_on_start_timer" in thread_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()
projects / mirror_frr.git / blobdiff