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 #####################################################
  53 ##
  54 ##   Network Topology Definition
  55 ##
  56 #####################################################
  57
  58
  59 class NetworkTopo(Topo):
  60     "evpn-pim Topology 1"
  61
  62     def build(self, **_opts):
  63         "Build function"
  64
  65         tgen = get_topogen(self)
  66
  67         tgen.add_router("spine")
  68         tgen.add_router("leaf1")
  69         tgen.add_router("leaf2")
  70         tgen.add_router("host1")
  71         tgen.add_router("host2")
  72
  73         # On main router
  74         # First switch is for a dummy interface (for local network)
  75         # spine-eth0 is connected to leaf1-eth0
  76         switch = tgen.add_switch("sw1")
  77         switch.add_link(tgen.gears["spine"])
  78         switch.add_link(tgen.gears["leaf1"])
  79
  80         # spine-eth1 is connected to leaf2-eth0
  81         switch = tgen.add_switch("sw2")
  82         switch.add_link(tgen.gears["spine"])
  83         switch.add_link(tgen.gears["leaf2"])
  84
  85         # leaf1-eth1 is connected to host1-eth0
  86         switch = tgen.add_switch("sw3")
  87         switch.add_link(tgen.gears["leaf1"])
  88         switch.add_link(tgen.gears["host1"])
  89
  90         # leaf2-eth1 is connected to host2-eth0
  91         switch = tgen.add_switch("sw4")
  92         switch.add_link(tgen.gears["leaf2"])
  93         switch.add_link(tgen.gears["host2"])
  94
  95
  96 #####################################################
  97 ##
  98 ##   Tests starting
  99 ##
 100 #####################################################
 101
 102
 103 def setup_module(module):
 104     "Setup topology"
 105     tgen = Topogen(NetworkTopo, module.__name__)
 106     tgen.start_topology()
 107
 108     leaf1 = tgen.gears["leaf1"]
 109     leaf2 = tgen.gears["leaf2"]
 110
 111     leaf1.run("brctl addbr brleaf1")
 112     leaf2.run("brctl addbr brleaf2")
 113     leaf1.run("ip link set dev brleaf1 up")
 114     leaf2.run("ip link set dev brleaf2 up")
 115     leaf1.run(
 116         "ip link add vxlan0 type vxlan id 42 group 239.1.1.1 dev leaf1-eth1 dstport 4789"
 117     )
 118     leaf2.run(
 119         "ip link add vxlan0 type vxlan id 42 group 239.1.1.1 dev leaf2-eth1 dstport 4789"
 120     )
 121     leaf1.run("brctl addif brleaf1 vxlan0")
 122     leaf2.run("brctl addif brleaf2 vxlan0")
 123     leaf1.run("ip link set up dev vxlan0")
 124     leaf2.run("ip link set up dev vxlan0")
 125     # tgen.mininet_cli()
 126     # This is a sample of configuration loading.
 127     router_list = tgen.routers()
 128     for rname, router in router_list.items():
 129         router.load_config(
 130             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 131         )
 132         router.load_config(
 133             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
 134         )
 135         router.load_config(
 136             TopoRouter.RD_PIM, os.path.join(CWD, "{}/pimd.conf".format(rname))
 137         )
 138     tgen.start_router()
 139     # tgen.mininet_cli()
 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 test_converge_protocols():
 151     "Wait for protocol convergence"
 152
 153     tgen = get_topogen()
 154     # Don't run this test if we have any failure.
 155     if tgen.routers_have_failure():
 156         pytest.skip(tgen.errors)
 157
 158     spine = tgen.gears["spine"]
 159     json_file = "{}/{}/bgp.summ.json".format(CWD, spine.name)
 160     expected = json.loads(open(json_file).read())
 161
 162     test_func = partial(
 163         topotest.router_json_cmp, spine, "show bgp ipv4 uni summ json", expected
 164     )
 165     _, result = topotest.run_and_expect(test_func, None, count=125, wait=1)
 166     assertmsg = '"{}" JSON output mismatches'.format(spine.name)
 167     assert result is None, assertmsg
 168     # tgen.mininet_cli()
 169
 170
 171 def test_multicast_groups_on_rp():
 172     "Ensure the multicast groups show up on the spine"
 173     # This test implicitly tests the auto mcast groups
 174     # of the created vlans and then the auto-joins that
 175     # pim will do to the RP( spine )
 176
 177     tgen = get_topogen()
 178
 179     if tgen.routers_have_failure():
 180         pytest.skip(tgen.errors)
 181
 182     spine = tgen.gears["spine"]
 183     json_file = "{}/{}/join-info.json".format(CWD, spine.name)
 184     expected = json.loads(open(json_file).read())
 185
 186     test_func = partial(
 187         topotest.router_json_cmp, spine, "show ip pim join json", expected
 188     )
 189     _, result = topotest.run_and_expect(test_func, None, count=30, wait=1)
 190     assertmsg = '"{}" JSON output mismatches'.format(spine.name)
 191     assert result is None, assertmsg
 192     # tgen.mininet_cli()
 193
 194
 195 def test_shutdown_check_stderr():
 196     if os.environ.get("TOPOTESTS_CHECK_STDERR") is None:
 197         pytest.skip("Skipping test for Stderr output and memory leaks")
 198
 199     tgen = get_topogen()
 200     # Don't run this test if we have any failure.
 201     if tgen.routers_have_failure():
 202         pytest.skip(tgen.errors)
 203
 204     logger.info("Verifying unexpected STDERR output from daemons")
 205
 206     router_list = tgen.routers().values()
 207     for router in router_list:
 208         router.stop()
 209
 210         log = tgen.net[router.name].getStdErr("pimd")
 211         if log:
 212             logger.error("PIMd StdErr Log:" + log)
 213         log = tgen.net[router.name].getStdErr("bgpd")
 214         if log:
 215             logger.error("BGPd StdErr Log:" + log)
 216         log = tgen.net[router.name].getStdErr("zebra")
 217         if log:
 218             logger.error("Zebra StdErr Log:" + log)
 219
 220
 221 if __name__ == "__main__":
 222     args = ["-s"] + sys.argv[1:]
 223     sys.exit(pytest.main(args))