Merge pull request #8172 from donaldsharp/more_pytest_bgp

More pytest stuff
This commit is contained in:
Donatas Abraitis 2021-03-02 09:30:06 +02:00 committed by GitHub
commit c103ac43de
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
80 changed files with 1226 additions and 750 deletions

View File

@ -352,7 +352,7 @@ def test_converge_protocols():
actual = (
net["r%s" % i]
.cmd(
'vtysh -c "show ip route" | sed -e \'/^Codes: /,/^\s*$/d\' | env LC_ALL=en_US.UTF-8 sort 2> /dev/null'
"vtysh -c \"show ip route\" | sed -e '/^Codes: /,/^\s*$/d' | env LC_ALL=en_US.UTF-8 sort 2> /dev/null"
)
.rstrip()
)
@ -383,7 +383,7 @@ def test_converge_protocols():
actual = (
net["r%s" % i]
.cmd(
'vtysh -c "show ipv6 route" | sed -e \'/^Codes: /,/^\s*$/d\' | env LC_ALL=en_US.UTF-8 sort 2> /dev/null'
"vtysh -c \"show ipv6 route\" | sed -e '/^Codes: /,/^\s*$/d' | env LC_ALL=en_US.UTF-8 sort 2> /dev/null"
)
.rstrip()
)

View File

@ -44,7 +44,8 @@ from lib.topolog import logger
# Required to instantiate the topology builder class.
from mininet.topo import Topo
pytestmark = [pytest.mark.bfdd]
pytestmark = [pytest.mark.bgpd, pytest.mark.bfdd]
class BFDTopo(Topo):
"Test topology builder"
@ -65,6 +66,7 @@ class BFDTopo(Topo):
switch.add_link(tgen.gears["r2"])
switch.add_link(tgen.gears["r3"])
def setup_module(mod):
"Sets up the pytest environment"
tgen = Topogen(BFDTopo, mod.__name__)

View File

@ -92,6 +92,7 @@ from mininet.topo import Topo
pytestmark = [pytest.mark.bfdd, pytest.mark.isisd]
class TemplateTopo(Topo):
"Test topology builder"

View File

@ -92,6 +92,7 @@ from mininet.topo import Topo
pytestmark = [pytest.mark.bfdd, pytest.mark.ospfd]
class TemplateTopo(Topo):
"Test topology builder"

View File

@ -45,7 +45,8 @@ from lib.topolog import logger
# Required to instantiate the topology builder class.
from mininet.topo import Topo
pytestmark = [pytest.mark.bfdd, pytest.mark.isisd, pytest.mark.ospfd]
pytestmark = [pytest.mark.bfdd, pytest.mark.bgpd, pytest.mark.isisd, pytest.mark.ospfd]
class BFDProfTopo(Topo):
"Test topology builder"

View File

@ -45,7 +45,8 @@ from lib.topolog import logger
# Required to instantiate the topology builder class.
from mininet.topo import Topo
pytestmark = [pytest.mark.bfdd]
pytestmark = [pytest.mark.bfdd, pytest.mark.bgpd]
class BFDTopo(Topo):
"Test topology builder"

View File

@ -46,7 +46,8 @@ from lib.topolog import logger
# Required to instantiate the topology builder class.
from mininet.topo import Topo
pytestmark = [pytest.mark.bfdd, pytest.mark.ospfd]
pytestmark = [pytest.mark.bfdd, pytest.mark.bgpd, pytest.mark.ospfd]
class BFDTopo(Topo):
"Test topology builder"

View File

@ -45,6 +45,8 @@ from lib.topolog import logger
# Required to instantiate the topology builder class.
from mininet.topo import Topo
pytestmark = [pytest.mark.bfdd, pytest.mark.bgpd]
class BFDTopo(Topo):
"Test topology builder"

View File

@ -46,7 +46,8 @@ from lib.topolog import logger
# Required to instantiate the topology builder class.
from mininet.topo import Topo
pytestmark = [pytest.mark.bfdd]
pytestmark = [pytest.mark.bfdd, pytest.mark.bgpd]
class BFDTopo(Topo):
"Test topology builder"

View File

@ -40,6 +40,8 @@ from lib.topogen import Topogen, TopoRouter, get_topogen
from lib.topolog import logger
from mininet.topo import Topo
pytestmark = [pytest.mark.bgpd]
class BgpAggregatorAsnZero(Topo):
def build(self, *_args, **_opts):

View File

@ -91,6 +91,9 @@ from lib.bgp import (
)
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 = "{}/bgp_basic_functionality.json".format(CWD)
try:

View File

@ -324,6 +324,7 @@ def test_modify_ecmp_max_paths(request, ecmp_num, test_type):
write_test_footer(tc_name)
@pytest.mark.parametrize("ecmp_num", ["8", "16", "32"])
@pytest.mark.parametrize("test_type", ["redist_static", "advertise_nw"])
def test_ecmp_after_clear_bgp(request, ecmp_num, test_type):
@ -349,7 +350,7 @@ def test_ecmp_after_clear_bgp(request, ecmp_num, test_type):
addr_type,
dut,
input_dict_1,
next_hop=NEXT_HOPS[addr_type][:int(ecmp_num)],
next_hop=NEXT_HOPS[addr_type][: int(ecmp_num)],
protocol=protocol,
)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
@ -372,7 +373,7 @@ def test_ecmp_after_clear_bgp(request, ecmp_num, test_type):
addr_type,
dut,
input_dict_1,
next_hop=NEXT_HOPS[addr_type][:int(ecmp_num)],
next_hop=NEXT_HOPS[addr_type][: int(ecmp_num)],
protocol=protocol,
)
assert result is True, "Testcase {} : Failed \n Error: {}".format(

View File

@ -325,6 +325,7 @@ def test_modify_ecmp_max_paths(request, ecmp_num, test_type):
write_test_footer(tc_name)
@pytest.mark.parametrize("ecmp_num", ["8", "16", "32"])
@pytest.mark.parametrize("test_type", ["redist_static", "advertise_nw"])
def test_ecmp_after_clear_bgp(request, ecmp_num, test_type):
@ -350,7 +351,7 @@ def test_ecmp_after_clear_bgp(request, ecmp_num, test_type):
addr_type,
dut,
input_dict_1,
next_hop=NEXT_HOPS[addr_type][:int(ecmp_num)],
next_hop=NEXT_HOPS[addr_type][: int(ecmp_num)],
protocol=protocol,
)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
@ -373,7 +374,7 @@ def test_ecmp_after_clear_bgp(request, ecmp_num, test_type):
addr_type,
dut,
input_dict_1,
next_hop=NEXT_HOPS[addr_type][:int(ecmp_num)],
next_hop=NEXT_HOPS[addr_type][: int(ecmp_num)],
protocol=protocol,
)
assert result is True, "Testcase {} : Failed \n Error: {}".format(

View File

@ -603,6 +603,7 @@ def ping_anycast_gw(tgen):
local_host.run(cmd_str)
remote_host.run(cmd_str)
def check_mac(dut, vni, mac, m_type, esi, intf, ping_gw=False, tgen=None):
"""
checks if mac is present and if desination matches the one provided

View File

@ -47,6 +47,7 @@ from mininet.topo import Topo
pytestmark = [pytest.mark.bgpd, pytest.mark.ospfd]
class TemplateTopo(Topo):
"Test topology builder"

View File

@ -47,6 +47,8 @@ from lib.snmptest import SnmpTester
# Required to instantiate the topology builder class.
from mininet.topo import Topo
pytestmark = [pytest.mark.bgpd, pytest.mark.isisd, pytest.mark.snmp]
class TemplateTopo(Topo):
"Test topology builder"
@ -266,7 +268,7 @@ def test_pe1_converge_evpn():
break
count += 1
sleep(1)
#tgen.mininet_cli()
# tgen.mininet_cli()
assertmsg = "BGP Peer 10.4.4.4 did not connect"
assert passed, assertmsg
@ -503,8 +505,10 @@ def test_r1_mplsvpn_VrfTable():
associated_int = r1_snmp.get(
"mplsL3VpnVrfAssociatedInterfaces.{}".format(snmp_str_to_oid("VRF-a"))
)
assertmsg = "mplsL3VpnVrfAssociatedInterfaces incorrect should be 3 value {}".format(
associated_int
assertmsg = (
"mplsL3VpnVrfAssociatedInterfaces incorrect should be 3 value {}".format(
associated_int
)
)
assert associated_int == "3", assertmsg
@ -620,7 +624,7 @@ rte_table_test = {
"unknown(0)",
"ipv4(1)",
"unknown(0)",
],
],
"mplsL3VpnVrfRteInetCidrNextHop": [
"C0 A8 64 0A",
"C0 A8 C8 0A",
@ -649,7 +653,15 @@ rte_table_test = {
"bgp(14)",
"local(2)",
],
"mplsL3VpnVrfRteInetCidrNextHopAS": ["65001", "65001", "0", "65001", "0", "65001", "0"],
"mplsL3VpnVrfRteInetCidrNextHopAS": [
"65001",
"65001",
"0",
"65001",
"0",
"65001",
"0",
],
"mplsL3VpnVrfRteInetCidrMetric1": ["0", "0", "20", "0", "0", "0", "0"],
"mplsL3VpnVrfRteInetCidrMetric2": ["-1", "-1", "-1", "-1", "-1", "-1", "-1"],
"mplsL3VpnVrfRteInetCidrMetric3": ["-1", "-1", "-1", "-1", "-1", "-1", "-1"],
@ -663,7 +675,7 @@ rte_table_test = {
"active(1)",
"active(1)",
"active(1)",
"active(1)",
"active(1)",
],
}

View File

@ -47,6 +47,8 @@ from lib.topogen import Topogen, TopoRouter, get_topogen
from lib.topolog import logger
from mininet.topo import Topo
pytestmark = [pytest.mark.bgpd]
class TemplateTopo(Topo):
def build(self, *_args, **_opts):

View File

@ -50,6 +50,8 @@ from lib.topogen import Topogen, TopoRouter, get_topogen
from lib.topolog import logger
from mininet.topo import Topo
pytestmark = [pytest.mark.bgpd]
class TemplateTopo(Topo):
def build(self, *_args, **_opts):

View File

@ -76,6 +76,9 @@ from lib.bgp import (
)
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 = "{}/bgp_as_allow_in.json".format(CWD)
try:

View File

@ -45,6 +45,8 @@ from lib.topogen import Topogen, TopoRouter, get_topogen
from lib.topolog import logger
from mininet.topo import Topo
pytestmark = [pytest.mark.bgpd]
class TemplateTopo(Topo):
def build(self, *_args, **_opts):

View File

@ -40,6 +40,8 @@ from lib.topolog import logger
from mininet.topo import Topo
from lib.common_config import step
pytestmark = [pytest.mark.bgpd]
class TemplateTopo(Topo):
def build(self, *_args, **_opts):

View File

@ -45,6 +45,8 @@ from lib.topogen import Topogen, TopoRouter, get_topogen
from lib.topolog import logger
from mininet.topo import Topo
pytestmark = [pytest.mark.bgpd]
class TemplateTopo(Topo):
def build(self, *_args, **_opts):

View File

@ -66,6 +66,9 @@ from lib.bgp import (
from lib.topojson import build_topo_from_json, build_config_from_json
from copy import deepcopy
pytestmark = [pytest.mark.bgpd, pytest.mark.staticd]
# Reading the data from JSON File for topology creation
jsonFile = "{}/bgp_communities.json".format(CWD)
try:

View File

@ -70,6 +70,9 @@ from lib.bgp import (
from lib.topojson import build_topo_from_json, build_config_from_json
from copy import deepcopy
pytestmark = [pytest.mark.bgpd, pytest.mark.staticd]
# Reading the data from JSON File for topology creation
jsonFile = "{}/bgp_communities_topo2.json".format(CWD)
try:

View File

@ -59,6 +59,8 @@ from mininet.topo import Topo
from lib.common_config import step
from time import sleep
pytestmark = [pytest.mark.bgpd, pytest.mark.bgpd]
class TemplateTopo(Topo):
def build(self, *_args, **_opts):

View File

@ -8,7 +8,7 @@ if len(sys.argv) != 2:
fifo = sys.argv[1]
while True:
pipe = open(fifo, 'r')
pipe = open(fifo, "r")
with pipe:
line = pipe.readline().strip()
if line != "":

View File

@ -8,7 +8,7 @@ if len(sys.argv) != 2:
fifo = sys.argv[1]
while True:
pipe = open(fifo, 'r')
pipe = open(fifo, "r")
with pipe:
line = pipe.readline().strip()
if line != "":

View File

@ -8,7 +8,7 @@ if len(sys.argv) != 2:
fifo = sys.argv[1]
while True:
pipe = open(fifo, 'r')
pipe = open(fifo, "r")
with pipe:
line = pipe.readline().strip()
if line != "":

View File

@ -8,7 +8,7 @@ if len(sys.argv) != 2:
fifo = sys.argv[1]
while True:
pipe = open(fifo, 'r')
pipe = open(fifo, "r")
with pipe:
line = pipe.readline().strip()
if line != "":

View File

@ -31,6 +31,7 @@ from lib.ltemplate import *
pytestmark = [pytest.mark.bgpd, pytest.mark.ospfd]
def test_adjacencies():
CliOnFail = None
# For debugging, uncomment the next line

View File

@ -31,6 +31,7 @@ from lib.ltemplate import *
pytestmark = [pytest.mark.bgpd, pytest.mark.ospfd]
def test_check_linux_vrf():
CliOnFail = None
# For debugging, uncomment the next line

View File

@ -142,7 +142,7 @@ def _bgp_converge_initial(router_name, peer_address, timeout=180):
"""
Waits for the BGP connection between a given router and a given peer
(specified by its IP address) to be established. If the connection is
not established within a given timeout, then an exception is raised.
not established within a given timeout, then an exception is raised.
"""
tgen = get_topogen()
router = tgen.routers()[router_name]

View File

@ -45,17 +45,18 @@ from lib.topolog import logger
# Required to instantiate the topology builder class.
from mininet.topo import Topo
#Basic scenario for BGP-LU. Nodes are directly connected.
#Node 3 is advertising many routes to 2, which advertises them
#as BGP-LU to 1; this way we get routes with actual labels, as
#opposed to implicit-null routes in the 2-node case.
# Basic scenario for BGP-LU. Nodes are directly connected.
# Node 3 is advertising many routes to 2, which advertises them
# as BGP-LU to 1; this way we get routes with actual labels, as
# opposed to implicit-null routes in the 2-node case.
#
# AS1 BGP-LU AS2 iBGP AS2
#+-----+ +-----+ +-----+
#| |.1 .2| |.2 .3| |
#| 1 +----------------+ 2 +-----------------+ 3 |
#| | 10.0.0.0/24 | | 10.0.1.0/24 | |
#+-----+ +-----+ +-----+
# +-----+ +-----+ +-----+
# | |.1 .2| |.2 .3| |
# | 1 +----------------+ 2 +-----------------+ 3 |
# | | 10.0.0.0/24 | | 10.0.1.0/24 | |
# +-----+ +-----+ +-----+
class TemplateTopo(Topo):
"Test topology builder"
@ -84,7 +85,6 @@ class TemplateTopo(Topo):
switch.add_link(tgen.gears["R3"])
def setup_module(mod):
"Sets up the pytest environment"
# This function initiates the topology build with Topogen...
@ -115,15 +115,19 @@ def teardown_module(mod):
# This function tears down the whole topology.
tgen.stop_topology()
def check_labelpool(router):
json_file = "{}/{}/labelpool.summ.json".format(CWD, router.name)
expected = json.loads(open(json_file).read())
test_func = partial(topotest.router_json_cmp, router, "show bgp labelpool summary json", expected)
test_func = partial(
topotest.router_json_cmp, router, "show bgp labelpool summary json", expected
)
_, result = topotest.run_and_expect(test_func, None, count=20, wait=1)
assertmsg = '"{}" JSON output mismatches - Did not converge'.format(router.name)
assert result is None, assertmsg
def test_converge_bgplu():
"Wait for protocol convergence"
@ -132,13 +136,14 @@ def test_converge_bgplu():
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
#tgen.mininet_cli();
# tgen.mininet_cli();
r1 = tgen.gears["R1"]
r2 = tgen.gears["R2"]
check_labelpool(r1)
check_labelpool(r2)
def test_clear_bgplu():
"Wait for protocol convergence"
@ -147,7 +152,7 @@ def test_clear_bgplu():
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
#tgen.mininet_cli();
# tgen.mininet_cli();
r1 = tgen.gears["R1"]
r2 = tgen.gears["R2"]
@ -164,6 +169,7 @@ def test_clear_bgplu():
check_labelpool(r1)
check_labelpool(r2)
def test_memory_leak():
"Run the memory leak test and report results."
tgen = get_topogen()

View File

@ -31,6 +31,7 @@ from lib.ltemplate import *
pytestmark = [pytest.mark.bgpd, pytest.mark.ospfd]
def test_add_routes():
CliOnFail = None
# For debugging, uncomment the next line

View File

@ -44,7 +44,7 @@ from lib.topolog import logger
from mininet.topo import Topo
#TODO: select markers based on daemons used during test
# TODO: select markers based on daemons used during test
# pytest module level markers
"""
pytestmark = pytest.mark.bfdd # single marker

View File

@ -54,7 +54,7 @@ from lib.bgp import verify_bgp_convergence
from lib.topojson import build_topo_from_json, build_config_from_json
#TODO: select markers based on daemons used during test
# TODO: select markers based on daemons used during test
# pytest module level markers
"""
pytestmark = pytest.mark.bfdd # single marker

View File

@ -53,7 +53,7 @@ from lib.bgp import verify_bgp_convergence
from lib.topojson import build_topo_from_json, build_config_from_json
#TODO: select markers based on daemons used during test
# TODO: select markers based on daemons used during test
# pytest module level markers
"""
pytestmark = pytest.mark.bfdd # single marker

View File

@ -55,7 +55,7 @@ from lib.bgp import verify_bgp_convergence
from lib.topojson import build_topo_from_json, build_config_from_json
#TODO: select markers based on daemons used during test
# TODO: select markers based on daemons used during test
# pytest module level markers
"""
pytestmark = pytest.mark.bfdd # single marker

View File

@ -165,6 +165,7 @@ class TemplateTopo(Topo):
f_in.close()
f_out.close()
def setup_module(mod):
"Sets up the pytest environment"
tgen = Topogen(TemplateTopo, mod.__name__)

View File

@ -73,7 +73,7 @@ from functools import partial
# Save the Current Working Directory to find configuration files.
CWD = os.path.dirname(os.path.realpath(__file__))
sys.path.append(os.path.join(CWD, '../'))
sys.path.append(os.path.join(CWD, "../"))
# pylint: disable=C0413
# Import topogen and topotest helpers
@ -87,8 +87,10 @@ from mininet.topo import Topo
# Global multi-dimensional dictionary containing all expected outputs
outputs = {}
class TemplateTopo(Topo):
"Test topology builder"
def build(self, *_args, **_opts):
"Build function"
tgen = get_topogen(self)
@ -96,36 +98,36 @@ class TemplateTopo(Topo):
#
# Define FRR Routers
#
for router in ['rt1', 'rt2', 'rt3', 'rt4', 'rt5', 'rt6']:
for router in ["rt1", "rt2", "rt3", "rt4", "rt5", "rt6"]:
tgen.add_router(router)
#
# Define connections
#
switch = tgen.add_switch('s1')
switch.add_link(tgen.gears['rt1'], nodeif="eth-sw1")
switch.add_link(tgen.gears['rt2'], nodeif="eth-sw1")
switch.add_link(tgen.gears['rt3'], nodeif="eth-sw1")
switch = tgen.add_switch("s1")
switch.add_link(tgen.gears["rt1"], nodeif="eth-sw1")
switch.add_link(tgen.gears["rt2"], nodeif="eth-sw1")
switch.add_link(tgen.gears["rt3"], nodeif="eth-sw1")
switch = tgen.add_switch('s2')
switch.add_link(tgen.gears['rt2'], nodeif="eth-rt4")
switch.add_link(tgen.gears['rt4'], nodeif="eth-rt2")
switch = tgen.add_switch("s2")
switch.add_link(tgen.gears["rt2"], nodeif="eth-rt4")
switch.add_link(tgen.gears["rt4"], nodeif="eth-rt2")
switch = tgen.add_switch('s4')
switch.add_link(tgen.gears['rt3'], nodeif="eth-rt5")
switch.add_link(tgen.gears['rt5'], nodeif="eth-rt3")
switch = tgen.add_switch("s4")
switch.add_link(tgen.gears["rt3"], nodeif="eth-rt5")
switch.add_link(tgen.gears["rt5"], nodeif="eth-rt3")
switch = tgen.add_switch('s6')
switch.add_link(tgen.gears['rt4'], nodeif="eth-rt5")
switch.add_link(tgen.gears['rt5'], nodeif="eth-rt4")
switch = tgen.add_switch("s6")
switch.add_link(tgen.gears["rt4"], nodeif="eth-rt5")
switch.add_link(tgen.gears["rt5"], nodeif="eth-rt4")
switch = tgen.add_switch('s7')
switch.add_link(tgen.gears['rt4'], nodeif="eth-rt6")
switch.add_link(tgen.gears['rt6'], nodeif="eth-rt4")
switch = tgen.add_switch("s7")
switch.add_link(tgen.gears["rt4"], nodeif="eth-rt6")
switch.add_link(tgen.gears["rt6"], nodeif="eth-rt4")
switch = tgen.add_switch('s8')
switch.add_link(tgen.gears['rt5'], nodeif="eth-rt6")
switch.add_link(tgen.gears['rt6'], nodeif="eth-rt5")
switch = tgen.add_switch("s8")
switch.add_link(tgen.gears["rt5"], nodeif="eth-rt6")
switch.add_link(tgen.gears["rt6"], nodeif="eth-rt5")
def setup_module(mod):
@ -138,16 +140,15 @@ def setup_module(mod):
# For all registered routers, load the zebra configuration file
for rname, router in router_list.items():
router.load_config(
TopoRouter.RD_ZEBRA,
os.path.join(CWD, '{}/zebra.conf'.format(rname))
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
)
router.load_config(
TopoRouter.RD_ISIS,
os.path.join(CWD, '{}/isisd.conf'.format(rname))
TopoRouter.RD_ISIS, os.path.join(CWD, "{}/isisd.conf".format(rname))
)
tgen.start_router()
def teardown_module(mod):
"Teardown the pytest environment"
tgen = get_topogen()
@ -155,6 +156,7 @@ def teardown_module(mod):
# This function tears down the whole topology.
tgen.stop_topology()
def router_compare_json_output(rname, command, reference):
"Compare router JSON output"
@ -170,6 +172,7 @@ def router_compare_json_output(rname, command, reference):
assertmsg = '"{}" JSON output mismatches the expected result'.format(rname)
assert diff is None, assertmsg
#
# Step 1
#
@ -184,13 +187,14 @@ def test_isis_adjacencies_step1():
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
for rname in ['rt1', 'rt2', 'rt3', 'rt4', 'rt5', 'rt6']:
for rname in ["rt1", "rt2", "rt3", "rt4", "rt5", "rt6"]:
router_compare_json_output(
rname,
rname,
"show yang operational-data /frr-interface:lib isisd",
"step1/show_yang_interface_isis_adjacencies.ref",
)
def test_rib_ipv4_step1():
logger.info("Test (step 1): verify IPv4 RIB")
tgen = get_topogen()
@ -199,11 +203,12 @@ def test_rib_ipv4_step1():
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
for rname in ['rt1', 'rt2', 'rt3', 'rt4', 'rt5', 'rt6']:
for rname in ["rt1", "rt2", "rt3", "rt4", "rt5", "rt6"]:
router_compare_json_output(
rname, "show ip route isis json", "step1/show_ip_route.ref"
)
def test_rib_ipv6_step1():
logger.info("Test (step 1): verify IPv6 RIB")
tgen = get_topogen()
@ -212,11 +217,12 @@ def test_rib_ipv6_step1():
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
for rname in ['rt1', 'rt2', 'rt3', 'rt4', 'rt5', 'rt6']:
for rname in ["rt1", "rt2", "rt3", "rt4", "rt5", "rt6"]:
router_compare_json_output(
rname, "show ipv6 route isis json", "step1/show_ipv6_route.ref"
)
#
# Step 2
#
@ -235,15 +241,20 @@ def test_rib_ipv4_step2():
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
logger.info('Disabling setting the attached-bit on RT2 and RT4')
tgen.net['rt2'].cmd('vtysh -c "conf t" -c "router isis 1" -c "no attached-bit send"')
tgen.net['rt4'].cmd('vtysh -c "conf t" -c "router isis 1" -c "no attached-bit send"')
logger.info("Disabling setting the attached-bit on RT2 and RT4")
tgen.net["rt2"].cmd(
'vtysh -c "conf t" -c "router isis 1" -c "no attached-bit send"'
)
tgen.net["rt4"].cmd(
'vtysh -c "conf t" -c "router isis 1" -c "no attached-bit send"'
)
for rname in ['rt1', 'rt6']:
for rname in ["rt1", "rt6"]:
router_compare_json_output(
rname, "show ip route isis json", "step2/show_ip_route.ref"
)
def test_rib_ipv6_step2():
logger.info("Test (step 2): verify IPv6 RIB")
tgen = get_topogen()
@ -252,11 +263,12 @@ def test_rib_ipv6_step2():
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
for rname in ['rt1', 'rt6']:
for rname in ["rt1", "rt6"]:
router_compare_json_output(
rname, "show ipv6 route isis json", "step2/show_ipv6_route.ref"
)
#
# Step 3
#
@ -265,7 +277,7 @@ def test_rib_ipv6_step2():
# -disble processing a LSP with attach bit set
#
# Expected changes:
# -RT1 and RT6 should not install a default route
# -RT1 and RT6 should not install a default route
#
def test_rib_ipv4_step3():
logger.info("Test (step 3): verify IPv4 RIB")
@ -275,19 +287,24 @@ def test_rib_ipv4_step3():
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
logger.info('Enable setting the attached-bit on RT2 and RT4')
tgen.net['rt2'].cmd('vtysh -c "conf t" -c "router isis 1" -c "attached-bit send"')
tgen.net['rt4'].cmd('vtysh -c "conf t" -c "router isis 1" -c "attached-bit send"')
logger.info("Enable setting the attached-bit on RT2 and RT4")
tgen.net["rt2"].cmd('vtysh -c "conf t" -c "router isis 1" -c "attached-bit send"')
tgen.net["rt4"].cmd('vtysh -c "conf t" -c "router isis 1" -c "attached-bit send"')
logger.info('Disable processing received attached-bit in LSP on RT1 and RT6')
tgen.net['rt1'].cmd('vtysh -c "conf t" -c "router isis 1" -c "attached-bit receive ignore"')
tgen.net['rt6'].cmd('vtysh -c "conf t" -c "router isis 1" -c "attached-bit receive ignore"')
logger.info("Disable processing received attached-bit in LSP on RT1 and RT6")
tgen.net["rt1"].cmd(
'vtysh -c "conf t" -c "router isis 1" -c "attached-bit receive ignore"'
)
tgen.net["rt6"].cmd(
'vtysh -c "conf t" -c "router isis 1" -c "attached-bit receive ignore"'
)
for rname in ['rt1', 'rt6']:
for rname in ["rt1", "rt6"]:
router_compare_json_output(
rname, "show ip route isis json", "step3/show_ip_route.ref"
)
def test_rib_ipv6_step3():
logger.info("Test (step 3): verify IPv6 RIB")
tgen = get_topogen()
@ -296,11 +313,12 @@ def test_rib_ipv6_step3():
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
for rname in ['rt1', 'rt6']:
for rname in ["rt1", "rt6"]:
router_compare_json_output(
rname, "show ipv6 route isis json", "step3/show_ipv6_route.ref"
)
#
# Step 4
#
@ -319,13 +337,21 @@ def test_rib_ipv4_step4():
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
logger.info('restore default processing on received attached-bit in LSP on RT1 and RT6')
tgen.net['rt1'].cmd('vtysh -c "conf t" -c "router isis 1" -c "no attached-bit receive ignore"')
tgen.net['rt6'].cmd('vtysh -c "conf t" -c "router isis 1" -c "no attached-bit receive ignore"')
logger.info(
"restore default processing on received attached-bit in LSP on RT1 and RT6"
)
tgen.net["rt1"].cmd(
'vtysh -c "conf t" -c "router isis 1" -c "no attached-bit receive ignore"'
)
tgen.net["rt6"].cmd(
'vtysh -c "conf t" -c "router isis 1" -c "no attached-bit receive ignore"'
)
for rname in ['rt1', 'rt6']:
for rname in ["rt1", "rt6"]:
router_compare_json_output(
rname, "show ip route isis json", "step4/show_ip_route.ref")
rname, "show ip route isis json", "step4/show_ip_route.ref"
)
def test_rib_ipv6_step4():
logger.info("Test (step 4): verify IPv6 RIB")
@ -335,19 +361,22 @@ def test_rib_ipv6_step4():
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
for rname in ['rt1', 'rt6']:
for rname in ["rt1", "rt6"]:
router_compare_json_output(
rname, "show ipv6 route isis json", "step4/show_ipv6_route.ref")
rname, "show ipv6 route isis json", "step4/show_ipv6_route.ref"
)
# Memory leak test template
def test_memory_leak():
"Run the memory leak test and report results."
tgen = get_topogen()
if not tgen.is_memleak_enabled():
pytest.skip('Memory leak test/report is disabled')
pytest.skip("Memory leak test/report is disabled")
tgen.report_memory_leaks()
if __name__ == '__main__':
if __name__ == "__main__":
args = ["-s"] + sys.argv[1:]
sys.exit(pytest.main(args))

View File

@ -167,6 +167,7 @@ class TemplateTopo(Topo):
f_in.close()
f_out.close()
def setup_module(mod):
"Sets up the pytest environment"
tgen = Topogen(TemplateTopo, mod.__name__)

View File

@ -85,7 +85,7 @@ from functools import partial
# Save the Current Working Directory to find configuration files.
CWD = os.path.dirname(os.path.realpath(__file__))
sys.path.append(os.path.join(CWD, '../'))
sys.path.append(os.path.join(CWD, "../"))
# pylint: disable=C0413
# Import topogen and topotest helpers
@ -101,6 +101,7 @@ pytestmark = [pytest.mark.bgpd, pytest.mark.isisd, pytest.mark.pathd]
class TemplateTopo(Topo):
"Test topology builder"
def build(self, *_args, **_opts):
"Build function"
tgen = get_topogen(self)
@ -108,48 +109,49 @@ class TemplateTopo(Topo):
#
# Define FRR Routers
#
for router in ['rt1', 'rt2', 'rt3', 'rt4', 'rt5', 'rt6', 'dst']:
for router in ["rt1", "rt2", "rt3", "rt4", "rt5", "rt6", "dst"]:
tgen.add_router(router)
#
# Define connections
#
switch = tgen.add_switch('s1')
switch.add_link(tgen.gears['rt1'], nodeif="eth-sw1")
switch.add_link(tgen.gears['rt2'], nodeif="eth-sw1")
switch.add_link(tgen.gears['rt3'], nodeif="eth-sw1")
switch = tgen.add_switch("s1")
switch.add_link(tgen.gears["rt1"], nodeif="eth-sw1")
switch.add_link(tgen.gears["rt2"], nodeif="eth-sw1")
switch.add_link(tgen.gears["rt3"], nodeif="eth-sw1")
switch = tgen.add_switch('s2')
switch.add_link(tgen.gears['rt2'], nodeif="eth-rt4-1")
switch.add_link(tgen.gears['rt4'], nodeif="eth-rt2-1")
switch = tgen.add_switch("s2")
switch.add_link(tgen.gears["rt2"], nodeif="eth-rt4-1")
switch.add_link(tgen.gears["rt4"], nodeif="eth-rt2-1")
switch = tgen.add_switch('s3')
switch.add_link(tgen.gears['rt2'], nodeif="eth-rt4-2")
switch.add_link(tgen.gears['rt4'], nodeif="eth-rt2-2")
switch = tgen.add_switch("s3")
switch.add_link(tgen.gears["rt2"], nodeif="eth-rt4-2")
switch.add_link(tgen.gears["rt4"], nodeif="eth-rt2-2")
switch = tgen.add_switch('s4')
switch.add_link(tgen.gears['rt3'], nodeif="eth-rt5-1")
switch.add_link(tgen.gears['rt5'], nodeif="eth-rt3-1")
switch = tgen.add_switch("s4")
switch.add_link(tgen.gears["rt3"], nodeif="eth-rt5-1")
switch.add_link(tgen.gears["rt5"], nodeif="eth-rt3-1")
switch = tgen.add_switch('s5')
switch.add_link(tgen.gears['rt3'], nodeif="eth-rt5-2")
switch.add_link(tgen.gears['rt5'], nodeif="eth-rt3-2")
switch = tgen.add_switch("s5")
switch.add_link(tgen.gears["rt3"], nodeif="eth-rt5-2")
switch.add_link(tgen.gears["rt5"], nodeif="eth-rt3-2")
switch = tgen.add_switch('s6')
switch.add_link(tgen.gears['rt4'], nodeif="eth-rt5")
switch.add_link(tgen.gears['rt5'], nodeif="eth-rt4")
switch = tgen.add_switch("s6")
switch.add_link(tgen.gears["rt4"], nodeif="eth-rt5")
switch.add_link(tgen.gears["rt5"], nodeif="eth-rt4")
switch = tgen.add_switch('s7')
switch.add_link(tgen.gears['rt4'], nodeif="eth-rt6")
switch.add_link(tgen.gears['rt6'], nodeif="eth-rt4")
switch = tgen.add_switch("s7")
switch.add_link(tgen.gears["rt4"], nodeif="eth-rt6")
switch.add_link(tgen.gears["rt6"], nodeif="eth-rt4")
switch = tgen.add_switch('s8')
switch.add_link(tgen.gears['rt5'], nodeif="eth-rt6")
switch.add_link(tgen.gears['rt6'], nodeif="eth-rt5")
switch = tgen.add_switch("s8")
switch.add_link(tgen.gears["rt5"], nodeif="eth-rt6")
switch.add_link(tgen.gears["rt6"], nodeif="eth-rt5")
switch = tgen.add_switch("s9")
switch.add_link(tgen.gears["rt6"], nodeif="eth-dst")
switch.add_link(tgen.gears["dst"], nodeif="eth-rt6")
switch = tgen.add_switch('s9')
switch.add_link(tgen.gears['rt6'], nodeif="eth-dst")
switch.add_link(tgen.gears['dst'], nodeif="eth-rt6")
def setup_module(mod):
"Sets up the pytest environment"
@ -167,24 +169,21 @@ def setup_module(mod):
# For all registered routers, load the zebra configuration file
for rname, router in router_list.iteritems():
router.load_config(
TopoRouter.RD_ZEBRA,
os.path.join(CWD, '{}/zebra.conf'.format(rname))
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
)
router.load_config(
TopoRouter.RD_ISIS,
os.path.join(CWD, '{}/isisd.conf'.format(rname))
TopoRouter.RD_ISIS, os.path.join(CWD, "{}/isisd.conf".format(rname))
)
router.load_config(
TopoRouter.RD_PATH,
os.path.join(CWD, '{}/pathd.conf'.format(rname))
TopoRouter.RD_PATH, os.path.join(CWD, "{}/pathd.conf".format(rname))
)
router.load_config(
TopoRouter.RD_BGP,
os.path.join(CWD, '{}/bgpd.conf'.format(rname))
TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
)
tgen.start_router()
def teardown_module(mod):
"Teardown the pytest environment"
tgen = get_topogen()
@ -192,6 +191,7 @@ def teardown_module(mod):
# This function tears down the whole topology.
tgen.stop_topology()
def setup_testcase(msg):
logger.info(msg)
tgen = get_topogen()
@ -202,9 +202,11 @@ def setup_testcase(msg):
return tgen
def print_cmd_result(rname, command):
print(get_topogen().gears[rname].vtysh_cmd(command, isjson=False))
def compare_json_test(router, command, reference, exact):
output = router.vtysh_cmd(command, isjson=True)
result = topotest.json_cmp(output, reference)
@ -212,9 +214,10 @@ def compare_json_test(router, command, reference, exact):
# Note: topotest.json_cmp() just checks on inclusion of keys.
# For exact matching also compare the other way around.
if not result and exact:
return topotest.json_cmp(reference, output)
return topotest.json_cmp(reference, output)
else:
return result
return result
def cmp_json_output(rname, command, reference, exact=False):
"Compare router JSON output"
@ -222,78 +225,136 @@ def cmp_json_output(rname, command, reference, exact=False):
logger.info('Comparing router "%s" "%s" output', rname, command)
tgen = get_topogen()
filename = '{}/{}/{}'.format(CWD, rname, reference)
filename = "{}/{}/{}".format(CWD, rname, reference)
expected = json.loads(open(filename).read())
# Run test function until we get an result. Wait at most 60 seconds.
test_func = partial(compare_json_test,
tgen.gears[rname], command, expected, exact)
test_func = partial(compare_json_test, tgen.gears[rname], command, expected, exact)
_, diff = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
assertmsg = '"{}" JSON output mismatches the expected result'.format(rname)
assert diff is None, assertmsg
def cmp_json_output_exact(rname, command, reference):
return cmp_json_output(rname, command, reference, True)
def add_candidate_path(rname, endpoint, pref, name, segment_list='default'):
get_topogen().net[rname].cmd(''' \
def add_candidate_path(rname, endpoint, pref, name, segment_list="default"):
get_topogen().net[rname].cmd(
""" \
vtysh -c "conf t" \
-c "segment-routing" \
-c "traffic-eng" \
-c "policy color 1 endpoint ''' + endpoint + '''" \
-c "candidate-path preference ''' + str(pref) + ''' name ''' + name + ''' explicit segment-list ''' + segment_list + '''"''')
-c "policy color 1 endpoint """
+ endpoint
+ """" \
-c "candidate-path preference """
+ str(pref)
+ """ name """
+ name
+ """ explicit segment-list """
+ segment_list
+ '''"'''
)
def delete_candidate_path(rname, endpoint, pref):
get_topogen().net[rname].cmd(''' \
get_topogen().net[rname].cmd(
""" \
vtysh -c "conf t" \
-c "segment-routing" \
-c "traffic-eng" \
-c "policy color 1 endpoint ''' + endpoint + '''" \
-c "no candidate-path preference ''' + str(pref) + '''"''')
-c "policy color 1 endpoint """
+ endpoint
+ """" \
-c "no candidate-path preference """
+ str(pref)
+ '''"'''
)
def add_segment(rname, name, index, label):
get_topogen().net[rname].cmd(''' \
get_topogen().net[rname].cmd(
""" \
vtysh -c "conf t" \
-c "segment-routing" \
-c "traffic-eng" \
-c "segment-list ''' + name + '''" \
-c "index ''' + str(index) + ''' mpls label ''' + str(label) + '''"''')
-c "segment-list """
+ name
+ """" \
-c "index """
+ str(index)
+ """ mpls label """
+ str(label)
+ '''"'''
)
def delete_segment(rname, name, index):
get_topogen().net[rname].cmd(''' \
get_topogen().net[rname].cmd(
""" \
vtysh -c "conf t" \
-c "segment-routing" \
-c "traffic-eng" \
-c "segment-list ''' + name + '''" \
-c "no index ''' + str(index) + '''"''')
-c "segment-list """
+ name
+ """" \
-c "no index """
+ str(index)
+ '''"'''
)
def create_sr_policy(rname, endpoint, bsid):
get_topogen().net[rname].cmd(''' \
get_topogen().net[rname].cmd(
""" \
vtysh -c "conf t" \
-c "segment-routing" \
-c "traffic-eng" \
-c "policy color 1 endpoint ''' + endpoint + '''" \
-c "policy color 1 endpoint """
+ endpoint
+ """" \
-c "name default" \
-c "binding-sid ''' + str(bsid) + '''"''')
-c "binding-sid """
+ str(bsid)
+ '''"'''
)
def delete_sr_policy(rname, endpoint):
get_topogen().net[rname].cmd(''' \
get_topogen().net[rname].cmd(
""" \
vtysh -c "conf t" \
-c "segment-routing" \
-c "traffic-eng" \
-c "no policy color 1 endpoint ''' + endpoint + '''"''')
-c "no policy color 1 endpoint """
+ endpoint
+ '''"'''
)
def create_prefix_sid(rname, prefix, sid):
get_topogen().net[rname].cmd(''' \
get_topogen().net[rname].cmd(
""" \
vtysh -c "conf t" \
-c "router isis 1" \
-c "segment-routing prefix ''' + prefix + " index " + str(sid) + '''"''')
-c "segment-routing prefix """
+ prefix
+ " index "
+ str(sid)
+ '''"'''
)
def delete_prefix_sid(rname, prefix):
get_topogen().net[rname].cmd(''' \
get_topogen().net[rname].cmd(
''' \
vtysh -c "conf t" \
-c "router isis 1" \
-c "no segment-routing prefix "''' + prefix)
-c "no segment-routing prefix "'''
+ prefix
)
#
# Step 1
@ -303,37 +364,42 @@ def delete_prefix_sid(rname, prefix):
def test_srte_init_step1():
setup_testcase("Test (step 1): wait for IS-IS convergence / label distribution")
for rname in ['rt1', 'rt6']:
cmp_json_output(rname,
"show mpls table json",
"step1/show_mpls_table_without_candidate.ref")
for rname in ["rt1", "rt6"]:
cmp_json_output(
rname, "show mpls table json", "step1/show_mpls_table_without_candidate.ref"
)
def test_srte_add_candidate_check_mpls_table_step1():
setup_testcase("Test (step 1): check MPLS table regarding the added Candidate Path")
for rname, endpoint in [('rt1', '6.6.6.6'), ('rt6', '1.1.1.1')]:
add_candidate_path(rname, endpoint, 100, 'default')
cmp_json_output(rname,
"show mpls table json",
"step1/show_mpls_table_with_candidate.ref")
for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
add_candidate_path(rname, endpoint, 100, "default")
cmp_json_output(
rname, "show mpls table json", "step1/show_mpls_table_with_candidate.ref"
)
delete_candidate_path(rname, endpoint, 100)
def test_srte_reinstall_sr_policy_check_mpls_table_step1():
setup_testcase("Test (step 1): check MPLS table after the SR Policy was removed and reinstalled")
setup_testcase(
"Test (step 1): check MPLS table after the SR Policy was removed and reinstalled"
)
for rname, endpoint, bsid in [('rt1', '6.6.6.6', 1111), ('rt6', '1.1.1.1', 6666)]:
add_candidate_path(rname, endpoint, 100, 'default')
for rname, endpoint, bsid in [("rt1", "6.6.6.6", 1111), ("rt6", "1.1.1.1", 6666)]:
add_candidate_path(rname, endpoint, 100, "default")
delete_sr_policy(rname, endpoint)
cmp_json_output(rname,
"show mpls table json",
"step1/show_mpls_table_without_candidate.ref")
cmp_json_output(
rname, "show mpls table json", "step1/show_mpls_table_without_candidate.ref"
)
create_sr_policy(rname, endpoint, bsid)
add_candidate_path(rname, endpoint, 100, 'default')
cmp_json_output(rname,
"show mpls table json",
"step1/show_mpls_table_with_candidate.ref")
add_candidate_path(rname, endpoint, 100, "default")
cmp_json_output(
rname, "show mpls table json", "step1/show_mpls_table_with_candidate.ref"
)
delete_candidate_path(rname, endpoint, 100)
#
# Step 2
#
@ -342,28 +408,41 @@ def test_srte_reinstall_sr_policy_check_mpls_table_step1():
def test_srte_bare_policy_step2():
setup_testcase("Test (step 2): bare SR Policy should not be operational")
for rname in ['rt1', 'rt6']:
cmp_json_output_exact(rname,
"show yang operational-data /frr-pathd:pathd pathd",
"step2/show_operational_data.ref")
for rname in ["rt1", "rt6"]:
cmp_json_output_exact(
rname,
"show yang operational-data /frr-pathd:pathd pathd",
"step2/show_operational_data.ref",
)
def test_srte_add_candidate_check_operational_data_step2():
setup_testcase("Test (step 2): add single Candidate Path, SR Policy should be operational")
setup_testcase(
"Test (step 2): add single Candidate Path, SR Policy should be operational"
)
for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
add_candidate_path(rname, endpoint, 100, "default")
cmp_json_output(
rname,
"show yang operational-data /frr-pathd:pathd pathd",
"step2/show_operational_data_with_candidate.ref",
)
for rname, endpoint in [('rt1', '6.6.6.6'), ('rt6', '1.1.1.1')]:
add_candidate_path(rname, endpoint, 100, 'default')
cmp_json_output(rname,
"show yang operational-data /frr-pathd:pathd pathd",
"step2/show_operational_data_with_candidate.ref")
def test_srte_config_remove_candidate_check_operational_data_step2():
setup_testcase("Test (step 2): remove single Candidate Path, SR Policy should not be operational anymore")
setup_testcase(
"Test (step 2): remove single Candidate Path, SR Policy should not be operational anymore"
)
for rname, endpoint in [('rt1', '6.6.6.6'), ('rt6', '1.1.1.1')]:
for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
delete_candidate_path(rname, endpoint, 100)
cmp_json_output_exact(rname,
"show yang operational-data /frr-pathd:pathd pathd",
"step2/show_operational_data.ref")
cmp_json_output_exact(
rname,
"show yang operational-data /frr-pathd:pathd pathd",
"step2/show_operational_data.ref",
)
#
# Step 3
@ -373,53 +452,62 @@ def test_srte_config_remove_candidate_check_operational_data_step2():
def test_srte_add_two_candidates_step3():
setup_testcase("Test (step 3): second Candidate Path has higher Priority")
for rname, endpoint in [('rt1', '6.6.6.6'), ('rt6', '1.1.1.1')]:
for pref, cand_name in [('100', 'first'), ('200', 'second')]:
for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
for pref, cand_name in [("100", "first"), ("200", "second")]:
add_candidate_path(rname, endpoint, pref, cand_name)
cmp_json_output(rname,
"show yang operational-data /frr-pathd:pathd pathd",
"step3/show_operational_data_with_two_candidates.ref")
cmp_json_output(
rname,
"show yang operational-data /frr-pathd:pathd pathd",
"step3/show_operational_data_with_two_candidates.ref",
)
# cleanup
for rname, endpoint in [('rt1', '6.6.6.6'), ('rt6', '1.1.1.1')]:
for pref in ['100', '200']:
for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
for pref in ["100", "200"]:
delete_candidate_path(rname, endpoint, pref)
def test_srte_add_two_candidates_with_reverse_priority_step3():
setup_testcase("Test (step 3): second Candidate Path has lower Priority")
# Use reversed priorities here
for rname, endpoint in [('rt1', '6.6.6.6'), ('rt6', '1.1.1.1')]:
for pref, cand_name in [('200', 'first'), ('100', 'second')]:
for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
for pref, cand_name in [("200", "first"), ("100", "second")]:
add_candidate_path(rname, endpoint, pref, cand_name)
cmp_json_output(rname,
"show yang operational-data /frr-pathd:pathd pathd",
"step3/show_operational_data_with_two_candidates.ref")
cmp_json_output(
rname,
"show yang operational-data /frr-pathd:pathd pathd",
"step3/show_operational_data_with_two_candidates.ref",
)
# cleanup
for rname, endpoint in [('rt1', '6.6.6.6'), ('rt6', '1.1.1.1')]:
for pref in ['100', '200']:
for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
for pref in ["100", "200"]:
delete_candidate_path(rname, endpoint, pref)
def test_srte_remove_best_candidate_step3():
setup_testcase("Test (step 3): delete the Candidate Path with higher priority")
for rname, endpoint in [('rt1', '6.6.6.6'), ('rt6', '1.1.1.1')]:
for pref, cand_name in [('100', 'first'), ('200', 'second')]:
for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
for pref, cand_name in [("100", "first"), ("200", "second")]:
add_candidate_path(rname, endpoint, pref, cand_name)
# Delete candidate with higher priority
for rname, endpoint in [('rt1', '6.6.6.6'), ('rt6', '1.1.1.1')]:
for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
delete_candidate_path(rname, endpoint, 200)
# Candidate with lower priority should get active now
for rname, endpoint in [('rt1', '6.6.6.6'), ('rt6', '1.1.1.1')]:
cmp_json_output(rname,
"show yang operational-data /frr-pathd:pathd pathd",
"step3/show_operational_data_with_single_candidate.ref")
for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
cmp_json_output(
rname,
"show yang operational-data /frr-pathd:pathd pathd",
"step3/show_operational_data_with_single_candidate.ref",
)
# cleanup
delete_candidate_path(rname, endpoint, 100)
#
# Step 4
#
@ -428,38 +516,38 @@ def test_srte_remove_best_candidate_step3():
def test_srte_change_segment_list_check_mpls_table_step4():
setup_testcase("Test (step 4): check MPLS table for changed Segment List")
for rname, endpoint in [('rt1', '6.6.6.6'), ('rt6', '1.1.1.1')]:
add_candidate_path(rname, endpoint, 100, 'default')
# now change the segment list name
add_candidate_path(rname, endpoint, 100, 'default', 'test')
cmp_json_output(rname,
"show mpls table json",
"step4/show_mpls_table.ref")
for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
add_candidate_path(rname, endpoint, 100, "default")
# now change the segment list name
add_candidate_path(rname, endpoint, 100, "default", "test")
cmp_json_output(rname, "show mpls table json", "step4/show_mpls_table.ref")
delete_candidate_path(rname, endpoint, 100)
def test_srte_segment_list_add_segment_check_mpls_table_step4():
setup_testcase("Test (step 4): check MPLS table for added (then changed and finally deleted) segment")
add_candidate_path('rt1', '6.6.6.6', 100, 'default', 'test')
def test_srte_segment_list_add_segment_check_mpls_table_step4():
setup_testcase(
"Test (step 4): check MPLS table for added (then changed and finally deleted) segment"
)
add_candidate_path("rt1", "6.6.6.6", 100, "default", "test")
# first add a new segment
add_segment('rt1', 'test', 25, 16050)
cmp_json_output('rt1',
"show mpls table json",
"step4/show_mpls_table_add_segment.ref")
add_segment("rt1", "test", 25, 16050)
cmp_json_output(
"rt1", "show mpls table json", "step4/show_mpls_table_add_segment.ref"
)
# ... then change it ...
add_segment('rt1', 'test', 25, 16030)
cmp_json_output('rt1',
"show mpls table json",
"step4/show_mpls_table_change_segment.ref")
add_segment("rt1", "test", 25, 16030)
cmp_json_output(
"rt1", "show mpls table json", "step4/show_mpls_table_change_segment.ref"
)
# ... and finally delete it
delete_segment('rt1', 'test', 25)
cmp_json_output('rt1',
"show mpls table json",
"step4/show_mpls_table.ref")
delete_candidate_path('rt1', '6.6.6.6', 100)
delete_segment("rt1", "test", 25)
cmp_json_output("rt1", "show mpls table json", "step4/show_mpls_table.ref")
delete_candidate_path("rt1", "6.6.6.6", 100)
#
# Step 5
@ -467,68 +555,81 @@ def test_srte_segment_list_add_segment_check_mpls_table_step4():
# Checking the nexthop using a single SR Policy and a Candidate Path with configured route-map
#
def test_srte_route_map_with_sr_policy_check_nextop_step5():
setup_testcase("Test (step 5): recursive nexthop learned through BGP neighbour should be aligned with SR Policy from route-map")
setup_testcase(
"Test (step 5): recursive nexthop learned through BGP neighbour should be aligned with SR Policy from route-map"
)
# (re-)build the SR Policy two times to ensure that reinstalling still works
for i in [1,2]:
cmp_json_output('rt1',
"show ip route bgp json",
"step5/show_ip_route_bgp_inactive_srte.ref")
for i in [1, 2]:
cmp_json_output(
"rt1", "show ip route bgp json", "step5/show_ip_route_bgp_inactive_srte.ref"
)
delete_sr_policy('rt1', '6.6.6.6')
cmp_json_output('rt1',
"show ip route bgp json",
"step5/show_ip_route_bgp_inactive_srte.ref")
delete_sr_policy("rt1", "6.6.6.6")
cmp_json_output(
"rt1", "show ip route bgp json", "step5/show_ip_route_bgp_inactive_srte.ref"
)
create_sr_policy('rt1', '6.6.6.6', 1111)
cmp_json_output('rt1',
"show ip route bgp json",
"step5/show_ip_route_bgp_inactive_srte.ref")
create_sr_policy("rt1", "6.6.6.6", 1111)
cmp_json_output(
"rt1", "show ip route bgp json", "step5/show_ip_route_bgp_inactive_srte.ref"
)
add_candidate_path('rt1', '6.6.6.6', 100, 'default')
cmp_json_output('rt1',
"show ip route bgp json",
"step5/show_ip_route_bgp_active_srte.ref")
add_candidate_path("rt1", "6.6.6.6", 100, "default")
cmp_json_output(
"rt1", "show ip route bgp json", "step5/show_ip_route_bgp_active_srte.ref"
)
delete_candidate_path("rt1", "6.6.6.6", 100)
delete_candidate_path('rt1', '6.6.6.6', 100)
def test_srte_route_map_with_sr_policy_reinstall_prefix_sid_check_nextop_step5():
setup_testcase("Test (step 5): remove and re-install prefix SID on fist path element and check SR Policy activity")
setup_testcase(
"Test (step 5): remove and re-install prefix SID on fist path element and check SR Policy activity"
)
# first add a candidate path so the SR Policy is active
add_candidate_path('rt1', '6.6.6.6', 100, 'default')
cmp_json_output('rt1',
"show yang operational-data /frr-pathd:pathd pathd",
"step5/show_operational_data_active.ref")
add_candidate_path("rt1", "6.6.6.6", 100, "default")
cmp_json_output(
"rt1",
"show yang operational-data /frr-pathd:pathd pathd",
"step5/show_operational_data_active.ref",
)
# delete prefix SID from first element of the configured path and check
# if the SR Policy is inactive since the label can't be resolved anymore
delete_prefix_sid('rt5', "5.5.5.5/32")
cmp_json_output('rt1',
"show yang operational-data /frr-pathd:pathd pathd",
"step5/show_operational_data_inactive.ref")
cmp_json_output('rt1',
"show ip route bgp json",
"step5/show_ip_route_bgp_inactive_srte.ref")
delete_prefix_sid("rt5", "5.5.5.5/32")
cmp_json_output(
"rt1",
"show yang operational-data /frr-pathd:pathd pathd",
"step5/show_operational_data_inactive.ref",
)
cmp_json_output(
"rt1", "show ip route bgp json", "step5/show_ip_route_bgp_inactive_srte.ref"
)
# re-create the prefix SID and check if the SR Policy is active
create_prefix_sid('rt5', "5.5.5.5/32", 50)
cmp_json_output('rt1',
"show yang operational-data /frr-pathd:pathd pathd",
"step5/show_operational_data_active.ref")
cmp_json_output('rt1',
"show ip route bgp json",
"step5/show_ip_route_bgp_active_srte.ref")
create_prefix_sid("rt5", "5.5.5.5/32", 50)
cmp_json_output(
"rt1",
"show yang operational-data /frr-pathd:pathd pathd",
"step5/show_operational_data_active.ref",
)
cmp_json_output(
"rt1", "show ip route bgp json", "step5/show_ip_route_bgp_active_srte.ref"
)
# Memory leak test template
def test_memory_leak():
"Run the memory leak test and report results."
tgen = get_topogen()
if not tgen.is_memleak_enabled():
pytest.skip('Memory leak test/report is disabled')
pytest.skip("Memory leak test/report is disabled")
tgen.report_memory_leaks()
if __name__ == '__main__':
if __name__ == "__main__":
args = ["-s"] + sys.argv[1:]
sys.exit(pytest.main(args))

View File

@ -86,6 +86,7 @@ from mininet.topo import Topo
pytestmark = [pytest.mark.isisd]
class TemplateTopo(Topo):
"Test topology builder"

View File

@ -179,6 +179,7 @@ class TemplateTopo(Topo):
f_in.close()
f_out.close()
def setup_module(mod):
"Sets up the pytest environment"
tgen = Topogen(TemplateTopo, mod.__name__)

View File

@ -45,6 +45,7 @@ from mininet.topo import Topo
pytestmark = [pytest.mark.isisd]
class ISISTopo1(Topo):
"Simple two layer ISIS vrf topology"

View File

@ -47,6 +47,7 @@ from mininet.topo import Topo
pytestmark = [pytest.mark.isisd]
class ISISTopo1(Topo):
"Simple two layer ISIS topology"

View File

@ -80,6 +80,7 @@ from mininet.topo import Topo
pytestmark = [pytest.mark.ldpd, pytest.mark.ospfd]
class TemplateTopo(Topo):
"Test topology builder"

View File

@ -80,6 +80,7 @@ from mininet.topo import Topo
pytestmark = [pytest.mark.ldpd, pytest.mark.ospfd]
class TemplateTopo(Topo):
"Test topology builder"

View File

@ -81,6 +81,7 @@ from mininet.topo import Topo
pytestmark = [pytest.mark.ldpd, pytest.mark.ospfd]
class TemplateTopo(Topo):
"Test topology builder"

View File

@ -82,6 +82,7 @@ from mininet.topo import Topo
pytestmark = [pytest.mark.ldpd, pytest.mark.ospfd]
class TemplateTopo(Topo):
"Test topology builder"

View File

@ -2516,8 +2516,9 @@ def verify_best_path_as_per_admin_distance(
@retry(attempts=5, wait=2, return_is_str=True, initial_wait=2)
def verify_bgp_rib(tgen, addr_type, dut, input_dict, next_hop=None,
aspath=None, multi_nh=None):
def verify_bgp_rib(
tgen, addr_type, dut, input_dict, next_hop=None, aspath=None, multi_nh=None
):
"""
This API is to verify whether bgp rib has any
matching route for a nexthop.
@ -2663,25 +2664,35 @@ aspath=None, multi_nh=None):
if not isinstance(next_hop, list):
next_hop = [next_hop]
list1 = next_hop
found_hops = [rib_r["ip"] for rib_r in
rib_routes_json["routes"][
st_rt][0]["nexthops"]]
found_hops = [
rib_r["ip"]
for rib_r in rib_routes_json["routes"][st_rt][0][
"nexthops"
]
]
list2 = found_hops
missing_list_of_nexthops = \
set(list2).difference(list1)
additional_nexthops_in_required_nhs = \
set(list1).difference(list2)
missing_list_of_nexthops = set(list2).difference(list1)
additional_nexthops_in_required_nhs = set(
list1
).difference(list2)
if list2:
if additional_nexthops_in_required_nhs:
logger.info("Missing nexthop %s for route"\
" %s in RIB of router %s\n", \
additional_nexthops_in_required_nhs, \
st_rt, dut)
errormsg=("Nexthop {} is Missing for "\
"route {} in RIB of router {}\n".format(
logger.info(
"Missing nexthop %s for route"
" %s in RIB of router %s\n",
additional_nexthops_in_required_nhs,
st_rt, dut))
st_rt,
dut,
)
errormsg = (
"Nexthop {} is Missing for "
"route {} in RIB of router {}\n".format(
additional_nexthops_in_required_nhs,
st_rt,
dut,
)
)
return errormsg
else:
nh_found = True

View File

@ -94,7 +94,9 @@ def create_router_ospf(tgen, topo, input_dict=None, build=False, load_config=Tru
return result
def __create_ospf_global(tgen, input_dict, router, build=False, load_config=True, ospf="ospf"):
def __create_ospf_global(
tgen, input_dict, router, build=False, load_config=True, ospf="ospf"
):
"""
Helper API to create ospf global configuration.

View File

@ -1564,26 +1564,30 @@ def verify_pim_interface(tgen, topo, dut, interface=None, interface_ip=None):
logger.info("[DUT: %s]: Verifying PIM interface status:", dut)
rnode = tgen.routers()[dut]
show_ip_pim_interface_json = rnode.\
vtysh_cmd("show ip pim interface json", isjson=True)
show_ip_pim_interface_json = rnode.vtysh_cmd(
"show ip pim interface json", isjson=True
)
logger.info("show_ip_pim_interface_json: \n %s",
show_ip_pim_interface_json)
logger.info("show_ip_pim_interface_json: \n %s", show_ip_pim_interface_json)
if interface_ip:
if interface in show_ip_pim_interface_json:
pim_intf_json = show_ip_pim_interface_json[interface]
if pim_intf_json["address"] != interface_ip:
errormsg = ("[DUT %s]: PIM interface "
"ip is not correct "
"[FAILED]!! Expected : %s, Found : %s"
%(dut, pim_intf_json["address"],interface_ip))
errormsg = (
"[DUT %s]: PIM interface "
"ip is not correct "
"[FAILED]!! Expected : %s, Found : %s"
% (dut, pim_intf_json["address"], interface_ip)
)
return errormsg
else:
logger.info("[DUT %s]: PIM interface "
"ip is correct "
"[Passed]!! Expected : %s, Found : %s"
%(dut, pim_intf_json["address"],interface_ip))
logger.info(
"[DUT %s]: PIM interface "
"ip is correct "
"[Passed]!! Expected : %s, Found : %s"
% (dut, pim_intf_json["address"], interface_ip)
)
return True
else:
for destLink, data in topo["routers"][dut]["links"].items():
@ -1595,24 +1599,36 @@ def verify_pim_interface(tgen, topo, dut, interface=None, interface_ip=None):
pim_intf_ip = data["ipv4"].split("/")[0]
if pim_interface in show_ip_pim_interface_json:
pim_intf_json = show_ip_pim_interface_json\
[pim_interface]
pim_intf_json = show_ip_pim_interface_json[pim_interface]
# Verifying PIM interface
if pim_intf_json["address"] != pim_intf_ip and \
pim_intf_json["state"] != "up":
errormsg = ("[DUT %s]: PIM interface: %s "
"PIM interface ip: %s, status check "
"[FAILED]!! Expected : %s, Found : %s"
%(dut, pim_interface, pim_intf_ip,
pim_interface, pim_intf_json["state"]))
if (
pim_intf_json["address"] != pim_intf_ip
and pim_intf_json["state"] != "up"
):
errormsg = (
"[DUT %s]: PIM interface: %s "
"PIM interface ip: %s, status check "
"[FAILED]!! Expected : %s, Found : %s"
% (
dut,
pim_interface,
pim_intf_ip,
pim_interface,
pim_intf_json["state"],
)
)
return errormsg
logger.info("[DUT %s]: PIM interface: %s, "
"interface ip: %s, status: %s"
" [PASSED]!!",
dut, pim_interface, pim_intf_ip,
pim_intf_json["state"])
logger.info(
"[DUT %s]: PIM interface: %s, "
"interface ip: %s, status: %s"
" [PASSED]!!",
dut,
pim_interface,
pim_intf_ip,
pim_intf_json["state"],
)
logger.debug("Exiting lib API: {}".format(sys._getframe().f_code.co_name))
return True
@ -3420,30 +3436,36 @@ def verify_igmp_interface(tgen, topo, dut, igmp_iface, interface_ip):
if router != dut:
continue
logger.info("[DUT: %s]: Verifying PIM interface status:",
dut)
logger.info("[DUT: %s]: Verifying PIM interface status:", dut)
rnode = tgen.routers()[dut]
show_ip_igmp_interface_json = \
run_frr_cmd(rnode, "show ip igmp interface json", isjson=True)
show_ip_igmp_interface_json = run_frr_cmd(
rnode, "show ip igmp interface json", isjson=True
)
if igmp_iface in show_ip_igmp_interface_json:
if igmp_iface in show_ip_igmp_interface_json:
igmp_intf_json = show_ip_igmp_interface_json[igmp_iface]
# Verifying igmp interface
if igmp_intf_json["address"] != interface_ip:
errormsg = ("[DUT %s]: igmp interface ip is not correct "
"[FAILED]!! Expected : %s, Found : %s"
%(dut, igmp_intf_json["address"], interface_ip))
if igmp_intf_json["address"] != interface_ip:
errormsg = (
"[DUT %s]: igmp interface ip is not correct "
"[FAILED]!! Expected : %s, Found : %s"
% (dut, igmp_intf_json["address"], interface_ip)
)
return errormsg
logger.info("[DUT %s]: igmp interface: %s, "
"interface ip: %s"
" [PASSED]!!",
dut, igmp_iface, interface_ip)
logger.info(
"[DUT %s]: igmp interface: %s, " "interface ip: %s" " [PASSED]!!",
dut,
igmp_iface,
interface_ip,
)
else:
errormsg = ("[DUT %s]: igmp interface: %s "
"igmp interface ip: %s, is not present "
%(dut, igmp_iface, interface_ip))
errormsg = (
"[DUT %s]: igmp interface: %s "
"igmp interface ip: %s, is not present "
% (dut, igmp_iface, interface_ip)
)
return errormsg
logger.debug("Exiting lib API: {}".format(sys._getframe().f_code.co_name))

View File

@ -86,12 +86,11 @@ class SnmpTester(object):
def _get_snmp_oid(snmp_output):
tokens = snmp_output.strip().split()
# if len(tokens) > 5:
# return None
# if len(tokens) > 5:
# return None
# third token is the value of the object
return tokens[0].split('.',1)[1]
return tokens[0].split(".", 1)[1]
def _parse_multiline(self, snmp_output):
results = snmp_output.strip().split("\r\n")
@ -142,7 +141,11 @@ class SnmpTester(object):
print("FAIL: missing oid key {}".format(oid))
return False
if results_dict[oid] != values[index]:
print("FAIL{} {} |{}| == |{}|".format(oid, index, results_dict[oid], values[index]))
print(
"FAIL{} {} |{}| == |{}|".format(
oid, index, results_dict[oid], values[index]
)
)
return False
index += 1
return True

View File

@ -706,7 +706,7 @@ def test_BSR_CRP_with_blackhole_address_p1(request):
"f1": {
"static_routes": [
{"network": [BSR1_ADDR, CRP], "next_hop": "blackhole", "delete": True},
{"network": BSR1_ADDR, "next_hop": NEXT_HOP1}
{"network": BSR1_ADDR, "next_hop": NEXT_HOP1},
]
}
}
@ -1118,7 +1118,10 @@ def test_static_rp_override_p1(request):
"l1": {
"pim": {
"rp": [
{"rp_addr": "33.33.33.33", "group_addr_range": ["225.1.1.1/32"],}
{
"rp_addr": "33.33.33.33",
"group_addr_range": ["225.1.1.1/32"],
}
]
}
}
@ -1294,7 +1297,8 @@ def test_bsmp_stress_add_del_restart_p2(request):
assert (
rp_add1 == rp2[group]
), "Testcase {} :Failed \n Error : rp expected {} rp received {}".format(
tc_name, rp_add1,
tc_name,
rp_add1,
)
# Verify if that rp is installed
@ -1632,7 +1636,14 @@ def test_iif_join_state_p0(request):
# Add back route for RP to make it reachable
step("Add back route for RP to make it reachable")
input_dict = {
"l1": {"static_routes": [{"network": rp_ip, "next_hop": next_hop_lhr,}]}
"l1": {
"static_routes": [
{
"network": rp_ip,
"next_hop": next_hop_lhr,
}
]
}
}
result = create_static_routes(tgen, input_dict)
assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)

View File

@ -695,7 +695,8 @@ def test_RP_priority_p0(request):
assert (
rp_add1 == rp2[group]
), "Testcase {} :Failed \n Error : rp expected {} rp received {}".format(
tc_name, rp_add1,
tc_name,
rp_add1,
)
# Verify if that rp is installed

View File

@ -102,7 +102,7 @@ from lib.pim import (
clear_ip_mroute,
clear_ip_pim_interface_traffic,
verify_igmp_config,
clear_ip_mroute_verify
clear_ip_mroute_verify,
)
from lib.topolog import logger
from lib.topojson import build_topo_from_json, build_config_from_json

View File

@ -877,7 +877,7 @@ def test_verify_SPT_switchover_when_RPT_and_SPT_path_is_different_p0(request):
data["src_address"],
_IGMP_JOIN_RANGE,
data["iif"],
data["oil"]
data["oil"],
)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
@ -1122,8 +1122,9 @@ def test_verify_mroute_after_shut_noshut_of_upstream_interface_p1(request):
done_flag = False
for retry in range(1, 11):
result = verify_upstream_iif(tgen, "l1", "Unknown", source, IGMP_JOIN_RANGE_2,
expected=False)
result = verify_upstream_iif(
tgen, "l1", "Unknown", source, IGMP_JOIN_RANGE_2, expected=False
)
if result is not True:
done_flag = True
else:
@ -1515,7 +1516,7 @@ def test_verify_mroute_when_FRR_is_FHR_and_LHR_p0(request):
_IGMP_JOIN_RANGE,
data["iif"],
data["oil"],
expected=False
expected=False,
)
if result is not True:
done_flag = True
@ -1928,9 +1929,10 @@ def test_verify_oil_iif_for_mroute_after_shut_noshut_source_interface_p1(request
"f1-i8-eth2",
expected=False,
)
assert result is not True, (
"Testcase {} : Failed \n mroutes are"
" still present \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n mroutes are" " still present \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behavior: {}".format(result))

View File

@ -597,9 +597,10 @@ def test_verify_oil_when_join_prune_sent_scenario_1_p1(request):
input_traffic = {"l1": {"traffic_sent": [intf_l1_i1]}}
result = verify_multicast_traffic(tgen, input_traffic, expected=False)
assert result is not True, (
"Testcase {} : Failed \n "
" Traffic is not stopped yet \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n " " Traffic is not stopped yet \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -612,9 +613,10 @@ def test_verify_oil_when_join_prune_sent_scenario_1_p1(request):
result = verify_igmp_groups(
tgen, dut, intf_l1_i1, IGMP_JOIN_RANGE_1, expected=False
)
assert result is not True, (
"Testcase {} : Failed \n "
"IGMP groups are not deleted \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n " "IGMP groups are not deleted \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -655,9 +657,10 @@ def test_verify_oil_when_join_prune_sent_scenario_1_p1(request):
data["oil"],
expected=False,
)
assert result is not True, (
"Testcase {} : Failed \n "
"mroutes are still present \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n " "mroutes are still present \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -722,9 +725,10 @@ def test_verify_oil_when_join_prune_sent_scenario_1_p1(request):
input_traffic = {"f1": {"traffic_sent": [intf_f1_i8]}}
result = verify_multicast_traffic(tgen, input_traffic, expected=False)
assert result is not True, (
"Testcase {} : Failed \n "
" Traffic is not stopped yet \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n " " Traffic is not stopped yet \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -737,9 +741,10 @@ def test_verify_oil_when_join_prune_sent_scenario_1_p1(request):
result = verify_igmp_groups(
tgen, dut, intf_f1_i8, IGMP_JOIN_RANGE_1, expected=False
)
assert result is not True, (
"Testcase {} : Failed \n "
"IGMP groups are not deleted \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n " "IGMP groups are not deleted \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -775,9 +780,10 @@ def test_verify_oil_when_join_prune_sent_scenario_1_p1(request):
data["oil"],
expected=False,
)
assert result is not True, (
"Testcase {} : Failed \n "
"mroutes are still present \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n " "mroutes are still present \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -959,9 +965,10 @@ def test_verify_oil_when_join_prune_sent_scenario_2_p1(request):
data["oil"],
expected=False,
)
assert result is not True, (
"Testcase {} : Failed \n "
"mroutes are still present \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n " "mroutes are still present \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -1022,9 +1029,10 @@ def test_verify_oil_when_join_prune_sent_scenario_2_p1(request):
data["oil"],
expected=False,
)
assert result is not True, (
"Testcase {} : Failed \n "
"mroutes are still present \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n " "mroutes are still present \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -1190,9 +1198,10 @@ def test_shut_noshut_source_interface_when_upstream_cleared_from_LHR_p1(request)
result = verify_ip_mroutes(
tgen, "f1", source_i2, IGMP_JOIN_RANGE_1, intf_f1_i2, intf_f1_r2, expected=False
)
assert result is not True, (
"Testcase {} : Failed \n mroutes are"
" still present \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n mroutes are" " still present \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behavior: {}".format(result))
@ -1630,7 +1639,14 @@ def test_verify_remove_add_igmp_config_to_receiver_interface_p0(request):
input_dict_2 = {
"l1": {
"igmp": {
"interfaces": {intf_l1_i1: {"igmp": {"version": "2", "delete": True,}}}
"interfaces": {
intf_l1_i1: {
"igmp": {
"version": "2",
"delete": True,
}
}
}
}
}
}
@ -1642,9 +1658,10 @@ def test_verify_remove_add_igmp_config_to_receiver_interface_p0(request):
dut = "l1"
interface = topo["routers"]["l1"]["links"]["i1"]["interface"]
result = verify_igmp_groups(tgen, dut, interface, IGMP_JOIN_RANGE_1, expected=False)
assert result is not True, (
"Testcase {} : Failed \n Groups are not"
" present \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n Groups are not" " present \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -1712,7 +1729,14 @@ def test_verify_remove_add_igmp_config_to_receiver_interface_p0(request):
input_dict_2 = {
"l1": {
"igmp": {
"interfaces": {intf_l1_i1: {"igmp": {"version": "2", "delete": True,}}}
"interfaces": {
intf_l1_i1: {
"igmp": {
"version": "2",
"delete": True,
}
}
}
}
}
}
@ -1725,9 +1749,10 @@ def test_verify_remove_add_igmp_config_to_receiver_interface_p0(request):
dut = "l1"
interface = topo["routers"]["l1"]["links"]["i1"]["interface"]
result = verify_igmp_groups(tgen, dut, interface, IGMP_JOIN_RANGE_1, expected=False)
assert result is not True, (
"Testcase {} : Failed \n Groups are not"
" present \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n Groups are not" " present \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -1811,7 +1836,14 @@ def test_verify_remove_add_igmp_config_to_receiver_interface_p0(request):
input_dict_2 = {
"l1": {
"igmp": {
"interfaces": {intf_l1_i1: {"igmp": {"version": "2", "delete": True,}}}
"interfaces": {
intf_l1_i1: {
"igmp": {
"version": "2",
"delete": True,
}
}
}
}
}
}
@ -1831,9 +1863,10 @@ def test_verify_remove_add_igmp_config_to_receiver_interface_p0(request):
result = verify_ip_mroutes(
tgen, dut, source, IGMP_JOIN_RANGE_1, iif, oil, expected=False
)
assert result is not True, (
"Testcase {} : Failed \n routes are still"
" present \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n routes are still" " present \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -1995,7 +2028,14 @@ def test_verify_remove_add_igmp_commands_when_pim_configured_p0(request):
input_dict_2 = {
"l1": {
"igmp": {
"interfaces": {intf_l1_i1: {"igmp": {"version": "2", "delete": True,}}}
"interfaces": {
intf_l1_i1: {
"igmp": {
"version": "2",
"delete": True,
}
}
}
}
}
}
@ -2009,9 +2049,10 @@ def test_verify_remove_add_igmp_commands_when_pim_configured_p0(request):
)
result = verify_igmp_config(tgen, input_dict_1, expected=False)
assert result is not True, (
"Testcase {} : Failed \n "
"IGMP interface is not removed \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n " "IGMP interface is not removed \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -2914,9 +2955,10 @@ def test_mroute_after_removing_RP_sending_IGMP_prune_p2(request):
data["oil"],
expected=False,
)
assert result is not True, (
"Testcase {} : Failed \n "
"mroute still present \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n " "mroute still present \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -3259,9 +3301,10 @@ def test_prune_sent_to_LHR_and_FHR_when_PIMnbr_down_p2(request):
data["oil"],
expected=False,
)
assert result is not True, (
"Testcase {} : Failed \n "
"mroute still present \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n " "mroute still present \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -3287,9 +3330,10 @@ def test_prune_sent_to_LHR_and_FHR_when_PIMnbr_down_p2(request):
IGMP_JOIN_RANGE_1,
expected=False,
)
assert result is not True, (
"Testcase {} : Failed \n "
"upstream still present \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n " "upstream still present \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -3311,9 +3355,10 @@ def test_prune_sent_to_LHR_and_FHR_when_PIMnbr_down_p2(request):
result = verify_pim_rp_info(
tgen, topo, dut, GROUP_RANGE_1, "Unknown", rp_address, SOURCE, expected=False
)
assert result is not True, (
"Testcase {} : Failed \n "
"RP iif is not updated \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n " "RP iif is not updated \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -3459,9 +3504,10 @@ def test_prune_sent_to_LHR_and_FHR_when_PIMnbr_down_p2(request):
result = verify_pim_rp_info(
tgen, topo, dut, GROUP_RANGE_1, "Unknown", rp_address, SOURCE, expected=False
)
assert result is not True, (
"Testcase {} : Failed \n "
"RP iif is not updated \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n " "RP iif is not updated \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -3606,9 +3652,10 @@ def test_prune_sent_to_LHR_and_FHR_when_PIMnbr_down_p2(request):
result = verify_pim_rp_info(
tgen, topo, dut, GROUP_RANGE_1, "Unknown", rp_address, SOURCE, expected=False
)
assert result is not True, (
"Testcase {} : Failed \n "
"RP iif is not updated \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n " "RP iif is not updated \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -3873,7 +3920,12 @@ def test_verify_multicast_traffic_when_LHR_connected_to_RP_p1(request):
"l1": {
"igmp": {
"interfaces": {
"l1-i1-eth1": {"igmp": {"version": "2", "delete": True,}}
"l1-i1-eth1": {
"igmp": {
"version": "2",
"delete": True,
}
}
}
}
}
@ -4107,9 +4159,10 @@ def test_verify_multicast_traffic_when_LHR_connected_to_RP_p1(request):
data["oil"],
expected=False,
)
assert result is not True, (
"Testcase {} : Failed \n"
"mroutes are cleared \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n" "mroutes are cleared \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -4181,9 +4234,10 @@ def test_verify_multicast_traffic_when_LHR_connected_to_RP_p1(request):
data["oil"],
expected=False,
)
assert result is not True, (
"Testcase {} : Failed \n"
" mroutes are cleared \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n" " mroutes are cleared \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -4248,9 +4302,10 @@ def test_verify_multicast_traffic_when_LHR_connected_to_RP_p1(request):
result = verify_ip_mroutes(
tgen, dut, src_address, _IGMP_JOIN_RANGE, iif, oil, expected=False
)
assert result is not True, (
"Testcase {} : Failed \n"
" mroutes are cleared \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n" " mroutes are cleared \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -4459,9 +4514,10 @@ def test_verify_multicast_traffic_when_FHR_connected_to_RP_p1(request):
data["oil"],
expected=False,
)
assert result is not True, (
"Testcase {} : Failed \n"
" mroutes are cleared \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n" " mroutes are cleared \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -4524,9 +4580,10 @@ def test_verify_multicast_traffic_when_FHR_connected_to_RP_p1(request):
data["oil"],
expected=False,
)
assert result is not True, (
"Testcase {} : Failed \n"
" mroutes are cleared \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n" " mroutes are cleared \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))
@ -4595,9 +4652,10 @@ def test_verify_multicast_traffic_when_FHR_connected_to_RP_p1(request):
data["oil"],
expected=False,
)
assert result is not True, (
"Testcase {} : Failed \n"
" mroutes are cleared \n Error: {}".format(tc_name, result)
assert (
result is not True
), "Testcase {} : Failed \n" " mroutes are cleared \n Error: {}".format(
tc_name, result
)
logger.info("Expected Behaviour: {}".format(result))

View File

@ -1009,7 +1009,11 @@ def test_PIM_hello_tx_rx_p1(request):
intf_c1_l1 = topo["routers"]["c1"]["links"]["l1"]["interface"]
step("verify before stats on C1")
state_dict = {"c1": {intf_c1_l1: ["helloTx", "helloRx"],}}
state_dict = {
"c1": {
intf_c1_l1: ["helloTx", "helloRx"],
}
}
c1_state_before = verify_pim_interface_traffic(tgen, state_dict)
assert isinstance(
@ -1040,7 +1044,11 @@ def test_PIM_hello_tx_rx_p1(request):
), "Testcase{} : Failed Error: {}" "stats incremented".format(tc_name, result)
step("verify before stats on l1")
l1_state_dict = {"l1": {intf_l1_c1: ["helloTx", "helloRx"],}}
l1_state_dict = {
"l1": {
intf_l1_c1: ["helloTx", "helloRx"],
}
}
l1_state_before = verify_pim_interface_traffic(tgen, l1_state_dict)
assert isinstance(
@ -1077,7 +1085,11 @@ def test_PIM_hello_tx_rx_p1(request):
l1_state_after = {}
step("verify before stats on C1")
state_dict = {"c1": {intf_c1_l1: ["helloTx", "helloRx"],}}
state_dict = {
"c1": {
intf_c1_l1: ["helloTx", "helloRx"],
}
}
c1_state_before = verify_pim_interface_traffic(tgen, state_dict)
assert isinstance(

View File

@ -281,6 +281,7 @@ def teardown_module():
#
#####################################################
def config_to_send_igmp_join_and_traffic(tgen, tc_name):
"""
API to do pre-configuration to send IGMP join and multicast
@ -961,7 +962,12 @@ def test_add_RP_after_join_received_p1(request):
input_dict = {
"r1": {
"pim": {
"rp": [{"rp_addr": "1.0.2.17", "group_addr_range": GROUP_RANGE_ALL,}]
"rp": [
{
"rp_addr": "1.0.2.17",
"group_addr_range": GROUP_RANGE_ALL,
}
]
}
}
}
@ -1147,32 +1153,32 @@ def test_reachable_static_RP_after_join_p0(request):
def test_send_join_on_higher_preffered_rp_p1(request):
"""
TC_11_P1 : Verify PIM join send towards the higher preferred RP
TC_12_P1 : Verify PIM prune send towards the lower preferred RP
TC_13_P1 : Verify RPF interface is updated in mroute (kernel) when higher
preferred overlapping RP configured
TC_14_P1 : Verify IIF and OIL in "show ip pim state" updated properly when
higher preferred overlapping RP configured
TC_15_P1 : Verify upstream interfaces(IIF) and join state are updated when
higher preferred overlapping RP is configured
TC_16_P1 : Verify join is send to lower preferred RP, when higher
preferred RP gets deleted
TC_17_P1 : Verify prune is send to higher preferred RP when higher
preferred RP gets deleted
TC_18_P1 : Verify RPF interface updated in mroute when higher preferred RP
gets deleted
TC_19_P1 : Verify IIF and OIL in "show ip pim state" updated when higher
preferred overlapping RP is deleted
TC_20_P1 : Verfiy PIM upstream IIF updated when higher preferred
overlapping RP deleted
TC_11_P1 : Verify PIM join send towards the higher preferred RP
TC_12_P1 : Verify PIM prune send towards the lower preferred RP
TC_13_P1 : Verify RPF interface is updated in mroute (kernel) when higher
preferred overlapping RP configured
TC_14_P1 : Verify IIF and OIL in "show ip pim state" updated properly when
higher preferred overlapping RP configured
TC_15_P1 : Verify upstream interfaces(IIF) and join state are updated when
higher preferred overlapping RP is configured
TC_16_P1 : Verify join is send to lower preferred RP, when higher
preferred RP gets deleted
TC_17_P1 : Verify prune is send to higher preferred RP when higher
preferred RP gets deleted
TC_18_P1 : Verify RPF interface updated in mroute when higher preferred RP
gets deleted
TC_19_P1 : Verify IIF and OIL in "show ip pim state" updated when higher
preferred overlapping RP is deleted
TC_20_P1 : Verfiy PIM upstream IIF updated when higher preferred
overlapping RP deleted
Topology used:
_______r2
|
iperf |
r0-----r1
|
|_______r4
Topology used:
_______r2
|
iperf |
r0-----r1
|
|_______r4
"""
tgen = get_topogen()
@ -1241,7 +1247,12 @@ def test_send_join_on_higher_preffered_rp_p1(request):
input_dict = {
"r4": {
"pim": {
"rp": [{"rp_addr": "1.0.4.17", "group_addr_range": ["225.1.1.1/32"],}]
"rp": [
{
"rp_addr": "1.0.4.17",
"group_addr_range": ["225.1.1.1/32"],
}
]
}
}
}
@ -1483,22 +1494,42 @@ def test_RP_configured_as_LHR_1_p1(request):
input_dict = {
"r1": {
"pim": {
"rp": [{"rp_addr": "1.0.1.17", "group_addr_range": GROUP_RANGE_ALL,}]
"rp": [
{
"rp_addr": "1.0.1.17",
"group_addr_range": GROUP_RANGE_ALL,
}
]
}
},
"r2": {
"pim": {
"rp": [{"rp_addr": "1.0.1.17", "group_addr_range": GROUP_RANGE_ALL,}]
"rp": [
{
"rp_addr": "1.0.1.17",
"group_addr_range": GROUP_RANGE_ALL,
}
]
}
},
"r3": {
"pim": {
"rp": [{"rp_addr": "1.0.1.17", "group_addr_range": GROUP_RANGE_ALL,}]
"rp": [
{
"rp_addr": "1.0.1.17",
"group_addr_range": GROUP_RANGE_ALL,
}
]
}
},
"r4": {
"pim": {
"rp": [{"rp_addr": "1.0.1.17", "group_addr_range": GROUP_RANGE_ALL,}]
"rp": [
{
"rp_addr": "1.0.1.17",
"group_addr_range": GROUP_RANGE_ALL,
}
]
}
},
}
@ -1677,22 +1708,42 @@ def test_RP_configured_as_LHR_2_p1(request):
input_dict = {
"r1": {
"pim": {
"rp": [{"rp_addr": "1.0.1.17", "group_addr_range": GROUP_RANGE_ALL,}]
"rp": [
{
"rp_addr": "1.0.1.17",
"group_addr_range": GROUP_RANGE_ALL,
}
]
}
},
"r2": {
"pim": {
"rp": [{"rp_addr": "1.0.1.17", "group_addr_range": GROUP_RANGE_ALL,}]
"rp": [
{
"rp_addr": "1.0.1.17",
"group_addr_range": GROUP_RANGE_ALL,
}
]
}
},
"r3": {
"pim": {
"rp": [{"rp_addr": "1.0.1.17", "group_addr_range": GROUP_RANGE_ALL,}]
"rp": [
{
"rp_addr": "1.0.1.17",
"group_addr_range": GROUP_RANGE_ALL,
}
]
}
},
"r4": {
"pim": {
"rp": [{"rp_addr": "1.0.1.17", "group_addr_range": GROUP_RANGE_ALL,}]
"rp": [
{
"rp_addr": "1.0.1.17",
"group_addr_range": GROUP_RANGE_ALL,
}
]
}
},
}
@ -1863,22 +1914,42 @@ def test_RP_configured_as_FHR_1_p1(request):
input_dict = {
"r1": {
"pim": {
"rp": [{"rp_addr": "1.0.3.17", "group_addr_range": GROUP_RANGE_ALL,}]
"rp": [
{
"rp_addr": "1.0.3.17",
"group_addr_range": GROUP_RANGE_ALL,
}
]
}
},
"r2": {
"pim": {
"rp": [{"rp_addr": "1.0.3.17", "group_addr_range": GROUP_RANGE_ALL,}]
"rp": [
{
"rp_addr": "1.0.3.17",
"group_addr_range": GROUP_RANGE_ALL,
}
]
}
},
"r3": {
"pim": {
"rp": [{"rp_addr": "1.0.3.17", "group_addr_range": GROUP_RANGE_ALL,}]
"rp": [
{
"rp_addr": "1.0.3.17",
"group_addr_range": GROUP_RANGE_ALL,
}
]
}
},
"r4": {
"pim": {
"rp": [{"rp_addr": "1.0.3.17", "group_addr_range": GROUP_RANGE_ALL,}]
"rp": [
{
"rp_addr": "1.0.3.17",
"group_addr_range": GROUP_RANGE_ALL,
}
]
}
},
}
@ -2050,22 +2121,42 @@ def test_RP_configured_as_FHR_2_p2(request):
input_dict = {
"r1": {
"pim": {
"rp": [{"rp_addr": "1.0.3.17", "group_addr_range": GROUP_RANGE_ALL,}]
"rp": [
{
"rp_addr": "1.0.3.17",
"group_addr_range": GROUP_RANGE_ALL,
}
]
}
},
"r2": {
"pim": {
"rp": [{"rp_addr": "1.0.3.17", "group_addr_range": GROUP_RANGE_ALL,}]
"rp": [
{
"rp_addr": "1.0.3.17",
"group_addr_range": GROUP_RANGE_ALL,
}
]
}
},
"r3": {
"pim": {
"rp": [{"rp_addr": "1.0.3.17", "group_addr_range": GROUP_RANGE_ALL,}]
"rp": [
{
"rp_addr": "1.0.3.17",
"group_addr_range": GROUP_RANGE_ALL,
}
]
}
},
"r4": {
"pim": {
"rp": [{"rp_addr": "1.0.3.17", "group_addr_range": GROUP_RANGE_ALL,}]
"rp": [
{
"rp_addr": "1.0.3.17",
"group_addr_range": GROUP_RANGE_ALL,
}
]
}
},
}
@ -2733,7 +2824,12 @@ def test_multiple_groups_same_RP_address_p2(request):
input_dict = {
"r1": {
"pim": {
"rp": [{"rp_addr": "1.0.2.17", "group_addr_range": GROUP_RANGE_ALL,}]
"rp": [
{
"rp_addr": "1.0.2.17",
"group_addr_range": GROUP_RANGE_ALL,
}
]
}
}
}
@ -2893,12 +2989,22 @@ def test_multiple_groups_different_RP_address_p2(request):
input_dict = {
"r2": {
"pim": {
"rp": [{"rp_addr": "1.0.2.17", "group_addr_range": GROUP_RANGE_LIST_1,}]
"rp": [
{
"rp_addr": "1.0.2.17",
"group_addr_range": GROUP_RANGE_LIST_1,
}
]
}
},
"r4": {
"pim": {
"rp": [{"rp_addr": "1.0.4.17", "group_addr_range": GROUP_RANGE_LIST_2,}]
"rp": [
{
"rp_addr": "1.0.4.17",
"group_addr_range": GROUP_RANGE_LIST_2,
}
]
}
},
}
@ -3148,12 +3254,22 @@ def test_multiple_groups_different_RP_address_p2(request):
input_dict = {
"r2": {
"pim": {
"rp": [{"rp_addr": "1.0.2.17", "group_addr_range": GROUP_RANGE_LIST_1,}]
"rp": [
{
"rp_addr": "1.0.2.17",
"group_addr_range": GROUP_RANGE_LIST_1,
}
]
}
},
"r4": {
"pim": {
"rp": [{"rp_addr": "1.0.4.17", "group_addr_range": GROUP_RANGE_LIST_2,}]
"rp": [
{
"rp_addr": "1.0.4.17",
"group_addr_range": GROUP_RANGE_LIST_2,
}
]
}
},
}

View File

@ -50,6 +50,7 @@ import pytest
pytestmark = [pytest.mark.ospfd]
class OspfSrTopo(Topo):
"Test topology builder"

View File

@ -166,6 +166,7 @@ def test_ospf_initial_convergence_step1():
"step1/show_ip_route_initial.ref",
)
def test_ospf_link_protection_step2():
logger.info("Test (step 2): check OSPF link protection")
tgen = get_topogen()
@ -175,9 +176,7 @@ def test_ospf_link_protection_step2():
pytest.skip(tgen.errors)
# enable TI-LFA link protection on all interfaces
tgen.net["rt1"].cmd(
'vtysh -c "conf t" -c "router ospf" -c "fast-reroute ti-lfa"'
)
tgen.net["rt1"].cmd('vtysh -c "conf t" -c "router ospf" -c "fast-reroute ti-lfa"')
router_compare_json_output(
"rt1",
@ -197,6 +196,7 @@ def test_ospf_link_protection_step2():
"step2/show_ip_route_initial.ref",
)
def test_ospf_node_protection_step3():
logger.info("Test (step 3): check OSPF node protection")
tgen = get_topogen()
@ -228,6 +228,7 @@ def test_ospf_node_protection_step3():
"step3/show_ip_route_initial.ref",
)
# Memory leak test template
def test_memory_leak():
"Run the memory leak test and report results."
@ -237,6 +238,7 @@ def test_memory_leak():
tgen.report_memory_leaks()
if __name__ == "__main__":
args = ["-s"] + sys.argv[1:]
sys.exit(pytest.main(args))

View File

@ -47,6 +47,7 @@ from mininet.topo import Topo
pytestmark = [pytest.mark.ospfd]
class OSPFTopo(Topo):
"Test topology builder"

View File

@ -47,6 +47,7 @@ from mininet.topo import Topo
pytestmark = [pytest.mark.ospfd]
class OSPFTopo(Topo):
"Test topology builder"

View File

@ -384,15 +384,14 @@ def test_linux_ipv6_kernel_routingTable():
% (i, diff)
)
else:
logger.error(
"r{} failed - no nhid ref file: {}".format(i, refTableFile)
)
logger.error("r{} failed - no nhid ref file: {}".format(i, refTableFile))
assert False, (
"Linux Kernel IPv6 Routing Table verification failed for router r%s\n"
% (i)
)
def test_shutdown_check_stderr():
tgen = get_topogen()

View File

@ -49,15 +49,14 @@ from lib.common_config import (
shutdown_bringup_interface,
topo_daemons,
verify_rib,
stop_router, start_router,
stop_router,
start_router,
create_static_routes,
start_router_daemons,
kill_router_daemons
kill_router_daemons,
)
from lib.ospf import (
verify_ospf_neighbor, verify_ospf_rib,
create_router_ospf)
from lib.ospf import verify_ospf_neighbor, verify_ospf_rib, create_router_ospf
from lib.topolog import logger
from lib.topojson import build_topo_from_json, build_config_from_json
@ -69,8 +68,13 @@ pytestmark = [pytest.mark.ospfd, pytest.mark.staticd]
topo = None
NETWORK = {
"ipv4": ["11.0.20.1/32", "11.0.20.2/32", "11.0.20.3/32", "11.0.20.4/32",
"11.0.20.5/32"]
"ipv4": [
"11.0.20.1/32",
"11.0.20.2/32",
"11.0.20.3/32",
"11.0.20.4/32",
"11.0.20.5/32",
]
}
"""
Topology:
@ -102,6 +106,7 @@ try:
except IOError:
assert False, "Could not read file {}".format(jsonFile)
class CreateTopo(Topo):
"""
Test topology builder.
@ -190,78 +195,70 @@ def test_ospf_chaos_tc31_p1(request):
step(
"Create static routes(10.0.20.1/32) in R1 and redistribute "
"to OSPF using route map.")
"to OSPF using route map."
)
# Create Static routes
input_dict = {
"r0": {
"static_routes": [
{
"network": NETWORK['ipv4'][0],
"network": NETWORK["ipv4"][0],
"no_of_ip": 5,
"next_hop": 'Null0',
"next_hop": "Null0",
}
]
}
}
result = create_static_routes(tgen, input_dict)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
ospf_red_r0 = {
"r0": {
"ospf": {
"redistribute": [{
"redist_type": "static"
}]
}
}
}
ospf_red_r0 = {"r0": {"ospf": {"redistribute": [{"redist_type": "static"}]}}}
result = create_router_ospf(tgen, topo, ospf_red_r0)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
step("Verify OSPF neighbors after base config is done.")
# Api call verify whether OSPF is converged
ospf_covergence = verify_ospf_neighbor(tgen, topo)
assert ospf_covergence is True, ("setup_module :Failed \n Error:"
" {}".format(ospf_covergence))
assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
ospf_covergence
)
step("Verify that route is advertised to R1.")
dut = 'r1'
protocol = 'ospf'
nh = topo['routers']['r0']['links']['r1']['ipv4'].split('/')[0]
dut = "r1"
protocol = "ospf"
nh = topo["routers"]["r0"]["links"]["r1"]["ipv4"].split("/")[0]
result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
result = verify_rib(
tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
step("Kill OSPFd daemon on R0.")
kill_router_daemons(tgen, "r0", ["ospfd"])
step("Verify OSPF neighbors are down after killing ospfd in R0")
dut = 'r0'
dut = "r0"
# Api call verify whether OSPF is converged
ospf_covergence = verify_ospf_neighbor(tgen, topo, dut=dut,
expected=False)
assert ospf_covergence is not True, ("setup_module :Failed \n Error:"
" {}".format(ospf_covergence))
ospf_covergence = verify_ospf_neighbor(tgen, topo, dut=dut, expected=False)
assert ospf_covergence is not True, "setup_module :Failed \n Error:" " {}".format(
ospf_covergence
)
step("Verify that route advertised to R1 are deleted from RIB and FIB.")
dut = 'r1'
protocol = 'ospf'
dut = "r1"
protocol = "ospf"
result = verify_ospf_rib(tgen, dut, input_dict, expected=False)
assert result is not True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol,
expected=False)
result = verify_rib(
tgen, "ipv4", dut, input_dict, protocol=protocol, expected=False
)
assert result is not True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
step("Bring up OSPFd daemon on R0.")
start_router_daemons(tgen, "r0", ["ospfd"])
@ -269,33 +266,32 @@ def test_ospf_chaos_tc31_p1(request):
step("Verify OSPF neighbors are up after bringing back ospfd in R0")
# Api call verify whether OSPF is converged
ospf_covergence = verify_ospf_neighbor(tgen, topo)
assert ospf_covergence is True, ("setup_module :Failed \n Error:"
" {}".format(ospf_covergence))
assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
ospf_covergence
)
step(
"All the neighbours are up and routes are installed before the"
" restart. Verify OSPF route table and ip route table.")
dut = 'r1'
protocol = 'ospf'
" restart. Verify OSPF route table and ip route table."
)
dut = "r1"
protocol = "ospf"
result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol,
next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
step("Kill OSPFd daemon on R1.")
kill_router_daemons(tgen, "r1", ["ospfd"])
step("Verify OSPF neighbors are down after killing ospfd in R1")
dut = 'r1'
dut = "r1"
# Api call verify whether OSPF is converged
ospf_covergence = verify_ospf_neighbor(tgen, topo, dut=dut,
expected=False)
assert ospf_covergence is not True, ("setup_module :Failed \n Error:"
" {}".format(ospf_covergence))
ospf_covergence = verify_ospf_neighbor(tgen, topo, dut=dut, expected=False)
assert ospf_covergence is not True, "setup_module :Failed \n Error:" " {}".format(
ospf_covergence
)
step("Bring up OSPFd daemon on R1.")
start_router_daemons(tgen, "r1", ["ospfd"])
@ -303,23 +299,22 @@ def test_ospf_chaos_tc31_p1(request):
step("Verify OSPF neighbors are up after bringing back ospfd in R1")
# Api call verify whether OSPF is converged
ospf_covergence = verify_ospf_neighbor(tgen, topo)
assert ospf_covergence is True, ("setup_module :Failed \n Error:"
" {}".format(ospf_covergence))
assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
ospf_covergence
)
step(
"All the neighbours are up and routes are installed before the"
" restart. Verify OSPF route table and ip route table.")
" restart. Verify OSPF route table and ip route table."
)
dut = 'r1'
protocol = 'ospf'
dut = "r1"
protocol = "ospf"
result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol,
next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
write_test_footer(tc_name)
@ -335,104 +330,91 @@ def test_ospf_chaos_tc32_p1(request):
step(
"Create static routes(10.0.20.1/32) in R1 and redistribute "
"to OSPF using route map.")
"to OSPF using route map."
)
# Create Static routes
input_dict = {
"r0": {
"static_routes": [
{
"network": NETWORK['ipv4'][0],
"network": NETWORK["ipv4"][0],
"no_of_ip": 5,
"next_hop": 'Null0',
"next_hop": "Null0",
}
]
}
}
result = create_static_routes(tgen, input_dict)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
ospf_red_r0 = {
"r0": {
"ospf": {
"redistribute": [{
"redist_type": "static"
}]
}
}
}
ospf_red_r0 = {"r0": {"ospf": {"redistribute": [{"redist_type": "static"}]}}}
result = create_router_ospf(tgen, topo, ospf_red_r0)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
step("Verify OSPF neighbors after base config is done.")
# Api call verify whether OSPF is converged
ospf_covergence = verify_ospf_neighbor(tgen, topo)
assert ospf_covergence is True, ("setup_module :Failed \n Error:"
" {}".format(ospf_covergence))
assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
ospf_covergence
)
step("Verify that route is advertised to R1.")
dut = 'r1'
protocol = 'ospf'
dut = "r1"
protocol = "ospf"
nh = topo['routers']['r0']['links']['r1']['ipv4'].split('/')[0]
nh = topo["routers"]["r0"]["links"]["r1"]["ipv4"].split("/")[0]
result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol,
next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
step("Restart frr on R0")
stop_router(tgen, 'r0')
start_router(tgen, 'r0')
stop_router(tgen, "r0")
start_router(tgen, "r0")
step("Verify OSPF neighbors are up after restarting R0")
# Api call verify whether OSPF is converged
ospf_covergence = verify_ospf_neighbor(tgen, topo)
assert ospf_covergence is True, ("setup_module :Failed \n Error:"
" {}".format(ospf_covergence))
assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
ospf_covergence
)
step(
"All the neighbours are up and routes are installed before the"
" restart. Verify OSPF route table and ip route table.")
dut = 'r1'
protocol = 'ospf'
" restart. Verify OSPF route table and ip route table."
)
dut = "r1"
protocol = "ospf"
result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol,
next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
step("Restart frr on R1")
stop_router(tgen, 'r1')
start_router(tgen, 'r1')
stop_router(tgen, "r1")
start_router(tgen, "r1")
step("Verify OSPF neighbors are up after restarting R1")
# Api call verify whether OSPF is converged
ospf_covergence = verify_ospf_neighbor(tgen, topo)
assert ospf_covergence is True, ("setup_module :Failed \n Error:"
" {}".format(ospf_covergence))
assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
ospf_covergence
)
step(
"All the neighbours are up and routes are installed before the"
" restart. Verify OSPF route table and ip route table.")
dut = 'r1'
protocol = 'ospf'
" restart. Verify OSPF route table and ip route table."
)
dut = "r1"
protocol = "ospf"
result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol,
next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
write_test_footer(tc_name)
@ -453,70 +435,62 @@ def test_ospf_chaos_tc34_p1(request):
step(
"Create static routes(10.0.20.1/32) in R1 and redistribute "
"to OSPF using route map.")
"to OSPF using route map."
)
# Create Static routes
input_dict = {
"r0": {
"static_routes": [
{
"network": NETWORK['ipv4'][0],
"network": NETWORK["ipv4"][0],
"no_of_ip": 5,
"next_hop": 'Null0',
"next_hop": "Null0",
}
]
}
}
result = create_static_routes(tgen, input_dict)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
ospf_red_r0 = {
"r0": {
"ospf": {
"redistribute": [{
"redist_type": "static"
}]
}
}
}
ospf_red_r0 = {"r0": {"ospf": {"redistribute": [{"redist_type": "static"}]}}}
result = create_router_ospf(tgen, topo, ospf_red_r0)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
step("Verify OSPF neighbors after base config is done.")
# Api call verify whether OSPF is converged
ospf_covergence = verify_ospf_neighbor(tgen, topo)
assert ospf_covergence is True, ("setup_module :Failed \n Error:"
" {}".format(ospf_covergence))
assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
ospf_covergence
)
step("Verify that route is advertised to R1.")
dut = 'r1'
protocol = 'ospf'
nh = topo['routers']['r0']['links']['r1']['ipv4'].split('/')[0]
dut = "r1"
protocol = "ospf"
nh = topo["routers"]["r0"]["links"]["r1"]["ipv4"].split("/")[0]
result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol,
next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
step("Kill staticd daemon on R0.")
kill_router_daemons(tgen, "r0", ["staticd"])
step("Verify that route advertised to R1 are deleted from RIB and FIB.")
dut = 'r1'
protocol = 'ospf'
dut = "r1"
protocol = "ospf"
result = verify_ospf_rib(tgen, dut, input_dict, expected=False)
assert result is not True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol,
expected=False)
result = verify_rib(
tgen, "ipv4", dut, input_dict, protocol=protocol, expected=False
)
assert result is not True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
step("Bring up staticd daemon on R0.")
start_router_daemons(tgen, "r0", ["staticd"])
@ -524,22 +498,21 @@ def test_ospf_chaos_tc34_p1(request):
step("Verify OSPF neighbors are up after bringing back ospfd in R0")
# Api call verify whether OSPF is converged
ospf_covergence = verify_ospf_neighbor(tgen, topo)
assert ospf_covergence is True, ("setup_module :Failed \n Error:"
" {}".format(ospf_covergence))
assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
ospf_covergence
)
step(
"All the neighbours are up and routes are installed before the"
" restart. Verify OSPF route table and ip route table.")
dut = 'r1'
protocol = 'ospf'
" restart. Verify OSPF route table and ip route table."
)
dut = "r1"
protocol = "ospf"
result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol,
next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
step("Kill staticd daemon on R1.")
kill_router_daemons(tgen, "r1", ["staticd"])
@ -550,23 +523,22 @@ def test_ospf_chaos_tc34_p1(request):
step("Verify OSPF neighbors are up after bringing back ospfd in R1")
# Api call verify whether OSPF is converged
ospf_covergence = verify_ospf_neighbor(tgen, topo)
assert ospf_covergence is True, ("setup_module :Failed \n Error:"
" {}".format(ospf_covergence))
assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
ospf_covergence
)
step(
"All the neighbours are up and routes are installed before the"
" restart. Verify OSPF route table and ip route table.")
" restart. Verify OSPF route table and ip route table."
)
dut = 'r1'
protocol = 'ospf'
dut = "r1"
protocol = "ospf"
result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol,
next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
write_test_footer(tc_name)

View File

@ -146,7 +146,6 @@ def setup_module(mod):
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
logger.info("Running setup_module() done")
@ -397,10 +396,7 @@ def test_ospf_p2mp_tc1_p0(request):
"links": {
"r3": {
"interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
"ospf": {
"area": "0.0.0.0",
"networkType":"POINTOMULTIPOINT"
},
"ospf": {"area": "0.0.0.0", "networkType": "POINTOMULTIPOINT"},
}
}
}

View File

@ -82,6 +82,7 @@ class NetworkTopo(Topo):
##
#####################################################
def setup_module(module):
"Setup topology"
tgen = Topogen(NetworkTopo, module.__name__)

View File

@ -45,6 +45,7 @@ from mininet.topo import Topo
pytestmark = [pytest.mark.pimd]
class PIMTopo(Topo):
def build(self, *_args, **_opts):
"Build function"

View File

@ -24,6 +24,7 @@ markers =
sharpd: Tests that run against SHARPD
staticd: Tests that run against STATICD
vrrpd: Tests that run against VRRPD
snmp: Tests that run against snmp changes
[topogen]
# Default configuration values

View File

@ -46,6 +46,8 @@ from lib.snmptest import SnmpTester
# Required to instantiate the topology builder class.
from mininet.topo import Topo
pytestmark = [pytest.mark.bgpd, pytest.mark.isisd, pytest.mark.snmp]
class TemplateTopo(Topo):
"Test topology builder"

View File

@ -35,6 +35,7 @@ import time
import os
import pytest
import platform
# Save the Current Working Directory to find configuration files.
CWD = os.path.dirname(os.path.realpath(__file__))
sys.path.append(os.path.join(CWD, "../"))
@ -44,6 +45,7 @@ sys.path.append(os.path.join(CWD, "../lib/"))
from mininet.topo import Topo
from lib.topogen import Topogen, get_topogen
from lib.topotest import version_cmp
# Import topoJson from lib, to create topology and initial configuration
from lib.common_config import (
start_topology,
@ -121,9 +123,11 @@ def setup_module(mod):
# Creating configuration from JSON
build_config_from_json(tgen, topo)
if version_cmp(platform.release(), '4.19') < 0:
error_msg = ('These tests will not run. (have kernel "{}", '
'requires kernel >= 4.19)'.format(platform.release()))
if version_cmp(platform.release(), "4.19") < 0:
error_msg = (
'These tests will not run. (have kernel "{}", '
"requires kernel >= 4.19)".format(platform.release())
)
pytest.skip(error_msg)
# Checking BGP convergence

View File

@ -163,9 +163,11 @@ def setup_module(mod):
# Creating configuration from JSON
build_config_from_json(tgen, topo)
if version_cmp(platform.release(), '4.19') < 0:
error_msg = ('These tests will not run. (have kernel "{}", '
'requires kernel >= 4.19)'.format(platform.release()))
if version_cmp(platform.release(), "4.19") < 0:
error_msg = (
'These tests will not run. (have kernel "{}", '
"requires kernel >= 4.19)".format(platform.release())
)
pytest.skip(error_msg)
# Checking BGP convergence
@ -854,12 +856,12 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc6_ebgp(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 still present in RIB".format(tc_name)
assert (
result is True
), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format(
tc_name
)
write_test_footer(tc_name)
@ -1129,9 +1131,10 @@ def test_static_route_8nh_diff_AD_ebgp_ecmp_p1_tc8_ebgp(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 still present in RIB".format(tc_name)
assert (
result is True
), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format(
tc_name
)
write_test_footer(tc_name)
@ -1339,7 +1342,15 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ebgp(request):
" value and all the nexthop populated in RIB and FIB again"
)
for addr_type in ADDR_TYPES:
input_dict_4 = {"r2": {"static_routes": [{"network": PREFIX1[addr_type],}]}}
input_dict_4 = {
"r2": {
"static_routes": [
{
"network": PREFIX1[addr_type],
}
]
}
}
nh = NEXT_HOP_IP["nh1"][addr_type]
result = verify_rib(
tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True
@ -1468,9 +1479,10 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ebgp(request):
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)

View File

@ -88,6 +88,7 @@ NEXT_HOP_IP = {}
pytestmark = [pytest.mark.bgpd, pytest.mark.staticd]
class CreateTopo(Topo):
"""
Test CreateTopo - topology 1.
@ -126,9 +127,11 @@ def setup_module(mod):
# Creating configuration from JSON
build_config_from_json(tgen, topo)
if version_cmp(platform.release(), '4.19') < 0:
error_msg = ('These tests will not run. (have kernel "{}", '
'requires kernel >= 4.19)'.format(platform.release()))
if version_cmp(platform.release(), "4.19") < 0:
error_msg = (
'These tests will not run. (have kernel "{}", '
"requires kernel >= 4.19)".format(platform.release())
)
pytest.skip(error_msg)
# Checking BGP convergence

View File

@ -125,9 +125,11 @@ def setup_module(mod):
# Creating configuration from JSON
build_config_from_json(tgen, topo)
if version_cmp(platform.release(), '4.19') < 0:
error_msg = ('These tests will not run. (have kernel "{}", '
'requires kernel >= 4.19)'.format(platform.release()))
if version_cmp(platform.release(), "4.19") < 0:
error_msg = (
'These tests will not run. (have kernel "{}", '
"requires kernel >= 4.19)".format(platform.release())
)
pytest.skip(error_msg)
# Checking BGP convergence

View File

@ -165,9 +165,11 @@ def setup_module(mod):
# Creating configuration from JSON
build_config_from_json(tgen, topo)
if version_cmp(platform.release(), '4.19') < 0:
error_msg = ('These tests will not run. (have kernel "{}", '
'requires kernel >= 4.19)'.format(platform.release()))
if version_cmp(platform.release(), "4.19") < 0:
error_msg = (
'These tests will not run. (have kernel "{}", '
"requires kernel >= 4.19)".format(platform.release())
)
pytest.skip(error_msg)
# Checking BGP convergence
@ -884,9 +886,10 @@ 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 still present in RIB".format(tc_name)
assert (
result is True
), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format(
tc_name
)
step("BGP neighbor remove and add")
@ -907,9 +910,10 @@ 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 still present in RIB".format(tc_name)
assert (
result is True
), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format(
tc_name
)
dut = "r3"
@ -917,9 +921,10 @@ 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 still present in RIB".format(tc_name)
assert (
result is True
), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format(
tc_name
)
step("Remove the redistribute static knob")
@ -1274,9 +1279,10 @@ 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 still present in RIB".format(tc_name)
assert (
result is True
), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format(
tc_name
)
step("BGP neighbor remove and add")
@ -1297,9 +1303,10 @@ 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 still present in RIB".format(tc_name)
assert (
result is True
), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format(
tc_name
)
dut = "r3"
@ -1307,9 +1314,10 @@ 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 still present in RIB".format(tc_name)
assert (
result is True
), "Testcase {} : Failed \n" "Error: Routes are still present in RIB".format(
tc_name
)
step("Remove the redistribute static knob")
@ -1557,7 +1565,15 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ibgp(request):
" value and all the nexthop populated in RIB and FIB again"
)
for addr_type in ADDR_TYPES:
input_dict_4 = {"r2": {"static_routes": [{"network": PREFIX1[addr_type],}]}}
input_dict_4 = {
"r2": {
"static_routes": [
{
"network": PREFIX1[addr_type],
}
]
}
}
nh = NEXT_HOP_IP["nh1"][addr_type]
result = verify_rib(
tgen, addr_type, dut, input_dict_4, next_hop=nh, protocol=protocol, fib=True
@ -1686,9 +1702,10 @@ def test_static_route_8nh_diff_AD_bgp_ecmp_p1_tc10_ibgp(request):
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 redistribute static knob")

View File

@ -149,9 +149,11 @@ def setup_module(mod):
# Creating configuration from JSON
build_config_from_json(tgen, topo)
if version_cmp(platform.release(), '4.19') < 0:
error_msg = ('These tests will not run. (have kernel "{}", '
'requires kernel >= 4.19)'.format(platform.release()))
if version_cmp(platform.release(), "4.19") < 0:
error_msg = (
'These tests will not run. (have kernel "{}", '
"requires kernel >= 4.19)".format(platform.release())
)
pytest.skip(error_msg)
# Checking BGP convergence

View File

@ -69,6 +69,7 @@ from lib.bgp import (
)
from lib.topojson import build_topo_from_json, build_config_from_json
from lib.topotest import version_cmp
# Reading the data from JSON File for topology creation
jsonFile = "{}/static_routes_topo4_ibgp.json".format(CWD)
try:
@ -85,6 +86,7 @@ NEXT_HOP_IP = {}
pytestmark = [pytest.mark.bgpd, pytest.mark.staticd]
class CreateTopo(Topo):
"""
Test CreateTopo - topology 1.
@ -123,9 +125,11 @@ def setup_module(mod):
# Creating configuration from JSON
build_config_from_json(tgen, topo)
if version_cmp(platform.release(), '4.19') < 0:
error_msg = ('These tests will not run. (have kernel "{}", '
'requires kernel >= 4.19)'.format(platform.release()))
if version_cmp(platform.release(), "4.19") < 0:
error_msg = (
'These tests will not run. (have kernel "{}", '
"requires kernel >= 4.19)".format(platform.release())
)
pytest.skip(error_msg)
# Checking BGP convergence

View File

@ -76,9 +76,11 @@ def setup_module(mod):
router_list = tgen.routers()
for rname, router in router_list.items():
router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)))
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
)
router.load_config(
TopoRouter.RD_SHARP, os.path.join(CWD, "{}/sharpd.conf".format(rname)))
TopoRouter.RD_SHARP, os.path.join(CWD, "{}/sharpd.conf".format(rname))
)
# Initialize all routers.
tgen.start_router()
@ -159,6 +161,7 @@ def test_zebra_kernel_override():
_, result = topotest.run_and_expect(test_func, None, count=2, wait=0.5)
assert result is None, '"r1" JSON output mismatches'
def test_route_map_usage():
"Test that FRR only reruns over routes associated with the routemap"
logger.info("Test that FRR runs on selected re's on route-map changes")
@ -174,7 +177,9 @@ def test_route_map_usage():
r1.vtysh_cmd("conf\nroute-map static permit 10\nset src 192.168.215.1")
r1.vtysh_cmd("conf\naccess-list 5 seq 5 permit 10.0.0.44/32")
r1.vtysh_cmd("conf\naccess-list 10 seq 5 permit 10.0.1.0/24")
r1.vtysh_cmd("conf\nroute-map sharp permit 10\nmatch ip address 10\nset src 192.168.214.1")
r1.vtysh_cmd(
"conf\nroute-map sharp permit 10\nmatch ip address 10\nset src 192.168.214.1"
)
r1.vtysh_cmd("conf\nroute-map sharp permit 20\nset src 192.168.213.1")
r1.vtysh_cmd("conf\nip protocol static route-map static")
r1.vtysh_cmd("conf\nip protocol sharp route-map sharp")
@ -186,47 +191,57 @@ def test_route_map_usage():
static_rmapfile = "%s/r1/static_rmap.ref" % (thisDir)
expected = open(static_rmapfile).read().rstrip()
expected = ('\n'.join(expected.splitlines()) + '\n').rstrip()
expected = ("\n".join(expected.splitlines()) + "\n").rstrip()
actual = r1.vtysh_cmd("show route-map static")
actual = ('\n'.join(actual.splitlines()) + '\n').rstrip()
logger.info("Does the show route-map static command run the correct number of times")
actual = ("\n".join(actual.splitlines()) + "\n").rstrip()
logger.info(
"Does the show route-map static command run the correct number of times"
)
diff = topotest.get_textdiff(actual, expected,
title1 = "Actual Route-map output",
title2 = "Expected Route-map output")
diff = topotest.get_textdiff(
actual,
expected,
title1="Actual Route-map output",
title2="Expected Route-map output",
)
if diff:
logger.info("Actual:")
logger.info(actual)
logger.info("Expected:")
logger.info(expected)
srun = r1.vtysh_cmd("show run")
srun = ('\n'.join(srun.splitlines()) + '\n').rstrip()
srun = ("\n".join(srun.splitlines()) + "\n").rstrip()
logger.info("Show run")
logger.info(srun)
assert 0, "r1 static route processing:\n"
sharp_rmapfile = "%s/r1/sharp_rmap.ref" % (thisDir)
expected = open(sharp_rmapfile).read().rstrip()
expected = ('\n'.join(expected.splitlines()) + '\n').rstrip()
expected = ("\n".join(expected.splitlines()) + "\n").rstrip()
actual = r1.vtysh_cmd("show route-map sharp")
actual = ('\n'.join(actual.splitlines()) + '\n').rstrip()
actual = ("\n".join(actual.splitlines()) + "\n").rstrip()
logger.info("Does the show route-map sharp command run the correct number of times")
diff = topotest.get_textdiff(actual, expected,
title1 = "Actual Route-map output",
title2 = "Expected Route-map output")
diff = topotest.get_textdiff(
actual,
expected,
title1="Actual Route-map output",
title2="Expected Route-map output",
)
if diff:
logger.info("Actual:")
logger.info(actual)
logger.info("Expected:")
logger.info(expected)
srun = r1.vtysh_cmd("show run")
srun = ('\n'.join(srun.splitlines()) + '\n').rstrip()
srun = ("\n".join(srun.splitlines()) + "\n").rstrip()
logger.info("Show run:")
logger.info(srun)
assert 0, "r1 sharp route-map processing:\n"
logger.info("Add a extension to the static route-map to see the static route go away")
logger.info(
"Add a extension to the static route-map to see the static route go away"
)
r1.vtysh_cmd("conf\nroute-map sharp deny 5\nmatch ip address 5")
sleep(2)
# we are only checking the kernel here as that this will give us the implied
@ -236,9 +251,9 @@ def test_route_map_usage():
logger.info("Test that the routes installed are correct")
sharp_ipfile = "%s/r1/iproute.ref" % (thisDir)
expected = open(sharp_ipfile).read().rstrip()
expected = ('\n'.join(expected.splitlines()) + '\n').rstrip()
expected = ("\n".join(expected.splitlines()) + "\n").rstrip()
actual = r1.run("ip route show")
actual = ('\n'.join(actual.splitlines()) + '\n').rstrip()
actual = ("\n".join(actual.splitlines()) + "\n").rstrip()
actual = re.sub(r" nhid [0-9][0-9]", "", actual)
actual = re.sub(r" proto sharp", " proto XXXX", actual)
actual = re.sub(r" proto static", " proto XXXX", actual)
@ -250,9 +265,9 @@ def test_route_map_usage():
actual = re.sub(r" proto XXXX ", " proto XXXX ", actual)
actual = re.sub(r" metric", " metric", actual)
actual = re.sub(r" link ", " link ", actual)
diff = topotest.get_textdiff(actual, expected,
title1 = "Actual ip route show",
title2 = "Expected ip route show")
diff = topotest.get_textdiff(
actual, expected, title1="Actual ip route show", title2="Expected ip route show"
)
if diff:
logger.info("Actual:")
@ -260,11 +275,12 @@ def test_route_map_usage():
logger.info("Expected:")
logger.info(expected)
srun = r1.vtysh_cmd("show run")
srun = ('\n'.join(srun.splitlines()) + '\n').rstrip()
srun = ("\n".join(srun.splitlines()) + "\n").rstrip()
logger.info("Show run:")
logger.info(srun)
assert 0, "r1 ip route show is not correct:"
def test_memory_leak():
"Run the memory leak test and report results."
tgen = get_topogen()