tests/topotests/bgp_community_change_update/test_bgp_community_change_update.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 # Copyright (c) 2020 by
   5 # Donatas Abraitis <donatas.abraitis@gmail.com>
   6 #
   7
   8 r"""
   9 Reference: https://www.cmand.org/communityexploration
  10
  11                      --y2--
  12                     /  |   \
  13   c1 ---- x1 ---- y1   |   z1
  14                     \  |   /
  15                      --y3--
  16
  17 1. z1 announces 192.168.255.254/32 to y2, y3.
  18 2. y2 and y3 tags this prefix at ingress with appropriate
  19 communities 65004:2 (y2) and 65004:3 (y3).
  20 3. x1 filters all communities at the egress to c1.
  21 4. Shutdown the link between y1 and y2.
  22 5. y1 will generate a BGP UPDATE message regarding the next-hop change.
  23 6. x1 will generate a BGP UPDATE message regarding community change.
  24
  25 To avoid sending duplicate BGP UPDATE messages we should make sure
  26 we send only actual route updates. In this example, x1 will skip
  27 BGP UPDATE to c1 because the actual route is the same
  28 (filtered communities - nothing changes).
  29 """
  30
  31 import os
  32 import sys
  33 import json
  34 import pytest
  35 import functools
  36
  37 CWD = os.path.dirname(os.path.realpath(__file__))
  38 sys.path.append(os.path.join(CWD, "../"))
  39
  40 # pylint: disable=C0413
  41 from lib import topotest
  42 from lib.topogen import Topogen, TopoRouter, get_topogen
  43
  44 from lib.common_config import step
  45 from time import sleep
  46
  47 pytestmark = [pytest.mark.bgpd]
  48
  49
  50 def build_topo(tgen):
  51     tgen.add_router("z1")
  52     tgen.add_router("y1")
  53     tgen.add_router("y2")
  54     tgen.add_router("y3")
  55     tgen.add_router("x1")
  56     tgen.add_router("c1")
  57
  58     # 10.0.1.0/30
  59     switch = tgen.add_switch("s1")
  60     switch.add_link(tgen.gears["c1"])
  61     switch.add_link(tgen.gears["x1"])
  62
  63     # 10.0.2.0/30
  64     switch = tgen.add_switch("s2")
  65     switch.add_link(tgen.gears["x1"])
  66     switch.add_link(tgen.gears["y1"])
  67
  68     # 10.0.3.0/30
  69     switch = tgen.add_switch("s3")
  70     switch.add_link(tgen.gears["y1"])
  71     switch.add_link(tgen.gears["y2"])
  72
  73     # 10.0.4.0/30
  74     switch = tgen.add_switch("s4")
  75     switch.add_link(tgen.gears["y1"])
  76     switch.add_link(tgen.gears["y3"])
  77
  78     # 10.0.5.0/30
  79     switch = tgen.add_switch("s5")
  80     switch.add_link(tgen.gears["y2"])
  81     switch.add_link(tgen.gears["y3"])
  82
  83     # 10.0.6.0/30
  84     switch = tgen.add_switch("s6")
  85     switch.add_link(tgen.gears["y2"])
  86     switch.add_link(tgen.gears["z1"])
  87
  88     # 10.0.7.0/30
  89     switch = tgen.add_switch("s7")
  90     switch.add_link(tgen.gears["y3"])
  91     switch.add_link(tgen.gears["z1"])
  92
  93
  94 def setup_module(mod):
  95     tgen = Topogen(build_topo, mod.__name__)
  96     tgen.start_topology()
  97
  98     router_list = tgen.routers()
  99
 100     for i, (rname, router) in enumerate(router_list.items(), 1):
 101         router.load_config(
 102             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 103         )
 104         router.load_config(
 105             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
 106         )
 107
 108     tgen.start_router()
 109
 110
 111 def teardown_module(mod):
 112     tgen = get_topogen()
 113     tgen.stop_topology()
 114
 115
 116 def test_bgp_community_update_path_change():
 117     tgen = get_topogen()
 118
 119     if tgen.routers_have_failure():
 120         pytest.skip(tgen.errors)
 121
 122     def _bgp_converge_initial():
 123         output = json.loads(
 124             tgen.gears["c1"].vtysh_cmd("show ip bgp neighbor 10.0.1.2 json")
 125         )
 126         expected = {
 127             "10.0.1.2": {
 128                 "bgpState": "Established",
 129                 "addressFamilyInfo": {"ipv4Unicast": {"acceptedPrefixCounter": 8}},
 130             }
 131         }
 132         return topotest.json_cmp(output, expected)
 133
 134     step("Check if an initial topology is converged")
 135     test_func = functools.partial(_bgp_converge_initial)
 136     success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 137     assert result is None, "Failed to see bgp convergence in c1"
 138
 139     step("Disable link between y1 and y2")
 140     tgen.gears["y1"].run("ip link set dev y1-eth1 down")
 141
 142     def _bgp_converge_link_disabled():
 143         output = json.loads(tgen.gears["y1"].vtysh_cmd("show ip bgp nei 10.0.3.2 json"))
 144         expected = {"10.0.3.2": {"bgpState": "Active"}}
 145         return topotest.json_cmp(output, expected)
 146
 147     step("Check if a topology is converged after a link down between y1 and y2")
 148     test_func = functools.partial(_bgp_converge_link_disabled)
 149     success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 150     assert result is None, "Failed to see bgp convergence in y1"
 151
 152     def _bgp_check_for_duplicate_updates():
 153         duplicate = False
 154         i = 0
 155         while i < 5:
 156             if (
 157                 len(
 158                     tgen.gears["c1"].run(
 159                         'grep "10.0.1.2(x1) rcvd 192.168.255.254/32 IPv4 unicast...duplicate ignored" bgpd.log'
 160                     )
 161                 )
 162                 > 0
 163             ):
 164                 duplicate = True
 165             i += 1
 166             sleep(0.5)
 167         return duplicate
 168
 169     step("Check if we see duplicate BGP UPDATE message in c1 (suppress-duplicates)")
 170     assert (
 171         _bgp_check_for_duplicate_updates() == False
 172     ), "Seen duplicate BGP UPDATE message in c1 from x1"
 173
 174     step("Disable bgp suppress-duplicates at x1")
 175     tgen.gears["x1"].run(
 176         "vtysh -c 'conf' -c 'router bgp' -c 'no bgp suppress-duplicates'"
 177     )
 178
 179     step("Enable link between y1 and y2")
 180     tgen.gears["y1"].run("ip link set dev y1-eth1 up")
 181
 182     def _bgp_converge_link_enabled():
 183         output = json.loads(tgen.gears["y1"].vtysh_cmd("show ip bgp nei 10.0.3.2 json"))
 184         expected = {
 185             "10.0.3.2": {
 186                 "bgpState": "Established",
 187                 "addressFamilyInfo": {
 188                     "ipv4Unicast": {"acceptedPrefixCounter": 5, "sentPrefixCounter": 4}
 189                 },
 190             }
 191         }
 192         return topotest.json_cmp(output, expected)
 193
 194     step("Check if a topology is converged after a link up between y1 and y2")
 195     test_func = functools.partial(_bgp_converge_link_enabled)
 196     success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 197     assert result is None, "Failed to see bgp convergence in y1"
 198
 199     step(
 200         "Check if we see duplicate BGP UPDATE message in c1 (no bgp suppress-duplicates)"
 201     )
 202     assert (
 203         _bgp_check_for_duplicate_updates() == True
 204     ), "Didn't see duplicate BGP UPDATE message in c1 from x1"
 205
 206
 207 if __name__ == "__main__":
 208     args = ["-s"] + sys.argv[1:]
 209     sys.exit(pytest.main(args))