tests/topotests/bgp_suppress_fib/test_bgp_suppress_fib.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_bgp_suppress_fib.py
   5 #
   6 # Copyright (c) 2019 by
   7 #
   8 # Permission to use, copy, modify, and/or distribute this software
   9 # for any purpose with or without fee is hereby granted, provided
  10 # that the above copyright notice and this permission notice appear
  11 # in all copies.
  12 #
  13 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  14 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  15 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  16 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  17 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  18 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  19 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  20 # OF THIS SOFTWARE.
  21 #
  22
  23 """
  24 """
  25
  26 import os
  27 import sys
  28 import json
  29 import pytest
  30 from functools import partial
  31 from time import sleep
  32 from lib.topolog import logger
  33
  34 CWD = os.path.dirname(os.path.realpath(__file__))
  35 sys.path.append(os.path.join(CWD, "../"))
  36
  37 # pylint: disable=C0413
  38 from lib import topotest
  39 from lib.topogen import Topogen, TopoRouter, get_topogen
  40
  41 pytestmark = [pytest.mark.bgpd]
  42
  43
  44 def build_topo(tgen):
  45     for routern in range(1, 4):
  46         tgen.add_router("r{}".format(routern))
  47
  48     switch = tgen.add_switch("s1")
  49     switch.add_link(tgen.gears["r1"])
  50     switch.add_link(tgen.gears["r2"])
  51
  52     switch = tgen.add_switch("s2")
  53     switch.add_link(tgen.gears["r2"])
  54     switch.add_link(tgen.gears["r3"])
  55
  56
  57 def setup_module(mod):
  58     tgen = Topogen(build_topo, mod.__name__)
  59     tgen.start_topology()
  60
  61     router_list = tgen.routers()
  62
  63     for i, (rname, router) in enumerate(router_list.items(), 1):
  64         router.load_config(
  65             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  66         )
  67         router.load_config(
  68             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
  69         )
  70
  71     tgen.start_router()
  72
  73
  74 def teardown_module(mod):
  75     tgen = get_topogen()
  76     tgen.stop_topology()
  77
  78
  79 def test_bgp_route():
  80     tgen = get_topogen()
  81
  82     if tgen.routers_have_failure():
  83         pytest.skip(tgen.errors)
  84
  85     r3 = tgen.gears["r3"]
  86
  87     json_file = "{}/r3/v4_route.json".format(CWD)
  88     expected = json.loads(open(json_file).read())
  89
  90     test_func = partial(
  91         topotest.router_json_cmp,
  92         r3,
  93         "show ip route 40.0.0.0 json",
  94         expected,
  95     )
  96     _, result = topotest.run_and_expect(test_func, None, count=20, wait=0.5)
  97     assertmsg = '"r3" JSON output mismatches'
  98     assert result is None, assertmsg
  99
 100     json_file = "{}/r3/v4_route2.json".format(CWD)
 101     expected = json.loads(open(json_file).read())
 102
 103     test_func = partial(
 104         topotest.router_json_cmp,
 105         r3,
 106         "show ip route 50.0.0.0 json",
 107         expected,
 108     )
 109     _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 110     assertmsg = '"r3" JSON output mismatches'
 111     assert result is None, assertmsg
 112
 113     json_file = "{}/r3/v4_route3.json".format(CWD)
 114     expected = json.loads(open(json_file).read())
 115
 116     test_func = partial(
 117         topotest.router_json_cmp,
 118         r3,
 119         "show ip route 10.0.0.3 json",
 120         expected,
 121     )
 122     _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 123
 124
 125 def test_bgp_better_admin_won():
 126     "A better Admin distance protocol may come along and knock us out"
 127
 128     tgen = get_topogen()
 129
 130     if tgen.routers_have_failure():
 131         pytest.skip(tgen.errors)
 132
 133     r2 = tgen.gears["r2"]
 134     r2.vtysh_cmd("conf\nip route 40.0.0.0/8 10.0.0.10")
 135
 136     json_file = "{}/r2/v4_override.json".format(CWD)
 137     expected = json.loads(open(json_file).read())
 138
 139     logger.info(expected)
 140     test_func = partial(
 141         topotest.router_json_cmp, r2, "show ip route 40.0.0.0 json", expected
 142     )
 143
 144     _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 145     assertmsg = '"r2" static route did not take over'
 146     assert result is None, assertmsg
 147
 148     r3 = tgen.gears["r3"]
 149
 150     json_file = "{}/r3/v4_override.json".format(CWD)
 151     expected = json.loads(open(json_file).read())
 152
 153     test_func = partial(
 154         topotest.router_json_cmp, r3, "show ip route 40.0.0.0 json", expected
 155     )
 156
 157     _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 158     assertmsg = '"r3" route to 40.0.0.0 should have been lost'
 159     assert result is None, assertmsg
 160
 161     r2.vtysh_cmd("conf\nno ip route 40.0.0.0/8 10.0.0.10")
 162
 163     json_file = "{}/r3/v4_route.json".format(CWD)
 164     expected = json.loads(open(json_file).read())
 165
 166     test_func = partial(
 167         topotest.router_json_cmp,
 168         r3,
 169         "show ip route 40.0.0.0 json",
 170         expected,
 171     )
 172     _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 173     assertmsg = '"r3" route to 40.0.0.0 did not come back'
 174     assert result is None, assertmsg
 175
 176
 177 def test_bgp_allow_as_in():
 178     tgen = get_topogen()
 179
 180     if tgen.routers_have_failure():
 181         pytest.skip(tgen.errors)
 182
 183     r2 = tgen.gears["r2"]
 184
 185     config_file = "{}/r2/bgpd.allowas_in.conf".format(CWD)
 186     r2.run("vtysh -f {}".format(config_file))
 187
 188     json_file = "{}/r2/bgp_ipv4_allowas.json".format(CWD)
 189     expected = json.loads(open(json_file).read())
 190
 191     test_func = partial(
 192         topotest.router_json_cmp,
 193         r2,
 194         "show bgp ipv4 uni 192.168.1.1/32 json",
 195         expected,
 196     )
 197     _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 198     assertmsg = '"r2" static redistribution failed into bgp'
 199     assert result is None, assertmsg
 200
 201     r1 = tgen.gears["r1"]
 202
 203     json_file = "{}/r1/bgp_ipv4_allowas.json".format(CWD)
 204     expected = json.loads(open(json_file).read())
 205
 206     test_func = partial(
 207         topotest.router_json_cmp,
 208         r1,
 209         "show bgp ipv4 uni 192.168.1.1/32 json",
 210         expected,
 211     )
 212
 213     _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 214     assertmsg = '"r1" 192.168.1.1/32 route should have arrived'
 215     assert result is None, assertmsg
 216
 217     r2.vtysh_cmd("conf\nno ip route 192.168.1.1/32 10.0.0.10")
 218
 219     json_file = "{}/r2/no_bgp_ipv4_allowas.json".format(CWD)
 220     expected = json.loads(open(json_file).read())
 221
 222     test_func = partial(
 223         topotest.router_json_cmp,
 224         r2,
 225         "show bgp ipv4 uni 192.168.1.1/32 json",
 226         expected,
 227     )
 228
 229     _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 230     assertmsg = '"r2" 192.168.1.1/32 route should be gone'
 231     assert result is None, assertmsg
 232
 233
 234 if __name__ == "__main__":
 235     args = ["-s"] + sys.argv[1:]
 236     sys.exit(pytest.main(args))