tests/topotests/bgp_aggregate_address_topo1/test_bgp_aggregate_address_topo1.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_bgp_aggregate_address_topo1.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2020 by
   8 # Network Device Education Foundation, Inc. ("NetDEF")
   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 BGP aggregate address features.
  27 """
  28
  29 import os
  30 import sys
  31 import pytest
  32 import functools
  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 from lib.topolog import logger
  41
  42 pytestmark = [pytest.mark.bgpd]
  43
  44
  45 def build_topo(tgen):
  46     r1 = tgen.add_router("r1")
  47     r2 = tgen.add_router("r2")
  48     peer1 = tgen.add_exabgp_peer("peer1", ip="10.0.0.2", defaultRoute="via 10.0.0.1")
  49
  50     switch = tgen.add_switch("s1")
  51     switch.add_link(r1)
  52     switch.add_link(peer1)
  53
  54     switch = tgen.add_switch("s2")
  55     switch.add_link(r1)
  56     switch.add_link(r2)
  57
  58
  59 def setup_module(mod):
  60     tgen = Topogen(build_topo, mod.__name__)
  61     tgen.start_topology()
  62
  63     router = tgen.gears["r1"]
  64     router.load_config(TopoRouter.RD_ZEBRA, os.path.join(CWD, "r1/zebra.conf"))
  65     router.load_config(TopoRouter.RD_BGP, os.path.join(CWD, "r1/bgpd.conf"))
  66     router.start()
  67
  68     router = tgen.gears["r2"]
  69     router.load_config(TopoRouter.RD_ZEBRA, os.path.join(CWD, "r2/zebra.conf"))
  70     router.load_config(TopoRouter.RD_BGP, os.path.join(CWD, "r2/bgpd.conf"))
  71     router.start()
  72
  73     peer = tgen.gears["peer1"]
  74     peer.start(os.path.join(CWD, "peer1"), os.path.join(CWD, "exabgp.env"))
  75
  76
  77 def teardown_module(mod):
  78     tgen = get_topogen()
  79     tgen.stop_topology()
  80
  81
  82 def expect_route(router_name, routes_expected):
  83     "Helper function to avoid repeated code."
  84     tgen = get_topogen()
  85     test_func = functools.partial(
  86         topotest.router_json_cmp,
  87         tgen.gears[router_name],
  88         "show ip route json",
  89         routes_expected,
  90     )
  91     _, result = topotest.run_and_expect(test_func, None, count=120, wait=1)
  92     assertmsg = '"{}" BGP convergence failure'.format(router_name)
  93     assert result is None, assertmsg
  94
  95
  96 def test_expect_convergence():
  97     "Test that BGP protocol converged."
  98
  99     tgen = get_topogen()
 100     if tgen.routers_have_failure():
 101         pytest.skip(tgen.errors)
 102
 103     logger.info("waiting for protocols to converge")
 104
 105     def expect_loopback_route(router, iptype, route, proto):
 106         "Wait until route is present on RIB for protocol."
 107         logger.info("waiting route {} in {}".format(route, router))
 108         test_func = functools.partial(
 109             topotest.router_json_cmp,
 110             tgen.gears[router],
 111             "show {} route json".format(iptype),
 112             {route: [{"protocol": proto}]},
 113         )
 114         _, result = topotest.run_and_expect(test_func, None, count=130, wait=1)
 115         assertmsg = '"{}" BGP convergence failure'.format(router)
 116         assert result is None, assertmsg
 117
 118     expect_loopback_route("r2", "ip", "10.254.254.1/32", "bgp")
 119     expect_loopback_route("r2", "ip", "10.254.254.3/32", "bgp")
 120
 121
 122 def test_bgp_aggregate_address_matching_med_only():
 123     "Test that the command matching-MED-only works."
 124
 125     tgen = get_topogen()
 126     if tgen.routers_have_failure():
 127         pytest.skip(tgen.errors)
 128
 129     routes_expected = {
 130         # All MED matches, aggregation must exist.
 131         "192.168.0.0/24": [{"protocol": "bgp", "metric": 0}],
 132         "192.168.0.1/32": [{"protocol": "bgp", "metric": 10}],
 133         "192.168.0.2/32": [{"protocol": "bgp", "metric": 10}],
 134         "192.168.0.3/32": [{"protocol": "bgp", "metric": 10}],
 135         # Non matching MED: aggregation must not exist.
 136         "192.168.1.0/24": None,
 137         "192.168.1.1/32": [{"protocol": "bgp", "metric": 10}],
 138         "192.168.1.2/32": [{"protocol": "bgp", "metric": 10}],
 139         "192.168.1.3/32": [{"protocol": "bgp", "metric": 20}],
 140     }
 141
 142     test_func = functools.partial(
 143         topotest.router_json_cmp,
 144         tgen.gears["r2"],
 145         "show ip route json",
 146         routes_expected,
 147     )
 148     _, result = topotest.run_and_expect(test_func, None, count=20, wait=1)
 149     assertmsg = '"r2" BGP convergence failure'
 150     assert result is None, assertmsg
 151
 152
 153 def test_bgp_aggregate_address_match_and_suppress():
 154     "Test that the command matching-MED-only with suppression works."
 155
 156     tgen = get_topogen()
 157     if tgen.routers_have_failure():
 158         pytest.skip(tgen.errors)
 159
 160     tgen.gears["r1"].vtysh_multicmd(
 161         """
 162 configure terminal
 163 router bgp 65000
 164 address-family ipv4 unicast
 165 no aggregate-address 192.168.0.0/24 matching-MED-only
 166 no aggregate-address 192.168.1.0/24 matching-MED-only
 167 aggregate-address 192.168.0.0/24 matching-MED-only summary-only
 168 aggregate-address 192.168.1.0/24 matching-MED-only summary-only
 169 """
 170     )
 171
 172     routes_expected = {
 173         # All MED matches, aggregation must exist.
 174         "192.168.0.0/24": [{"protocol": "bgp", "metric": 0}],
 175         "192.168.0.1/32": None,
 176         "192.168.0.2/32": None,
 177         "192.168.0.3/32": None,
 178         # Non matching MED: aggregation must not exist.
 179         "192.168.1.0/24": None,
 180         "192.168.1.1/32": [{"protocol": "bgp", "metric": 10}],
 181         "192.168.1.2/32": [{"protocol": "bgp", "metric": 10}],
 182         "192.168.1.3/32": [{"protocol": "bgp", "metric": 20}],
 183     }
 184
 185     test_func = functools.partial(
 186         topotest.router_json_cmp,
 187         tgen.gears["r2"],
 188         "show ip route json",
 189         routes_expected,
 190     )
 191     _, result = topotest.run_and_expect(test_func, None, count=120, wait=1)
 192     assertmsg = '"r2" BGP convergence failure'
 193     assert result is None, assertmsg
 194
 195
 196 def test_bgp_aggregate_address_suppress_map():
 197     "Test that the command suppress-map works."
 198
 199     tgen = get_topogen()
 200     if tgen.routers_have_failure():
 201         pytest.skip(tgen.errors)
 202
 203     expect_route(
 204         "r2",
 205         {
 206             "192.168.2.0/24": [{"protocol": "bgp"}],
 207             "192.168.2.1/32": None,
 208             "192.168.2.2/32": [{"protocol": "bgp"}],
 209             "192.168.2.3/32": [{"protocol": "bgp"}],
 210         },
 211     )
 212
 213     # Change route map and test again.
 214     tgen.gears["r1"].vtysh_multicmd(
 215         """
 216 configure terminal
 217 router bgp 65000
 218 address-family ipv4 unicast
 219 no aggregate-address 192.168.2.0/24 suppress-map rm-sup-one
 220 aggregate-address 192.168.2.0/24 suppress-map rm-sup-two
 221 """
 222     )
 223
 224     expect_route(
 225         "r2",
 226         {
 227             "192.168.2.0/24": [{"protocol": "bgp"}],
 228             "192.168.2.1/32": [{"protocol": "bgp"}],
 229             "192.168.2.2/32": None,
 230             "192.168.2.3/32": [{"protocol": "bgp"}],
 231         },
 232     )
 233
 234
 235 def test_bgp_aggregate_address_suppress_map_update_route_map():
 236     "Test that the suppress-map late route map creation works."
 237     tgen = get_topogen()
 238     if tgen.routers_have_failure():
 239         pytest.skip(tgen.errors)
 240
 241     tgen.gears["r1"].vtysh_multicmd(
 242         """
 243 configure terminal
 244 router bgp 65000
 245 address-family ipv4 unicast
 246 no aggregate-address 192.168.2.0/24 suppress-map rm-sup-two
 247 aggregate-address 192.168.2.0/24 suppress-map rm-sup-three
 248 """
 249     )
 250
 251     expect_route(
 252         "r2",
 253         {
 254             "192.168.2.0/24": [{"protocol": "bgp"}],
 255             "192.168.2.1/32": [{"protocol": "bgp"}],
 256             "192.168.2.2/32": [{"protocol": "bgp"}],
 257             "192.168.2.3/32": [{"protocol": "bgp"}],
 258         },
 259     )
 260
 261     # Create missing route map and test again.
 262     tgen.gears["r1"].vtysh_multicmd(
 263         """
 264 configure terminal
 265 route-map rm-sup-three permit 10
 266 match ip address acl-sup-three
 267 """
 268     )
 269
 270     expect_route(
 271         "r2",
 272         {
 273             "192.168.2.0/24": [{"protocol": "bgp"}],
 274             "192.168.2.1/32": [{"protocol": "bgp"}],
 275             "192.168.2.2/32": [{"protocol": "bgp"}],
 276             "192.168.2.3/32": None,
 277         },
 278     )
 279
 280
 281 def test_memory_leak():
 282     "Run the memory leak test and report results."
 283     tgen = get_topogen()
 284     if not tgen.is_memleak_enabled():
 285         pytest.skip("Memory leak test/report is disabled")
 286
 287     tgen.report_memory_leaks()
 288
 289
 290 if __name__ == "__main__":
 291     args = ["-s"] + sys.argv[1:]
 292     sys.exit(pytest.main(args))