tests/topotests/evpn_pim_1/test_evpn_pim_topo1.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_evpn-pim_topo1.py
   5 #
   6 # Copyright (c) 2017 by
   7 # Cumulus Networks, Inc.
   8 # Donald Sharp
   9 #
  10 # Permission to use, copy, modify, and/or distribute this software
  11 # for any purpose with or without fee is hereby granted, provided
  12 # that the above copyright notice and this permission notice appear
  13 # in all copies.
  14 #
  15 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  16 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  17 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  18 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  19 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  20 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  21 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  22 # OF THIS SOFTWARE.
  23 #
  24
  25 """
  26 test_evpn_pim_topo1.py: Testing evpn-pim
  27
  28 """
  29
  30 import os
  31 import re
  32 import sys
  33 import pytest
  34 import json
  35 from functools import partial
  36
  37 pytestmark = [pytest.mark.pimd]
  38
  39 # Save the Current Working Directory to find configuration files.
  40 CWD = os.path.dirname(os.path.realpath(__file__))
  41 sys.path.append(os.path.join(CWD, "../"))
  42
  43 # pylint: disable=C0413
  44 # Import topogen and topotest helpers
  45 from lib import topotest
  46 from lib.topogen import Topogen, TopoRouter, get_topogen
  47 from lib.topolog import logger
  48
  49 # Required to instantiate the topology builder class.
  50 from mininet.topo import Topo
  51
  52 pytestmark = [pytest.mark.bgpd, pytest.mark.bgpd]
  53
  54
  55 #####################################################
  56 ##
  57 ##   Network Topology Definition
  58 ##
  59 #####################################################
  60
  61
  62 class NetworkTopo(Topo):
  63     "evpn-pim Topology 1"
  64
  65     def build(self, **_opts):
  66         "Build function"
  67
  68         tgen = get_topogen(self)
  69
  70         tgen.add_router("spine")
  71         tgen.add_router("leaf1")
  72         tgen.add_router("leaf2")
  73         tgen.add_router("host1")
  74         tgen.add_router("host2")
  75
  76         # On main router
  77         # First switch is for a dummy interface (for local network)
  78         # spine-eth0 is connected to leaf1-eth0
  79         switch = tgen.add_switch("sw1")
  80         switch.add_link(tgen.gears["spine"])
  81         switch.add_link(tgen.gears["leaf1"])
  82
  83         # spine-eth1 is connected to leaf2-eth0
  84         switch = tgen.add_switch("sw2")
  85         switch.add_link(tgen.gears["spine"])
  86         switch.add_link(tgen.gears["leaf2"])
  87
  88         # leaf1-eth1 is connected to host1-eth0
  89         switch = tgen.add_switch("sw3")
  90         switch.add_link(tgen.gears["leaf1"])
  91         switch.add_link(tgen.gears["host1"])
  92
  93         # leaf2-eth1 is connected to host2-eth0
  94         switch = tgen.add_switch("sw4")
  95         switch.add_link(tgen.gears["leaf2"])
  96         switch.add_link(tgen.gears["host2"])
  97
  98
  99 #####################################################
 100 ##
 101 ##   Tests starting
 102 ##
 103 #####################################################
 104
 105
 106 def setup_module(module):
 107     "Setup topology"
 108     tgen = Topogen(NetworkTopo, module.__name__)
 109     tgen.start_topology()
 110
 111     leaf1 = tgen.gears["leaf1"]
 112     leaf2 = tgen.gears["leaf2"]
 113
 114     leaf1.run("brctl addbr brleaf1")
 115     leaf2.run("brctl addbr brleaf2")
 116     leaf1.run("ip link set dev brleaf1 up")
 117     leaf2.run("ip link set dev brleaf2 up")
 118     leaf1.run(
 119         "ip link add vxlan0 type vxlan id 42 group 239.1.1.1 dev leaf1-eth1 dstport 4789"
 120     )
 121     leaf2.run(
 122         "ip link add vxlan0 type vxlan id 42 group 239.1.1.1 dev leaf2-eth1 dstport 4789"
 123     )
 124     leaf1.run("brctl addif brleaf1 vxlan0")
 125     leaf2.run("brctl addif brleaf2 vxlan0")
 126     leaf1.run("ip link set up dev vxlan0")
 127     leaf2.run("ip link set up dev vxlan0")
 128     # tgen.mininet_cli()
 129     # This is a sample of configuration loading.
 130     router_list = tgen.routers()
 131     for rname, router in router_list.items():
 132         router.load_config(
 133             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 134         )
 135         router.load_config(
 136             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
 137         )
 138         router.load_config(
 139             TopoRouter.RD_PIM, os.path.join(CWD, "{}/pimd.conf".format(rname))
 140         )
 141     tgen.start_router()
 142     # tgen.mininet_cli()
 143
 144
 145 def teardown_module(_mod):
 146     "Teardown the pytest environment"
 147     tgen = get_topogen()
 148
 149     # This function tears down the whole topology.
 150     tgen.stop_topology()
 151
 152
 153 def test_converge_protocols():
 154     "Wait for protocol convergence"
 155
 156     tgen = get_topogen()
 157     # Don't run this test if we have any failure.
 158     if tgen.routers_have_failure():
 159         pytest.skip(tgen.errors)
 160
 161     spine = tgen.gears["spine"]
 162     json_file = "{}/{}/bgp.summ.json".format(CWD, spine.name)
 163     expected = json.loads(open(json_file).read())
 164
 165     test_func = partial(
 166         topotest.router_json_cmp, spine, "show bgp ipv4 uni summ json", expected
 167     )
 168     _, result = topotest.run_and_expect(test_func, None, count=125, wait=1)
 169     assertmsg = '"{}" JSON output mismatches'.format(spine.name)
 170     assert result is None, assertmsg
 171     # tgen.mininet_cli()
 172
 173
 174 def test_multicast_groups_on_rp():
 175     "Ensure the multicast groups show up on the spine"
 176     # This test implicitly tests the auto mcast groups
 177     # of the created vlans and then the auto-joins that
 178     # pim will do to the RP( spine )
 179
 180     tgen = get_topogen()
 181
 182     if tgen.routers_have_failure():
 183         pytest.skip(tgen.errors)
 184
 185     spine = tgen.gears["spine"]
 186     json_file = "{}/{}/join-info.json".format(CWD, spine.name)
 187     expected = json.loads(open(json_file).read())
 188
 189     test_func = partial(
 190         topotest.router_json_cmp, spine, "show ip pim join json", expected
 191     )
 192     _, result = topotest.run_and_expect(test_func, None, count=30, wait=1)
 193     assertmsg = '"{}" JSON output mismatches'.format(spine.name)
 194     assert result is None, assertmsg
 195     # tgen.mininet_cli()
 196
 197
 198 def test_shutdown_check_stderr():
 199     if os.environ.get("TOPOTESTS_CHECK_STDERR") is None:
 200         pytest.skip("Skipping test for Stderr output and memory leaks")
 201
 202     tgen = get_topogen()
 203     # Don't run this test if we have any failure.
 204     if tgen.routers_have_failure():
 205         pytest.skip(tgen.errors)
 206
 207     logger.info("Verifying unexpected STDERR output from daemons")
 208
 209     router_list = tgen.routers().values()
 210     for router in router_list:
 211         router.stop()
 212
 213         log = tgen.net[router.name].getStdErr("pimd")
 214         if log:
 215             logger.error("PIMd StdErr Log:" + log)
 216         log = tgen.net[router.name].getStdErr("bgpd")
 217         if log:
 218             logger.error("BGPd StdErr Log:" + log)
 219         log = tgen.net[router.name].getStdErr("zebra")
 220         if log:
 221             logger.error("Zebra StdErr Log:" + log)
 222
 223
 224 if __name__ == "__main__":
 225     args = ["-s"] + sys.argv[1:]
 226     sys.exit(pytest.main(args))