tests/topotests/bgp_evpn_rt5/test_bgp_evpn.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # test_bgp_evpn.py
   6 # Part of NetDEF Topology Tests
   7 #
   8 # Copyright (c) 2019 by 6WIND
   9 #
  10
  11 """
  12  test_bgp_evpn.py: Test the FRR BGP daemon with BGP IPv6 interface
  13  with route advertisements on a separate netns.
  14 """
  15
  16 import os
  17 import sys
  18 import pytest
  19 import platform
  20
  21 # Save the Current Working Directory to find configuration files.
  22 CWD = os.path.dirname(os.path.realpath(__file__))
  23 sys.path.append(os.path.join(CWD, "../"))
  24
  25 # pylint: disable=C0413
  26 # Import topogen and topotest helpers
  27 from lib import topotest
  28 from lib.topogen import Topogen, TopoRouter, get_topogen
  29 from lib.topolog import logger
  30
  31 # Required to instantiate the topology builder class.
  32
  33 pytestmark = [pytest.mark.bgpd]
  34
  35
  36 def build_topo(tgen):
  37     "Build function"
  38
  39     tgen.add_router("r1")
  40     tgen.add_router("r2")
  41
  42     switch = tgen.add_switch("s1")
  43     switch.add_link(tgen.gears["r1"])
  44     switch.add_link(tgen.gears["r2"])
  45
  46     switch = tgen.add_switch("s2")
  47     switch.add_link(tgen.gears["r1"])
  48
  49     switch = tgen.add_switch("s3")
  50     switch.add_link(tgen.gears["r2"])
  51
  52
  53 def setup_module(mod):
  54     "Sets up the pytest environment"
  55
  56     tgen = Topogen(build_topo, mod.__name__)
  57     tgen.start_topology()
  58
  59     router_list = tgen.routers()
  60
  61     krel = platform.release()
  62     if topotest.version_cmp(krel, "4.18") < 0:
  63         logger.info(
  64             'BGP EVPN RT5 NETNS tests will not run (have kernel "{}", but it requires 4.18)'.format(
  65                 krel
  66             )
  67         )
  68         return pytest.skip("Skipping BGP EVPN RT5 NETNS Test. Kernel not supported")
  69
  70     # create VRF vrf-101 on R1 and R2
  71     # create loop101
  72     cmds_vrflite = [
  73         "ip link add {}-vrf-101 type vrf table 101",
  74         "ip ru add oif {}-vrf-101 table 101",
  75         "ip ru add iif {}-vrf-101 table 101",
  76         "ip link set dev {}-vrf-101 up",
  77         "ip link add loop101 type dummy",
  78         "ip link set dev loop101 master {}-vrf-101",
  79         "ip link set dev loop101 up",
  80     ]
  81
  82     cmds_r2 = [  # config routing 101
  83         "ip link add name bridge-101 up type bridge stp_state 0",
  84         "ip link set bridge-101 master {}-vrf-101",
  85         "ip link set dev bridge-101 up",
  86         "ip link add name vxlan-101 type vxlan id 101 dstport 4789 dev r2-eth0 local 192.168.100.41",
  87         "ip link set dev vxlan-101 master bridge-101",
  88         "ip link set vxlan-101 up type bridge_slave learning off flood off mcast_flood off",
  89     ]
  90
  91     # cmds_r1_netns_method3 = [
  92     #     "ip link add name vxlan-{1} type vxlan id {1} dstport 4789 dev {0}-eth0 local 192.168.100.21",
  93     #     "ip link set dev vxlan-{1} netns {0}-vrf-{1}",
  94     #     "ip netns exec {0}-vrf-{1} ip li set dev lo up",
  95     #     "ip netns exec {0}-vrf-{1} ip link add name bridge-{1} up type bridge stp_state 0",
  96     #     "ip netns exec {0}-vrf-{1} ip link set dev vxlan-{1} master bridge-{1}",
  97     #     "ip netns exec {0}-vrf-{1} ip link set bridge-{1} up",
  98     #     "ip netns exec {0}-vrf-{1} ip link set vxlan-{1} up",
  99     # ]
 100
 101     router = tgen.gears["r1"]
 102
 103     ns = "r1-vrf-101"
 104     tgen.net["r1"].add_netns(ns)
 105     tgen.net["r1"].cmd_raises("ip link add loop101 type dummy")
 106     tgen.net["r1"].set_intf_netns("loop101", ns, up=True)
 107
 108     router = tgen.gears["r2"]
 109     for cmd in cmds_vrflite:
 110         logger.info("cmd to r2: " + cmd.format("r2"))
 111         output = router.cmd_raises(cmd.format("r2"))
 112         logger.info("result: " + output)
 113
 114     for cmd in cmds_r2:
 115         logger.info("cmd to r2: " + cmd.format("r2"))
 116         output = router.cmd_raises(cmd.format("r2"))
 117         logger.info("result: " + output)
 118
 119     tgen.net["r1"].cmd_raises(
 120         "ip link add name vxlan-101 type vxlan id 101 dstport 4789 dev r1-eth0 local 192.168.100.21"
 121     )
 122     tgen.net["r1"].set_intf_netns("vxlan-101", "r1-vrf-101", up=True)
 123     tgen.net["r1"].cmd_raises("ip -n r1-vrf-101 link set lo up")
 124     tgen.net["r1"].cmd_raises(
 125         "ip -n r1-vrf-101 link add name bridge-101 up type bridge stp_state 0"
 126     )
 127     tgen.net["r1"].cmd_raises(
 128         "ip -n r1-vrf-101 link set dev vxlan-101 master bridge-101"
 129     )
 130     tgen.net["r1"].cmd_raises("ip -n r1-vrf-101 link set bridge-101 up")
 131     tgen.net["r1"].cmd_raises("ip -n r1-vrf-101 link set vxlan-101 up")
 132
 133     for rname, router in router_list.items():
 134         if rname == "r1":
 135             router.load_config(
 136                 TopoRouter.RD_ZEBRA,
 137                 os.path.join(CWD, "{}/zebra.conf".format(rname)),
 138                 "--vrfwnetns",
 139             )
 140         else:
 141             router.load_config(
 142                 TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 143             )
 144         router.load_config(
 145             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
 146         )
 147
 148     # Initialize all routers.
 149     tgen.start_router()
 150
 151
 152 def teardown_module(_mod):
 153     "Teardown the pytest environment"
 154     tgen = get_topogen()
 155
 156     tgen.net["r1"].delete_netns("r1-vrf-101")
 157     tgen.stop_topology()
 158
 159
 160 def test_protocols_convergence():
 161     """
 162     Assert that all protocols have converged
 163     statuses as they depend on it.
 164     """
 165     tgen = get_topogen()
 166     if tgen.routers_have_failure():
 167         pytest.skip(tgen.errors)
 168     topotest.sleep(4, "waiting 4 seconds for bgp convergence")
 169     # Check IPv4/IPv6 routing tables.
 170     output = tgen.gears["r1"].vtysh_cmd("show bgp l2vpn evpn", isjson=False)
 171     logger.info("==== result from show bgp l2vpn evpn")
 172     logger.info(output)
 173     output = tgen.gears["r1"].vtysh_cmd(
 174         "show bgp l2vpn evpn route detail", isjson=False
 175     )
 176     logger.info("==== result from show bgp l2vpn evpn route detail")
 177     logger.info(output)
 178     output = tgen.gears["r1"].vtysh_cmd("show bgp vrf r1-vrf-101 ipv4", isjson=False)
 179     logger.info("==== result from show bgp vrf r1-vrf-101 ipv4")
 180     logger.info(output)
 181     output = tgen.gears["r1"].vtysh_cmd("show bgp vrf r1-vrf-101", isjson=False)
 182     logger.info("==== result from show bgp vrf r1-vrf-101 ")
 183     logger.info(output)
 184     output = tgen.gears["r1"].vtysh_cmd("show ip route vrf r1-vrf-101", isjson=False)
 185     logger.info("==== result from show ip route vrf r1-vrf-101")
 186     logger.info(output)
 187     output = tgen.gears["r1"].vtysh_cmd("show evpn vni detail", isjson=False)
 188     logger.info("==== result from show evpn vni detail")
 189     logger.info(output)
 190     output = tgen.gears["r1"].vtysh_cmd("show evpn next-hops vni all", isjson=False)
 191     logger.info("==== result from show evpn next-hops vni all")
 192     logger.info(output)
 193     output = tgen.gears["r1"].vtysh_cmd("show evpn rmac vni all", isjson=False)
 194     logger.info("==== result from show evpn next-hops vni all")
 195     logger.info(output)
 196     # Check IPv4 and IPv6 connectivity between r1 and r2 ( routing vxlan evpn)
 197     pingrouter = tgen.gears["r1"]
 198     logger.info(
 199         "Check Ping IPv4 from  R1(r1-vrf-101) to R2(r2-vrf-101 = 192.168.101.41)"
 200     )
 201     output = pingrouter.run("ip netns exec r1-vrf-101 ping 192.168.101.41 -f -c 1000")
 202     logger.info(output)
 203     if "1000 packets transmitted, 1000 received" not in output:
 204         assertmsg = (
 205             "expected ping IPv4 from R1(r1-vrf-101) to R2(192.168.101.41) should be ok"
 206         )
 207         assert 0, assertmsg
 208     else:
 209         logger.info("Check Ping IPv4 from R1(r1-vrf-101) to R2(192.168.101.41) OK")
 210
 211
 212 def test_memory_leak():
 213     "Run the memory leak test and report results."
 214     tgen = get_topogen()
 215     if not tgen.is_memleak_enabled():
 216         pytest.skip("Memory leak test/report is disabled")
 217
 218     tgen.report_memory_leaks()
 219
 220
 221 if __name__ == "__main__":
 222     args = ["-s"] + sys.argv[1:]
 223     sys.exit(pytest.main(args))