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