tests/topotests/bgp_evpn_vxlan_topo1/test_bgp_evpn_vxlan.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_bgp_evpn_vxlan.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2020 by Volta Networks
   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_vxlan.py: Test VXLAN EVPN MAC a route signalling over BGP.
  26 """
  27
  28 import os
  29 import sys
  30 import json
  31 from functools import partial
  32 from time import sleep
  33 import pytest
  34
  35 # Save the Current Working Directory to find configuration files.
  36 CWD = os.path.dirname(os.path.realpath(__file__))
  37 sys.path.append(os.path.join(CWD, "../"))
  38
  39 # pylint: disable=C0413
  40 # Import topogen and topotest helpers
  41 from lib import topotest
  42 from lib.topogen import Topogen, TopoRouter, get_topogen
  43 from lib.topolog import logger
  44
  45 # Required to instantiate the topology builder class.
  46 from mininet.topo import Topo
  47
  48 pytestmark = [pytest.mark.bgpd, pytest.mark.ospfd]
  49
  50
  51 class TemplateTopo(Topo):
  52     "Test topology builder"
  53
  54     def build(self, *_args, **_opts):
  55         "Build function"
  56         tgen = get_topogen(self)
  57
  58         # This function only purpose is to define allocation and relationship
  59         # between routers, switches and hosts.
  60         #
  61         #
  62         # Create routers
  63         tgen.add_router("P1")
  64         tgen.add_router("PE1")
  65         tgen.add_router("PE2")
  66         tgen.add_router("host1")
  67         tgen.add_router("host2")
  68
  69         # Host1-PE1
  70         switch = tgen.add_switch("s1")
  71         switch.add_link(tgen.gears["host1"])
  72         switch.add_link(tgen.gears["PE1"])
  73
  74         # PE1-P1
  75         switch = tgen.add_switch("s2")
  76         switch.add_link(tgen.gears["PE1"])
  77         switch.add_link(tgen.gears["P1"])
  78
  79         # P1-PE2
  80         switch = tgen.add_switch("s3")
  81         switch.add_link(tgen.gears["P1"])
  82         switch.add_link(tgen.gears["PE2"])
  83
  84         # PE2-host2
  85         switch = tgen.add_switch("s4")
  86         switch.add_link(tgen.gears["PE2"])
  87         switch.add_link(tgen.gears["host2"])
  88
  89
  90 def setup_module(mod):
  91     "Sets up the pytest environment"
  92     # This function initiates the topology build with Topogen...
  93     tgen = Topogen(TemplateTopo, mod.__name__)
  94     # ... and here it calls Mininet initialization functions.
  95     tgen.start_topology()
  96
  97     pe1 = tgen.gears["PE1"]
  98     pe2 = tgen.gears["PE2"]
  99     p1 = tgen.gears["P1"]
 100
 101     # set up PE bridges with the EVPN member interfaces facing the CE hosts
 102     pe1.run("ip link add name br101 type bridge stp_state 0")
 103     pe1.run("ip addr add 10.10.1.1/24 dev br101")
 104     pe1.run("ip link set dev br101 up")
 105     pe1.run(
 106         "ip link add vxlan101 type vxlan id 101 dstport 4789 local 10.10.10.10 nolearning"
 107     )
 108     pe1.run("ip link set dev vxlan101 master br101")
 109     pe1.run("ip link set up dev vxlan101")
 110     pe1.run("ip link set dev PE1-eth0 master br101")
 111
 112     pe2.run("ip link add name br101 type bridge stp_state 0")
 113     pe2.run("ip addr add 10.10.1.3/24 dev br101")
 114     pe2.run("ip link set dev br101 up")
 115     pe2.run(
 116         "ip link add vxlan101 type vxlan id 101 dstport 4789 local 10.30.30.30 nolearning"
 117     )
 118     pe2.run("ip link set dev vxlan101 master br101")
 119     pe2.run("ip link set up dev vxlan101")
 120     pe2.run("ip link set dev PE2-eth1 master br101")
 121     p1.run("sysctl -w net.ipv4.ip_forward=1")
 122
 123     # This is a sample of configuration loading.
 124     router_list = tgen.routers()
 125
 126     # For all registred routers, load the zebra configuration file
 127     for rname, router in router_list.items():
 128         router.load_config(
 129             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 130         )
 131         router.load_config(
 132             TopoRouter.RD_OSPF, os.path.join(CWD, "{}/ospfd.conf".format(rname))
 133         )
 134         router.load_config(
 135             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
 136         )
 137
 138     # After loading the configurations, this function loads configured daemons.
 139     tgen.start_router()
 140
 141
 142 def teardown_module(mod):
 143     "Teardown the pytest environment"
 144     tgen = get_topogen()
 145
 146     # This function tears down the whole topology.
 147     tgen.stop_topology()
 148
 149
 150 def show_vni_json_elide_ifindex(pe, vni, expected):
 151     output_json = pe.vtysh_cmd("show evpn vni {} json".format(vni), isjson=True)
 152
 153     if "ifindex" in output_json:
 154         output_json.pop("ifindex")
 155
 156     return topotest.json_cmp(output_json, expected)
 157
 158
 159 def test_pe1_converge_evpn():
 160     "Wait for protocol convergence"
 161
 162     tgen = get_topogen()
 163     # Don't run this test if we have any failure.
 164     if tgen.routers_have_failure():
 165         pytest.skip(tgen.errors)
 166
 167     pe1 = tgen.gears["PE1"]
 168     json_file = "{}/{}/evpn.vni.json".format(CWD, pe1.name)
 169     expected = json.loads(open(json_file).read())
 170
 171     test_func = partial(show_vni_json_elide_ifindex, pe1, 101, expected)
 172     _, result = topotest.run_and_expect(test_func, None, count=125, wait=1)
 173     assertmsg = '"{}" JSON output mismatches'.format(pe1.name)
 174     assert result is None, assertmsg
 175     # tgen.mininet_cli()
 176
 177
 178 def test_pe2_converge_evpn():
 179     "Wait for protocol convergence"
 180
 181     tgen = get_topogen()
 182     # Don't run this test if we have any failure.
 183     if tgen.routers_have_failure():
 184         pytest.skip(tgen.errors)
 185
 186     pe2 = tgen.gears["PE2"]
 187     json_file = "{}/{}/evpn.vni.json".format(CWD, pe2.name)
 188     expected = json.loads(open(json_file).read())
 189
 190     test_func = partial(show_vni_json_elide_ifindex, pe2, 101, expected)
 191     _, result = topotest.run_and_expect(test_func, None, count=125, wait=1)
 192     assertmsg = '"{}" JSON output mismatches'.format(pe2.name)
 193     assert result is None, assertmsg
 194     # tgen.mininet_cli()
 195
 196
 197 def mac_learn_test(host, local):
 198     "check the host MAC gets learned by the VNI"
 199
 200     host_output = host.vtysh_cmd("show interface {}-eth0".format(host.name))
 201     int_lines = host_output.splitlines()
 202     for line in int_lines:
 203         line_items = line.split(": ")
 204         if "HWaddr" in line_items[0]:
 205             mac = line_items[1]
 206             break
 207
 208     mac_output = local.vtysh_cmd("show evpn mac vni 101 mac {} json".format(mac))
 209     mac_output_json = json.loads(mac_output)
 210     assertmsg = "Local MAC output does not match interface mac {}".format(mac)
 211     assert mac_output_json[mac]["type"] == "local", assertmsg
 212
 213
 214 def mac_test_local_remote(local, remote):
 215     "test MAC transfer between local and remote"
 216
 217     local_output = local.vtysh_cmd("show evpn mac vni all json")
 218     remote_output = remote.vtysh_cmd("show evpn mac vni all json")
 219     local_output_vni = local.vtysh_cmd("show evpn vni detail json")
 220     local_output_json = json.loads(local_output)
 221     remote_output_json = json.loads(remote_output)
 222     local_output_vni_json = json.loads(local_output_vni)
 223
 224     for vni in local_output_json:
 225         mac_list = local_output_json[vni]["macs"]
 226         for mac in mac_list:
 227             if mac_list[mac]["type"] == "local" and mac_list[mac]["intf"] != "br101":
 228                 assertmsg = "JSON output mismatches local: {} remote: {}".format(
 229                     local_output_vni_json[0]["vtepIp"],
 230                     remote_output_json[vni]["macs"][mac]["remoteVtep"],
 231                 )
 232                 assert (
 233                     remote_output_json[vni]["macs"][mac]["remoteVtep"]
 234                     == local_output_vni_json[0]["vtepIp"]
 235                 ), assertmsg
 236
 237
 238 def test_learning_pe1():
 239     "test MAC learning on PE1"
 240
 241     tgen = get_topogen()
 242     # Don't run this test if we have any failure.
 243     if tgen.routers_have_failure():
 244         pytest.skip(tgen.errors)
 245
 246     host1 = tgen.gears["host1"]
 247     pe1 = tgen.gears["PE1"]
 248     mac_learn_test(host1, pe1)
 249
 250
 251 def test_learning_pe2():
 252     "test MAC learning on PE2"
 253
 254     tgen = get_topogen()
 255     # Don't run this test if we have any failure.
 256     if tgen.routers_have_failure():
 257         pytest.skip(tgen.errors)
 258
 259     host2 = tgen.gears["host2"]
 260     pe2 = tgen.gears["PE2"]
 261     mac_learn_test(host2, pe2)
 262
 263
 264 def test_local_remote_mac_pe1():
 265     " Test MAC transfer PE1 local and PE2 remote"
 266
 267     tgen = get_topogen()
 268     # Don't run this test if we have any failure.
 269     if tgen.routers_have_failure():
 270         pytest.skip(tgen.errors)
 271
 272     pe1 = tgen.gears["PE1"]
 273     pe2 = tgen.gears["PE2"]
 274     mac_test_local_remote(pe1, pe2)
 275
 276
 277 def test_local_remote_mac_pe2():
 278     " Test MAC transfer PE2 local and PE1 remote"
 279
 280     tgen = get_topogen()
 281     # Don't run this test if we have any failure.
 282     if tgen.routers_have_failure():
 283         pytest.skip(tgen.errors)
 284
 285     pe1 = tgen.gears["PE1"]
 286     pe2 = tgen.gears["PE2"]
 287     mac_test_local_remote(pe2, pe1)
 288
 289     # Memory leak test template
 290
 291
 292 def ip_learn_test(tgen, host, local, remote, ip_addr):
 293     "check the host IP gets learned by the VNI"
 294     host_output = host.vtysh_cmd("show interface {}-eth0".format(host.name))
 295     int_lines = host_output.splitlines()
 296     for line in int_lines:
 297         line_items = line.split(": ")
 298         if "HWaddr" in line_items[0]:
 299             mac = line_items[1]
 300             break
 301     print(host_output)
 302
 303     # check we have a local association between the MAC and IP
 304     local_output = local.vtysh_cmd("show evpn mac vni 101 mac {} json".format(mac))
 305     print(local_output)
 306     local_output_json = json.loads(local_output)
 307     mac_type = local_output_json[mac]["type"]
 308     assertmsg = "Failed to learn local IP address on host {}".format(host.name)
 309     assert local_output_json[mac]["neighbors"] != "none", assertmsg
 310     learned_ip = local_output_json[mac]["neighbors"]["active"][0]
 311
 312     assertmsg = "local learned mac wrong type: {} ".format(mac_type)
 313     assert mac_type == "local", assertmsg
 314
 315     assertmsg = (
 316         "learned address mismatch with configured address host: {} learned: {}".format(
 317             ip_addr, learned_ip
 318         )
 319     )
 320     assert ip_addr == learned_ip, assertmsg
 321
 322     # now lets check the remote
 323     count = 0
 324     converged = False
 325     while count < 30:
 326         remote_output = remote.vtysh_cmd(
 327             "show evpn mac vni 101 mac {} json".format(mac)
 328         )
 329         print(remote_output)
 330         remote_output_json = json.loads(remote_output)
 331         type = remote_output_json[mac]["type"]
 332         if not remote_output_json[mac]["neighbors"] == "none":
 333             # due to a kernel quirk, learned IPs can be inactive
 334             if (
 335                 remote_output_json[mac]["neighbors"]["active"]
 336                 or remote_output_json[mac]["neighbors"]["inactive"]
 337             ):
 338                 converged = True
 339                 break
 340         count += 1
 341         sleep(1)
 342
 343     print("tries: {}".format(count))
 344     assertmsg = "{} remote learned mac no address: {} ".format(host.name, mac)
 345     # some debug for this failure
 346     if not converged == True:
 347         log_output = remote.run("cat zebra.log")
 348         print(log_output)
 349
 350     assert converged == True, assertmsg
 351     if remote_output_json[mac]["neighbors"]["active"]:
 352         learned_ip = remote_output_json[mac]["neighbors"]["active"][0]
 353     else:
 354         learned_ip = remote_output_json[mac]["neighbors"]["inactive"][0]
 355     assertmsg = "remote learned mac wrong type: {} ".format(type)
 356     assert type == "remote", assertmsg
 357
 358     assertmsg = "remote learned address mismatch with configured address host: {} learned: {}".format(
 359         ip_addr, learned_ip
 360     )
 361     assert ip_addr == learned_ip, assertmsg
 362
 363
 364 def test_ip_pe1_learn():
 365     "run the IP learn test for PE1"
 366
 367     tgen = get_topogen()
 368     host1 = tgen.gears["host1"]
 369     pe1 = tgen.gears["PE1"]
 370     pe2 = tgen.gears["PE2"]
 371     pe2.vtysh_cmd("debug zebra vxlan")
 372     pe2.vtysh_cmd("debug zebra kernel")
 373     # lets populate that arp cache
 374     host1.run("ping -c1 10.10.1.1")
 375     ip_learn_test(tgen, host1, pe1, pe2, "10.10.1.55")
 376     # tgen.mininet_cli()
 377
 378
 379 def test_ip_pe2_learn():
 380     "run the IP learn test for PE2"
 381
 382     tgen = get_topogen()
 383     host2 = tgen.gears["host2"]
 384     pe1 = tgen.gears["PE1"]
 385     pe2 = tgen.gears["PE2"]
 386     pe1.vtysh_cmd("debug zebra vxlan")
 387     pe1.vtysh_cmd("debug zebra kernel")
 388     # lets populate that arp cache
 389     host2.run("ping -c1 10.10.1.3")
 390     ip_learn_test(tgen, host2, pe2, pe1, "10.10.1.56")
 391     # tgen.mininet_cli()
 392
 393
 394 def test_memory_leak():
 395     "Run the memory leak test and report results."
 396     tgen = get_topogen()
 397     if not tgen.is_memleak_enabled():
 398         pytest.skip("Memory leak test/report is disabled")
 399
 400     tgen.report_memory_leaks()
 401
 402
 403 if __name__ == "__main__":
 404     args = ["-s"] + sys.argv[1:]
 405     sys.exit(pytest.main(args))