diff options
Diffstat (limited to 'tests/topotests/ldp-topo1/test_ldp_topo1.py')
| -rwxr-xr-x | tests/topotests/ldp-topo1/test_ldp_topo1.py | 432 |
1 files changed, 291 insertions, 141 deletions
diff --git a/tests/topotests/ldp-topo1/test_ldp_topo1.py b/tests/topotests/ldp-topo1/test_ldp_topo1.py index c0d11fd5e0..cef4d6587e 100755 --- a/tests/topotests/ldp-topo1/test_ldp_topo1.py +++ b/tests/topotests/ldp-topo1/test_ldp_topo1.py @@ -57,7 +57,7 @@ r3-eth1 .3 | | .3 r3-eth0 | .4 r4-eth0 | r3 | | r4 | | 3.3.3.3 | | 4.4.4.4 | +-----------+ +---------+ -""" +""" import os import re @@ -83,6 +83,7 @@ fatal_error = "" ## ##################################################### + class NetworkTopo(Topo): "LDP Test Topology 1" @@ -91,23 +92,65 @@ class NetworkTopo(Topo): # Setup Routers router = {} for i in range(1, 5): - router[i] = topotest.addRouter(self, 'r%s' % i) + router[i] = topotest.addRouter(self, "r%s" % i) # Setup Switches, add Interfaces and Connections switch = {} # First switch - switch[0] = self.addSwitch('sw0', cls=topotest.LegacySwitch) - self.addLink(switch[0], router[1], intfName2='r1-eth0', addr1='80:AA:00:00:00:00', addr2='00:11:00:01:00:00') - self.addLink(switch[0], router[2], intfName2='r2-eth0', addr1='80:AA:00:00:00:01', addr2='00:11:00:02:00:00') + switch[0] = self.addSwitch("sw0", cls=topotest.LegacySwitch) + self.addLink( + switch[0], + router[1], + intfName2="r1-eth0", + addr1="80:AA:00:00:00:00", + addr2="00:11:00:01:00:00", + ) + self.addLink( + switch[0], + router[2], + intfName2="r2-eth0", + addr1="80:AA:00:00:00:01", + addr2="00:11:00:02:00:00", + ) # Second switch - switch[1] = self.addSwitch('sw1', cls=topotest.LegacySwitch) - self.addLink(switch[1], router[2], intfName2='r2-eth1', addr1='80:AA:00:01:00:00', addr2='00:11:00:02:00:01') - self.addLink(switch[1], router[3], intfName2='r3-eth0', addr1='80:AA:00:01:00:01', addr2='00:11:00:03:00:00') - self.addLink(switch[1], router[4], intfName2='r4-eth0', addr1='80:AA:00:01:00:02', addr2='00:11:00:04:00:00') + switch[1] = self.addSwitch("sw1", cls=topotest.LegacySwitch) + self.addLink( + switch[1], + router[2], + intfName2="r2-eth1", + addr1="80:AA:00:01:00:00", + addr2="00:11:00:02:00:01", + ) + self.addLink( + switch[1], + router[3], + intfName2="r3-eth0", + addr1="80:AA:00:01:00:01", + addr2="00:11:00:03:00:00", + ) + self.addLink( + switch[1], + router[4], + intfName2="r4-eth0", + addr1="80:AA:00:01:00:02", + addr2="00:11:00:04:00:00", + ) # Third switch - switch[2] = self.addSwitch('sw2', cls=topotest.LegacySwitch) - self.addLink(switch[2], router[2], intfName2='r2-eth2', addr1='80:AA:00:02:00:00', addr2='00:11:00:02:00:02') - self.addLink(switch[2], router[3], intfName2='r3-eth1', addr1='80:AA:00:02:00:01', addr2='00:11:00:03:00:01') + switch[2] = self.addSwitch("sw2", cls=topotest.LegacySwitch) + self.addLink( + switch[2], + router[2], + intfName2="r2-eth2", + addr1="80:AA:00:02:00:00", + addr2="00:11:00:02:00:02", + ) + self.addLink( + switch[2], + router[3], + intfName2="r3-eth1", + addr1="80:AA:00:02:00:01", + addr2="00:11:00:03:00:01", + ) ##################################################### @@ -116,6 +159,7 @@ class NetworkTopo(Topo): ## ##################################################### + def setup_module(module): global topo, net global fatal_error @@ -124,7 +168,7 @@ def setup_module(module): print("******************************************\n") print("Cleanup old Mininet runs") - os.system('sudo mn -c > /dev/null 2>&1') + os.system("sudo mn -c > /dev/null 2>&1") thisDir = os.path.dirname(os.path.realpath(__file__)) topo = NetworkTopo() @@ -134,10 +178,10 @@ def setup_module(module): # 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)) - fatal_error = net['r%s' % i].startRouter() + 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)) + fatal_error = net["r%s" % i].startRouter() if fatal_error != "": break @@ -145,6 +189,7 @@ def setup_module(module): # For debugging after starting FRR/Quagga daemons, uncomment the next line # CLI(net) + def teardown_module(module): global net @@ -160,7 +205,7 @@ def test_router_running(): global net # Skip if previous fatal error condition is raised - if (fatal_error != ""): + if fatal_error != "": pytest.skip(fatal_error) print("\n\n** Check if FRR/Quagga is running on each Router node") @@ -169,18 +214,19 @@ def test_router_running(): # Starting Routers for i in range(1, 5): - fatal_error = net['r%s' % i].checkRouterRunning() + fatal_error = net["r%s" % i].checkRouterRunning() assert fatal_error == "", fatal_error # For debugging after starting FRR/Quagga daemons, uncomment the next line # CLI(net) + def test_mpls_interfaces(): global fatal_error global net # Skip if previous fatal error condition is raised - if (fatal_error != ""): + if fatal_error != "": pytest.skip(fatal_error) thisDir = os.path.dirname(os.path.realpath(__file__)) @@ -190,40 +236,51 @@ def test_mpls_interfaces(): print("******************************************\n") failures = 0 for i in range(1, 5): - refTableFile = '%s/r%s/show_mpls_ldp_interface.ref' + 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) + 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() + 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) + actual = ("\n".join(actual.splitlines()) + "\n").splitlines(1) # Generate Diff - diff = topotest.get_textdiff(actual, expected, + diff = topotest.get_textdiff( + actual, + expected, title1="actual MPLS LDP interface status", - title2="expected 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)) + 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: + 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) + 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() + fatal_error = net["r%s" % i].checkRouterRunning() assert fatal_error == "", fatal_error # For debugging after starting FRR/Quagga daemons, uncomment the next line @@ -235,7 +292,7 @@ def test_mpls_ldp_neighbor_establish(): global net # Skip if previous fatal error condition is raised - if (fatal_error != ""): + if fatal_error != "": pytest.skip(fatal_error) # Wait for MPLS LDP neighbors to establish. @@ -247,17 +304,23 @@ def test_mpls_ldp_neighbor_establish(): 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() + 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) + 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.*' + 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])): + 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 @@ -265,14 +328,14 @@ def test_mpls_ldp_neighbor_establish(): # Need at least one operational neighbor established = "" # Empty string shows NOT established if not established: - print('Waiting for r%s' %i) + print("Waiting for r%s" % i) sys.stdout.flush() break if not established: sleep(5) timeout -= 5 else: - print('Done') + print("Done") break else: # Bail out with error if a router fails to converge @@ -285,10 +348,10 @@ def test_mpls_ldp_neighbor_establish(): # 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() + fatal_error = net["r%s" % i].checkRouterRunning() assert fatal_error == "", fatal_error @@ -297,7 +360,7 @@ def test_mpls_ldp_discovery(): global net # Skip if previous fatal error condition is raised - if (fatal_error != ""): + if fatal_error != "": pytest.skip(fatal_error) thisDir = os.path.dirname(os.path.realpath(__file__)) @@ -307,39 +370,54 @@ def test_mpls_ldp_discovery(): print("******************************************\n") failures = 0 for i in range(1, 5): - refTableFile = '%s/r%s/show_mpls_ldp_discovery.ref' % (thisDir, i) + 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() + 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) + 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() + 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) + actual = ("\n".join(actual.splitlines()) + "\n").splitlines(1) # Generate Diff - diff = topotest.get_textdiff(actual, expected, + diff = topotest.get_textdiff( + actual, + expected, title1="actual MPLS LDP discovery output", - title2="expected 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)) + 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) + 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() + fatal_error = net["r%s" % i].checkRouterRunning() assert fatal_error == "", fatal_error # For debugging after starting FRR/Quagga daemons, uncomment the next line @@ -351,7 +429,7 @@ def test_mpls_ldp_neighbor(): global net # Skip if previous fatal error condition is raised - if (fatal_error != ""): + if fatal_error != "": pytest.skip(fatal_error) thisDir = os.path.dirname(os.path.realpath(__file__)) @@ -361,44 +439,59 @@ def test_mpls_ldp_neighbor(): print("******************************************\n") failures = 0 for i in range(1, 5): - refTableFile = '%s/r%s/show_mpls_ldp_neighbor.ref' % (thisDir, i) + 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) + 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() + 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) + 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) + actual = ("\n".join(actual.splitlines()) + "\n").splitlines(1) # Generate Diff - diff = topotest.get_textdiff(actual, expected, + diff = topotest.get_textdiff( + actual, + expected, title1="actual MPLS LDP neighbor output", - title2="expected 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)) + 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) + 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() + fatal_error = net["r%s" % i].checkRouterRunning() assert fatal_error == "", fatal_error # For debugging after starting FRR/Quagga daemons, uncomment the next line - #CLI(net) + # CLI(net) def test_mpls_ldp_binding(): @@ -410,7 +503,7 @@ def test_mpls_ldp_binding(): # pytest.skip("Skipping test_mpls_ldp_binding") # Skip if previous fatal error condition is raised - if (fatal_error != ""): + if fatal_error != "": pytest.skip(fatal_error) thisDir = os.path.dirname(os.path.realpath(__file__)) @@ -420,58 +513,77 @@ def test_mpls_ldp_binding(): print("******************************************\n") failures = 0 for i in range(1, 5): - refTableFile = '%s/r%s/show_mpls_ldp_binding.ref' % (thisDir, i) + 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) + 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() + 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) + 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) + 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+' + 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] + 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, + diff = topotest.get_textdiff( + actual, + expected, title1="actual MPLS LDP binding output", - title2="expected 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)) + 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 Interface binding output for router r%s:\n%s" % (i, diff) + assert ( + failures == 0 + ), "MPLS LDP Interface 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() + fatal_error = net["r%s" % i].checkRouterRunning() assert fatal_error == "", fatal_error # For debugging after starting FRR/Quagga daemons, uncomment the next line - #CLI(net) + # CLI(net) def test_zebra_ipv4_routingTable(): @@ -479,7 +591,7 @@ def test_zebra_ipv4_routingTable(): global net # Skip if previous fatal error condition is raised - if (fatal_error != ""): + if fatal_error != "": pytest.skip(fatal_error) thisDir = os.path.dirname(os.path.realpath(__file__)) @@ -489,13 +601,17 @@ def test_zebra_ipv4_routingTable(): print("******************************************\n") failures = 0 for i in range(1, 5): - refTableFile = '%s/r%s/show_ipv4_route.ref' % (thisDir, i) + 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() + actual = ( + net["r%s" % i] + .cmd('vtysh -c "show ip route" 2> /dev/null | grep "^O"') + .rstrip() + ) # Drop timers on end of line (older Quagga Versions) actual = re.sub(r", [0-2][0-9]:[0-5][0-9]:[0-5][0-9]", "", actual) @@ -505,32 +621,40 @@ def test_zebra_ipv4_routingTable(): # 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)): + 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) + expected = ("\n".join(expected.splitlines()) + "\n").splitlines(1) # Fix newlines (make them all the same) - actual = ('\n'.join(actual.splitlines()) + '\n').splitlines(1) + actual = ("\n".join(actual.splitlines()) + "\n").splitlines(1) # Generate Diff - diff = topotest.get_textdiff(actual, expected, + diff = topotest.get_textdiff( + actual, + expected, title1="actual IPv4 zebra routing table", - title2="expected 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)) + 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) + 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() + fatal_error = net["r%s" % i].checkRouterRunning() assert fatal_error == "", fatal_error # For debugging after starting FRR/Quagga daemons, uncomment the next line @@ -542,7 +666,7 @@ def test_mpls_table(): global net # Skip if previous fatal error condition is raised - if (fatal_error != ""): + if fatal_error != "": pytest.skip(fatal_error) thisDir = os.path.dirname(os.path.realpath(__file__)) @@ -553,45 +677,54 @@ def test_mpls_table(): failures = 0 for i in range(1, 5): - refTableFile = '%s/r%s/show_mpls_table.ref' % (thisDir, i) + 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) + 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') + 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) + 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) + actual = ("\n".join(actual.splitlines()) + "\n").splitlines(1) # Sort lines which start with " XX LDP" - pattern = r'^\s+[0-9X]+\s+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] + 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, + diff = topotest.get_textdiff( + actual, + expected, title1="actual MPLS table output", - title2="expected 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)) + sys.stderr.write( + "r%s failed MPLS table output Check:\n%s\n" % (i, diff) + ) failures += 1 else: print("r%s ok" % i) @@ -600,7 +733,7 @@ def test_mpls_table(): # Make sure that all daemons are running for i in range(1, 5): - fatal_error = net['r%s' % i].checkRouterRunning() + fatal_error = net["r%s" % i].checkRouterRunning() assert fatal_error == "", fatal_error # For debugging after starting FRR/Quagga daemons, uncomment the next line @@ -611,8 +744,8 @@ def test_linux_mpls_routes(): global fatal_error global net - # Skip if previous fatal error condition is raised - if (fatal_error != ""): + # Skip if previous fatal error condition is raised + if fatal_error != "": pytest.skip(fatal_error) thisDir = os.path.dirname(os.path.realpath(__file__)) @@ -622,15 +755,17 @@ def test_linux_mpls_routes(): print("******************************************\n") failures = 0 for i in range(1, 5): - refTableFile = '%s/r%s/ip_mpls_route.ref' % (thisDir, i) + 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) + 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() + 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) @@ -641,39 +776,48 @@ def test_linux_mpls_routes(): # 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()) + 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 = "\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) + actual = ("\n".join(actual.splitlines()) + "\n").splitlines(1) # Generate Diff - diff = topotest.get_textdiff(actual, expected, + diff = topotest.get_textdiff( + actual, + expected, title1="actual Linux Kernel MPLS route", - title2="expected 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)) + 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) + 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() + fatal_error = net["r%s" % i].checkRouterRunning() assert fatal_error == "", fatal_error # For debugging after starting FRR/Quagga daemons, uncomment the next line @@ -685,12 +829,14 @@ def test_shutdown_check_stderr(): global net # Skip if previous fatal error condition is raised - if (fatal_error != ""): + 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') + 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__)) @@ -698,14 +844,14 @@ def test_shutdown_check_stderr(): print("******************************************\n") for i in range(1, 5): - net['r%s' % i].stopRouter() - log = net['r%s' % i].getStdErr('ldpd') + 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') + 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') + log = net["r%s" % i].getStdErr("zebra") if log: print("\nRouter r%s Zebra StdErr Log:\n%s" % (i, log)) @@ -715,23 +861,27 @@ def test_shutdown_check_memleak(): global net # Skip if previous fatal error condition is raised - if (fatal_error != ""): + 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') - + 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__)) + 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__': +if __name__ == "__main__": - setLogLevel('info') + setLogLevel("info") # 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"]) |