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