summaryrefslogtreecommitdiffstats
path: root/tests/topotests/ldp_topo1/test_ldp_topo1.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/topotests/ldp_topo1/test_ldp_topo1.py')
-rw-r--r--tests/topotests/ldp_topo1/test_ldp_topo1.py857
1 files changed, 857 insertions, 0 deletions
diff --git a/tests/topotests/ldp_topo1/test_ldp_topo1.py b/tests/topotests/ldp_topo1/test_ldp_topo1.py
new file mode 100644
index 0000000..cb8adfb
--- /dev/null
+++ b/tests/topotests/ldp_topo1/test_ldp_topo1.py
@@ -0,0 +1,857 @@
+#!/usr/bin/env python
+
+#
+# test_bgp_multiview_topo1.py
+# Part of NetDEF Topology Tests
+#
+# Copyright (c) 2016 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.
+#
+
+r"""
+test_ldp_topo1.py: Simple FRR LDP Test
+
+ +---------+
+ | r1 |
+ | 1.1.1.1 |
+ +----+----+
+ | .1 r1-eth0
+ |
+ ~~~~~~~~~~~~~
+ ~~ sw0 ~~
+ ~~ 10.0.1.0/24 ~~
+ ~~~~~~~~~~~~~
+ |10.0.1.0/24
+ |
+ | .2 r2-eth0
+ +----+----+
+ | r2 |
+ | 2.2.2.2 |
+ +--+---+--+
+ r2-eth2 .2 | | .2 r2-eth1
+ ______/ \______
+ / \
+ ~~~~~~~~~~~~~ ~~~~~~~~~~~~~
+~~ sw2 ~~ ~~ sw1 ~~
+~~ 10.0.3.0/24 ~~ ~~ 10.0.2.0/24 ~~
+ ~~~~~~~~~~~~~ ~~~~~~~~~~~~~
+ | / |
+ \ _________/ |
+ \ / \
+r3-eth1 .3 | | .3 r3-eth0 | .4 r4-eth0
+ +----+--+---+ +----+----+
+ | r3 | | r4 |
+ | 3.3.3.3 | | 4.4.4.4 |
+ +-----------+ +---------+
+"""
+
+import os
+import re
+import sys
+import pytest
+import json
+from functools import partial
+from time import sleep
+from lib.topolog import logger
+
+# Save the Current Working Directory to find configuration files.
+CWD = os.path.dirname(os.path.realpath(__file__))
+sys.path.append(os.path.join(CWD, "../"))
+
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from lib import topotest
+from lib.topogen import Topogen, get_topogen
+
+fatal_error = ""
+
+pytestmark = [pytest.mark.ldpd, pytest.mark.ospfd]
+
+#####################################################
+##
+## Network Topology Definition
+##
+#####################################################
+
+
+def build_topo(tgen):
+
+ # Setup Routers
+ for i in range(1, 5):
+ tgen.add_router("r%s" % i)
+
+ # First switch
+ switch = tgen.add_switch("sw0")
+ switch.add_link(tgen.gears["r1"])
+ switch.add_link(tgen.gears["r2"])
+ # Second switch
+ switch = tgen.add_switch("sw1")
+ switch.add_link(tgen.gears["r2"])
+ switch.add_link(tgen.gears["r3"])
+ switch.add_link(tgen.gears["r4"])
+ # Third switch
+ switch = tgen.add_switch("sw2")
+ switch.add_link(tgen.gears["r2"])
+ switch.add_link(tgen.gears["r3"])
+
+
+#####################################################
+##
+## Helper functions
+##
+#####################################################
+
+
+def router_compare_json_output(rname, command, reference, count=60, wait=1):
+ "Compare router JSON output"
+
+ logger.info('Comparing router "%s" "%s" output', rname, command)
+
+ tgen = get_topogen()
+ filename = "{}/{}/{}".format(CWD, rname, reference)
+ expected = json.loads(open(filename).read())
+
+ # Run test function until we get an result.
+ test_func = partial(topotest.router_json_cmp, tgen.gears[rname], command, expected)
+ _, diff = topotest.run_and_expect(test_func, None, count, wait)
+ assertmsg = '"{}" JSON output mismatches the expected result'.format(rname)
+ assert diff is None, assertmsg
+
+
+#####################################################
+##
+## Tests starting
+##
+#####################################################
+
+
+def setup_module(module):
+ print("\n\n** %s: Setup Topology" % module.__name__)
+ print("******************************************\n")
+
+ thisDir = os.path.dirname(os.path.realpath(__file__))
+ tgen = Topogen(build_topo, module.__name__)
+ tgen.start_topology()
+
+ net = tgen.net
+
+ # Starting Routers
+ for i in range(1, 5):
+ net["r%s" % i].loadConf("zebra", "%s/r%s/zebra.conf" % (thisDir, i))
+ net["r%s" % i].loadConf("ospfd", "%s/r%s/ospfd.conf" % (thisDir, i))
+ net["r%s" % i].loadConf("ldpd", "%s/r%s/ldpd.conf" % (thisDir, i))
+ tgen.gears["r%s" % i].start()
+
+ # For debugging after starting FRR daemons, uncomment the next line
+ # tgen.mininet_cli()
+
+
+def teardown_module(module):
+ print("\n\n** %s: Shutdown Topology" % module.__name__)
+ print("******************************************\n")
+ tgen = get_topogen()
+ tgen.stop_topology()
+
+
+def test_router_running():
+ global fatal_error
+ net = get_topogen().net
+
+ # Skip if previous fatal error condition is raised
+ if fatal_error != "":
+ pytest.skip(fatal_error)
+
+ print("\n\n** Check if FRR is running on each Router node")
+ print("******************************************\n")
+ sleep(5)
+
+ # Starting Routers
+ for i in range(1, 5):
+ fatal_error = net["r%s" % i].checkRouterRunning()
+ assert fatal_error == "", fatal_error
+
+
+def test_mpls_interfaces():
+ global fatal_error
+ net = get_topogen().net
+
+ # Skip if previous fatal error condition is raised
+ if fatal_error != "":
+ pytest.skip(fatal_error)
+
+ thisDir = os.path.dirname(os.path.realpath(__file__))
+
+ # Verify MPLS Interfaces
+ print("\n\n** Verifying MPLS Interfaces")
+ print("******************************************\n")
+ failures = 0
+ for i in range(1, 5):
+ refTableFile = "%s/r%s/show_mpls_ldp_interface.ref"
+ if os.path.isfile(refTableFile):
+ # Read expected result from file
+ expected = open(refTableFile).read().rstrip()
+ # Fix newlines (make them all the same)
+ expected = ("\n".join(expected.splitlines()) + "\n").splitlines(1)
+
+ # Actual output from router
+ actual = (
+ net["r%s" % i]
+ .cmd('vtysh -c "show mpls ldp interface" 2> /dev/null')
+ .rstrip()
+ )
+ # Mask out Timer in Uptime
+ actual = re.sub(r" [0-9][0-9]:[0-9][0-9]:[0-9][0-9] ", " xx:xx:xx ", actual)
+ # Fix newlines (make them all the same)
+ actual = ("\n".join(actual.splitlines()) + "\n").splitlines(1)
+
+ # Generate Diff
+ diff = topotest.get_textdiff(
+ actual,
+ expected,
+ title1="actual MPLS LDP interface status",
+ title2="expected MPLS LDP interface status",
+ )
+
+ # Empty string if it matches, otherwise diff contains unified diff
+ if diff:
+ sys.stderr.write(
+ "r%s failed MPLS LDP Interface status Check:\n%s\n" % (i, diff)
+ )
+ failures += 1
+ else:
+ print("r%s ok" % i)
+
+ if failures > 0:
+ fatal_error = "MPLS LDP Interface status failed"
+
+ assert (
+ failures == 0
+ ), "MPLS LDP Interface status failed for router r%s:\n%s" % (i, diff)
+
+ # Make sure that all daemons are running
+ for i in range(1, 5):
+ fatal_error = net["r%s" % i].checkRouterRunning()
+ assert fatal_error == "", fatal_error
+
+
+def test_ospf_convergence():
+ logger.info("Test: check OSPF adjacencies")
+
+ # Skip if previous fatal error condition is raised
+ if fatal_error != "":
+ pytest.skip(fatal_error)
+
+ for rname in ["r1", "r2", "r3", "r4"]:
+ router_compare_json_output(
+ rname, "show ip ospf neighbor json", "show_ip_ospf_neighbor.json"
+ )
+
+
+def test_mpls_ldp_neighbor_establish():
+ global fatal_error
+ net = get_topogen().net
+
+ # Skip if previous fatal error condition is raised
+ if fatal_error != "":
+ pytest.skip(fatal_error)
+
+ neighbors_operational = {
+ 1: 1,
+ 2: 3,
+ 3: 2,
+ 4: 2,
+ }
+
+ # Wait for MPLS LDP neighbors to establish.
+ print("\n\n** Verify MPLS LDP neighbors to establish")
+ print("******************************************\n")
+ timeout = 90
+ while timeout > 0:
+ print("Timeout in %s: " % timeout),
+ sys.stdout.flush()
+ # Look for any node not yet converged
+ for i in range(1, 5):
+ established = (
+ net["r%s" % i]
+ .cmd('vtysh -c "show mpls ldp neighbor" 2> /dev/null')
+ .rstrip()
+ )
+
+ # On current version, we need to make sure they all turn to OPERATIONAL on all lines
+ #
+ lines = ("\n".join(established.splitlines()) + "\n").splitlines(1)
+ # Check all lines to be either table header (starting with ^AF or show OPERATIONAL)
+ header = r"^AF.*"
+ operational = r"^ip.*OPERATIONAL.*"
+ found_operational = 0
+ for j in range(1, len(lines)):
+ if (not re.search(header, lines[j])) and (
+ not re.search(operational, lines[j])
+ ):
+ established = "" # Empty string shows NOT established
+ if re.search(operational, lines[j]):
+ found_operational += 1
+
+ logger.info("Found operational %d" % found_operational)
+ if found_operational < 1:
+ # Need at least one operational neighbor
+ established = "" # Empty string shows NOT established
+ else:
+ if found_operational != neighbors_operational[i]:
+ established = ""
+ if not established:
+ print("Waiting for r%s" % i)
+ sys.stdout.flush()
+ break
+ if not established:
+ sleep(5)
+ timeout -= 5
+ else:
+ print("Done")
+ break
+ else:
+ # Bail out with error if a router fails to converge
+ fatal_error = "MPLS LDP neighbors did not establish"
+ assert False, "MPLS LDP neighbors did not establish"
+
+ print("MPLS LDP neighbors established.")
+
+ if timeout < 60:
+ # Only wait if we actually went through a convergence
+ print("\nwaiting 15s for LDP sessions to establish")
+ sleep(15)
+
+ # Make sure that all daemons are running
+ for i in range(1, 5):
+ fatal_error = net["r%s" % i].checkRouterRunning()
+ assert fatal_error == "", fatal_error
+
+
+def test_mpls_ldp_discovery():
+ global fatal_error
+ net = get_topogen().net
+
+ # Skip if previous fatal error condition is raised
+ if fatal_error != "":
+ pytest.skip(fatal_error)
+
+ thisDir = os.path.dirname(os.path.realpath(__file__))
+
+ # Verify MPLS LDP discovery
+ print("\n\n** Verifying MPLS LDP discovery")
+ print("******************************************\n")
+ failures = 0
+ for i in range(1, 5):
+ refTableFile = "%s/r%s/show_mpls_ldp_discovery.ref" % (thisDir, i)
+ if os.path.isfile(refTableFile):
+ # Actual output from router
+ actual = (
+ net["r%s" % i]
+ .cmd('vtysh -c "show mpls ldp discovery" 2> /dev/null')
+ .rstrip()
+ )
+
+ # Read expected result from file
+ expected = open(refTableFile).read().rstrip()
+ # Fix newlines (make them all the same)
+ expected = ("\n".join(expected.splitlines()) + "\n").splitlines(1)
+
+ # Actual output from router
+ actual = (
+ net["r%s" % i]
+ .cmd('vtysh -c "show mpls ldp discovery" 2> /dev/null')
+ .rstrip()
+ )
+
+ # Fix newlines (make them all the same)
+ actual = ("\n".join(actual.splitlines()) + "\n").splitlines(1)
+
+ # Generate Diff
+ diff = topotest.get_textdiff(
+ actual,
+ expected,
+ title1="actual MPLS LDP discovery output",
+ title2="expected MPLS LDP discovery output",
+ )
+
+ # Empty string if it matches, otherwise diff contains unified diff
+ if diff:
+ sys.stderr.write(
+ "r%s failed MPLS LDP discovery output Check:\n%s\n" % (i, diff)
+ )
+ failures += 1
+ else:
+ print("r%s ok" % i)
+
+ assert (
+ failures == 0
+ ), "MPLS LDP Interface discovery output for router r%s:\n%s" % (i, diff)
+
+ # Make sure that all daemons are running
+ for i in range(1, 5):
+ fatal_error = net["r%s" % i].checkRouterRunning()
+ assert fatal_error == "", fatal_error
+
+
+def test_mpls_ldp_neighbor():
+ global fatal_error
+ net = get_topogen().net
+
+ # Skip if previous fatal error condition is raised
+ if fatal_error != "":
+ pytest.skip(fatal_error)
+
+ thisDir = os.path.dirname(os.path.realpath(__file__))
+
+ # Verify MPLS LDP neighbor
+ print("\n\n** Verifying MPLS LDP neighbor")
+ print("******************************************\n")
+ failures = 0
+ for i in range(1, 5):
+ refTableFile = "%s/r%s/show_mpls_ldp_neighbor.ref" % (thisDir, i)
+ if os.path.isfile(refTableFile):
+ # Read expected result from file
+ expected = open(refTableFile).read().rstrip()
+ # Fix newlines (make them all the same)
+ expected = ("\n".join(expected.splitlines()) + "\n").splitlines(1)
+
+ # Actual output from router
+ actual = (
+ net["r%s" % i]
+ .cmd('vtysh -c "show mpls ldp neighbor" 2> /dev/null')
+ .rstrip()
+ )
+
+ # Mask out changing parts in output
+ # Mask out Timer in Uptime
+ actual = re.sub(
+ r"(ipv4 [0-9\.]+ +OPERATIONAL [0-9\.]+ +)[0-9][0-9]:[0-9][0-9]:[0-9][0-9]",
+ r"\1xx:xx:xx",
+ actual,
+ )
+
+ # Fix newlines (make them all the same)
+ actual = ("\n".join(actual.splitlines()) + "\n").splitlines(1)
+
+ # Generate Diff
+ diff = topotest.get_textdiff(
+ actual,
+ expected,
+ title1="actual MPLS LDP neighbor output",
+ title2="expected MPLS LDP neighbor output",
+ )
+
+ # Empty string if it matches, otherwise diff contains unified diff
+ if diff:
+ sys.stderr.write(
+ "r%s failed MPLS LDP neighbor output Check:\n%s\n" % (i, diff)
+ )
+ failures += 1
+ else:
+ print("r%s ok" % i)
+
+ assert (
+ failures == 0
+ ), "MPLS LDP Interface neighbor output for router r%s:\n%s" % (i, diff)
+
+ # Make sure that all daemons are running
+ for i in range(1, 5):
+ fatal_error = net["r%s" % i].checkRouterRunning()
+ assert fatal_error == "", fatal_error
+
+
+def test_mpls_ldp_binding():
+ global fatal_error
+ net = get_topogen().net
+
+ # Skip this test for now until proper sorting of the output
+ # is implemented
+ # pytest.skip("Skipping test_mpls_ldp_binding")
+
+ # Skip if previous fatal error condition is raised
+ if fatal_error != "":
+ pytest.skip(fatal_error)
+
+ thisDir = os.path.dirname(os.path.realpath(__file__))
+
+ # Verify MPLS LDP binding
+ print("\n\n** Verifying MPLS LDP binding")
+ print("******************************************\n")
+ failures = 0
+ for i in range(1, 5):
+ refTableFile = "%s/r%s/show_mpls_ldp_binding.ref" % (thisDir, i)
+ if os.path.isfile(refTableFile):
+ # Read expected result from file
+ expected = open(refTableFile).read().rstrip()
+ # Fix newlines (make them all the same)
+ expected = ("\n".join(expected.splitlines()) + "\n").splitlines(1)
+
+ # Actual output from router
+ actual = (
+ net["r%s" % i]
+ .cmd('vtysh -c "show mpls ldp binding" 2> /dev/null')
+ .rstrip()
+ )
+
+ # Mask out changing parts in output
+ # Mask out label
+ actual = re.sub(
+ r"(ipv4 [0-9\./]+ +[0-9\.]+ +)[0-9][0-9] (.*)", r"\1xxx\2", actual
+ )
+ actual = re.sub(
+ r"(ipv4 [0-9\./]+ +[0-9\.]+ +[a-z\-]+ +)[0-9][0-9] (.*)",
+ r"\1xxx\2",
+ actual,
+ )
+
+ # Fix newlines (make them all the same)
+ actual = ("\n".join(actual.splitlines()) + "\n").splitlines(1)
+
+ # Sort lines which start with "xx via inet "
+ pattern = r"^\s+[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+\s+"
+ swapped = True
+ while swapped:
+ swapped = False
+ for j in range(1, len(actual)):
+ if re.search(pattern, actual[j]) and re.search(
+ pattern, actual[j - 1]
+ ):
+ if actual[j - 1] > actual[j]:
+ temp = actual[j - 1]
+ actual[j - 1] = actual[j]
+ actual[j] = temp
+ swapped = True
+
+ # Generate Diff
+ diff = topotest.get_textdiff(
+ actual,
+ expected,
+ title1="actual MPLS LDP binding output",
+ title2="expected MPLS LDP binding output",
+ )
+
+ # Empty string if it matches, otherwise diff contains unified diff
+ if diff:
+ sys.stderr.write(
+ "r%s failed MPLS LDP binding output Check:\n%s\n" % (i, diff)
+ )
+ failures += 1
+ else:
+ print("r%s ok" % i)
+
+ assert failures == 0, "MPLS LDP binding output for router r%s:\n%s" % (
+ i,
+ diff,
+ )
+
+ # Make sure that all daemons are running
+ for i in range(1, 5):
+ fatal_error = net["r%s" % i].checkRouterRunning()
+ assert fatal_error == "", fatal_error
+
+
+def test_zebra_ipv4_routingTable():
+ global fatal_error
+ net = get_topogen().net
+
+ # Skip if previous fatal error condition is raised
+ if fatal_error != "":
+ pytest.skip(fatal_error)
+
+ thisDir = os.path.dirname(os.path.realpath(__file__))
+
+ # Verify Zebra IPv4 Routing Table
+ print("\n\n** Verifying Zebra IPv4 Routing Table")
+ print("******************************************\n")
+ failures = 0
+ for i in range(1, 5):
+ refTableFile = "%s/r%s/show_ipv4_route.ref" % (thisDir, i)
+ if os.path.isfile(refTableFile):
+ # Read expected result from file
+ expected = open(refTableFile).read().rstrip()
+
+ # Actual output from router
+ actual = (
+ net["r%s" % i]
+ .cmd('vtysh -c "show ip route" 2> /dev/null | grep "^O"')
+ .rstrip()
+ )
+ # Drop timers on end of line
+ actual = re.sub(r", [0-2][0-9]:[0-5][0-9]:[0-5][0-9]", "", actual)
+
+ # Mask out label - all LDP labels should be >= 10 (2-digit)
+ # leaving the implicit labels unmasked
+ actual = re.sub(r" label [0-9][0-9]+", " label xxx", actual)
+ # and translating remaining implicit (single-digit) labels to label implicit-null
+ actual = re.sub(r" label [0-9]+", " label implicit-null", actual)
+ # Check if we have implicit labels - if not, then remove them from reference
+ if not re.search(r" label implicit-null", actual):
+ expected = re.sub(r", label implicit-null", "", expected)
+
+ # now fix newlines of expected (make them all the same)
+ expected = ("\n".join(expected.splitlines()) + "\n").splitlines(1)
+
+ # Fix newlines (make them all the same)
+ actual = ("\n".join(actual.splitlines()) + "\n").splitlines(1)
+
+ # Generate Diff
+ diff = topotest.get_textdiff(
+ actual,
+ expected,
+ title1="actual IPv4 zebra routing table",
+ title2="expected IPv4 zebra routing table",
+ )
+
+ # Empty string if it matches, otherwise diff contains unified diff
+ if diff:
+ sys.stderr.write(
+ "r%s failed IPv4 Zebra Routing Table Check:\n%s\n" % (i, diff)
+ )
+ failures += 1
+ else:
+ print("r%s ok" % i)
+
+ assert (
+ failures == 0
+ ), "IPv4 Zebra Routing Table verification failed for router r%s:\n%s" % (
+ i,
+ diff,
+ )
+
+ # Make sure that all daemons are running
+ for i in range(1, 5):
+ fatal_error = net["r%s" % i].checkRouterRunning()
+ assert fatal_error == "", fatal_error
+
+
+def test_mpls_table():
+ global fatal_error
+ net = get_topogen().net
+
+ # Skip if previous fatal error condition is raised
+ if fatal_error != "":
+ pytest.skip(fatal_error)
+
+ thisDir = os.path.dirname(os.path.realpath(__file__))
+
+ # Verify MPLS table
+ print("\n\n** Verifying MPLS table")
+ print("******************************************\n")
+ failures = 0
+
+ for i in range(1, 5):
+ refTableFile = "%s/r%s/show_mpls_table.ref" % (thisDir, i)
+ if os.path.isfile(refTableFile):
+ # Read expected result from file
+ expected = open(refTableFile).read()
+ # Fix newlines (make them all the same)
+ expected = ("\n".join(expected.splitlines()) + "\n").splitlines(1)
+
+ # Actual output from router
+ actual = net["r%s" % i].cmd('vtysh -c "show mpls table" 2> /dev/null')
+
+ # Fix inconsistent Label numbers at beginning of line
+ actual = re.sub(r"(\s+)[0-9]+(\s+LDP)", r"\1XX\2", actual)
+ # Fix inconsistent Label numbers at end of line
+ actual = re.sub(
+ r"(\s+[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+\s+)[0-9][0-9]", r"\1XX", actual
+ )
+
+ # Fix newlines (make them all the same)
+ actual = ("\n".join(actual.splitlines()) + "\n").splitlines(1)
+
+ # Sort lines which start with " XX LDP"
+ pattern = r"^\s+[0-9X]+\s+LDP"
+ swapped = True
+ while swapped:
+ swapped = False
+ for j in range(1, len(actual)):
+ if re.search(pattern, actual[j]) and re.search(
+ pattern, actual[j - 1]
+ ):
+ if actual[j - 1] > actual[j]:
+ temp = actual[j - 1]
+ actual[j - 1] = actual[j]
+ actual[j] = temp
+ swapped = True
+
+ # Generate Diff
+ diff = topotest.get_textdiff(
+ actual,
+ expected,
+ title1="actual MPLS table output",
+ title2="expected MPLS table output",
+ )
+
+ # Empty string if it matches, otherwise diff contains unified diff
+ if diff:
+ sys.stderr.write(
+ "r%s failed MPLS table output Check:\n%s\n" % (i, diff)
+ )
+ failures += 1
+ else:
+ print("r%s ok" % i)
+
+ assert failures == 0, "MPLS table output for router r%s:\n%s" % (i, diff)
+
+ # Make sure that all daemons are running
+ for i in range(1, 5):
+ fatal_error = net["r%s" % i].checkRouterRunning()
+ assert fatal_error == "", fatal_error
+
+
+def test_linux_mpls_routes():
+ global fatal_error
+ net = get_topogen().net
+
+ # Skip if previous fatal error condition is raised
+ if fatal_error != "":
+ pytest.skip(fatal_error)
+
+ thisDir = os.path.dirname(os.path.realpath(__file__))
+
+ # Verify Linux Kernel MPLS routes
+ print("\n\n** Verifying Linux Kernel MPLS routes")
+ print("******************************************\n")
+ failures = 0
+ for i in range(1, 5):
+ refTableFile = "%s/r%s/ip_mpls_route.ref" % (thisDir, i)
+ if os.path.isfile(refTableFile):
+ # Read expected result from file
+ expected = open(refTableFile).read().rstrip()
+ # Fix newlines (make them all the same)
+ expected = ("\n".join(expected.splitlines()) + "\n").splitlines(1)
+
+ # Actual output from router
+ actual = (
+ net["r%s" % i].cmd("ip -o -family mpls route 2> /dev/null").rstrip()
+ )
+
+ # Mask out label and protocol
+ actual = re.sub(r"[0-9][0-9] via inet ", "xx via inet ", actual)
+ actual = re.sub(r"[0-9][0-9] +proto", "xx proto", actual)
+ actual = re.sub(r"[0-9][0-9] as to ", "xx as to ", actual)
+ actual = re.sub(r"[ ]+proto \w+", " proto xx", actual)
+
+ # Sort nexthops
+ nexthop_sorted = []
+ for line in actual.splitlines():
+ tokens = re.split(r"\\\t", line.strip())
+ nexthop_sorted.append(
+ "{} {}".format(
+ tokens[0].strip(),
+ " ".join([token.strip() for token in sorted(tokens[1:])]),
+ ).strip()
+ )
+
+ # Sort lines and fixup differences between old and new iproute
+ actual = "\n".join(sorted(nexthop_sorted))
+ actual = re.sub(r"nexthop via", "nexthopvia", actual)
+ actual = re.sub(r" nexthop as to xx via inet ", " nexthopvia inet ", actual)
+ actual = re.sub(r" weight 1", "", actual)
+ actual = re.sub(r" [ ]+", " ", actual)
+
+ # put \n back at line ends
+ actual = ("\n".join(actual.splitlines()) + "\n").splitlines(1)
+
+ # Generate Diff
+ diff = topotest.get_textdiff(
+ actual,
+ expected,
+ title1="actual Linux Kernel MPLS route",
+ title2="expected Linux Kernel MPLS route",
+ )
+
+ # Empty string if it matches, otherwise diff contains unified diff
+ if diff:
+ sys.stderr.write(
+ "r%s failed Linux Kernel MPLS route output Check:\n%s\n" % (i, diff)
+ )
+ failures += 1
+ else:
+ print("r%s ok" % i)
+
+ assert (
+ failures == 0
+ ), "Linux Kernel MPLS route output for router r%s:\n%s" % (i, diff)
+
+ # Make sure that all daemons are running
+ for i in range(1, 5):
+ fatal_error = net["r%s" % i].checkRouterRunning()
+ assert fatal_error == "", fatal_error
+
+
+def test_shutdown_check_stderr():
+ global fatal_error
+ net = get_topogen().net
+
+ # Skip if previous fatal error condition is raised
+ if fatal_error != "":
+ pytest.skip(fatal_error)
+
+ if os.environ.get("TOPOTESTS_CHECK_STDERR") is None:
+ print(
+ "SKIPPED final check on StdErr output: Disabled (TOPOTESTS_CHECK_STDERR undefined)\n"
+ )
+ pytest.skip("Skipping test for Stderr output")
+
+ thisDir = os.path.dirname(os.path.realpath(__file__))
+
+ print("\n\n** Verifying unexpected STDERR output from daemons")
+ print("******************************************\n")
+
+ for i in range(1, 5):
+ net["r%s" % i].stopRouter()
+ log = net["r%s" % i].getStdErr("ldpd")
+ if log:
+ print("\nRouter r%s LDPd StdErr Log:\n%s" % (i, log))
+ log = net["r%s" % i].getStdErr("ospfd")
+ if log:
+ print("\nRouter r%s OSPFd StdErr Log:\n%s" % (i, log))
+ log = net["r%s" % i].getStdErr("zebra")
+ if log:
+ print("\nRouter r%s Zebra StdErr Log:\n%s" % (i, log))
+
+
+def test_shutdown_check_memleak():
+ global fatal_error
+ net = get_topogen().net
+
+ # Skip if previous fatal error condition is raised
+ if fatal_error != "":
+ pytest.skip(fatal_error)
+
+ if os.environ.get("TOPOTESTS_CHECK_MEMLEAK") is None:
+ print(
+ "SKIPPED final check on Memory leaks: Disabled (TOPOTESTS_CHECK_MEMLEAK undefined)\n"
+ )
+ pytest.skip("Skipping test for memory leaks")
+
+ thisDir = os.path.dirname(os.path.realpath(__file__))
+
+ for i in range(1, 5):
+ net["r%s" % i].stopRouter()
+ net["r%s" % i].report_memory_leaks(
+ os.environ.get("TOPOTESTS_CHECK_MEMLEAK"), os.path.basename(__file__)
+ )
+
+
+if __name__ == "__main__":
+
+ # To suppress tracebacks, either use the following pytest call or add "--tb=no" to cli
+ # retval = pytest.main(["-s", "--tb=no"])
+ retval = pytest.main(["-s"])
+ sys.exit(retval)
generated by cgit v1.2.3 (git 2.39.1) at 2024-07-14 05:25:12 +0000