diff options
Diffstat (limited to 'tests')
19 files changed, 1663 insertions, 1017 deletions
diff --git a/tests/lib/test_nexthop.c b/tests/lib/test_nexthop.c new file mode 100644 index 0000000000..659d207b4e --- /dev/null +++ b/tests/lib/test_nexthop.c @@ -0,0 +1,201 @@ +/* + * Nexthop module test. + * + * Copyright (C) 2021 by Volta Networks, Inc. + * + * This file is part of FRR. + * + * FRR is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2, or (at your option) any + * later version. + * + * FRR is distributed in the hope that it will be useful, but + * WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * General Public License for more details. + * + * You should have received a copy of the GNU General Public License along + * with this program; see the file COPYING; if not, write to the Free Software + * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA + */ + +#include <zebra.h> +#include <nexthop.h> + +static bool verbose; + +static void test_run_first(void) +{ + int ret, i; + struct nexthop *nh1, *nh2; + struct in_addr addr; + struct in6_addr addr6; + mpls_label_t labels[MPLS_MAX_LABELS]; + + /* Test comparison apis */ + + /* ifindex comparisons */ + nh1 = nexthop_from_ifindex(11, 0); + nh2 = nexthop_from_ifindex(12, 0); + + ret = nexthop_cmp_basic(nh1, nh2); + assert(ret < 0); + + nexthop_free(nh1); + nh1 = nexthop_from_ifindex(12, 0); + + ret = nexthop_cmp_basic(nh1, nh2); + assert(ret == 0); + + nexthop_free(nh1); + nexthop_free(nh2); + + /* ipv4, vrf */ + addr.s_addr = 0x04030201; + nh1 = nexthop_from_ipv4(&addr, NULL, 0); + nh2 = nexthop_from_ipv4(&addr, NULL, 111); + + ret = nexthop_cmp_basic(nh1, nh2); + assert(ret != 0); + + nexthop_free(nh2); + + addr.s_addr = 0x04030202; + nh2 = nexthop_from_ipv4(&addr, NULL, 0); + + ret = nexthop_cmp_basic(nh1, nh2); + assert(ret != 0); + + nexthop_free(nh2); + + addr.s_addr = 0x04030201; + nh2 = nexthop_from_ipv4(&addr, NULL, 0); + + ret = nexthop_cmp_basic(nh1, nh2); + assert(ret == 0); + + /* Weight */ + nh2->weight = 20; + + ret = nexthop_cmp_basic(nh1, nh2); + assert(ret != 0); + + nexthop_free(nh1); + nexthop_free(nh2); + + /* ipv6 */ + memset(addr6.s6_addr, 0, sizeof(addr6.s6_addr)); + nh1 = nexthop_from_ipv6(&addr6, 0); + nh2 = nexthop_from_ipv6(&addr6, 0); + + ret = nexthop_cmp_basic(nh1, nh2); + assert(ret == 0); + + nexthop_free(nh2); + + nh2 = nexthop_from_ipv6(&addr6, 1); + + ret = nexthop_cmp_basic(nh1, nh2); + assert(ret != 0); + + nexthop_free(nh2); + + addr6.s6_addr[14] = 1; + addr6.s6_addr[15] = 1; + nh2 = nexthop_from_ipv6(&addr6, 0); + + ret = nexthop_cmp_basic(nh1, nh2); + assert(ret != 0); + + nexthop_free(nh1); + nexthop_free(nh2); + + /* Blackhole */ + nh1 = nexthop_from_blackhole(BLACKHOLE_REJECT); + nh2 = nexthop_from_blackhole(BLACKHOLE_REJECT); + + ret = nexthop_cmp_basic(nh1, nh2); + assert(ret == 0); + + nexthop_free(nh2); + + nh2 = nexthop_from_blackhole(BLACKHOLE_NULL); + + ret = nexthop_cmp_basic(nh1, nh2); + assert(ret != 0); + + /* Labels */ + addr.s_addr = 0x04030201; + nh1 = nexthop_from_ipv4(&addr, NULL, 0); + nh2 = nexthop_from_ipv4(&addr, NULL, 0); + + memset(labels, 0, sizeof(labels)); + labels[0] = 111; + labels[1] = 222; + + nexthop_add_labels(nh1, ZEBRA_LSP_STATIC, 2, labels); + + ret = nexthop_cmp_basic(nh1, nh2); + assert(ret != 0); + + nexthop_add_labels(nh2, ZEBRA_LSP_STATIC, 2, labels); + + ret = nexthop_cmp_basic(nh1, nh2); + assert(ret == 0); + + nexthop_free(nh2); + + /* LSP type isn't included */ + nh2 = nexthop_from_ipv4(&addr, NULL, 0); + nexthop_add_labels(nh2, ZEBRA_LSP_LDP, 2, labels); + + ret = nexthop_cmp_basic(nh1, nh2); + assert(ret == 0); + + nexthop_free(nh2); + + labels[2] = 333; + nh2 = nexthop_from_ipv4(&addr, NULL, 0); + nexthop_add_labels(nh2, ZEBRA_LSP_LDP, 3, labels); + + ret = nexthop_cmp_basic(nh1, nh2); + assert(ret != 0); + + nexthop_free(nh1); + nexthop_free(nh2); + + nh1 = nexthop_from_ipv4(&addr, NULL, 0); + nh2 = nexthop_from_ipv4(&addr, NULL, 0); + + for (i = 0; i < MPLS_MAX_LABELS; i++) + labels[i] = 111 * (i + 1); + + nexthop_add_labels(nh1, ZEBRA_LSP_LDP, MPLS_MAX_LABELS, labels); + nexthop_add_labels(nh2, ZEBRA_LSP_LDP, MPLS_MAX_LABELS, labels); + + ret = nexthop_cmp_basic(nh1, nh2); + assert(ret == 0); + + nexthop_free(nh2); + + /* Test very last label in stack */ + labels[15] = 999; + nh2 = nexthop_from_ipv4(&addr, NULL, 0); + nexthop_add_labels(nh2, ZEBRA_LSP_LDP, MPLS_MAX_LABELS, labels); + + ret = nexthop_cmp_basic(nh1, nh2); + assert(ret != 0); + + /* End */ + nexthop_free(nh1); + nexthop_free(nh2); +} + +int main(int argc, char **argv) +{ + if (argc >= 2 && !strcmp("-v", argv[1])) + verbose = true; + test_run_first(); + printf("Simple test passed.\n"); +} diff --git a/tests/lib/test_nexthop.py b/tests/lib/test_nexthop.py new file mode 100644 index 0000000000..81bfa43ab5 --- /dev/null +++ b/tests/lib/test_nexthop.py @@ -0,0 +1,8 @@ +import frrtest + + +class TestNexthopIter(frrtest.TestMultiOut): + program = "./test_nexthop" + + +TestNexthopIter.onesimple("Simple test passed.") diff --git a/tests/subdir.am b/tests/subdir.am index 139f4878c8..43fad29fa2 100644 --- a/tests/subdir.am +++ b/tests/subdir.am @@ -76,6 +76,7 @@ check_PROGRAMS = \ tests/lib/test_idalloc \ tests/lib/test_memory \ tests/lib/test_nexthop_iter \ + tests/lib/test_nexthop \ tests/lib/test_ntop \ tests/lib/test_prefix2str \ tests/lib/test_printfrr \ @@ -293,6 +294,10 @@ tests_lib_test_nexthop_iter_CFLAGS = $(TESTS_CFLAGS) tests_lib_test_nexthop_iter_CPPFLAGS = $(TESTS_CPPFLAGS) tests_lib_test_nexthop_iter_LDADD = $(ALL_TESTS_LDADD) tests_lib_test_nexthop_iter_SOURCES = tests/lib/test_nexthop_iter.c tests/helpers/c/prng.c +tests_lib_test_nexthop_CFLAGS = $(TESTS_CFLAGS) +tests_lib_test_nexthop_CPPFLAGS = $(TESTS_CPPFLAGS) +tests_lib_test_nexthop_LDADD = $(ALL_TESTS_LDADD) +tests_lib_test_nexthop_SOURCES = tests/lib/test_nexthop.c tests_lib_test_ntop_CFLAGS = $(TESTS_CFLAGS) tests_lib_test_ntop_CPPFLAGS = $(CPPFLAGS_BASE) # no assert override tests_lib_test_ntop_LDADD = # none @@ -412,6 +417,7 @@ EXTRA_DIST += \ tests/lib/test_assert.py \ tests/lib/test_atomlist.py \ tests/lib/test_nexthop_iter.py \ + tests/lib/test_nexthop.py \ tests/lib/test_ntop.py \ tests/lib/test_prefix2str.py \ tests/lib/test_printfrr.py \ diff --git a/tests/topotests/bgp-auth/test_bgp_auth.py b/tests/topotests/bgp-auth/test_bgp_auth.py index 521ca332d0..b2cdef1c93 100644 --- a/tests/topotests/bgp-auth/test_bgp_auth.py +++ b/tests/topotests/bgp-auth/test_bgp_auth.py @@ -357,9 +357,10 @@ def check_neigh_state(router, peer, state, vrf=""): "show bgp vrf {} neighbors {} json".format(vrf, peer) ) neigh_output_json = json.loads(neigh_output) - if neigh_output_json[peer]["bgpState"] == state: - matched = True - break + if peer in neigh_output_json.keys(): + if neigh_output_json[peer]["bgpState"] == state: + matched = True + break count += 1 sleep(1) diff --git a/tests/topotests/isis-topo1/r1/r1_topology.json b/tests/topotests/isis-topo1/r1/r1_topology.json index 6d2bbb80c9..337d6bf5ef 100644 --- a/tests/topotests/isis-topo1/r1/r1_topology.json +++ b/tests/topotests/isis-topo1/r1/r1_topology.json @@ -18,9 +18,9 @@ "vertex": "r1" }, { - "metric": "internal", - "parent": "0", - "type": "IP", + "metric": "0", + "parent": "r1(4)", + "type": "IP internal", "vertex": "10.0.20.0/24" }, { @@ -32,27 +32,27 @@ "vertex": "r3" }, { - "interface": "r3", - "metric": "TE", - "next-hop": "10", - "parent": "r1-eth0", - "type": "IP", + "interface": "r1-eth0", + "metric": "10", + "next-hop": "r3", + "parent": "r3(4)", + "type": "IP TE", "vertex": "10.0.10.0/24" }, { - "interface": "r3", - "metric": "TE", - "next-hop": "10", - "parent": "r1-eth0", - "type": "IP", + "interface": "r1-eth0", + "metric": "10", + "next-hop": "r3", + "parent": "r3(4)", + "type": "IP TE", "vertex": "10.0.20.0/24" }, { - "interface": "r3", - "metric": "TE", - "next-hop": "10", - "parent": "r1-eth0", - "type": "IP", + "interface": "r1-eth0", + "metric": "10", + "next-hop": "r3", + "parent": "r3(4)", + "type": "IP TE", "vertex": "10.254.0.3/32" } ], @@ -61,9 +61,9 @@ "vertex": "r1" }, { - "metric": "internal", - "parent": "0", - "type": "IP6", + "metric": "0", + "parent": "r1(4)", + "type": "IP6 internal", "vertex": "2001:db8:1:1::/64" }, { @@ -75,17 +75,19 @@ "vertex": "r3" }, { - "interface": "r3", - "next-hop": "10", - "parent": "r1-eth0", - "type": "IP6", + "metric": "10", + "interface": "r1-eth0", + "next-hop": "r3", + "parent": "r3(4)", + "type": "IP6 internal", "vertex": "2001:db8:2:1::/64" }, { - "interface": "r3", - "next-hop": "10", - "parent": "r1-eth0", - "type": "IP6", + "metric": "10", + "interface": "r1-eth0", + "next-hop": "r3", + "parent": "r3(4)", + "type": "IP6 internal", "vertex": "2001:db8:f::3/128" } ] diff --git a/tests/topotests/isis-topo1/r2/r2_topology.json b/tests/topotests/isis-topo1/r2/r2_topology.json index 396c618caa..de90fb5a32 100644 --- a/tests/topotests/isis-topo1/r2/r2_topology.json +++ b/tests/topotests/isis-topo1/r2/r2_topology.json @@ -18,9 +18,9 @@ "vertex": "r2" }, { - "metric": "internal", - "parent": "0", - "type": "IP", + "metric": "0", + "parent": "r2(4)", + "type": "IP internal", "vertex": "10.0.21.0/24" }, { @@ -32,27 +32,27 @@ "vertex": "r4" }, { - "interface": "r4", - "metric": "TE", - "next-hop": "10", - "parent": "r2-eth0", - "type": "IP", + "interface": "r2-eth0", + "metric": "10", + "next-hop": "r4", + "parent": "r4(4)", + "type": "IP TE", "vertex": "10.0.11.0/24" }, { - "interface": "r4", - "metric": "TE", - "next-hop": "10", - "parent": "r2-eth0", - "type": "IP", + "interface": "r2-eth0", + "metric": "10", + "next-hop": "r4", + "parent": "r4(4)", + "type": "IP TE", "vertex": "10.0.21.0/24" }, { - "interface": "r4", - "metric": "TE", - "next-hop": "10", - "parent": "r2-eth0", - "type": "IP", + "interface": "r2-eth0", + "metric": "10", + "next-hop": "r4", + "parent": "r4(4)", + "type": "IP TE", "vertex": "10.254.0.4/32" } ], @@ -61,9 +61,9 @@ "vertex": "r2" }, { - "metric": "internal", - "parent": "0", - "type": "IP6", + "metric": "0", + "parent": "r2(4)", + "type": "IP6 internal", "vertex": "2001:db8:1:2::/64" }, { @@ -75,17 +75,19 @@ "vertex": "r4" }, { - "interface": "r4", - "next-hop": "10", - "parent": "r2-eth0", - "type": "IP6", + "metric": "10", + "interface": "r2-eth0", + "next-hop": "r4", + "parent": "r4(4)", + "type": "IP6 internal", "vertex": "2001:db8:2:2::/64" }, { - "interface": "r4", - "next-hop": "10", - "parent": "r2-eth0", - "type": "IP6", + "metric": "10", + "interface": "r2-eth0", + "next-hop": "r4", + "parent": "r4(4)", + "type": "IP6 internal", "vertex": "2001:db8:f::4/128" } ] diff --git a/tests/topotests/isis-topo1/r3/r3_topology.json b/tests/topotests/isis-topo1/r3/r3_topology.json index 5ab58c41a6..2d36f9b427 100644 --- a/tests/topotests/isis-topo1/r3/r3_topology.json +++ b/tests/topotests/isis-topo1/r3/r3_topology.json @@ -6,9 +6,9 @@ "vertex": "r3" }, { - "metric": "internal", - "parent": "0", - "type": "IP", + "metric": "0", + "parent": "r3(4)", + "type": "IP internal", "vertex": "10.0.10.0/24" }, { @@ -20,27 +20,27 @@ "vertex": "r5" }, { - "interface": "r5", - "metric": "TE", - "next-hop": "10", - "parent": "r3-eth1", - "type": "IP", + "interface": "r3-eth1", + "metric": "10", + "next-hop": "r5", + "parent": "r5(4)", + "type": "IP TE", "vertex": "10.0.10.0/24" }, { - "interface": "r5", - "metric": "TE", - "next-hop": "10", - "parent": "r3-eth1", - "type": "IP", + "interface": "r3-eth1", + "metric": "10", + "next-hop": "r5", + "parent": "r5(4)", + "type": "IP TE", "vertex": "10.0.11.0/24" }, { - "interface": "r5", - "metric": "TE", - "next-hop": "10", - "parent": "r3-eth1", - "type": "IP", + "interface": "r3-eth1", + "metric": "10", + "next-hop": "r5", + "parent": "r5(4)", + "type": "IP TE", "vertex": "10.254.0.5/32" }, { @@ -51,19 +51,19 @@ "vertex": "r4" }, { - "interface": "r5", - "metric": "TE", - "next-hop": "20", - "parent": "r3-eth1", - "type": "IP", + "interface": "r3-eth1", + "metric": "20", + "next-hop": "r5", + "parent": "r4(4)", + "type": "IP TE", "vertex": "10.0.21.0/24" }, { - "interface": "r5", - "metric": "TE", - "next-hop": "20", - "parent": "r3-eth1", - "type": "IP", + "interface": "r3-eth1", + "metric": "20", + "next-hop": "r5", + "parent": "r4(4)", + "type": "IP TE", "vertex": "10.254.0.4/32" } ], @@ -72,9 +72,9 @@ "vertex": "r3" }, { - "metric": "internal", - "parent": "0", - "type": "IP6", + "metric": "0", + "parent": "r3(4)", + "type": "IP6 internal", "vertex": "2001:db8:2:1::/64" }, { @@ -86,17 +86,19 @@ "vertex": "r5" }, { - "interface": "r5", - "next-hop": "10", - "parent": "r3-eth1", - "type": "IP6", + "metric": "10", + "interface": "r3-eth1", + "next-hop": "r5", + "parent": "r5(4)", + "type": "IP6 internal", "vertex": "2001:db8:2:2::/64" }, { - "interface": "r5", - "next-hop": "10", - "parent": "r3-eth1", - "type": "IP6", + "metric": "10", + "interface": "r3-eth1", + "next-hop": "r5", + "parent": "r5(4)", + "type": "IP6 internal", "vertex": "2001:db8:f::5/128" }, { @@ -107,17 +109,19 @@ "vertex": "r4" }, { - "interface": "r5", - "next-hop": "20", - "parent": "r3-eth1", - "type": "IP6", + "metric": "20", + "interface": "r3-eth1", + "next-hop": "r5", + "parent": "r4(4)", + "type": "IP6 internal", "vertex": "2001:db8:1:2::/64" }, { - "interface": "r5", - "next-hop": "20", - "parent": "r3-eth1", - "type": "IP6", + "metric": "20", + "interface": "r3-eth1", + "next-hop": "r5", + "parent": "r4(4)", + "type": "IP6 internal", "vertex": "2001:db8:f::4/128" } ] @@ -128,9 +132,9 @@ "vertex": "r3" }, { - "metric": "internal", - "parent": "0", - "type": "IP", + "metric": "0", + "parent": "r3(4)", + "type": "IP internal", "vertex": "10.0.20.0/24" }, { @@ -142,19 +146,19 @@ "vertex": "r1" }, { - "interface": "r1", - "metric": "TE", - "next-hop": "10", - "parent": "r3-eth0", - "type": "IP", + "interface": "r3-eth0", + "metric": "10", + "next-hop": "r1", + "parent": "r1(4)", + "type": "IP TE", "vertex": "10.0.20.0/24" }, { - "interface": "r1", - "metric": "TE", - "next-hop": "10", - "parent": "r3-eth0", - "type": "IP", + "interface": "r3-eth0", + "metric": "10", + "next-hop": "r1", + "parent": "r1(4)", + "type": "IP TE", "vertex": "10.254.0.1/32" } ], @@ -163,9 +167,9 @@ "vertex": "r3" }, { - "metric": "internal", - "parent": "0", - "type": "IP6", + "metric": "0", + "parent": "r3(4)", + "type": "IP6 internal", "vertex": "2001:db8:1:1::/64" }, { @@ -177,10 +181,11 @@ "vertex": "r1" }, { - "interface": "r1", - "next-hop": "10", - "parent": "r3-eth0", - "type": "IP6", + "metric": "10", + "interface": "r3-eth0", + "next-hop": "r1", + "parent": "r1(4)", + "type": "IP6 internal", "vertex": "2001:db8:f::1/128" } ] diff --git a/tests/topotests/isis-topo1/r4/r4_topology.json b/tests/topotests/isis-topo1/r4/r4_topology.json index ae74a60948..e7d7841912 100644 --- a/tests/topotests/isis-topo1/r4/r4_topology.json +++ b/tests/topotests/isis-topo1/r4/r4_topology.json @@ -6,9 +6,9 @@ "vertex": "r4" }, { - "metric": "internal", - "parent": "0", - "type": "IP", + "metric": "0", + "parent": "r4(4)", + "type": "IP internal", "vertex": "10.0.11.0/24" }, { @@ -20,27 +20,27 @@ "vertex": "r5" }, { - "interface": "r5", - "metric": "TE", - "next-hop": "10", - "parent": "r4-eth1", - "type": "IP", + "interface": "r4-eth1", + "metric": "10", + "next-hop": "r5", + "parent": "r5(4)", + "type": "IP TE", "vertex": "10.0.10.0/24" }, { - "interface": "r5", - "metric": "TE", - "next-hop": "10", - "parent": "r4-eth1", - "type": "IP", + "interface": "r4-eth1", + "metric": "10", + "next-hop": "r5", + "parent": "r5(4)", + "type": "IP TE", "vertex": "10.0.11.0/24" }, { - "interface": "r5", - "metric": "TE", - "next-hop": "10", - "parent": "r4-eth1", - "type": "IP", + "interface": "r4-eth1", + "metric": "10", + "next-hop": "r5", + "parent": "r5(4)", + "type": "IP TE", "vertex": "10.254.0.5/32" }, { @@ -51,19 +51,19 @@ "vertex": "r3" }, { - "interface": "r5", - "metric": "TE", - "next-hop": "20", - "parent": "r4-eth1", - "type": "IP", + "interface": "r4-eth1", + "metric": "20", + "next-hop": "r5", + "parent": "r3(4)", + "type": "IP TE", "vertex": "10.0.20.0/24" }, { - "interface": "r5", - "metric": "TE", - "next-hop": "20", - "parent": "r4-eth1", - "type": "IP", + "interface": "r4-eth1", + "metric": "20", + "next-hop": "r5", + "parent": "r3(4)", + "type": "IP TE", "vertex": "10.254.0.3/32" } ], @@ -72,9 +72,9 @@ "vertex": "r4" }, { - "metric": "internal", - "parent": "0", - "type": "IP6", + "metric": "0", + "parent": "r4(4)", + "type": "IP6 internal", "vertex": "2001:db8:2:2::/64" }, { @@ -86,17 +86,19 @@ "vertex": "r5" }, { - "interface": "r5", - "next-hop": "10", - "parent": "r4-eth1", - "type": "IP6", + "metric": "10", + "interface": "r4-eth1", + "next-hop": "r5", + "parent": "r5(4)", + "type": "IP6 internal", "vertex": "2001:db8:2:1::/64" }, { - "interface": "r5", - "next-hop": "10", - "parent": "r4-eth1", - "type": "IP6", + "metric": "10", + "interface": "r4-eth1", + "next-hop": "r5", + "parent": "r5(4)", + "type": "IP6 internal", "vertex": "2001:db8:f::5/128" }, { @@ -107,17 +109,19 @@ "vertex": "r3" }, { - "interface": "r5", - "next-hop": "20", - "parent": "r4-eth1", - "type": "IP6", + "metric": "20", + "interface": "r4-eth1", + "next-hop": "r5", + "parent": "r3(4)", + "type": "IP6 internal", "vertex": "2001:db8:1:1::/64" }, { - "interface": "r5", - "next-hop": "20", - "parent": "r4-eth1", - "type": "IP6", + "metric": "20", + "interface": "r4-eth1", + "next-hop": "r5", + "parent": "r3(4)", + "type": "IP6 internal", "vertex": "2001:db8:f::3/128" } ] @@ -128,9 +132,9 @@ "vertex": "r4" }, { - "metric": "internal", - "parent": "0", - "type": "IP", + "metric": "0", + "parent": "r4(4)", + "type": "IP internal", "vertex": "10.0.21.0/24" }, { @@ -142,19 +146,19 @@ "vertex": "r2" }, { - "interface": "r2", - "metric": "TE", - "next-hop": "10", - "parent": "r4-eth0", - "type": "IP", + "interface": "r4-eth0", + "metric": "10", + "next-hop": "r2", + "parent": "r2(4)", + "type": "IP TE", "vertex": "10.0.21.0/24" }, { - "interface": "r2", - "metric": "TE", - "next-hop": "10", - "parent": "r4-eth0", - "type": "IP", + "interface": "r4-eth0", + "metric": "10", + "next-hop": "r2", + "parent": "r2(4)", + "type": "IP TE", "vertex": "10.254.0.2/32" } ], @@ -163,9 +167,9 @@ "vertex": "r4" }, { - "metric": "internal", - "parent": "0", - "type": "IP6", + "metric": "0", + "parent": "r4(4)", + "type": "IP6 internal", "vertex": "2001:db8:1:2::/64" }, { @@ -177,10 +181,11 @@ "vertex": "r2" }, { - "interface": "r2", - "next-hop": "10", - "parent": "r4-eth0", - "type": "IP6", + "metric": "10", + "interface": "r4-eth0", + "next-hop": "r2", + "parent": "r2(4)", + "type": "IP6 internal", "vertex": "2001:db8:f::2/128" } ] diff --git a/tests/topotests/isis-topo1/r5/r5_topology.json b/tests/topotests/isis-topo1/r5/r5_topology.json index 0224661411..3d887b7cea 100644 --- a/tests/topotests/isis-topo1/r5/r5_topology.json +++ b/tests/topotests/isis-topo1/r5/r5_topology.json @@ -6,15 +6,15 @@ "vertex": "r5" }, { - "metric": "internal", - "parent": "0", - "type": "IP", + "metric": "0", + "parent": "r5(4)", + "type": "IP internal", "vertex": "10.0.10.0/24" }, { - "metric": "internal", - "parent": "0", - "type": "IP", + "metric": "0", + "parent": "r5(4)", + "type": "IP internal", "vertex": "10.0.11.0/24" }, { @@ -34,51 +34,51 @@ "vertex": "r4" }, { - "interface": "r3", - "metric": "TE", - "next-hop": "10", - "parent": "r5-eth0", - "type": "IP", + "interface": "r5-eth0", + "metric": "10", + "next-hop": "r3", + "parent": "r3(4)", + "type": "IP TE", "vertex": "10.0.10.0/24" }, { - "interface": "r3", - "metric": "TE", - "next-hop": "10", - "parent": "r5-eth0", - "type": "IP", + "interface": "r5-eth0", + "metric": "10", + "next-hop": "r3", + "parent": "r3(4)", + "type": "IP TE", "vertex": "10.0.20.0/24" }, { - "interface": "r3", - "metric": "TE", - "next-hop": "10", - "parent": "r5-eth0", - "type": "IP", + "interface": "r5-eth0", + "metric": "10", + "next-hop": "r3", + "parent": "r3(4)", + "type": "IP TE", "vertex": "10.254.0.3/32" }, { - "interface": "r4", - "metric": "TE", - "next-hop": "10", - "parent": "r5-eth1", - "type": "IP", + "interface": "r5-eth1", + "metric": "10", + "next-hop": "r4", + "parent": "r4(4)", + "type": "IP TE", "vertex": "10.0.11.0/24" }, { - "interface": "r4", - "metric": "TE", - "next-hop": "10", - "parent": "r5-eth1", - "type": "IP", + "interface": "r5-eth1", + "metric": "10", + "next-hop": "r4", + "parent": "r4(4)", + "type": "IP TE", "vertex": "10.0.21.0/24" }, { - "interface": "r4", - "metric": "TE", - "next-hop": "10", - "parent": "r5-eth1", - "type": "IP", + "interface": "r5-eth1", + "metric": "10", + "next-hop": "r4", + "parent": "r4(4)", + "type": "IP TE", "vertex": "10.254.0.4/32" } ], @@ -87,15 +87,15 @@ "vertex": "r5" }, { - "metric": "internal", - "parent": "0", - "type": "IP6", + "metric": "0", + "parent": "r5(4)", + "type": "IP6 internal", "vertex": "2001:db8:2:1::/64" }, { - "metric": "internal", - "parent": "0", - "type": "IP6", + "metric": "0", + "parent": "r5(4)", + "type": "IP6 internal", "vertex": "2001:db8:2:2::/64" }, { @@ -115,31 +115,35 @@ "vertex": "r4" }, { - "interface": "r3", - "next-hop": "10", - "parent": "r5-eth0", - "type": "IP6", + "metric": "10", + "interface": "r5-eth0", + "next-hop": "r3", + "parent": "r3(4)", + "type": "IP6 internal", "vertex": "2001:db8:1:1::/64" }, { - "interface": "r3", - "next-hop": "10", - "parent": "r5-eth0", - "type": "IP6", + "metric": "10", + "interface": "r5-eth0", + "next-hop": "r3", + "parent": "r3(4)", + "type": "IP6 internal", "vertex": "2001:db8:f::3/128" }, { - "interface": "r4", - "next-hop": "10", - "parent": "r5-eth1", - "type": "IP6", + "metric": "10", + "interface": "r5-eth1", + "next-hop": "r4", + "parent": "r4(4)", + "type": "IP6 internal", "vertex": "2001:db8:1:2::/64" }, { - "interface": "r4", - "next-hop": "10", - "parent": "r5-eth1", - "type": "IP6", + "metric": "10", + "interface": "r5-eth1", + "next-hop": "r4", + "parent": "r4(4)", + "type": "IP6 internal", "vertex": "2001:db8:f::4/128" } ] diff --git a/tests/topotests/isis-topo1/test_isis_topo1.py b/tests/topotests/isis-topo1/test_isis_topo1.py index e7618c7bce..083a8b1e8d 100644 --- a/tests/topotests/isis-topo1/test_isis_topo1.py +++ b/tests/topotests/isis-topo1/test_isis_topo1.py @@ -47,6 +47,20 @@ from mininet.topo import Topo 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", +] + class ISISTopo1(Topo): "Simple two layer ISIS topology" @@ -263,6 +277,7 @@ def parse_topology(lines, level): areas = {} area = None ipv = None + vertex_type_regex = "|".join(VERTEX_TYPE_LIST) for line in lines: area_match = re.match(r"Area (.+):", line) @@ -282,44 +297,57 @@ def parse_topology(lines, level): ipv = "ipv4" continue - item_match = re.match(r"([^ ]+) ([^ ]+) ([^ ]+) ([^ ]+) ([^ ]+) ([^ ]+)", line) - if item_match is not None: + 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 - if ( - 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" - ): - continue + 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(4), - "interface": item_match.group(5), - "parent": item_match.group(6), + "next-hop": item_match.group(5), + "interface": item_match.group(6), + "parent": item_match.group(7), } ) continue - item_match = re.match(r"([^ ]+) ([^ ]+) ([^ ]+) ([^ ]+)", line) + 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(4), + "parent": item_match.group(5), } ) continue - item_match = re.match(r"([^ ]+)", line) + item_match = re.match(r"([^\s]+)", line) if item_match is not None: areas[area][level][ipv].append({"vertex": item_match.group(1)}) continue diff --git a/tests/topotests/multicast-pim-static-rp-topo1/test_multicast_pim_static_rp.py b/tests/topotests/multicast-pim-static-rp-topo1/test_multicast_pim_static_rp.py index e90230eb3b..7bef57b629 100755 --- a/tests/topotests/multicast-pim-static-rp-topo1/test_multicast_pim_static_rp.py +++ b/tests/topotests/multicast-pim-static-rp-topo1/test_multicast_pim_static_rp.py @@ -101,9 +101,9 @@ import os import sys import json import time -import pytest from time import sleep import datetime +import pytest # Save the Current Working Directory to find configuration files. CWD = os.path.dirname(os.path.realpath(__file__)) @@ -114,9 +114,12 @@ sys.path.append(os.path.join(CWD, "../lib/")) # pylint: disable=C0413 # Import topogen and topotest helpers -from lib.topogen import Topogen, get_topogen from mininet.topo import Topo +from lib.topogen import Topogen, get_topogen +from lib.topolog import logger +from lib.topojson import build_topo_from_json, build_config_from_json + from lib.common_config import ( start_topology, write_test_header, @@ -149,16 +152,14 @@ from lib.pim import ( clear_ip_mroute, clear_ip_mroute_verify, ) -from lib.topolog import logger -from lib.topojson import build_topo_from_json, build_config_from_json # Reading the data from JSON File for topology and configuration creation jsonFile = "{}/multicast_pim_static_rp.json".format(CWD) try: with open(jsonFile, "r") as topoJson: - topo = json.load(topoJson) + TOPO = json.load(topoJson) except IOError: - logger.info("Could not read file:", jsonFile) + logger.info("Could not read file: %s", jsonFile) # Global variables GROUP_RANGE_ALL = "224.0.0.0/4" @@ -203,7 +204,11 @@ class CreateTopo(Topo): tgen = get_topogen(self) # Building topology from json file - build_topo_from_json(tgen, topo) + build_topo_from_json(tgen, TOPO) + + def dumdum(self): + """ Dummy """ + print("%s", self.name) def setup_module(mod): @@ -212,10 +217,9 @@ def setup_module(mod): * `mod`: module name """ - global topo testsuite_run_time = time.asctime(time.localtime(time.time())) - logger.info("Testsuite start time: {}".format(testsuite_run_time)) + logger.info("Testsuite start time: %s", testsuite_run_time) logger.info("=" * 40) topology = """ @@ -229,7 +233,7 @@ def setup_module(mod): r4 """ - logger.info("Master Topology: \n {}".format(topology)) + logger.info("Master Topology: \n %s", topology) logger.info("Running setup_module to create topology") @@ -239,7 +243,7 @@ def setup_module(mod): # ... and here it calls Mininet initialization functions. # get list of daemons needs to be started for this suite. - daemons = topo_daemons(tgen, topo) + daemons = topo_daemons(tgen, TOPO) # Starting topology, create tmp files which are loaded to routers # to start deamons and then start routers @@ -250,10 +254,10 @@ def setup_module(mod): pytest.skip(tgen.errors) # Creating configuration from JSON - build_config_from_json(tgen, topo) + build_config_from_json(tgen, TOPO) # Verify PIM neighbors - result = verify_pim_neighbors(tgen, topo) + result = verify_pim_neighbors(tgen, TOPO) assert result is True, "setup_module :Failed \n Error:" " {}".format(result) logger.info("Running setup_module() done") @@ -269,9 +273,7 @@ def teardown_module(): # Stop toplogy and Remove tmp files tgen.stop_topology() - logger.info( - "Testsuite end time: {}".format(time.asctime(time.localtime(time.time()))) - ) + logger.info("Testsuite end time: %s", time.asctime(time.localtime(time.time()))) logger.info("=" * 40) @@ -335,9 +337,9 @@ def verify_mroute_repopulated(uptime_before, uptime_after): ) return errormsg - d1 = datetime.datetime.strptime(uptime_before[group][source], "%H:%M:%S") - d2 = datetime.datetime.strptime(uptime_after[group][source], "%H:%M:%S") - if d2 >= d1: + d_1 = datetime.datetime.strptime(uptime_before[group][source], "%H:%M:%S") + d_2 = datetime.datetime.strptime(uptime_after[group][source], "%H:%M:%S") + if d_2 >= d_1: errormsg = "mroute (%s, %s) is not " "repopulated [FAILED!!]" % ( source, group, @@ -360,7 +362,7 @@ def verify_state_incremented(state_before, state_after): """ for router, state_data in state_before.items(): - for state, value in state_data.items(): + for state, _ in state_data.items(): if state_before[router][state] >= state_after[router][state]: errormsg = ( "[DUT: %s]: state %s value has not" @@ -422,7 +424,7 @@ def test_add_delete_static_RP_p0(request): step("r1: Verify show ip igmp group without any IGMP join") dut = "r1" interface = "r1-r0-eth0" - result = verify_igmp_groups(tgen, dut, interface, GROUP_ADDRESS, expected=False) + result = verify_igmp_groups(tgen, dut, interface, GROUP_ADDRESS) assert result is not True, ( "Testcase {} : Failed \n " "r1: igmp group present without any IGMP join \n Error: {}".format( @@ -436,8 +438,9 @@ def test_add_delete_static_RP_p0(request): state_before = verify_pim_interface_traffic(tgen, state_dict) assert isinstance( state_before, dict - ), "Testcase{} : Failed \n state_before is not dictionary \n " - "Error: {}".format(tc_name, result) + ), "Testcase {} : Failed \n state_before is not dictionary\n Error: {}".format( + tc_name, result + ) step("r0 : Send IGMP join") result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS, join_interval=1) @@ -453,7 +456,7 @@ def test_add_delete_static_RP_p0(request): iif = "r1-r2-eth1" rp_address = "1.0.2.17" result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE + tgen, TOPO, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) @@ -487,12 +490,12 @@ def test_add_delete_static_RP_p0(request): } } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify RP info") result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE, expected=False + tgen, TOPO, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE ) assert ( result is not True @@ -501,16 +504,14 @@ def test_add_delete_static_RP_p0(request): ) step("r1: Verify upstream IIF interface") - result = verify_upstream_iif(tgen, dut, iif, STAR, GROUP_ADDRESS, expected=False) + result = verify_upstream_iif(tgen, dut, iif, STAR, GROUP_ADDRESS) assert result is not True, ( "Testcase {} : Failed \n " "r1: upstream IIF interface present \n Error: {}".format(tc_name, result) ) step("r1: Verify upstream join state and join timer") - result = verify_join_state_and_timer( - tgen, dut, iif, STAR, GROUP_ADDRESS, expected=False - ) + result = verify_join_state_and_timer(tgen, dut, iif, STAR, GROUP_ADDRESS) assert result is not True, ( "Testcase {} : Failed \n " "r1: upstream join state is up and join timer is running \n Error: {}".format( @@ -519,13 +520,13 @@ def test_add_delete_static_RP_p0(request): ) step("r1: Verify PIM state") - result = verify_pim_state(tgen, dut, iif, oif, GROUP_ADDRESS, expected=False) + result = verify_pim_state(tgen, dut, iif, oif, GROUP_ADDRESS) assert result is not True, "Testcase {} :Failed \n Error: {}".format( tc_name, result ) step("r1: Verify ip mroutes") - result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False) + result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif) assert ( result is not True ), "Testcase {} : Failed \n " "r1: mroutes are still present \n Error: {}".format( @@ -536,8 +537,9 @@ def test_add_delete_static_RP_p0(request): state_after = verify_pim_interface_traffic(tgen, state_dict) assert isinstance( state_after, dict - ), "Testcase{} : Failed \n state_before is not dictionary \n " - "Error: {}".format(tc_name, result) + ), "Testcase {} : Failed \n state_before is not dictionary \n Error: {}".format( + tc_name, result + ) result = verify_state_incremented(state_before, state_after) assert result is True, "Testcase{} : Failed Error: {}".format(tc_name, result) @@ -576,7 +578,7 @@ def test_SPT_RPT_path_same_p1(request): reset_config_on_routers(tgen) kill_iperf(tgen) clear_ip_mroute(tgen) - clear_ip_pim_interface_traffic(tgen, topo) + clear_ip_pim_interface_traffic(tgen, TOPO) step("pre-configuration to send IGMP join and multicast traffic") result = config_to_send_igmp_join_and_traffic(tgen, tc_name) @@ -604,7 +606,7 @@ def test_SPT_RPT_path_same_p1(request): rp_address = "1.0.2.17" iif = "lo" result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE + tgen, TOPO, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) @@ -684,9 +686,7 @@ def test_SPT_RPT_path_same_p1(request): assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r3: Verify (S, G) upstream join state and join timer") - result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS, expected=False - ) + result = verify_join_state_and_timer(tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS) assert result is not True, ( "Testcase {} : Failed \n " "r3: (S, G) upstream join state is up and join timer is running\n Error: {}".format( @@ -732,7 +732,7 @@ def test_not_reachable_static_RP_p0(request): reset_config_on_routers(tgen) kill_iperf(tgen) clear_ip_mroute(tgen) - clear_ip_pim_interface_traffic(tgen, topo) + clear_ip_pim_interface_traffic(tgen, TOPO) step("pre-configuration to send IGMP join and multicast traffic") result = config_to_send_igmp_join_and_traffic(tgen, tc_name) @@ -754,8 +754,9 @@ def test_not_reachable_static_RP_p0(request): state_before = verify_pim_interface_traffic(tgen, state_dict) assert isinstance( state_before, dict - ), "Testcase{} : Failed \n state_before is not dictionary \n " - "Error: {}".format(tc_name, result) + ), "Testcase{} : Failed \n state_before is not dictionary \n " "Error: {}".format( + tc_name, result + ) step("Enable IGMP on r1 interface and send IGMP " "join (225.1.1.1) to r1") step("Configure r2 loopback interface as RP") @@ -770,7 +771,7 @@ def test_not_reachable_static_RP_p0(request): iif = "r1-r2-eth1" rp_address = "1.0.2.17" result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE + tgen, TOPO, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) @@ -810,7 +811,7 @@ def test_not_reachable_static_RP_p0(request): step("r1: Check RP detail using show ip pim rp-info OIF should be unknown") result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_ALL, "Unknown", rp_address, SOURCE + tgen, TOPO, dut, GROUP_RANGE_ALL, "Unknown", rp_address, SOURCE ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) @@ -818,7 +819,7 @@ def test_not_reachable_static_RP_p0(request): "r1 : OIL should be same and IIF should be cleared on R1 verify" "using show ip pim state" ) - result = verify_pim_state(tgen, dut, iif, oif, GROUP_ADDRESS, expected=False) + result = verify_pim_state(tgen, dut, iif, oif, GROUP_ADDRESS) assert result is not True, ( "Testcase {} : Failed \n " "OIL is not same and IIF is not cleared on R1 \n Error: {}".format( @@ -827,7 +828,7 @@ def test_not_reachable_static_RP_p0(request): ) step("r1: upstream IIF should be unknown , verify using show ip pim" "upstream") - result = verify_upstream_iif(tgen, dut, iif, STAR, GROUP_ADDRESS, expected=False) + result = verify_upstream_iif(tgen, dut, iif, STAR, GROUP_ADDRESS) assert result is not True, ( "Testcase {} : Failed \n " "r1: upstream IIF is not unknown \n Error: {}".format(tc_name, result) @@ -837,9 +838,7 @@ def test_not_reachable_static_RP_p0(request): "r1: join state should not be joined and join timer should stop," "verify using show ip pim upstream" ) - result = verify_join_state_and_timer( - tgen, dut, iif, STAR, GROUP_ADDRESS, expected=False - ) + result = verify_join_state_and_timer(tgen, dut, iif, STAR, GROUP_ADDRESS) assert result is not True, ( "Testcase {} : Failed \n " "r1: join state is joined and timer is not stopped \n Error: {}".format( @@ -854,21 +853,22 @@ def test_not_reachable_static_RP_p0(request): state_after = verify_pim_interface_traffic(tgen, state_dict) assert isinstance( state_after, dict - ), "Testcase{} : Failed \n state_before is not dictionary \n " - "Error: {}".format(tc_name, result) + ), "Testcase{} : Failed \n state_before is not dictionary \n " "Error: {}".format( + tc_name, result + ) result = verify_state_incremented(state_before, state_after) assert result is True, "Testcase{} : Failed Error: {}".format(tc_name, result) step("r1: (*, G) cleared from mroute table using show ip mroute") - result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False) + result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif) assert result is not True, ( "Testcase {} : Failed \n " "r1: (*, G) are not cleared from mroute table \n Error: {}".format( tc_name, result ) ) - logger.info("Expected behavior: {}".format(result)) + logger.info("Expected behavior: %s", result) # Uncomment next line for debugging # tgen.mininet_cli() @@ -898,7 +898,7 @@ def test_add_RP_after_join_received_p1(request): reset_config_on_routers(tgen) kill_iperf(tgen) clear_ip_mroute(tgen) - clear_ip_pim_interface_traffic(tgen, topo) + clear_ip_pim_interface_traffic(tgen, TOPO) step("pre-configuration to send IGMP join and multicast traffic") result = config_to_send_igmp_join_and_traffic(tgen, tc_name) @@ -924,7 +924,7 @@ def test_add_RP_after_join_received_p1(request): } } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify rp-info") @@ -932,7 +932,7 @@ def test_add_RP_after_join_received_p1(request): rp_address = "1.0.2.17" iif = "r1-r2-eth1" result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE, expected=False + tgen, TOPO, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE ) assert ( result is not True @@ -945,8 +945,9 @@ def test_add_RP_after_join_received_p1(request): state_before = verify_pim_interface_traffic(tgen, state_dict) assert isinstance( state_before, dict - ), "Testcase{} : Failed \n state_before is not dictionary \n " - "Error: {}".format(tc_name, result) + ), "Testcase{} : Failed \n state_before is not dictionary \n " "Error: {}".format( + tc_name, result + ) step("r0 : Send IGMP join (225.1.1.1) to r1, when rp is not configured" "in r1") result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS, join_interval=1) @@ -958,7 +959,7 @@ def test_add_RP_after_join_received_p1(request): assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify upstream IIF interface") - result = verify_upstream_iif(tgen, dut, iif, STAR, GROUP_ADDRESS, expected=False) + result = verify_upstream_iif(tgen, dut, iif, STAR, GROUP_ADDRESS) assert result is not True, ( "Testcase {} : Failed \n " "r1: upstream IFF interface is present \n Error: {}".format(tc_name, result) @@ -966,9 +967,7 @@ def test_add_RP_after_join_received_p1(request): step("r1: Verify upstream join state and join timer") - result = verify_join_state_and_timer( - tgen, dut, iif, STAR, GROUP_ADDRESS, expected=False - ) + result = verify_join_state_and_timer(tgen, dut, iif, STAR, GROUP_ADDRESS) assert result is not True, ( "Testcase {} : Failed \n " "r1: upstream join state is joined and timer is running \n Error: {}".format( @@ -977,7 +976,7 @@ def test_add_RP_after_join_received_p1(request): ) step("r1: Verify PIM state") - result = verify_pim_state(tgen, dut, iif, oif, GROUP_ADDRESS, expected=False) + result = verify_pim_state(tgen, dut, iif, oif, GROUP_ADDRESS) assert ( result is not True ), "Testcase {} : Failed \n " "r1: PIM state is up\n Error: {}".format( @@ -985,7 +984,7 @@ def test_add_RP_after_join_received_p1(request): ) step("r1: Verify ip mroutes") - result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False) + result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif) assert ( result is not True ), "Testcase {} : Failed \n " "r1: mroutes are still present\n Error: {}".format( @@ -1006,12 +1005,12 @@ def test_add_RP_after_join_received_p1(request): } } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify rp-info") result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE + tgen, TOPO, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) @@ -1030,13 +1029,14 @@ def test_add_RP_after_join_received_p1(request): step("r1 : Verify ip mroutes") result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) - logger.info("Expected behavior: {}".format(result)) + logger.info("Expected behavior: %s", result) state_after = verify_pim_interface_traffic(tgen, state_dict) assert isinstance( state_after, dict - ), "Testcase{} : Failed \n state_before is not dictionary \n " - "Error: {}".format(tc_name, result) + ), "Testcase{} : Failed \n state_before is not dictionary \n " "Error: {}".format( + tc_name, result + ) result = verify_state_incremented(state_before, state_after) assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result) @@ -1068,7 +1068,7 @@ def test_reachable_static_RP_after_join_p0(request): reset_config_on_routers(tgen) kill_iperf(tgen) clear_ip_mroute(tgen) - clear_ip_pim_interface_traffic(tgen, topo) + clear_ip_pim_interface_traffic(tgen, TOPO) step("pre-configuration to send IGMP join and multicast traffic") result = config_to_send_igmp_join_and_traffic(tgen, tc_name) @@ -1083,8 +1083,9 @@ def test_reachable_static_RP_after_join_p0(request): state_before = verify_pim_interface_traffic(tgen, state_dict) assert isinstance( state_before, dict - ), "Testcase{} : Failed \n state_before is not dictionary \n " - "Error: {}".format(tc_name, result) + ), "Testcase{} : Failed \n state_before is not dictionary \n " "Error: {}".format( + tc_name, result + ) step("r1: Make RP un-reachable") dut = "r1" @@ -1098,7 +1099,7 @@ def test_reachable_static_RP_after_join_p0(request): step("r1: Verify rp-info") rp_address = "1.0.2.17" result = verify_pim_rp_info( - tgen, topo, dut, GROUP_ADDRESS, "Unknown", rp_address, SOURCE + tgen, TOPO, dut, GROUP_ADDRESS, "Unknown", rp_address, SOURCE ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) @@ -1113,16 +1114,14 @@ def test_reachable_static_RP_after_join_p0(request): step("r1 : Verify upstream IIF interface") iif = "r1-r2-eth1" - result = verify_upstream_iif(tgen, dut, iif, STAR, GROUP_ADDRESS, expected=False) + result = verify_upstream_iif(tgen, dut, iif, STAR, GROUP_ADDRESS) assert result is not True, ( "Testcase {} : Failed \n " "r1: upstream IIF interface is present\n Error: {}".format(tc_name, result) ) step("r1 : Verify upstream join state and join timer") - result = verify_join_state_and_timer( - tgen, dut, iif, STAR, GROUP_ADDRESS, expected=False - ) + result = verify_join_state_and_timer(tgen, dut, iif, STAR, GROUP_ADDRESS) assert result is not True, ( "Testcase {} : Failed \n " "r1: upstream join state is joined and timer is running\n Error: {}".format( @@ -1131,7 +1130,7 @@ def test_reachable_static_RP_after_join_p0(request): ) step("r1 : Verify PIM state") - result = verify_pim_state(tgen, dut, iif, oif, GROUP_ADDRESS, expected=False) + result = verify_pim_state(tgen, dut, iif, oif, GROUP_ADDRESS) assert ( result is not True ), "Testcase {} : Failed \n " "r1: PIM state is up\n Error: {}".format( @@ -1139,7 +1138,7 @@ def test_reachable_static_RP_after_join_p0(request): ) step("r1 : Verify ip mroutes") - result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False) + result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif) assert ( result is not True ), "Testcase {} : Failed \n " "r1: mroutes are still present\n Error: {}".format( @@ -1156,7 +1155,7 @@ def test_reachable_static_RP_after_join_p0(request): step("r1 : Verify rp-info") result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE + tgen, TOPO, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) @@ -1175,14 +1174,15 @@ def test_reachable_static_RP_after_join_p0(request): step("r1 : Verify ip mroutes") result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) - logger.info("Expected behavior: {}".format(result)) + logger.info("Expected behavior: %s", result) step("r1 : Verify pim interface traffic") state_after = verify_pim_interface_traffic(tgen, state_dict) assert isinstance( state_after, dict - ), "Testcase{} : Failed \n state_before is not dictionary \n " - "Error: {}".format(tc_name, result) + ), "Testcase{} : Failed \n state_before is not dictionary \n " "Error: {}".format( + tc_name, result + ) result = verify_state_incremented(state_before, state_after) assert result is True, "Testcase{} : Failed Error: {}".format(tc_name, result) @@ -1235,7 +1235,7 @@ def test_send_join_on_higher_preffered_rp_p1(request): reset_config_on_routers(tgen) kill_iperf(tgen) clear_ip_mroute(tgen) - clear_ip_pim_interface_traffic(tgen, topo) + clear_ip_pim_interface_traffic(tgen, TOPO) step("pre-configuration to send IGMP join and multicast traffic") result = config_to_send_igmp_join_and_traffic(tgen, tc_name) @@ -1272,8 +1272,9 @@ def test_send_join_on_higher_preffered_rp_p1(request): state_before = verify_pim_interface_traffic(tgen, state_dict) assert isinstance( state_before, dict - ), "Testcase{} : Failed \n state_before is not dictionary \n " - "Error: {}".format(tc_name, result) + ), "Testcase{} : Failed \n state_before is not dictionary \n " "Error: {}".format( + tc_name, result + ) step("r0 : Send IGMP join for 225.1.1.1") result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS, join_interval=1) @@ -1299,21 +1300,21 @@ def test_send_join_on_higher_preffered_rp_p1(request): } } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1 : Verify RP info for group 224.0.0.0/4") rp_address_1 = "1.0.2.17" iif = "r1-r2-eth1" result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_ALL, iif, rp_address_1, SOURCE + tgen, TOPO, dut, GROUP_RANGE_ALL, iif, rp_address_1, SOURCE ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1 : Verify RP info for group 225.1.1.1") rp_address_2 = "1.0.4.17" iif = "r1-r4-eth3" - result = verify_pim_rp_info(tgen, topo, dut, GROUP_RANGE, iif, rp_address_2, SOURCE) + result = verify_pim_rp_info(tgen, TOPO, dut, GROUP_RANGE, iif, rp_address_2, SOURCE) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1 : Verify join is sent to higher preferred RP") @@ -1321,8 +1322,9 @@ def test_send_join_on_higher_preffered_rp_p1(request): state_after = verify_pim_interface_traffic(tgen, state_dict) assert isinstance( state_after, dict - ), "Testcase{} : Failed \n state_before is not dictionary \n " - "Error: {}".format(tc_name, result) + ), "Testcase{} : Failed \n state_before is not dictionary \n " "Error: {}".format( + tc_name, result + ) result = verify_state_incremented(state_before, state_after) assert result is True, "Testcase{} : Failed Error: {}".format(tc_name, result) @@ -1344,7 +1346,7 @@ def test_send_join_on_higher_preffered_rp_p1(request): result = verify_join_state_and_timer(tgen, dut, iif, STAR, GROUP_ADDRESS) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) - clear_ip_pim_interface_traffic(tgen, topo) + clear_ip_pim_interface_traffic(tgen, TOPO) step("r1 : Verify joinTx, pruneTx count before RP gets deleted") state_dict = {"r1": {"r1-r2-eth1": ["joinTx"], "r1-r4-eth3": ["pruneTx"]}} @@ -1352,8 +1354,9 @@ def test_send_join_on_higher_preffered_rp_p1(request): state_before = verify_pim_interface_traffic(tgen, state_dict) assert isinstance( state_before, dict - ), "Testcase{} : Failed \n state_before is not dictionary \n " - "Error: {}".format(tc_name, result) + ), "Testcase{} : Failed \n state_before is not dictionary \n " "Error: {}".format( + tc_name, result + ) step("r1 : Delete RP configuration for 225.1.1.1") input_dict = { @@ -1370,21 +1373,19 @@ def test_send_join_on_higher_preffered_rp_p1(request): } } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1 : Verify rp-info for group 224.0.0.0/4") iif = "r1-r2-eth1" result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_ALL, iif, rp_address_1, SOURCE + tgen, TOPO, dut, GROUP_RANGE_ALL, iif, rp_address_1, SOURCE ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1 : Verify rp-info for group 225.1.1.1") iif = "r1-r4-eth3" - result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE, oif, rp_address_2, SOURCE, expected=False - ) + result = verify_pim_rp_info(tgen, TOPO, dut, GROUP_RANGE, oif, rp_address_2, SOURCE) assert result is not True, ( "Testcase {} : Failed \n " "r1: rp-info is present for group 225.1.1.1 \n Error: {}".format( @@ -1399,7 +1400,7 @@ def test_send_join_on_higher_preffered_rp_p1(request): iif = "r1-r2-eth1" result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) - logger.info("Expected behavior: {}".format(result)) + logger.info("Expected behavior: %s", result) step( "r1 : Verify IIF and OIL in show ip pim state updated when higher" @@ -1433,8 +1434,9 @@ def test_send_join_on_higher_preffered_rp_p1(request): state_after = verify_pim_interface_traffic(tgen, state_dict) assert isinstance( state_after, dict - ), "Testcase{} : Failed \n state_before is not dictionary \n " - "Error: {}".format(tc_name, result) + ), "Testcase{} : Failed \n state_before is not dictionary \n " "Error: {}".format( + tc_name, result + ) result = verify_state_incremented(state_before, state_after) assert result is True, "Testcase{} : Failed Error: {}".format(tc_name, result) @@ -1472,7 +1474,7 @@ def test_RP_configured_as_LHR_1_p1(request): reset_config_on_routers(tgen) kill_iperf(tgen) clear_ip_mroute(tgen) - clear_ip_pim_interface_traffic(tgen, topo) + clear_ip_pim_interface_traffic(tgen, TOPO) step("pre-configuration to send IGMP join and multicast traffic") result = config_to_send_igmp_join_and_traffic(tgen, tc_name) @@ -1532,7 +1534,7 @@ def test_RP_configured_as_LHR_1_p1(request): }, } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Configure r1(LHR) as RP") @@ -1579,7 +1581,7 @@ def test_RP_configured_as_LHR_1_p1(request): }, } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) shutdown_bringup_interface(tgen, "r1", "lo", False) @@ -1592,7 +1594,7 @@ def test_RP_configured_as_LHR_1_p1(request): rp_address = "1.0.1.17" iif = "lo" result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE + tgen, TOPO, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) @@ -1641,9 +1643,7 @@ def test_RP_configured_as_LHR_1_p1(request): assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r3: Verify (S, G) upstream join state and join timer") - result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS, expected=False - ) + result = verify_join_state_and_timer(tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS) assert result is not True, ( "Testcase {} : Failed \n " "r3: (S, G) upstream join state is joined and join" @@ -1688,7 +1688,7 @@ def test_RP_configured_as_LHR_2_p1(request): reset_config_on_routers(tgen) kill_iperf(tgen) clear_ip_mroute(tgen) - clear_ip_pim_interface_traffic(tgen, topo) + clear_ip_pim_interface_traffic(tgen, TOPO) step("pre-configuration to send IGMP join and multicast traffic") result = config_to_send_igmp_join_and_traffic(tgen, tc_name) @@ -1748,7 +1748,7 @@ def test_RP_configured_as_LHR_2_p1(request): }, } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1, r2, r3, r4: Configure r1(LHR) as RP") @@ -1795,14 +1795,14 @@ def test_RP_configured_as_LHR_2_p1(request): }, } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify RP info") dut = "r1" rp_address = "1.0.1.17" iif = "lo" - result = verify_pim_rp_info(tgen, topo, dut, GROUP_ADDRESS, iif, rp_address, SOURCE) + result = verify_pim_rp_info(tgen, TOPO, dut, GROUP_ADDRESS, iif, rp_address, SOURCE) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r5: Send multicast traffic for group 225.1.1.1") @@ -1850,9 +1850,7 @@ def test_RP_configured_as_LHR_2_p1(request): assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r3: Verify (S, G) upstream join state and join timer") - result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS, expected=False - ) + result = verify_join_state_and_timer(tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS) assert result is not True, ( "Testcase {} : Failed \n " "r3: (S,G) upstream state is joined and join timer is running\n Error: {}".format( @@ -1898,7 +1896,7 @@ def test_RP_configured_as_FHR_1_p1(request): reset_config_on_routers(tgen) kill_iperf(tgen) clear_ip_mroute(tgen) - clear_ip_pim_interface_traffic(tgen, topo) + clear_ip_pim_interface_traffic(tgen, TOPO) step("pre-configuration to send IGMP join and multicast traffic") result = config_to_send_igmp_join_and_traffic(tgen, tc_name) @@ -1957,7 +1955,7 @@ def test_RP_configured_as_FHR_1_p1(request): } }, } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1, r2, r3, r4: Configure r3(FHR) as RP") @@ -2004,7 +2002,7 @@ def test_RP_configured_as_FHR_1_p1(request): }, } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify RP info") @@ -2012,7 +2010,7 @@ def test_RP_configured_as_FHR_1_p1(request): rp_address = "1.0.3.17" iif = "r1-r3-eth2" result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE + tgen, TOPO, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) @@ -2060,9 +2058,7 @@ def test_RP_configured_as_FHR_1_p1(request): assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r3: Verify (S, G) upstream join state and join timer") - result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS, expected=False - ) + result = verify_join_state_and_timer(tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS) assert result is not True, ( "Testcase {} : Failed \n " "r3: (S,G) upstream state is joined and join timer is running\n Error: {}".format( @@ -2107,7 +2103,7 @@ def test_RP_configured_as_FHR_2_p2(request): reset_config_on_routers(tgen) kill_iperf(tgen) clear_ip_mroute(tgen) - clear_ip_pim_interface_traffic(tgen, topo) + clear_ip_pim_interface_traffic(tgen, TOPO) step("pre-configuration to send IGMP join and multicast traffic") result = config_to_send_igmp_join_and_traffic(tgen, tc_name) @@ -2167,7 +2163,7 @@ def test_RP_configured_as_FHR_2_p2(request): }, } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1, r2, r3, r4: Configure r3(FHR) as RP") @@ -2214,7 +2210,7 @@ def test_RP_configured_as_FHR_2_p2(request): }, } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify RP info") @@ -2222,7 +2218,7 @@ def test_RP_configured_as_FHR_2_p2(request): rp_address = "1.0.3.17" iif = "r1-r3-eth2" result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE + tgen, TOPO, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) @@ -2271,9 +2267,7 @@ def test_RP_configured_as_FHR_2_p2(request): assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r3: Verify (S, G) upstream join state and join timer") - result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS, expected=False - ) + result = verify_join_state_and_timer(tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS) assert result is not True, ( "Testcase {} : Failed \n " "r3: (S,G) upstream state is joined and join timer is running\n Error: {}".format( @@ -2320,7 +2314,7 @@ def test_SPT_RPT_path_different_p1(request): reset_config_on_routers(tgen) kill_iperf(tgen) clear_ip_mroute(tgen) - clear_ip_pim_interface_traffic(tgen, topo) + clear_ip_pim_interface_traffic(tgen, TOPO) step("pre-configuration to send IGMP join and multicast traffic") result = config_to_send_igmp_join_and_traffic(tgen, tc_name) @@ -2335,7 +2329,7 @@ def test_SPT_RPT_path_different_p1(request): dut = "r2" rp_address = "1.0.2.17" iif = "lo" - result = verify_pim_rp_info(tgen, topo, dut, GROUP_ADDRESS, iif, rp_address, SOURCE) + result = verify_pim_rp_info(tgen, TOPO, dut, GROUP_ADDRESS, iif, rp_address, SOURCE) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r0: Send IGMP join") @@ -2400,9 +2394,7 @@ def test_SPT_RPT_path_different_p1(request): assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r3: Verify (S, G) upstream join state and join timer") - result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS, expected=False - ) + result = verify_join_state_and_timer(tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS) assert result is not True, ( "Testcase {} : Failed \n " "r3: (S,G) upstream state is joined and join timer is running\n Error: {}".format( @@ -2424,9 +2416,7 @@ def test_SPT_RPT_path_different_p1(request): assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r2: Verify (S, G) upstream join state and join timer") - result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS, expected=False - ) + result = verify_join_state_and_timer(tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS) assert result is not True, ( "Testcase {} : Failed \n " "r2: (S,G) upstream state is joined and join timer is running\n Error: {}".format( @@ -2475,7 +2465,7 @@ def test_clear_pim_configuration_p1(request): reset_config_on_routers(tgen) kill_iperf(tgen) clear_ip_mroute(tgen) - clear_ip_pim_interface_traffic(tgen, topo) + clear_ip_pim_interface_traffic(tgen, TOPO) step("pre-configuration to send IGMP join and multicast traffic") result = config_to_send_igmp_join_and_traffic(tgen, tc_name) @@ -2492,7 +2482,7 @@ def test_clear_pim_configuration_p1(request): rp_address = "1.0.2.17" oif = "lo" result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_ALL, oif, rp_address, SOURCE + tgen, TOPO, dut, GROUP_RANGE_ALL, oif, rp_address, SOURCE ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) @@ -2572,7 +2562,7 @@ def test_restart_pimd_process_p2(request): reset_config_on_routers(tgen) kill_iperf(tgen) clear_ip_mroute(tgen) - clear_ip_pim_interface_traffic(tgen, topo) + clear_ip_pim_interface_traffic(tgen, TOPO) step("pre-configuration to send IGMP join and multicast traffic") result = config_to_send_igmp_join_and_traffic(tgen, tc_name) @@ -2589,7 +2579,7 @@ def test_restart_pimd_process_p2(request): rp_address = "1.0.2.17" oif = "lo" result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_ALL, oif, rp_address, SOURCE + tgen, TOPO, dut, GROUP_RANGE_ALL, oif, rp_address, SOURCE ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) @@ -2655,9 +2645,7 @@ def test_restart_pimd_process_p2(request): assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r3: Verify (S, G) upstream join state and join timer") - result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS, expected=False - ) + result = verify_join_state_and_timer(tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS) assert result is not True, ( "Testcase {} : Failed \n " "r3: (S,G) upstream state is joined and join timer is running\n Error: {}".format( @@ -2676,7 +2664,7 @@ def test_restart_pimd_process_p2(request): logger.info("waiting for 10 sec to make sure old mroute time is higher") sleep(10) uptime_before = verify_ip_mroutes( - tgen, dut, STAR, GROUP_ADDRESS, iif, oil, return_uptime=True, wait=60 + tgen, dut, STAR, GROUP_ADDRESS, iif, oil, return_uptime=True, mwait=60 ) assert isinstance(uptime_before, dict), "Testcase{} : Failed Error: {}".format( tc_name, result @@ -2692,7 +2680,7 @@ def test_restart_pimd_process_p2(request): sleep(5) uptime_after = verify_ip_mroutes( - tgen, dut, STAR, GROUP_ADDRESS, iif, oil, return_uptime=True, wait=10 + tgen, dut, STAR, GROUP_ADDRESS, iif, oil, return_uptime=True, mwait=10 ) assert isinstance(uptime_after, dict), "Testcase{} : Failed Error: {}".format( tc_name, result @@ -2731,7 +2719,7 @@ def test_multiple_groups_same_RP_address_p2(request): reset_config_on_routers(tgen) kill_iperf(tgen) clear_ip_mroute(tgen) - clear_ip_pim_interface_traffic(tgen, topo) + clear_ip_pim_interface_traffic(tgen, TOPO) step("pre-configuration to send IGMP join and multicast traffic") result = config_to_send_igmp_join_and_traffic(tgen, tc_name) @@ -2754,79 +2742,79 @@ def test_multiple_groups_same_RP_address_p2(request): rp_address = "1.0.2.17" oif = "lo" result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_ALL, oif, rp_address, SOURCE + tgen, TOPO, dut, GROUP_RANGE_ALL, oif, rp_address, SOURCE ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) - GROUP_ADDRESS_LIST = GROUP_ADDRESS_LIST_1 + GROUP_ADDRESS_LIST_2 + group_address_list = GROUP_ADDRESS_LIST_1 + GROUP_ADDRESS_LIST_2 step("r0: Send IGMP join for 10 groups") - result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS_LIST, join_interval=1) + result = iperfSendIGMPJoin(tgen, "r0", group_address_list, join_interval=1) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify IGMP groups") dut = "r1" oif = "r1-r0-eth0" - result = verify_igmp_groups(tgen, dut, oif, GROUP_ADDRESS_LIST) + result = verify_igmp_groups(tgen, dut, oif, group_address_list) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r5: Send multicast traffic for group 225.1.1.1") - result = iperfSendTraffic(tgen, "r5", GROUP_ADDRESS_LIST, 32, 2500) + result = iperfSendTraffic(tgen, "r5", group_address_list, 32, 2500) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify (*, G) upstream IIF interface") dut = "r1" iif = "r1-r2-eth1" - result = verify_upstream_iif(tgen, dut, iif, STAR, GROUP_ADDRESS_LIST) + result = verify_upstream_iif(tgen, dut, iif, STAR, group_address_list) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify (*, G) upstream join state and join timer") - result = verify_join_state_and_timer(tgen, dut, iif, STAR, GROUP_ADDRESS_LIST) + result = verify_join_state_and_timer(tgen, dut, iif, STAR, group_address_list) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify (*, G) ip mroutes") oif = "r1-r0-eth0" - result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST, iif, oif) + result = verify_ip_mroutes(tgen, dut, STAR, group_address_list, iif, oif) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify (S, G) upstream IIF interface") iif = "r1-r3-eth2" - result = verify_upstream_iif(tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST) + result = verify_upstream_iif(tgen, dut, iif, SOURCE_ADDRESS, group_address_list) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify (S, G) upstream join state and join timer") result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST + tgen, dut, iif, SOURCE_ADDRESS, group_address_list ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify (S, G) ip mroutes") - result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST, iif, oif) + result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r2: Verify (*, G) upstream IIF interface") dut = "r2" iif = "lo" - result = verify_upstream_iif(tgen, dut, iif, STAR, GROUP_ADDRESS_LIST) + result = verify_upstream_iif(tgen, dut, iif, STAR, group_address_list) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r2: Verify (*, G) upstream join state and join timer") - result = verify_join_state_and_timer(tgen, dut, iif, STAR, GROUP_ADDRESS_LIST) + result = verify_join_state_and_timer(tgen, dut, iif, STAR, group_address_list) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r2: Verify (*, G) ip mroutes") oif = "r2-r1-eth0" - result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST, iif, oif) + result = verify_ip_mroutes(tgen, dut, STAR, group_address_list, iif, oif) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r3: Verify (S, G) upstream IIF interface") dut = "r3" iif = "r3-r5-eth3" - result = verify_upstream_iif(tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST) + result = verify_upstream_iif(tgen, dut, iif, SOURCE_ADDRESS, group_address_list) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r3: Verify (S, G) upstream join state and join timer") result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST, expected=False + tgen, dut, iif, SOURCE_ADDRESS, group_address_list ) assert result is not True, ( "Testcase {} : Failed \n " @@ -2837,20 +2825,20 @@ def test_multiple_groups_same_RP_address_p2(request): step("r3: Verify (S, G) ip mroutes") oif = "r3-r1-eth0" - result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST, iif, oif) + result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r2: Verify (S, G) upstream IIF interface") dut = "r2" iif = "r2-r3-eth1" result = verify_upstream_iif( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST, joinState="NotJoined" + tgen, dut, iif, SOURCE_ADDRESS, group_address_list, joinState="NotJoined" ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r2: Verify (S, G) upstream join state and join timer") result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST, expected=False + tgen, dut, iif, SOURCE_ADDRESS, group_address_list ) assert result is not True, ( "Testcase {} : Failed \n " @@ -2861,7 +2849,7 @@ def test_multiple_groups_same_RP_address_p2(request): step("r2: Verify (S, G) ip mroutes") oif = "none" - result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST, iif, oif) + result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Delete RP configuration") @@ -2879,7 +2867,7 @@ def test_multiple_groups_same_RP_address_p2(request): } } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Shut the interface r1-r2-eth1 from R1 to R2") @@ -2905,7 +2893,7 @@ def test_multiple_groups_same_RP_address_p2(request): } } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Shut the interface r1-r0-eth0 from R1 to R2") @@ -2919,59 +2907,59 @@ def test_multiple_groups_same_RP_address_p2(request): step("r1: Verify (*, G) upstream IIF interface") dut = "r1" iif = "r1-r2-eth1" - result = verify_upstream_iif(tgen, dut, iif, STAR, GROUP_ADDRESS_LIST) + result = verify_upstream_iif(tgen, dut, iif, STAR, group_address_list) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify (*, G) upstream join state and join timer") - result = verify_join_state_and_timer(tgen, dut, iif, STAR, GROUP_ADDRESS_LIST) + result = verify_join_state_and_timer(tgen, dut, iif, STAR, group_address_list) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify (*, G) ip mroutes") oif = "r1-r0-eth0" - result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST, iif, oif) + result = verify_ip_mroutes(tgen, dut, STAR, group_address_list, iif, oif) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify (S, G) upstream IIF interface") iif = "r1-r3-eth2" - result = verify_upstream_iif(tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST) + result = verify_upstream_iif(tgen, dut, iif, SOURCE_ADDRESS, group_address_list) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify (S, G) upstream join state and join timer") result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST + tgen, dut, iif, SOURCE_ADDRESS, group_address_list ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify (S, G) ip mroutes") - result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST, iif, oif) + result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r2: Verify (*, G) upstream IIF interface") dut = "r2" iif = "lo" - result = verify_upstream_iif(tgen, dut, iif, STAR, GROUP_ADDRESS_LIST) + result = verify_upstream_iif(tgen, dut, iif, STAR, group_address_list) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r2: Verify (*, G) upstream join state and join timer") - result = verify_join_state_and_timer(tgen, dut, iif, STAR, GROUP_ADDRESS_LIST) + result = verify_join_state_and_timer(tgen, dut, iif, STAR, group_address_list) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r2: Verify (*, G) ip mroutes") oif = "r2-r1-eth0" - result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST, iif, oif) + result = verify_ip_mroutes(tgen, dut, STAR, group_address_list, iif, oif) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r2: Verify (S, G) upstream IIF interface") dut = "r2" iif = "r2-r3-eth1" result = verify_upstream_iif( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST, joinState="NotJoined" + tgen, dut, iif, SOURCE_ADDRESS, group_address_list, joinState="NotJoined" ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r2: Verify (S, G) upstream join state and join timer") result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST, expected=False + tgen, dut, iif, SOURCE_ADDRESS, group_address_list ) assert result is not True, ( "Testcase {} : Failed \n " @@ -2982,18 +2970,18 @@ def test_multiple_groups_same_RP_address_p2(request): step("r2: Verify (S, G) ip mroutes") oif = "none" - result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST, iif, oif) + result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r3: Verify (S, G) upstream IIF interface") dut = "r3" iif = "r3-r5-eth3" - result = verify_upstream_iif(tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST) + result = verify_upstream_iif(tgen, dut, iif, SOURCE_ADDRESS, group_address_list) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r3: Verify (S, G) upstream join state and join timer") result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST, expected=False + tgen, dut, iif, SOURCE_ADDRESS, group_address_list ) assert result is not True, ( "Testcase {} : Failed \n " @@ -3004,7 +2992,7 @@ def test_multiple_groups_same_RP_address_p2(request): step("r3: Verify (S, G) ip mroutes") oif = "r3-r1-eth0" - result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST, iif, oif) + result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) write_test_footer(tc_name) @@ -3040,7 +3028,7 @@ def test_multiple_groups_different_RP_address_p2(request): reset_config_on_routers(tgen) kill_iperf(tgen) clear_ip_mroute(tgen) - clear_ip_pim_interface_traffic(tgen, topo) + clear_ip_pim_interface_traffic(tgen, TOPO) step("pre-configuration to send IGMP join and multicast traffic") result = config_to_send_igmp_join_and_traffic(tgen, tc_name) @@ -3060,7 +3048,7 @@ def test_multiple_groups_different_RP_address_p2(request): } } } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) input_dict = { @@ -3085,7 +3073,7 @@ def test_multiple_groups_different_RP_address_p2(request): } }, } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r2: Verify RP info") @@ -3093,7 +3081,7 @@ def test_multiple_groups_different_RP_address_p2(request): rp_address = "1.0.2.17" oif = "lo" result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_LIST_1, oif, rp_address, SOURCE + tgen, TOPO, dut, GROUP_RANGE_LIST_1, oif, rp_address, SOURCE ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) @@ -3101,23 +3089,23 @@ def test_multiple_groups_different_RP_address_p2(request): dut = "r4" rp_address = "1.0.4.17" result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_LIST_2, oif, rp_address, SOURCE + tgen, TOPO, dut, GROUP_RANGE_LIST_2, oif, rp_address, SOURCE ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) - GROUP_ADDRESS_LIST = GROUP_ADDRESS_LIST_1 + GROUP_ADDRESS_LIST_2 + group_address_list = GROUP_ADDRESS_LIST_1 + GROUP_ADDRESS_LIST_2 step("r0: Send IGMP join") - result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS_LIST, join_interval=1) + result = iperfSendIGMPJoin(tgen, "r0", group_address_list, join_interval=1) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify IGMP groups") dut = "r1" oif = "r1-r0-eth0" - result = verify_igmp_groups(tgen, dut, oif, GROUP_ADDRESS_LIST) + result = verify_igmp_groups(tgen, dut, oif, group_address_list) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r5: Send multicast traffic for group 225.1.1.1") - result = iperfSendTraffic(tgen, "r5", GROUP_ADDRESS_LIST, 32, 2500) + result = iperfSendTraffic(tgen, "r5", group_address_list, 32, 2500) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Verify (*, G) upstream IIF interface") @@ -3175,7 +3163,7 @@ def test_multiple_groups_different_RP_address_p2(request): step("r2: Verify (S, G) upstream join state and join timer") result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, expected=False + tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1 ) assert result is not True, ( "Testcase {} : Failed \n " @@ -3199,7 +3187,7 @@ def test_multiple_groups_different_RP_address_p2(request): step("r3: Verify (S, G) upstream join state and join timer") result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, expected=False + tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1 ) assert result is not True, ( "Testcase {} : Failed \n " @@ -3271,7 +3259,7 @@ def test_multiple_groups_different_RP_address_p2(request): step("r4: Verify (S, G) upstream join state and join timer") result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, expected=False + tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2 ) assert result is not True, ( "Testcase {} : Failed \n " @@ -3295,7 +3283,7 @@ def test_multiple_groups_different_RP_address_p2(request): step("r3: Verify (S, G) upstream join state and join timer") result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, expected=False + tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2 ) assert result is not True, "Testcase {} :Failed \n Error: {}".format( tc_name, result @@ -3333,7 +3321,7 @@ def test_multiple_groups_different_RP_address_p2(request): } }, } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1, r2, r3, r4: Re-configure RP") @@ -3359,7 +3347,7 @@ def test_multiple_groups_different_RP_address_p2(request): } }, } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Shut the interface r1-r2-eth1 from R1 to R2") @@ -3448,7 +3436,7 @@ def test_multiple_groups_different_RP_address_p2(request): step("r2: Verify (S, G) upstream join state and join timer") result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, expected=False + tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1 ) assert result is not True, ( "Testcase {} : Failed \n " @@ -3472,7 +3460,7 @@ def test_multiple_groups_different_RP_address_p2(request): step("r3: Verify (S, G) upstream join state and join timer") result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, expected=False + tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1 ) assert result is not True, ( "Testcase {} : Failed \n " @@ -3544,7 +3532,7 @@ def test_multiple_groups_different_RP_address_p2(request): step("r4: Verify (S, G) upstream join state and join timer") result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, expected=False + tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2 ) assert result is not True, ( "Testcase {} : Failed \n " @@ -3568,7 +3556,7 @@ def test_multiple_groups_different_RP_address_p2(request): step("r3: Verify (S, G) upstream join state and join timer") result = verify_join_state_and_timer( - tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, expected=False + tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2 ) assert result is not True, ( "Testcase {} : Failed \n " @@ -3611,7 +3599,7 @@ def test_shutdown_primary_path_p1(request): reset_config_on_routers(tgen) kill_iperf(tgen) clear_ip_mroute(tgen) - clear_ip_pim_interface_traffic(tgen, topo) + clear_ip_pim_interface_traffic(tgen, TOPO) step("pre-configuration to send IGMP join and multicast traffic") result = config_to_send_igmp_join_and_traffic(tgen, tc_name) @@ -3630,7 +3618,7 @@ def test_shutdown_primary_path_p1(request): rp_address = "1.0.2.17" iif = "r1-r2-eth1" result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE + tgen, TOPO, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) @@ -3701,7 +3689,7 @@ def test_shutdown_primary_path_p1(request): dut = "r1" iif = "r1-r3-eth2" oif = "r1-r0-eth0" - result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False) + result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif) assert result is not True, ( "Testcase {} : Failed \n " "r1: (*,G) mroutes are not cleared after shut of R1 to R3 link\n Error: {}".format( @@ -3713,7 +3701,7 @@ def test_shutdown_primary_path_p1(request): dut = "r2" iif = "lo" oif = "r2-r3-eth1" - result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False) + result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif) assert result is not True, ( "Testcase {} : Failed \n " "r2: (*,G) mroutes are not cleared after shut of R1 to R3 link\n Error: {}".format( @@ -3725,7 +3713,7 @@ def test_shutdown_primary_path_p1(request): dut = "r3" iif = "r3-r2-eth1" oif = "r3-r1-eth0" - result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False) + result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif) assert result is not True, ( "Testcase {} : Failed \n " "r3: (*,G) mroutes are not cleared after shut of R1 to R3 link\n Error: {}".format( @@ -3804,7 +3792,7 @@ def test_delete_RP_shut_noshut_upstream_interface_p1(request): reset_config_on_routers(tgen) kill_iperf(tgen) clear_ip_mroute(tgen) - clear_ip_pim_interface_traffic(tgen, topo) + clear_ip_pim_interface_traffic(tgen, TOPO) step("pre-configuration to send IGMP join and multicast traffic") result = config_to_send_igmp_join_and_traffic(tgen, tc_name) @@ -3821,7 +3809,7 @@ def test_delete_RP_shut_noshut_upstream_interface_p1(request): rp_address = "1.0.2.17" iif = "r1-r2-eth1" result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE + tgen, TOPO, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) @@ -3863,7 +3851,7 @@ def test_delete_RP_shut_noshut_upstream_interface_p1(request): } } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r1: Shut the interface r1-r2-eth1 from R1 to R2") @@ -3890,7 +3878,7 @@ def test_delete_RP_shut_noshut_upstream_interface_p1(request): dut = "r1" iif = "r1-r2-eth1" oif = "r1-r0-eth0" - result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False) + result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif) assert result is not True, ( "Testcase {} : Failed \n " "r1: (*,G) mroutes are not cleared after shut of R1 to R0 link\n Error: {}".format( @@ -3902,7 +3890,7 @@ def test_delete_RP_shut_noshut_upstream_interface_p1(request): dut = "r2" iif = "lo" oif = "r2-r1-eth0" - result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False) + result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif) assert result is not True, ( "Testcase {} : Failed \n " "r2: (*,G) mroutes are not cleared after shut of R1 to R0 link\n Error: {}".format( @@ -3937,7 +3925,7 @@ def test_delete_RP_shut_noshut_RP_interface_p1(request): reset_config_on_routers(tgen) kill_iperf(tgen) clear_ip_mroute(tgen) - clear_ip_pim_interface_traffic(tgen, topo) + clear_ip_pim_interface_traffic(tgen, TOPO) step("pre-configuration to send IGMP join and multicast traffic") result = config_to_send_igmp_join_and_traffic(tgen, tc_name) @@ -3954,7 +3942,7 @@ def test_delete_RP_shut_noshut_RP_interface_p1(request): rp_address = "1.0.2.17" iif = "r1-r2-eth1" result = verify_pim_rp_info( - tgen, topo, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE + tgen, TOPO, dut, GROUP_RANGE_ALL, iif, rp_address, SOURCE ) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) @@ -3995,7 +3983,7 @@ def test_delete_RP_shut_noshut_RP_interface_p1(request): } } - result = create_pim_config(tgen, topo, input_dict) + result = create_pim_config(tgen, TOPO, input_dict) assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) step("r2: Shut the RP interface lo") @@ -4017,7 +4005,7 @@ def test_delete_RP_shut_noshut_RP_interface_p1(request): dut = "r1" iif = "r1-r2-eth1" oif = "r1-r0-eth0" - result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False) + result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif) assert result is not True, ( "Testcase {} : Failed \n " "r1: (*,G) mroutes are not cleared after shut of R1 to R2 and R3 link\n Error: {}".format( @@ -4029,7 +4017,7 @@ def test_delete_RP_shut_noshut_RP_interface_p1(request): dut = "r2" iif = "lo" oif = "r2-r1-eth0" - result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False) + result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif) assert result is not True, ( "Testcase {} : Failed \n " "r2: (*,G) mroutes are not cleared after shut of R1 to R2 and R3 link\n Error: {}".format( @@ -4041,5 +4029,5 @@ def test_delete_RP_shut_noshut_RP_interface_p1(request): if __name__ == "__main__": - args = ["-s"] + sys.argv[1:] - sys.exit(pytest.main(args)) + ARGS = ["-s"] + sys.argv[1:] + sys.exit(pytest.main(ARGS)) diff --git a/tests/topotests/static_routing_with_ebgp/test_static_routes_topo1_ebgp.py b/tests/topotests/static_routing_with_ebgp/test_static_routes_topo1_ebgp.py index a4cc8e8e7a..a16c4ae297 100644 --- a/tests/topotests/static_routing_with_ebgp/test_static_routes_topo1_ebgp.py +++ b/tests/topotests/static_routing_with_ebgp/test_static_routes_topo1_ebgp.py @@ -40,10 +40,11 @@ import platform CWD = os.path.dirname(os.path.realpath(__file__)) sys.path.append(os.path.join(CWD, "../")) sys.path.append(os.path.join(CWD, "../lib/")) + # pylint: disable=C0413 # Import topogen and topotest helpers -from mininet.topo import Topo from lib.topogen import Topogen, get_topogen +from mininet.topo import Topo from lib.topotest import version_cmp # Import topoJson from lib, to create topology and initial configuration @@ -68,15 +69,14 @@ from lib.topojson import build_topo_from_json, build_config_from_json pytestmark = [pytest.mark.bgpd, pytest.mark.staticd] # Reading the data from JSON File for topology creation -jsonFile = "{}/static_routes_topo1_ebgp.json".format(CWD) +JSONFILE = "{}/static_routes_topo1_ebgp.json".format(CWD) try: - with open(jsonFile, "r") as topoJson: + with open(JSONFILE, "r") as topoJson: topo = json.load(topoJson) except IOError: - assert False, "Could not read file {}".format(jsonFile) + assert False, "Could not read file {}".format(JSONFILE) # Global variables -BGP_CONVERGENCE = False ADDR_TYPES = check_address_types() NETWORK = {"ipv4": ["11.0.20.1/32", "11.0.20.2/32"], "ipv6": ["2::1/128", "2::2/128"]} NETWORK2 = {"ipv4": "11.0.20.1/32", "ipv6": "2::1/128"} @@ -98,6 +98,10 @@ class CreateTopo(Topo): # Building topology from json file build_topo_from_json(tgen, topo) + def dumdum(self): + """ Dummy """ + print("%s", self.name) + def setup_module(mod): """ @@ -105,7 +109,7 @@ def setup_module(mod): * `mod`: module name """ - global topo + testsuite_run_time = time.asctime(time.localtime(time.time())) logger.info("Testsuite start time: {}".format(testsuite_run_time)) logger.info("=" * 40) @@ -131,16 +135,14 @@ def setup_module(mod): pytest.skip(error_msg) # Checking BGP convergence - global BGP_CONVERGENCE - global ADDR_TYPES + # Don't run this test if we have any failure. if tgen.routers_have_failure(): pytest.skip(tgen.errors) + # Api call verify whether BGP is converged - BGP_CONVERGENCE = verify_bgp_convergence(tgen, topo) - assert BGP_CONVERGENCE is True, "setup_module :Failed \n Error: {}".format( - BGP_CONVERGENCE - ) + converged = verify_bgp_convergence(tgen, topo) + assert converged is True, "setup_module :Failed \n Error: {}".format(converged) logger.info("Running setup_module() done") @@ -152,7 +154,7 @@ def teardown_module(mod): * `mod`: module name """ - logger.info("Running teardown_module to delete topology") + logger.info("Running teardown_module to delete topology: %s", mod) tgen = get_topogen() @@ -166,7 +168,11 @@ def teardown_module(mod): def populate_nh(): - NEXT_HOP_IP = { + """ + Populate nexthops. + """ + + next_hop_ip = { "nh1": { "ipv4": topo["routers"]["r1"]["links"]["r2-link0"]["ipv4"].split("/")[0], "ipv6": topo["routers"]["r1"]["links"]["r2-link0"]["ipv6"].split("/")[0], @@ -176,7 +182,7 @@ def populate_nh(): "ipv6": topo["routers"]["r1"]["links"]["r2-link1"]["ipv6"].split("/")[0], }, } - return NEXT_HOP_IP + return next_hop_ip ##################################################### @@ -199,7 +205,7 @@ def test_static_route_2nh_p0_tc_1_ebgp(request): pytest.skip(tgen.errors) reset_config_on_routers(tgen) - NEXT_HOP_IP = populate_nh() + next_hop_ip = populate_nh() step( "Configure IPv4 static route (10.1.1.1) in R2 with next hop N1" @@ -213,11 +219,11 @@ def test_static_route_2nh_p0_tc_1_ebgp(request): "static_routes": [ { "network": NETWORK[addr_type], - "next_hop": NEXT_HOP_IP["nh1"][addr_type], + "next_hop": next_hop_ip["nh1"][addr_type], }, { "network": NETWORK[addr_type], - "next_hop": NEXT_HOP_IP["nh2"][addr_type], + "next_hop": next_hop_ip["nh2"][addr_type], }, ] } @@ -233,7 +239,7 @@ def test_static_route_2nh_p0_tc_1_ebgp(request): "On R2, static route installed in RIB using show ip route" " with 2 ECMP next hop " ) - nh = [NEXT_HOP_IP["nh1"][addr_type], NEXT_HOP_IP["nh2"][addr_type]] + nh = [next_hop_ip["nh1"][addr_type], next_hop_ip["nh2"][addr_type]] dut = "r2" protocol = "static" result = verify_rib( @@ -268,7 +274,7 @@ def test_static_route_2nh_p0_tc_1_ebgp(request): "static_routes": [ { "network": NETWORK[addr_type], - "next_hop": NEXT_HOP_IP["nh1"][addr_type], + "next_hop": next_hop_ip["nh1"][addr_type], "delete": True, } ] @@ -284,7 +290,7 @@ def test_static_route_2nh_p0_tc_1_ebgp(request): "On R2, after removing the static route with N1 , " "route become active with nexthop N2 and vice versa." ) - nh = NEXT_HOP_IP["nh1"][addr_type] + nh = next_hop_ip["nh1"][addr_type] result = verify_rib( tgen, addr_type, @@ -300,12 +306,13 @@ def test_static_route_2nh_p0_tc_1_ebgp(request): tc_name ) - nh = [NEXT_HOP_IP["nh2"][addr_type]] + nh = [next_hop_ip["nh2"][addr_type]] result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) step("Configure the static route with nexthop N1") @@ -314,7 +321,7 @@ def test_static_route_2nh_p0_tc_1_ebgp(request): "static_routes": [ { "network": NETWORK[addr_type], - "next_hop": NEXT_HOP_IP["nh1"][addr_type], + "next_hop": next_hop_ip["nh1"][addr_type], } ] } @@ -333,7 +340,7 @@ def test_static_route_2nh_p0_tc_1_ebgp(request): "static_routes": [ { "network": NETWORK[addr_type], - "next_hop": NEXT_HOP_IP["nh2"][addr_type], + "next_hop": next_hop_ip["nh2"][addr_type], "delete": True, } ] @@ -350,7 +357,7 @@ def test_static_route_2nh_p0_tc_1_ebgp(request): "On R2, after removing the static route with N2 , " "route become active with nexthop N1 and vice versa." ) - nh = NEXT_HOP_IP["nh2"][addr_type] + nh = next_hop_ip["nh2"][addr_type] result = verify_rib( tgen, addr_type, @@ -360,15 +367,19 @@ def test_static_route_2nh_p0_tc_1_ebgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" " still present in RIB".format( + tc_name + ) - nh = [NEXT_HOP_IP["nh1"][addr_type]] + nh = [next_hop_ip["nh1"][addr_type]] result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) step("Configure the static route with nexthop N2") input_dict_4 = { @@ -376,7 +387,7 @@ def test_static_route_2nh_p0_tc_1_ebgp(request): "static_routes": [ { "network": NETWORK[addr_type], - "next_hop": NEXT_HOP_IP["nh2"][addr_type], + "next_hop": next_hop_ip["nh2"][addr_type], } ] } @@ -395,14 +406,15 @@ def test_static_route_2nh_p0_tc_1_ebgp(request): step("Only one the nexthops should be active in RIB.") - nh = NEXT_HOP_IP["nh2"][addr_type] + nh = next_hop_ip["nh2"][addr_type] result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) - nh = NEXT_HOP_IP["nh1"][addr_type] + nh = next_hop_ip["nh1"][addr_type] result = verify_rib( tgen, addr_type, @@ -412,15 +424,21 @@ def test_static_route_2nh_p0_tc_1_ebgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" " still present in RIB".format( + tc_name + ) dut = "r3" result = verify_bgp_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, expected=False ) - assert result is not True, "Testcase {} : Failed \nError: Route is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Route is" " still present in RIB".format( + tc_name + ) result = verify_rib( tgen, @@ -431,27 +449,35 @@ def test_static_route_2nh_p0_tc_1_ebgp(request): next_hop=nh, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Route is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Route is" " still present in RIB".format( + tc_name + ) dut = "r2" - nh = [NEXT_HOP_IP["nh2"][addr_type]] + nh = [next_hop_ip["nh2"][addr_type]] result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) dut = "r3" result = verify_bgp_rib(tgen, addr_type, dut, input_dict_4) - assert result is True, "Testcase {} : Failed \nError: Route is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Route is" " missing in RIB".format(tc_name) result = verify_rib( tgen, addr_type, dut, input_dict_4, protocol=protocol, expected=False ) - assert result is not True, "Testcase {} : Failed \nError: Route is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Route is" " still present in RIB".format( + tc_name + ) dut = "r2" step("No shut the nexthop interface N1") @@ -461,13 +487,14 @@ def test_static_route_2nh_p0_tc_1_ebgp(request): "after shut of nexthop N1 , route become active " "with nexthop N2 and vice versa." ) - nh = [NEXT_HOP_IP["nh1"][addr_type], NEXT_HOP_IP["nh2"][addr_type]] + nh = [next_hop_ip["nh1"][addr_type], next_hop_ip["nh2"][addr_type]] result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) step("Shut nexthop interface N2") intf = topo["routers"]["r2"]["links"]["r1-link1"]["interface"] @@ -478,7 +505,7 @@ def test_static_route_2nh_p0_tc_1_ebgp(request): " after shut of nexthop N1 , route become active with " "nexthop N2 and vice versa." ) - nh = NEXT_HOP_IP["nh2"][addr_type] + nh = next_hop_ip["nh2"][addr_type] result = verify_rib( tgen, @@ -489,28 +516,36 @@ def test_static_route_2nh_p0_tc_1_ebgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" " still present in RIB".format( + tc_name + ) - nh = [NEXT_HOP_IP["nh1"][addr_type]] + nh = [next_hop_ip["nh1"][addr_type]] dut = "r2" protocol = "static" result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) dut = "r3" result = verify_bgp_rib(tgen, addr_type, dut, input_dict_4) - assert result is True, "Testcase {} : Failed \nError: Route is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Route is" " missing in RIB".format(tc_name) result = verify_rib( tgen, addr_type, dut, input_dict_4, protocol=protocol, expected=False ) - assert result is not True, "Testcase {} : Failed \nError: Route is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Route is" " still present in RIB".format( + tc_name + ) step("No shut nexthop interface N2") dut = "r2" @@ -520,24 +555,29 @@ def test_static_route_2nh_p0_tc_1_ebgp(request): "after shut of nexthop N1 , route become active " "with nexthop N2 and vice versa." ) - nh = [NEXT_HOP_IP["nh1"][addr_type], NEXT_HOP_IP["nh2"][addr_type]] + nh = [next_hop_ip["nh1"][addr_type], next_hop_ip["nh2"][addr_type]] result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) dut = "r3" result = verify_bgp_rib(tgen, addr_type, dut, input_dict_4) - assert result is True, "Testcase {} : Failed \nError: Route is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Route is" " missing in RIB".format(tc_name) result = verify_rib( tgen, addr_type, dut, input_dict_4, protocol=protocol, expected=False ) - assert result is not True, "Testcase {} : Failed \nError: Route is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Route is" " still present in RIB".format( + tc_name + ) step("Reload the FRR router") # stop/start -> restart FRR router and verify @@ -553,19 +593,26 @@ def test_static_route_2nh_p0_tc_1_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) dut = "r3" result = verify_bgp_rib(tgen, addr_type, dut, input_dict_4) - assert result is True, "Testcase {} : Failed \nError: Route is" - " still present in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Route is" " still present in RIB".format( + tc_name + ) result = verify_rib( tgen, addr_type, dut, input_dict_4, protocol=protocol, expected=False ) - assert result is not True, "Testcase {} : Failed \nError: Route is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Route is" " still present in RIB".format( + tc_name + ) write_test_footer(tc_name) @@ -583,7 +630,7 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): pytest.skip(tgen.errors) reset_config_on_routers(tgen) - NEXT_HOP_IP = populate_nh() + step( "Configure IPv4 static route (10.1.1.1) in R2 with next hop N1" "(28.1.1.2 ) AD 10 and N2 (29.1.1.2) AD 20 , Static route next-hop" @@ -592,19 +639,19 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): ) reset_config_on_routers(tgen) - NEXT_HOP_IP = populate_nh() + next_hop_ip = populate_nh() for addr_type in ADDR_TYPES: input_dict_4 = { "r2": { "static_routes": [ { "network": NETWORK2[addr_type], - "next_hop": NEXT_HOP_IP["nh1"][addr_type], + "next_hop": next_hop_ip["nh1"][addr_type], "admin_distance": 10, }, { "network": NETWORK2[addr_type], - "next_hop": NEXT_HOP_IP["nh2"][addr_type], + "next_hop": next_hop_ip["nh2"][addr_type], "admin_distance": 20, }, ] @@ -625,33 +672,34 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): "static_routes": [ { "network": NETWORK2[addr_type], - "next_hop": NEXT_HOP_IP["nh1"][addr_type], + "next_hop": next_hop_ip["nh1"][addr_type], "admin_distance": 10, } ] } } - nh = [NEXT_HOP_IP["nh1"][addr_type]] + nh = [next_hop_ip["nh1"][addr_type]] dut = "r2" protocol = "static" result = verify_rib( tgen, addr_type, dut, rte1_nh1, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - "missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" "missing in RIB".format(tc_name) rte2_nh2 = { "r2": { "static_routes": [ { "network": NETWORK2[addr_type], - "next_hop": NEXT_HOP_IP["nh2"][addr_type], + "next_hop": next_hop_ip["nh2"][addr_type], "admin_distance": 20, } ] } } - nh = [NEXT_HOP_IP["nh2"][addr_type]] + nh = [next_hop_ip["nh2"][addr_type]] dut = "r2" protocol = "static" result = verify_rib( @@ -664,8 +712,9 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - "not active in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" "not active in RIB".format(tc_name) step("Configure IBGP IPv4 peering between R2 and R3 router.") step("Explicit route is added in R3 for R2 nexthop rechability") @@ -673,11 +722,11 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): "r3": { "static_routes": [ { - "network": NEXT_HOP_IP["nh1"][addr_type] + "/32", + "network": next_hop_ip["nh1"][addr_type] + "/32", "next_hop": topo["routers"]["r2"]["links"]["r3"][addr_type], }, { - "network": NEXT_HOP_IP["nh2"][addr_type] + "/32", + "network": next_hop_ip["nh2"][addr_type] + "/32", "next_hop": topo["routers"]["r2"]["links"]["r3"][addr_type], }, ] @@ -712,7 +761,7 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): "static_routes": [ { "network": NETWORK[addr_type], - "next_hop": NEXT_HOP_IP["nh1"][addr_type], + "next_hop": next_hop_ip["nh1"][addr_type], "admin_distance": 10, "delete": True, } @@ -735,13 +784,13 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): "static_routes": [ { "network": NETWORK2[addr_type], - "next_hop": NEXT_HOP_IP["nh1"][addr_type], + "next_hop": next_hop_ip["nh1"][addr_type], "admin_distance": 10, } ] } } - nh = [NEXT_HOP_IP["nh1"][addr_type]] + nh = [next_hop_ip["nh1"][addr_type]] dut = "r2" protocol = "static" result = verify_rib( @@ -754,26 +803,28 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - "missing in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" "missing in RIB".format(tc_name) rte2_nh2 = { "r2": { "static_routes": [ { "network": NETWORK2[addr_type], - "next_hop": NEXT_HOP_IP["nh2"][addr_type], + "next_hop": next_hop_ip["nh2"][addr_type], "admin_distance": 20, } ] } } - nh = [NEXT_HOP_IP["nh2"][addr_type]] + nh = [next_hop_ip["nh2"][addr_type]] result = verify_rib( tgen, addr_type, dut, rte2_nh2, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - "not active in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" "not active in RIB".format(tc_name) step("Configure the static route with nexthop N1") rte1_nh1 = { @@ -781,7 +832,7 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): "static_routes": [ { "network": NETWORK[addr_type], - "next_hop": NEXT_HOP_IP["nh1"][addr_type], + "next_hop": next_hop_ip["nh1"][addr_type], "admin_distance": 10, } ] @@ -799,7 +850,7 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): "static_routes": [ { "network": NETWORK[addr_type], - "next_hop": NEXT_HOP_IP["nh2"][addr_type], + "next_hop": next_hop_ip["nh2"][addr_type], "admin_distance": 20, "delete": True, } @@ -816,7 +867,7 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): "On R2, after removing the static route with N2 , " "route become active with nexthop N1 and vice versa." ) - nh = NEXT_HOP_IP["nh2"][addr_type] + nh = next_hop_ip["nh2"][addr_type] result = verify_rib( tgen, addr_type, @@ -826,15 +877,19 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" " still present in RIB".format( + tc_name + ) - nh = [NEXT_HOP_IP["nh1"][addr_type]] + nh = [next_hop_ip["nh1"][addr_type]] result = verify_rib( tgen, addr_type, dut, rte1_nh1, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) step("Configure the static route with nexthop N2") rte2_nh2 = { @@ -842,7 +897,7 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): "static_routes": [ { "network": NETWORK[addr_type], - "next_hop": NEXT_HOP_IP["nh2"][addr_type], + "next_hop": next_hop_ip["nh2"][addr_type], "admin_distance": 20, } ] @@ -862,7 +917,7 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): step("after shut of nexthop N1 , route become active with nexthop N2") - nh = NEXT_HOP_IP["nh1"][addr_type] + nh = next_hop_ip["nh1"][addr_type] result = verify_rib( tgen, addr_type, @@ -872,15 +927,19 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" " still present in RIB".format( + tc_name + ) - nh = [NEXT_HOP_IP["nh2"][addr_type]] + nh = [next_hop_ip["nh2"][addr_type]] result = verify_rib( tgen, addr_type, dut, rte2_nh2, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) step("No shut the nexthop interface N1") shutdown_bringup_interface(tgen, dut, intf, True) @@ -889,13 +948,14 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): "after shut of nexthop N1 , route become active " "with nexthop N2 and vice versa." ) - nh = [NEXT_HOP_IP["nh1"][addr_type]] + nh = [next_hop_ip["nh1"][addr_type]] result = verify_rib( tgen, addr_type, dut, rte1_nh1, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) step("Shut nexthop interface N2") intf = topo["routers"]["r2"]["links"]["r1-link1"]["interface"] @@ -906,7 +966,7 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): " after shut of nexthop N1 , route become active with " "nexthop N2 and vice versa." ) - nh = NEXT_HOP_IP["nh2"][addr_type] + nh = next_hop_ip["nh2"][addr_type] result = verify_rib( tgen, @@ -917,15 +977,19 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" " still present in RIB".format( + tc_name + ) - nh = [NEXT_HOP_IP["nh1"][addr_type]] + nh = [next_hop_ip["nh1"][addr_type]] result = verify_rib( tgen, addr_type, dut, rte1_nh1, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) step("No shut nexthop interface N2") shutdown_bringup_interface(tgen, dut, intf, True) @@ -939,33 +1003,34 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): "static_routes": [ { "network": NETWORK2[addr_type], - "next_hop": NEXT_HOP_IP["nh1"][addr_type], + "next_hop": next_hop_ip["nh1"][addr_type], "admin_distance": 10, } ] } } - nh = [NEXT_HOP_IP["nh1"][addr_type]] + nh = [next_hop_ip["nh1"][addr_type]] dut = "r2" protocol = "static" result = verify_rib( tgen, addr_type, dut, rte1_nh1, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - "missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" "missing in RIB".format(tc_name) rte2_nh2 = { "r2": { "static_routes": [ { "network": NETWORK2[addr_type], - "next_hop": NEXT_HOP_IP["nh2"][addr_type], + "next_hop": next_hop_ip["nh2"][addr_type], "admin_distance": 20, } ] } } - nh = [NEXT_HOP_IP["nh2"][addr_type]] + nh = [next_hop_ip["nh2"][addr_type]] dut = "r2" protocol = "static" result = verify_rib( @@ -978,8 +1043,9 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - "not active in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" "not active in RIB".format(tc_name) dut = "r3" protocol = "bgp" @@ -994,8 +1060,9 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - "not active in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" "not active in RIB".format(tc_name) dut = "r2" step("Reload the FRR router") @@ -1013,39 +1080,41 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): "static_routes": [ { "network": NETWORK2[addr_type], - "next_hop": NEXT_HOP_IP["nh1"][addr_type], + "next_hop": next_hop_ip["nh1"][addr_type], "admin_distance": 10, } ] } } - nh = [NEXT_HOP_IP["nh1"][addr_type]] + nh = [next_hop_ip["nh1"][addr_type]] dut = "r2" protocol = "static" result = verify_rib( tgen, addr_type, dut, rte1_nh1, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - "missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" "missing in RIB".format(tc_name) dut = "r3" protocol = "bgp" result = verify_bgp_rib(tgen, addr_type, dut, rte1_nh1, next_hop=nh) - assert result is True, "Testcase {} : Failed \nError: Routes is" - "missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" "missing in RIB".format(tc_name) rte2_nh2 = { "r2": { "static_routes": [ { "network": NETWORK2[addr_type], - "next_hop": NEXT_HOP_IP["nh2"][addr_type], + "next_hop": next_hop_ip["nh2"][addr_type], "admin_distance": 20, } ] } } - nh = [NEXT_HOP_IP["nh2"][addr_type]] + nh = [next_hop_ip["nh2"][addr_type]] dut = "r2" protocol = "static" result = verify_rib( @@ -1058,14 +1127,16 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - "not active in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" "not active in RIB".format(tc_name) dut = "r3" protocol = "bgp" result = verify_bgp_rib(tgen, addr_type, dut, rte2_nh2, next_hop=nh) - assert result is True, "Testcase {} : Failed \nError: Routes is" - "not active in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" "not active in RIB".format(tc_name) result = verify_rib( tgen, @@ -1077,8 +1148,9 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ebgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - "not active in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" "not active in RIB".format(tc_name) write_test_footer(tc_name) @@ -1148,14 +1220,20 @@ def test_same_rte_from_bgp_static_p0_tc5_ebgp(request): step("Verify on R3 , route receive on R3 BGP table ") dut = "r3" result = verify_bgp_rib(tgen, addr_type, dut, input_dict_4) - assert result is True, "Testcase {} : Failed \nError: Route is" - " still present in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Route is" " still present in RIB".format( + tc_name + ) step("Verify route installed in the RIB and FIB of R3") protocol = "bgp" result = verify_rib(tgen, addr_type, dut, input_dict_4, protocol=protocol) - assert result is True, "Testcase {} : Failed \nError: Route is" - " still present in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Route is" " still present in RIB".format( + tc_name + ) step( "Configure 2 links/interfaces between R1 and R3 , keep one" @@ -1206,15 +1284,19 @@ def test_same_rte_from_bgp_static_p0_tc5_ebgp(request): ) dut = "r3" result = verify_bgp_rib(tgen, addr_type, dut, input_dict_4) - assert result is True, "Testcase {} : Failed \nError: Route is" - " missing in BGP RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Route is" " missing in BGP RIB".format( + tc_name + ) protocol = "bgp" result = verify_rib( tgen, addr_type, dut, input_dict_4, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Route is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Route is" " missing in RIB".format(tc_name) step("Remove the static route on R3 configured with active" "interface") for addr_type in ADDR_TYPES: @@ -1249,15 +1331,19 @@ def test_same_rte_from_bgp_static_p0_tc5_ebgp(request): ) dut = "r3" result = verify_bgp_rib(tgen, addr_type, dut, input_dict_4) - assert result is True, "Testcase {} : Failed \nError: Route is" - " missing in BGP RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Route is" " missing in BGP RIB".format( + tc_name + ) protocol = "bgp" result = verify_rib( tgen, addr_type, dut, input_dict_4, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Route is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Route is" " missing in RIB".format(tc_name) write_test_footer(tc_name) diff --git a/tests/topotests/static_routing_with_ebgp/test_static_routes_topo2_ebgp.py b/tests/topotests/static_routing_with_ebgp/test_static_routes_topo2_ebgp.py index 6649915dec..bbb4370a93 100644 --- a/tests/topotests/static_routing_with_ebgp/test_static_routes_topo2_ebgp.py +++ b/tests/topotests/static_routing_with_ebgp/test_static_routes_topo2_ebgp.py @@ -314,8 +314,9 @@ def test_static_rte_with_8ecmp_nh_p1_tc9_ebgp(request): next_hop=nh_all[addr_type], protocol=protocol, ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) step("Configure redistribute static in BGP on R2 router") for addr_type in ADDR_TYPES: @@ -338,8 +339,9 @@ def test_static_rte_with_8ecmp_nh_p1_tc9_ebgp(request): dut = "r3" protocol = "bgp" result = verify_bgp_rib(tgen, addr_type, dut, input_dict_4) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) step( "Remove the static route configured with nexthop N1 to N8, one" @@ -384,8 +386,11 @@ def test_static_rte_with_8ecmp_nh_p1_tc9_ebgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed\nError: Routes is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed\nError: Routes is" " still present in RIB".format( + tc_name + ) step("Configure the static route with nexthop N1 to N8, one by one") for addr_type in ADDR_TYPES: @@ -410,8 +415,11 @@ def test_static_rte_with_8ecmp_nh_p1_tc9_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed\nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed\nError: Routes are" " missing in RIB".format( + tc_name + ) protocol = "static" step("Random shut of the nexthop interfaces") @@ -455,8 +463,9 @@ def test_static_rte_with_8ecmp_nh_p1_tc9_ebgp(request): next_hop=nh_all[addr_type], protocol=protocol, ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) step("Remove random static route with all the nexthop") dut = "r2" @@ -493,8 +502,9 @@ def test_static_rte_with_8ecmp_nh_p1_tc9_ebgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) for addr_type in ADDR_TYPES: input_dict_4 = { @@ -546,8 +556,9 @@ def test_static_rte_with_8ecmp_nh_p1_tc9_ebgp(request): next_hop=nh_all[addr_type], protocol=protocol, ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) step("Remove the redistribute static knob") for addr_type in ADDR_TYPES: @@ -580,8 +591,11 @@ def test_static_rte_with_8ecmp_nh_p1_tc9_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, protocol=protocol, expected=False ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes are" " still present in RIB".format( + tc_name + ) write_test_footer(tc_name) @@ -661,8 +675,9 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) nh = [] for nhp in range(2, 9): @@ -679,8 +694,9 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ebgp(request): wait=2, attempts=3, ) - assert result is not True, "Testcase {} : Failed \nError: Routes " - " are missing in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes " " are missing in RIB".format(tc_name) step( "Remove the static route configured with nexthop N1 to N8, one" @@ -723,8 +739,11 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ebgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes are" " still present in RIB".format( + tc_name + ) step("Configure the static route with nexthop N1 to N8, one by one") @@ -770,8 +789,9 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) nh = [] for nhp in range(2, 9): nh.append(NEXT_HOP_IP["nh" + str(nhp)][addr_type]) @@ -787,8 +807,9 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ebgp(request): wait=2, attempts=3, ) - assert result is not True, "Testcase {} : Failed \nError: Routes " - " are missing in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes " " are missing in RIB".format(tc_name) step("Random shut of the nexthop interfaces") randnum = random.randint(0, 7) @@ -815,8 +836,11 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ebgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \n" - "Error: Routes are still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format( + tc_name + ) step("Random no shut of the nexthop interfaces") for addr_type in ADDR_TYPES: @@ -826,8 +850,9 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_5, next_hop=nhip, protocol=protocol ) - assert result is True, "Testcase {} : Failed \n" - "Error: Routes are missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \n" "Error: Routes are missing in RIB".format(tc_name) dut = "r2" protocol = "static" @@ -955,8 +980,9 @@ def test_static_route_8nh_diff_AD_ebgp_ecmp_p1_tc8_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) nh = [] for nhp in range(2, 9): @@ -971,8 +997,9 @@ def test_static_route_8nh_diff_AD_ebgp_ecmp_p1_tc8_ebgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes " - " are missing in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes " " are missing in RIB".format(tc_name) step( "Remove the static route configured with nexthop N1 to N8, one" @@ -1015,8 +1042,11 @@ def test_static_route_8nh_diff_AD_ebgp_ecmp_p1_tc8_ebgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes are" " still present in RIB".format( + tc_name + ) step("Configure the static route with nexthop N1 to N8, one by one") @@ -1062,8 +1092,9 @@ def test_static_route_8nh_diff_AD_ebgp_ecmp_p1_tc8_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) nh = [] for nhp in range(2, 9): nh.append(NEXT_HOP_IP["nh" + str(nhp)][addr_type]) @@ -1077,8 +1108,9 @@ def test_static_route_8nh_diff_AD_ebgp_ecmp_p1_tc8_ebgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes " - " are missing in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes " " are missing in RIB".format(tc_name) step("Random shut of the nexthop interfaces") randnum = random.randint(0, 7) @@ -1105,8 +1137,11 @@ def test_static_route_8nh_diff_AD_ebgp_ecmp_p1_tc8_ebgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \n" - "Error: Routes are still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format( + tc_name + ) step("Random no shut of the nexthop interfaces") for addr_type in ADDR_TYPES: @@ -1116,8 +1151,9 @@ def test_static_route_8nh_diff_AD_ebgp_ecmp_p1_tc8_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_5, next_hop=nhip, protocol=protocol ) - assert result is True, "Testcase {} : Failed \n" - "Error: Routes are missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \n" "Error: Routes are missing in RIB".format(tc_name) dut = "r2" protocol = "static" @@ -1230,8 +1266,9 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) step("Verify that highest AD nexthop are inactive") nh = [] @@ -1249,8 +1286,9 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ebgp(request): wait=2, attempts=3, ) - assert result is not True, "Testcase {} : Failed \nError: Routes " - " are missing in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes " " are missing in RIB".format(tc_name) step("Configure redistribute static in BGP on R2 router") for addr_type in ADDR_TYPES: @@ -1312,8 +1350,11 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ebgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes are" " still present in RIB".format( + tc_name + ) step("Configure the static route with nexthop N1 to N8, one by one") for addr_type in ADDR_TYPES: @@ -1355,8 +1396,10 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Route with " - "lowest AD is missing in RIB".format(tc_name) + assert result is True, ( + "Testcase {} : Failed \nError: Route with " + "lowest AD is missing in RIB".format(tc_name) + ) step("Random shut of the nexthop interfaces") randnum = random.randint(0, 7) @@ -1383,8 +1426,11 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ebgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \n" - "Error: Routes are still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format( + tc_name + ) step("Random no shut of the nexthop interfaces") for addr_type in ADDR_TYPES: @@ -1394,8 +1440,9 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_5, next_hop=nhip, protocol=protocol ) - assert result is True, "Testcase {} : Failed \n" - "Error: Routes are missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \n" "Error: Routes are missing in RIB".format(tc_name) step("Remove random static route with all the nexthop") for addr_type in ADDR_TYPES: @@ -1434,8 +1481,10 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ebgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Route " - " is still present in RIB".format(tc_name) + assert result is not True, ( + "Testcase {} : Failed \nError: Route " + " is still present in RIB".format(tc_name) + ) step("Reconfigure the deleted routes and verify they are installed") for nhp in range(1, 9): @@ -1460,8 +1509,10 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ebgp(request): protocol = "static" nh = NEXT_HOP_IP["nh1"][addr_type] result = verify_rib(tgen, addr_type, dut, input_dict_4, protocol=protocol) - assert result is True, "Testcase {} : Failed \nError: Route " - " is still present in RIB".format(tc_name) + assert result is True, ( + "Testcase {} : Failed \nError: Route " + " is still present in RIB".format(tc_name) + ) step("Reload the FRR router") # stop/start -> restart FRR router and verify @@ -1577,8 +1628,9 @@ def test_static_route_delete_p0_tc11_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) step("Verify that highest AD nexthop are inactive") nh = [] @@ -1594,8 +1646,9 @@ def test_static_route_delete_p0_tc11_ebgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes " - " are missing in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes " " are missing in RIB".format(tc_name) step("Configure redistribute static in BGP on R2 router") for addr_type in ADDR_TYPES: @@ -1647,8 +1700,11 @@ def test_static_route_delete_p0_tc11_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, protocol=protocol, expected=False ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes are" " still present in RIB".format( + tc_name + ) for addr_type in ADDR_TYPES: for nhp in range(1, 9): @@ -1715,8 +1771,12 @@ def test_static_route_delete_p0_tc11_ebgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " still active in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes are" " still active in RIB".format( + tc_name + ) + write_test_footer(tc_name) diff --git a/tests/topotests/static_routing_with_ebgp/test_static_routes_topo3_ebgp.py b/tests/topotests/static_routing_with_ebgp/test_static_routes_topo3_ebgp.py index 175a1123d7..8525e3655c 100644 --- a/tests/topotests/static_routing_with_ebgp/test_static_routes_topo3_ebgp.py +++ b/tests/topotests/static_routing_with_ebgp/test_static_routes_topo3_ebgp.py @@ -297,8 +297,9 @@ def test_staticroute_with_ecmp_p0_tc3_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) step("Configure redistribute static in BGP on R2 router") for addr_type in ADDR_TYPES: input_dict_2 = { @@ -351,8 +352,11 @@ def test_staticroute_with_ecmp_p0_tc3_ebgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes are" " still present in RIB".format( + tc_name + ) step("Configure the static route with nexthop N1 to N8, one by" "one") @@ -379,8 +383,9 @@ def test_staticroute_with_ecmp_p0_tc3_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) step("Random shut of the nexthop interfaces") randnum = random.randint(0, 7) @@ -407,8 +412,11 @@ def test_staticroute_with_ecmp_p0_tc3_ebgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \n" - "Error: Routes are still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format( + tc_name + ) step("Random no shut of the nexthop interfaces") for addr_type in ADDR_TYPES: @@ -418,8 +426,9 @@ def test_staticroute_with_ecmp_p0_tc3_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_5, next_hop=nhip, protocol=protocol ) - assert result is True, "Testcase {} : Failed \n" - "Error: Routes are missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \n" "Error: Routes are missing in RIB".format(tc_name) step("Reload the FRR router") # stop/start -> restart FRR router and verify @@ -429,8 +438,9 @@ def test_staticroute_with_ecmp_p0_tc3_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) write_test_footer(tc_name) @@ -509,8 +519,10 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Route with " - " lowest AD is missing in RIB".format(tc_name) + assert result is True, ( + "Testcase {} : Failed \nError: Route with " + " lowest AD is missing in RIB".format(tc_name) + ) nh = [] for nhp in range(2, 9): @@ -525,8 +537,10 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ebgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes " - " with high AD are active in RIB".format(tc_name) + assert result is not True, ( + "Testcase {} : Failed \nError: Routes " + " with high AD are active in RIB".format(tc_name) + ) step("Configure redistribute static in BGP on R2 router") for addr_type in ADDR_TYPES: @@ -569,8 +583,10 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Route with " - " lowest AD is missing in RIB".format(tc_name) + assert result is True, ( + "Testcase {} : Failed \nError: Route with " + " lowest AD is missing in RIB".format(tc_name) + ) step( "Remove the static route configured with nexthop N1 to N8, one" @@ -613,8 +629,11 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ebgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes are" " still present in RIB".format( + tc_name + ) step("Configure the static route with nexthop N1 to N8, one by" "one") for addr_type in ADDR_TYPES: @@ -655,8 +674,10 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Route with " - " lowest AD is missing in RIB".format(tc_name) + assert result is True, ( + "Testcase {} : Failed \nError: Route with " + " lowest AD is missing in RIB".format(tc_name) + ) nh = [] for nhp in range(2, 9): @@ -671,8 +692,10 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ebgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes " - " with high AD are active in RIB".format(tc_name) + assert result is not True, ( + "Testcase {} : Failed \nError: Routes " + " with high AD are active in RIB".format(tc_name) + ) step("Random shut of the nexthop interfaces") randnum = random.randint(0, 7) @@ -699,8 +722,11 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ebgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \n" - "Error: Routes are still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format( + tc_name + ) step("Random no shut of the nexthop interfaces") for addr_type in ADDR_TYPES: @@ -710,8 +736,9 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_5, next_hop=nhip, protocol=protocol ) - assert result is True, "Testcase {} : Failed \n" - "Error: Routes are missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \n" "Error: Routes are missing in RIB".format(tc_name) step("Reload the FRR router") # stop/start -> restart FRR router and verify @@ -740,8 +767,10 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Route with " - " lowest AD is missing in RIB".format(tc_name) + assert result is True, ( + "Testcase {} : Failed \nError: Route with " + " lowest AD is missing in RIB".format(tc_name) + ) nh = [] for nhp in range(2, 9): @@ -756,8 +785,10 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ebgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes " - " with high AD are active in RIB".format(tc_name) + assert result is not True, ( + "Testcase {} : Failed \nError: Routes " + " with high AD are active in RIB".format(tc_name) + ) write_test_footer(tc_name) @@ -818,18 +849,25 @@ def test_bgp_local_nexthop_p1_tc14_ebgp(request): step("Verify R2 BGP table has IPv4 route") dut = "r2" result = verify_rib(tgen, addr_type, dut, input_dict_4) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB of R2".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB of R2".format( + tc_name + ) step(" Verify route did not install in the R3 BGP table, RIB/FIB") dut = "r3" result = verify_bgp_rib(tgen, addr_type, dut, input_dict_4, expected=False) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - " still present in BGP RIB of R2".format(tc_name) + assert result is not True, ( + "Testcase {} : Failed \nError: Routes is" + " still present in BGP RIB of R2".format(tc_name) + ) result = verify_rib(tgen, addr_type, dut, input_dict_4, expected=False) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - " still present in RIB of R2".format(tc_name) + assert result is not True, ( + "Testcase {} : Failed \nError: Routes is" + " still present in RIB of R2".format(tc_name) + ) write_test_footer(tc_name) @@ -889,8 +927,9 @@ def test_frr_intf_name_as_gw_gap_tc4_ebgp_p0(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, protocol=protocol, next_hop=nh ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) input_dict_nh = { "r1": { @@ -902,8 +941,9 @@ def test_frr_intf_name_as_gw_gap_tc4_ebgp_p0(request): } } result = verify_ip_nht(tgen, input_dict_nh) - assert result is True, "Testcase {} : Failed \nError: Nexthop is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Nexthop is" " missing in RIB".format(tc_name) step( "Shut / no shut IPv4 and IPv6 static next hop interface from" @@ -925,8 +965,9 @@ def test_frr_intf_name_as_gw_gap_tc4_ebgp_p0(request): next_hop=nh, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) shutdown_bringup_interface(tgen, dut, intf, True) @@ -970,8 +1011,11 @@ def test_frr_intf_name_as_gw_gap_tc4_ebgp_p0(request): next_hop=nh, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes" " still present in RIB".format( + tc_name + ) write_test_footer(tc_name) @@ -1034,8 +1078,9 @@ def test_static_route_with_tag_p0_tc_13_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) step("Configure route-map on R2 with allow tag1 and deny tag2") @@ -1116,8 +1161,10 @@ def test_static_route_with_tag_p0_tc_13_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_0, protocol=protocol, expected=False ) - assert result is not True, "Testcase {} : Failed \nError: Route with " - "tag 4002 is still present in RIB".format(tc_name) + assert result is not True, ( + "Testcase {} : Failed \nError: Route with " + "tag 4002 is still present in RIB".format(tc_name) + ) dut = "r2" input_dict_1 = { @@ -1125,8 +1172,10 @@ def test_static_route_with_tag_p0_tc_13_ebgp(request): } result = verify_rib(tgen, addr_type, dut, input_dict_0, protocol=protocol) - assert result is True, "Testcase {} : Failed \nError: Route with " - "tag 4001 is missing in RIB".format(tc_name) + assert result is True, ( + "Testcase {} : Failed \nError: Route with " + "tag 4001 is missing in RIB".format(tc_name) + ) step("Modify the route-map to allow tag2 and deny tag1") # Create route map @@ -1164,8 +1213,10 @@ def test_static_route_with_tag_p0_tc_13_ebgp(request): } result = verify_rib(tgen, addr_type, dut, input_dict_0, protocol=protocol) - assert result is True, "Testcase {} : Failed \nError: Route with " - "tag 4002 is missing in RIB".format(tc_name) + assert result is True, ( + "Testcase {} : Failed \nError: Route with " + "tag 4002 is missing in RIB".format(tc_name) + ) input_dict_1 = { "r2": {"static_routes": [{"network": NETWORK[addr_type], "tag": 4001}]} @@ -1173,8 +1224,10 @@ def test_static_route_with_tag_p0_tc_13_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_1, protocol=protocol, expected=False ) - assert result is not True, "Testcase {} : Failed \nError: Route with " - "tag 4001 is still present in RIB".format(tc_name, result) + assert result is not True, ( + "Testcase {} : Failed \nError: Route with " + "tag 4001 is still present in RIB".format(tc_name) + ) step("Configure one static route with 2 ECMP nexthop N1 and N2") step("For N1 configure tag 1 and for N2 configure tag 2") @@ -1213,8 +1266,9 @@ def test_static_route_with_tag_p0_tc_13_ebgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) step("shut/no shut of tag1 and tag2 nexthop") diff --git a/tests/topotests/static_routing_with_ebgp/test_static_routes_topo4_ebgp.py b/tests/topotests/static_routing_with_ebgp/test_static_routes_topo4_ebgp.py index 5d4950a70e..812b39797f 100644 --- a/tests/topotests/static_routing_with_ebgp/test_static_routes_topo4_ebgp.py +++ b/tests/topotests/static_routing_with_ebgp/test_static_routes_topo4_ebgp.py @@ -109,7 +109,7 @@ def setup_module(mod): Set up the pytest environment. * `mod`: module name """ - global topo + testsuite_run_time = time.asctime(time.localtime(time.time())) logger.info("Testsuite start time: {}".format(testsuite_run_time)) logger.info("=" * 40) @@ -135,8 +135,7 @@ def setup_module(mod): pytest.skip(error_msg) # Checking BGP convergence - global BGP_CONVERGENCE - global ADDR_TYPES + # Don't run this test if we have any failure. if tgen.routers_have_failure(): pytest.skip(tgen.errors) @@ -173,7 +172,7 @@ def teardown_module(mod): # Tests starting # ##################################################### -def static_routes_rmap_pfxlist_p0_tc7_ebgp(request): +def test_static_routes_rmap_pfxlist_p0_tc7_ebgp(request): """ Verify static route are blocked from route-map & prefix-list applied in BGP nbrs @@ -200,7 +199,7 @@ def static_routes_rmap_pfxlist_p0_tc7_ebgp(request): ) for addr_type in ADDR_TYPES: - # Api call to modfiy BGP timerse + # Api call to modify BGP timers input_dict = { "r2": { "bgp": { @@ -241,8 +240,11 @@ def static_routes_rmap_pfxlist_p0_tc7_ebgp(request): step(" All BGP nbrs are down as authentication is mismatch on both" " the sides") bgp_convergence = verify_bgp_convergence(tgen, topo, expected=False) - assert bgp_convergence is not True, "Testcase {} : " - "Failed \n BGP nbrs must be down. Error: {}".format(tc_name, bgp_convergence) + assert ( + bgp_convergence is not True + ), "Testcase {} : " "Failed \n BGP nbrs must be down. Error: {}".format( + tc_name, bgp_convergence + ) step( "Configure 4 IPv4 and 4 IPv6 nbrs with macthing password " @@ -337,8 +339,9 @@ def static_routes_rmap_pfxlist_p0_tc7_ebgp(request): "show ip prefix list" ) result = verify_prefix_lists(tgen, input_dict_2) - assert result is not True, "Testcase {} : Failed \n" - " Error: {}".format(tc_name, result) + assert result is not True, "Testcase {} : Failed \n" " Error: {}".format( + tc_name, result + ) step("Redistribute all the routes (connected, static)") input_dict_2_r1 = { @@ -588,8 +591,10 @@ def static_routes_rmap_pfxlist_p0_tc7_ebgp(request): result4 = verify_rib( tgen, addr_type, dut, input_dict, protocol=protocol, expected=False ) - assert result4 is not True, "Testcase {} : Failed , VM1 route is " - "not filtered out via prefix list. \n Error: {}".format(tc_name, result4) + assert result4 is not True, ( + "Testcase {} : Failed , VM1 route is " + "not filtered out via prefix list. \n Error: {}".format(tc_name, result4) + ) step( "VM4 and VM6 IPV4 and IPv6 address are present in local and " @@ -964,8 +969,10 @@ def static_routes_rmap_pfxlist_p0_tc7_ebgp(request): ) input_dict = {"r1": {"static_routes": [{"network": ntwk_r2_vm1}]}} result4 = verify_rib(tgen, addr_type, dut, input_dict) - assert result4 is True, "Testcase {} : Failed , VM1 route is " - "not filtered out via prefix list. \n Error: {}".format(tc_name, result4) + assert result4 is True, ( + "Testcase {} : Failed , VM1 route is " + "not filtered out via prefix list. \n Error: {}".format(tc_name, result4) + ) step("vm4 should be present in FRR2") dut = "r2" @@ -976,8 +983,10 @@ def static_routes_rmap_pfxlist_p0_tc7_ebgp(request): ) input_dict = {"r1": {"static_routes": [{"network": ntwk_r2_vm1}]}} result4 = verify_rib(tgen, addr_type, dut, input_dict) - assert result4 is True, "Testcase {} : Failed , VM1 route is " - "not filtered out via prefix list. \n Error: {}".format(tc_name, result4) + assert result4 is True, ( + "Testcase {} : Failed , VM1 route is " + "not filtered out via prefix list. \n Error: {}".format(tc_name, result4) + ) dut = "r3" protocol = "bgp" diff --git a/tests/topotests/static_routing_with_ibgp/test_static_routes_topo1_ibgp.py b/tests/topotests/static_routing_with_ibgp/test_static_routes_topo1_ibgp.py index 8c2fdfca13..4e23a72423 100644 --- a/tests/topotests/static_routing_with_ibgp/test_static_routes_topo1_ibgp.py +++ b/tests/topotests/static_routing_with_ibgp/test_static_routes_topo1_ibgp.py @@ -242,8 +242,9 @@ def test_static_route_2nh_p0_tc_1_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) step("Configure IBGP IPv4 peering between R2 and R3 router.") step("Configure redistribute static in BGP on R2 router") @@ -296,15 +297,19 @@ def test_static_route_2nh_p0_tc_1_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" " still present in RIB".format( + tc_name + ) nh = [NEXT_HOP_IP["nh2"][addr_type]] result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) step("Configure the static route with nexthop N1") @@ -359,15 +364,19 @@ def test_static_route_2nh_p0_tc_1_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" " still present in RIB".format( + tc_name + ) nh = [NEXT_HOP_IP["nh1"][addr_type]] result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) step("Configure the static route with nexthop N2") input_dict_4 = { @@ -398,8 +407,9 @@ def test_static_route_2nh_p0_tc_1_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) nh = NEXT_HOP_IP["nh1"][addr_type] result = verify_rib( @@ -411,15 +421,21 @@ def test_static_route_2nh_p0_tc_1_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" " still present in RIB".format( + tc_name + ) dut = "r3" result = verify_bgp_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, expected=False ) - assert result is not True, "Testcase {} : Failed \nError: Route is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Route is" " still present in RIB".format( + tc_name + ) result = verify_rib( tgen, @@ -430,27 +446,35 @@ def test_static_route_2nh_p0_tc_1_ibgp(request): next_hop=nh, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Route is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Route is" " still present in RIB".format( + tc_name + ) dut = "r2" nh = [NEXT_HOP_IP["nh2"][addr_type]] result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) dut = "r3" result = verify_bgp_rib(tgen, addr_type, dut, input_dict_4) - assert result is True, "Testcase {} : Failed \nError: Route is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Route is" " missing in RIB".format(tc_name) result = verify_rib( tgen, addr_type, dut, input_dict_4, protocol=protocol, expected=False ) - assert result is not True, "Testcase {} : Failed \nError: Route is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Route is" " still present in RIB".format( + tc_name + ) dut = "r2" step("No shut the nexthop interface N1") @@ -465,8 +489,9 @@ def test_static_route_2nh_p0_tc_1_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) step("Shut nexthop interface N2") intf = topo["routers"]["r2"]["links"]["r1-link1"]["interface"] @@ -488,8 +513,11 @@ def test_static_route_2nh_p0_tc_1_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" " still present in RIB".format( + tc_name + ) nh = [NEXT_HOP_IP["nh1"][addr_type]] dut = "r2" @@ -497,19 +525,24 @@ def test_static_route_2nh_p0_tc_1_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) dut = "r3" result = verify_bgp_rib(tgen, addr_type, dut, input_dict_4) - assert result is True, "Testcase {} : Failed \nError: Route is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Route is" " missing in RIB".format(tc_name) result = verify_rib( tgen, addr_type, dut, input_dict_4, protocol=protocol, expected=False ) - assert result is not True, "Testcase {} : Failed \nError: Route is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Route is" " still present in RIB".format( + tc_name + ) step("No shut nexthop interface N2") dut = "r2" @@ -524,19 +557,24 @@ def test_static_route_2nh_p0_tc_1_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) dut = "r3" result = verify_bgp_rib(tgen, addr_type, dut, input_dict_4) - assert result is True, "Testcase {} : Failed \nError: Route is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Route is" " missing in RIB".format(tc_name) result = verify_rib( tgen, addr_type, dut, input_dict_4, protocol=protocol, expected=False ) - assert result is not True, "Testcase {} : Failed \nError: Route is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Route is" " still present in RIB".format( + tc_name + ) step("Reload the FRR router") # stop/start -> restart FRR router and verify @@ -552,19 +590,26 @@ def test_static_route_2nh_p0_tc_1_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) dut = "r3" result = verify_bgp_rib(tgen, addr_type, dut, input_dict_4) - assert result is True, "Testcase {} : Failed \nError: Route is" - " still present in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Route is" " still present in RIB".format( + tc_name + ) result = verify_rib( tgen, addr_type, dut, input_dict_4, protocol=protocol, expected=False ) - assert result is not True, "Testcase {} : Failed \nError: Route is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Route is" " still present in RIB".format( + tc_name + ) write_test_footer(tc_name) @@ -636,8 +681,9 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ibgp(request): result = verify_rib( tgen, addr_type, dut, rte1_nh1, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - "missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" "missing in RIB".format(tc_name) rte2_nh2 = { "r2": { @@ -663,8 +709,9 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ibgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - "not active in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" "not active in RIB".format(tc_name) step("Configure IBGP IPv4 peering between R2 and R3 router.") step("Explicit route is added in R3 for R2 nexthop rechability") @@ -753,8 +800,9 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ibgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - "missing in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" "missing in RIB".format(tc_name) rte2_nh2 = { "r2": { @@ -771,8 +819,9 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ibgp(request): result = verify_rib( tgen, addr_type, dut, rte2_nh2, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - "not active in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" "not active in RIB".format(tc_name) step("Configure the static route with nexthop N1") rte1_nh1 = { @@ -825,15 +874,19 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" " still present in RIB".format( + tc_name + ) nh = [NEXT_HOP_IP["nh1"][addr_type]] result = verify_rib( tgen, addr_type, dut, rte1_nh1, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) step("Configure the static route with nexthop N2") rte2_nh2 = { @@ -871,15 +924,19 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" " still present in RIB".format( + tc_name + ) nh = [NEXT_HOP_IP["nh2"][addr_type]] result = verify_rib( tgen, addr_type, dut, rte2_nh2, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) step("No shut the nexthop interface N1") shutdown_bringup_interface(tgen, dut, intf, True) @@ -893,8 +950,9 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ibgp(request): result = verify_rib( tgen, addr_type, dut, rte1_nh1, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) step("Shut nexthop interface N2") intf = topo["routers"]["r2"]["links"]["r1-link1"]["interface"] @@ -916,15 +974,19 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" " still present in RIB".format( + tc_name + ) nh = [NEXT_HOP_IP["nh1"][addr_type]] result = verify_rib( tgen, addr_type, dut, rte1_nh1, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB".format(tc_name) step("No shut nexthop interface N2") shutdown_bringup_interface(tgen, dut, intf, True) @@ -950,8 +1012,9 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ibgp(request): result = verify_rib( tgen, addr_type, dut, rte1_nh1, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - "missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" "missing in RIB".format(tc_name) rte2_nh2 = { "r2": { @@ -977,8 +1040,9 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ibgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - "not active in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" "not active in RIB".format(tc_name) dut = "r3" protocol = "bgp" @@ -993,8 +1057,9 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ibgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - "not active in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" "not active in RIB".format(tc_name) dut = "r2" step("Reload the FRR router") @@ -1024,14 +1089,16 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ibgp(request): result = verify_rib( tgen, addr_type, dut, rte1_nh1, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes is" - "missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" "missing in RIB".format(tc_name) dut = "r3" protocol = "bgp" result = verify_bgp_rib(tgen, addr_type, dut, rte1_nh1, next_hop=nh) - assert result is True, "Testcase {} : Failed \nError: Routes is" - "missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" "missing in RIB".format(tc_name) rte2_nh2 = { "r2": { @@ -1057,14 +1124,16 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ibgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - "not active in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" "not active in RIB".format(tc_name) dut = "r3" protocol = "bgp" result = verify_bgp_rib(tgen, addr_type, dut, rte2_nh2, next_hop=nh) - assert result is True, "Testcase {} : Failed \nError: Routes is" - "not active in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" "not active in RIB".format(tc_name) result = verify_rib( tgen, @@ -1076,8 +1145,9 @@ def test_static_route_2nh_admin_dist_p0_tc_2_ibgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - "not active in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes is" "not active in RIB".format(tc_name) write_test_footer(tc_name) diff --git a/tests/topotests/static_routing_with_ibgp/test_static_routes_topo2_ibgp.py b/tests/topotests/static_routing_with_ibgp/test_static_routes_topo2_ibgp.py index 644ddc02d4..ee0e01b411 100644 --- a/tests/topotests/static_routing_with_ibgp/test_static_routes_topo2_ibgp.py +++ b/tests/topotests/static_routing_with_ibgp/test_static_routes_topo2_ibgp.py @@ -318,8 +318,9 @@ def test_static_rte_with_8ecmp_nh_p1_tc9_ibgp(request): next_hop=nh_all[addr_type], protocol=protocol, ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) step("Configure redistribute static in BGP on R2 router") for addr_type in ADDR_TYPES: @@ -342,8 +343,9 @@ def test_static_rte_with_8ecmp_nh_p1_tc9_ibgp(request): dut = "r3" protocol = "bgp" result = verify_bgp_rib(tgen, addr_type, dut, input_dict_4) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) step( "Remove the static route configured with nexthop N1 to N8, one" @@ -388,8 +390,11 @@ def test_static_rte_with_8ecmp_nh_p1_tc9_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed\nError: Routes is" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed\nError: Routes is" " still present in RIB".format( + tc_name + ) step("Configure the static route with nexthop N1 to N8, one by one") for addr_type in ADDR_TYPES: @@ -414,8 +419,11 @@ def test_static_rte_with_8ecmp_nh_p1_tc9_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed\nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed\nError: Routes are" " missing in RIB".format( + tc_name + ) protocol = "static" step("Random shut of the nexthop interfaces") @@ -423,7 +431,7 @@ def test_static_rte_with_8ecmp_nh_p1_tc9_ibgp(request): # Shutdown interface dut = "r2" step( - " interface which is about to be shut no shut between r1 and r2 is " "%s", + " interface which is about to be shut no shut between r1 and r2 is %s", topo["routers"]["r2"]["links"]["r1-link{}".format(randnum)]["interface"], ) intf = topo["routers"]["r2"]["links"]["r1-link{}".format(randnum)]["interface"] @@ -459,8 +467,9 @@ def test_static_rte_with_8ecmp_nh_p1_tc9_ibgp(request): next_hop=nh_all[addr_type], protocol=protocol, ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) step("Remove random static route with all the nexthop") dut = "r2" @@ -497,8 +506,9 @@ def test_static_rte_with_8ecmp_nh_p1_tc9_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) for addr_type in ADDR_TYPES: input_dict_4 = { @@ -550,8 +560,9 @@ def test_static_rte_with_8ecmp_nh_p1_tc9_ibgp(request): next_hop=nh_all[addr_type], protocol=protocol, ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) step("Remove the redistribute static knob") for addr_type in ADDR_TYPES: @@ -584,8 +595,11 @@ def test_static_rte_with_8ecmp_nh_p1_tc9_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, protocol=protocol, expected=False ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes are" " still present in RIB".format( + tc_name + ) write_test_footer(tc_name) @@ -665,8 +679,9 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) nh = [] for nhp in range(2, 9): @@ -683,8 +698,9 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ibgp(request): wait=2, attempts=3, ) - assert result is not True, "Testcase {} : Failed \nError: Routes " - " are missing in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes " " are missing in RIB".format(tc_name) step( "Remove the static route configured with nexthop N1 to N8, one" @@ -727,8 +743,11 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes are" " still present in RIB".format( + tc_name + ) step("Configure the static route with nexthop N1 to N8, one by one") @@ -774,8 +793,9 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) nh = [] for nhp in range(2, 9): nh.append(NEXT_HOP_IP["nh" + str(nhp)][addr_type]) @@ -791,8 +811,9 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ibgp(request): wait=2, attempts=3, ) - assert result is not True, "Testcase {} : Failed \nError: Routes " - " are missing in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes " " are missing in RIB".format(tc_name) step("Random shut of the nexthop interfaces") randnum = random.randint(0, 7) @@ -819,8 +840,11 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \n" - "Error: Routes are still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format( + tc_name + ) step("Random no shut of the nexthop interfaces") for addr_type in ADDR_TYPES: @@ -830,8 +854,9 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_5, next_hop=nhip, protocol=protocol ) - assert result is True, "Testcase {} : Failed \n" - "Error: Routes are missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \n" "Error: Routes are missing in RIB".format(tc_name) protocol = "bgp" # this is next hop reachability route in r3 as we are using ibgp @@ -856,8 +881,9 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ibgp(request): for addr_type in ADDR_TYPES: input_dict_4 = {"r2": {"static_routes": [{"network": PREFIX1[addr_type]}]}} result = verify_rib(tgen, addr_type, dut, input_dict_4, protocol=protocol) - assert result is True, "Testcase {} : Failed \n" - "Error: Routes are missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \n" "Error: Routes are missing in RIB".format(tc_name) protocol = "static" dut = "r2" @@ -866,7 +892,7 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ibgp(request): result = verify_rib(tgen, addr_type, dut, input_dict_4, protocol=protocol) assert ( result is True - ), "Testcase {}: Failed \n " "Error: Routes are missing in RIB".format(tc_name) + ), "Testcase {}: Failed \n Error: Routes are missing in RIB".format(tc_name) protocol = "bgp" dut = "r3" @@ -875,7 +901,7 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ibgp(request): result = verify_rib(tgen, addr_type, dut, input_dict_4, protocol=protocol) assert ( result is True - ), "Testcase {}: Failed \n " "Error: Routes are missing in RIB".format(tc_name) + ), "Testcase {}: Failed \n Error: Routes are missing in RIB".format(tc_name) step("Reload the FRR router") # stop/start -> restart FRR router and verify @@ -888,7 +914,7 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ibgp(request): result = verify_rib(tgen, addr_type, dut, input_dict_4, protocol=protocol) assert ( result is True - ), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format( + ), "Testcase {} : Failed \nError: Routes are still present in RIB".format( tc_name ) @@ -912,7 +938,7 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ibgp(request): result = verify_rib(tgen, addr_type, dut, input_dict_4, protocol=protocol) assert ( result is True - ), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format( + ), "Testcase {} : Failed \nError: Routes are still present in RIB".format( tc_name ) @@ -923,7 +949,7 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ibgp(request): result = verify_rib(tgen, addr_type, dut, input_dict_4, protocol=protocol) assert ( result is True - ), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format( + ), "Testcase {} : Failed \nError: Routes are still present in RIB".format( tc_name ) @@ -964,8 +990,10 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " strill present in RIB of R3".format(tc_name) + assert result is not True, ( + "Testcase {} : Failed \nError: Routes are" + " strill present in RIB of R3".format(tc_name) + ) write_test_footer(tc_name) @@ -1060,8 +1088,9 @@ def test_static_route_8nh_diff_AD_ibgp_ecmp_p1_tc7_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) nh = [] for nhp in range(2, 9): @@ -1076,8 +1105,9 @@ def test_static_route_8nh_diff_AD_ibgp_ecmp_p1_tc7_ibgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes " - " are missing in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes " " are missing in RIB".format(tc_name) step( "Remove the static route configured with nexthop N1 to N8, one" @@ -1120,8 +1150,11 @@ def test_static_route_8nh_diff_AD_ibgp_ecmp_p1_tc7_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes are" " still present in RIB".format( + tc_name + ) step("Configure the static route with nexthop N1 to N8, one by one") @@ -1167,8 +1200,9 @@ def test_static_route_8nh_diff_AD_ibgp_ecmp_p1_tc7_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) nh = [] for nhp in range(2, 9): nh.append(NEXT_HOP_IP["nh" + str(nhp)][addr_type]) @@ -1182,8 +1216,9 @@ def test_static_route_8nh_diff_AD_ibgp_ecmp_p1_tc7_ibgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes " - " are missing in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes " " are missing in RIB".format(tc_name) step("Random shut of the nexthop interfaces") randnum = random.randint(0, 7) @@ -1210,8 +1245,11 @@ def test_static_route_8nh_diff_AD_ibgp_ecmp_p1_tc7_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \n" - "Error: Routes are still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format( + tc_name + ) step("Random no shut of the nexthop interfaces") for addr_type in ADDR_TYPES: @@ -1221,8 +1259,9 @@ def test_static_route_8nh_diff_AD_ibgp_ecmp_p1_tc7_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_5, next_hop=nhip, protocol=protocol ) - assert result is True, "Testcase {} : Failed \n" - "Error: Routes are missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \n" "Error: Routes are missing in RIB".format(tc_name) dut = "r2" protocol = "bgp" @@ -1249,8 +1288,9 @@ def test_static_route_8nh_diff_AD_ibgp_ecmp_p1_tc7_ibgp(request): for addr_type in ADDR_TYPES: input_dict_4 = {"r2": {"static_routes": [{"network": PREFIX1[addr_type]}]}} result = verify_rib(tgen, addr_type, dut, input_dict_4, protocol=protocol) - assert result is True, "Testcase {} : Failed \n" - "Error: Routes are missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \n" "Error: Routes are missing in RIB".format(tc_name) protocol = "static" dut = "r2" @@ -1259,7 +1299,7 @@ def test_static_route_8nh_diff_AD_ibgp_ecmp_p1_tc7_ibgp(request): result = verify_rib(tgen, addr_type, dut, input_dict_4, protocol=protocol) assert ( result is True - ), "Testcase {}: Failed \n " "Error: Routes are missing in RIB".format(tc_name) + ), "Testcase {}: Failed \n Error: Routes are missing in RIB".format(tc_name) protocol = "bgp" dut = "r3" @@ -1268,7 +1308,7 @@ def test_static_route_8nh_diff_AD_ibgp_ecmp_p1_tc7_ibgp(request): result = verify_rib(tgen, addr_type, dut, input_dict_4, protocol=protocol) assert ( result is True - ), "Testcase {}: Failed \n " "Error: Routes are missing in RIB".format(tc_name) + ), "Testcase {}: Failed \n Error: Routes are missing in RIB".format(tc_name) step("Reload the FRR router") # stop/start -> restart FRR router and verify @@ -1281,7 +1321,7 @@ def test_static_route_8nh_diff_AD_ibgp_ecmp_p1_tc7_ibgp(request): result = verify_rib(tgen, addr_type, dut, input_dict_4, protocol=protocol) assert ( result is True - ), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format( + ), "Testcase {} : Failed \n Error: Routes are still present in RIB".format( tc_name ) @@ -1305,7 +1345,7 @@ def test_static_route_8nh_diff_AD_ibgp_ecmp_p1_tc7_ibgp(request): result = verify_rib(tgen, addr_type, dut, input_dict_4, protocol=protocol) assert ( result is True - ), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format( + ), "Testcase {} : Failed \n Error: Routes are still present in RIB".format( tc_name ) @@ -1316,7 +1356,7 @@ def test_static_route_8nh_diff_AD_ibgp_ecmp_p1_tc7_ibgp(request): result = verify_rib(tgen, addr_type, dut, input_dict_4, protocol=protocol) assert ( result is True - ), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format( + ), "Testcase {} : Failed \n Error: Routes are still present in RIB".format( tc_name ) @@ -1357,8 +1397,10 @@ def test_static_route_8nh_diff_AD_ibgp_ecmp_p1_tc7_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " still present in RIB of R3".format(tc_name) + assert result is not True, ( + "Testcase {} : Failed \nError: Routes are" + " still present in RIB of R3".format(tc_name) + ) write_test_footer(tc_name) @@ -1453,8 +1495,9 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) step("Verify that highest AD nexthop are inactive") nh = [] @@ -1472,8 +1515,9 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ibgp(request): wait=2, attempts=3, ) - assert result is not True, "Testcase {} : Failed \nError: Routes " - " are missing in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes " " are missing in RIB".format(tc_name) step("Configure redistribute static in BGP on R2 router") for addr_type in ADDR_TYPES: @@ -1535,8 +1579,11 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes are" " still present in RIB".format( + tc_name + ) step("Configure the static route with nexthop N1 to N8, one by one") for addr_type in ADDR_TYPES: @@ -1578,8 +1625,10 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Route with " - "lowest AD is missing in RIB".format(tc_name) + assert result is True, ( + "Testcase {} : Failed \nError: Route with " + "lowest AD is missing in RIB".format(tc_name) + ) step("Random shut of the nexthop interfaces") randnum = random.randint(0, 7) @@ -1606,8 +1655,11 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \n" - "Error: Routes are still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format( + tc_name + ) step("Random no shut of the nexthop interfaces") for addr_type in ADDR_TYPES: @@ -1617,8 +1669,9 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_5, next_hop=nhip, protocol=protocol ) - assert result is True, "Testcase {} : Failed \n" - "Error: Routes are missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \n" "Error: Routes are missing in RIB".format(tc_name) step("Remove random static route with all the nexthop") for addr_type in ADDR_TYPES: @@ -1657,8 +1710,10 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Route " - " is still present in RIB".format(tc_name) + assert result is not True, ( + "Testcase {} : Failed \nError: Route " + " is still present in RIB".format(tc_name) + ) step("Reconfigure the deleted routes and verify they are installed") for nhp in range(1, 9): @@ -1683,8 +1738,10 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ibgp(request): protocol = "static" nh = NEXT_HOP_IP["nh1"][addr_type] result = verify_rib(tgen, addr_type, dut, input_dict_4, protocol=protocol) - assert result is True, "Testcase {} : Failed \nError: Route " - " is still present in RIB".format(tc_name) + assert result is True, ( + "Testcase {} : Failed \nError: Route " + " is still present in RIB".format(tc_name) + ) step("Reload the FRR router") # stop/start -> restart FRR router and verify @@ -1704,9 +1761,7 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ibgp(request): ) assert ( result is True - ), "Testcase {} : Failed \nError: Route " " is missing in RIB".format( - tc_name - ) + ), "Testcase {} : Failed \nError: Route is missing in RIB".format(tc_name) step("Remove the redistribute static knob") for addr_type in ADDR_TYPES: @@ -1750,8 +1805,11 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, protocol=protocol, expected=False ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes are" " still present in RIB".format( + tc_name + ) write_test_footer(tc_name) @@ -1845,8 +1903,9 @@ def test_static_route_delete_p0_tc11_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) step("Verify that highest AD nexthop are inactive") nh = [] @@ -1862,8 +1921,9 @@ def test_static_route_delete_p0_tc11_ibgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes " - " are missing in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes " " are missing in RIB".format(tc_name) step("Configure redistribute static in BGP on R2 router") for addr_type in ADDR_TYPES: @@ -1915,8 +1975,11 @@ def test_static_route_delete_p0_tc11_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, protocol=protocol, expected=False ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes are" " still present in RIB".format( + tc_name + ) for addr_type in ADDR_TYPES: for nhp in range(1, 9): @@ -1983,8 +2046,12 @@ def test_static_route_delete_p0_tc11_ibgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " still active in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes are" " still active in RIB".format( + tc_name + ) + write_test_footer(tc_name) diff --git a/tests/topotests/static_routing_with_ibgp/test_static_routes_topo3_ibgp.py b/tests/topotests/static_routing_with_ibgp/test_static_routes_topo3_ibgp.py index 8f9d88a442..c84c88ac35 100644 --- a/tests/topotests/static_routing_with_ibgp/test_static_routes_topo3_ibgp.py +++ b/tests/topotests/static_routing_with_ibgp/test_static_routes_topo3_ibgp.py @@ -298,8 +298,9 @@ def test_staticroute_with_ecmp_p0_tc3_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) step("Configure redistribute static in BGP on R2 router") for addr_type in ADDR_TYPES: input_dict_2 = { @@ -352,8 +353,11 @@ def test_staticroute_with_ecmp_p0_tc3_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes are" " still present in RIB".format( + tc_name + ) step("Configure the static route with nexthop N1 to N8, one by" "one") @@ -380,8 +384,9 @@ def test_staticroute_with_ecmp_p0_tc3_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) step("Random shut of the nexthop interfaces") randnum = random.randint(0, 7) @@ -408,8 +413,11 @@ def test_staticroute_with_ecmp_p0_tc3_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \n" - "Error: Routes are still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format( + tc_name + ) step("Random no shut of the nexthop interfaces") for addr_type in ADDR_TYPES: @@ -419,8 +427,9 @@ def test_staticroute_with_ecmp_p0_tc3_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_5, next_hop=nhip, protocol=protocol ) - assert result is True, "Testcase {} : Failed \n" - "Error: Routes are missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \n" "Error: Routes are missing in RIB".format(tc_name) step("Reload the FRR router") # stop/start -> restart FRR router and verify @@ -430,8 +439,9 @@ def test_staticroute_with_ecmp_p0_tc3_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol ) - assert result is True, "Testcase {} : Failed \nError: Routes are" - " missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes are" " missing in RIB".format(tc_name) write_test_footer(tc_name) @@ -510,8 +520,10 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Route with " - " lowest AD is missing in RIB".format(tc_name) + assert result is True, ( + "Testcase {} : Failed \nError: Route with " + " lowest AD is missing in RIB".format(tc_name) + ) nh = [] for nhp in range(2, 9): @@ -526,8 +538,10 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ibgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes " - " with high AD are active in RIB".format(tc_name) + assert result is not True, ( + "Testcase {} : Failed \nError: Routes " + " with high AD are active in RIB".format(tc_name) + ) step("Configure redistribute static in BGP on R2 router") for addr_type in ADDR_TYPES: @@ -570,8 +584,10 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Route with " - " lowest AD is missing in RIB".format(tc_name) + assert result is True, ( + "Testcase {} : Failed \nError: Route with " + " lowest AD is missing in RIB".format(tc_name) + ) step( "Remove the static route configured with nexthop N1 to N8, one" @@ -614,8 +630,11 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \nError: Routes are" " still present in RIB".format( + tc_name + ) step("Configure the static route with nexthop N1 to N8, one by" "one") for addr_type in ADDR_TYPES: @@ -656,8 +675,10 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Route with " - " lowest AD is missing in RIB".format(tc_name) + assert result is True, ( + "Testcase {} : Failed \nError: Route with " + " lowest AD is missing in RIB".format(tc_name) + ) nh = [] for nhp in range(2, 9): @@ -672,8 +693,10 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ibgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes " - " with high AD are active in RIB".format(tc_name) + assert result is not True, ( + "Testcase {} : Failed \nError: Routes " + " with high AD are active in RIB".format(tc_name) + ) step("Random shut of the nexthop interfaces") randnum = random.randint(0, 7) @@ -700,8 +723,11 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \n" - "Error: Routes are still present in RIB".format(tc_name) + assert ( + result is not True + ), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format( + tc_name + ) step("Random no shut of the nexthop interfaces") for addr_type in ADDR_TYPES: @@ -711,8 +737,9 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_5, next_hop=nhip, protocol=protocol ) - assert result is True, "Testcase {} : Failed \n" - "Error: Routes are missing in RIB".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \n" "Error: Routes are missing in RIB".format(tc_name) step("Reload the FRR router") # stop/start -> restart FRR router and verify @@ -741,8 +768,10 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ibgp(request): result = verify_rib( tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True ) - assert result is True, "Testcase {} : Failed \nError: Route with " - " lowest AD is missing in RIB".format(tc_name) + assert result is True, ( + "Testcase {} : Failed \nError: Route with " + " lowest AD is missing in RIB".format(tc_name) + ) nh = [] for nhp in range(2, 9): @@ -757,8 +786,10 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ibgp(request): fib=True, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes " - " with high AD are active in RIB".format(tc_name) + assert result is not True, ( + "Testcase {} : Failed \nError: Routes " + " with high AD are active in RIB".format(tc_name) + ) step("Remove the redistribute static knob") @@ -796,8 +827,10 @@ def test_staticroute_with_ecmp_with_diff_AD_p0_tc4_ibgp(request): protocol=protocol, expected=False, ) - assert result is not True, "Testcase {} : Failed \nError: Routes are" - " strill present in RIB of R3".format(tc_name) + assert result is not True, ( + "Testcase {} : Failed \nError: Routes are" + " still present in RIB of R3".format(tc_name) + ) write_test_footer(tc_name) @@ -858,18 +891,25 @@ def test_bgp_local_nexthop_p1_tc14_ibgp(request): step("Verify R2 BGP table has IPv4 route") dut = "r2" result = verify_rib(tgen, addr_type, dut, input_dict_4) - assert result is True, "Testcase {} : Failed \nError: Routes is" - " missing in RIB of R2".format(tc_name) + assert ( + result is True + ), "Testcase {} : Failed \nError: Routes is" " missing in RIB of R2".format( + tc_name + ) step(" Verify route did not install in the R3 BGP table, RIB/FIB") dut = "r3" result = verify_bgp_rib(tgen, addr_type, dut, input_dict_4, expected=False) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - " still present in BGP RIB of R2".format(tc_name) + assert result is not True, ( + "Testcase {} : Failed \nError: Routes is" + " still present in BGP RIB of R2".format(tc_name) + ) result = verify_rib(tgen, addr_type, dut, input_dict_4, expected=False) - assert result is not True, "Testcase {} : Failed \nError: Routes is" - " still present in RIB of R2".format(tc_name) + assert result is not True, ( + "Testcase {} : Failed \nError: Routes is" + " still present in RIB of R2".format(tc_name) + ) write_test_footer(tc_name) diff --git a/tests/topotests/static_routing_with_ibgp/test_static_routes_topo4_ibgp.py b/tests/topotests/static_routing_with_ibgp/test_static_routes_topo4_ibgp.py index 09c437c3c4..a82ee64538 100644 --- a/tests/topotests/static_routing_with_ibgp/test_static_routes_topo4_ibgp.py +++ b/tests/topotests/static_routing_with_ibgp/test_static_routes_topo4_ibgp.py @@ -239,8 +239,11 @@ def test_static_routes_rmap_pfxlist_p0_tc7_ibgp(request): step(" All BGP nbrs are down as authentication is mismatch on both" " the sides") bgp_convergence = verify_bgp_convergence(tgen, topo, expected=False) - assert bgp_convergence is not True, "Testcase {} : " - "Failed \n BGP nbrs must be down. Error: {}".format(tc_name, bgp_convergence) + assert ( + bgp_convergence is not True + ), "Testcase {} : " "Failed \n BGP nbrs must be down. Error: {}".format( + tc_name, bgp_convergence + ) step( "Configure 4 IPv4 and 4 IPv6 nbrs with macthing password " @@ -335,8 +338,9 @@ def test_static_routes_rmap_pfxlist_p0_tc7_ibgp(request): "show ip prefix list" ) result = verify_prefix_lists(tgen, input_dict_2) - assert result is not True, "Testcase {} : Failed \n" - " Error: {}".format(tc_name, result) + assert result is not True, "Testcase {} : Failed \n" " Error: {}".format( + tc_name, result + ) step("Redistribute all the routes (connected, static)") input_dict_2_r1 = { @@ -586,8 +590,10 @@ def test_static_routes_rmap_pfxlist_p0_tc7_ibgp(request): result4 = verify_rib( tgen, addr_type, dut, input_dict, protocol=protocol, expected=False ) - assert result4 is not True, "Testcase {} : Failed , VM1 route is " - "not filtered out via prefix list. \n Error: {}".format(tc_name, result4) + assert result4 is not True, ( + "Testcase {} : Failed , VM1 route is " + "not filtered out via prefix list. \n Error: {}".format(tc_name, result4) + ) step( "VM4 and VM6 IPV4 and IPv6 address are present in local and " @@ -962,8 +968,10 @@ def test_static_routes_rmap_pfxlist_p0_tc7_ibgp(request): ) input_dict = {"r1": {"static_routes": [{"network": ntwk_r2_vm1}]}} result4 = verify_rib(tgen, addr_type, dut, input_dict) - assert result4 is True, "Testcase {} : Failed , VM1 route is " - "not filtered out via prefix list. \n Error: {}".format(tc_name, result4) + assert result4 is True, ( + "Testcase {} : Failed , VM1 route is " + "not filtered out via prefix list. \n Error: {}".format(tc_name, result4) + ) step("vm4 should be present in FRR2") dut = "r2" @@ -974,8 +982,10 @@ def test_static_routes_rmap_pfxlist_p0_tc7_ibgp(request): ) input_dict = {"r1": {"static_routes": [{"network": ntwk_r2_vm1}]}} result4 = verify_rib(tgen, addr_type, dut, input_dict) - assert result4 is True, "Testcase {} : Failed , VM1 route is " - "not filtered out via prefix list. \n Error: {}".format(tc_name, result4) + assert result4 is True, ( + "Testcase {} : Failed , VM1 route is " + "not filtered out via prefix list. \n Error: {}".format(tc_name, result4) + ) dut = "r3" protocol = "bgp" |