tests/topotests/bgp_evpn_rt5/test_bgp_evpn.py

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