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