tests/topotests/bgp_gr_notification/test_bgp_gr_notification.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # bgp_gr_notification.py
   6 #
   7 # Copyright (c) 2022 by
   8 # Donatas Abraitis <donatas@opensourcerouting.org>
   9 #
  10
  11 """
  12 TC1: Disable the link between R1-R2 and wait for HoldTimerExpire notification:
  13     1) Check if R2 sent HoldTimerExpired notification
  14     2) Check if the routes are retained at R2
  15 TC2: Trigger `clear bgp` (Administrative Reset):
  16     `bgp hard-administrative-reset` disabled:
  17         a) Check if Administrative Reset notification was sent from R2
  18         b) Routes should be retained on R1
  19 """
  20
  21 import os
  22 import sys
  23 import json
  24 import pytest
  25 import functools
  26
  27 CWD = os.path.dirname(os.path.realpath(__file__))
  28 sys.path.append(os.path.join(CWD, "../"))
  29
  30 # pylint: disable=C0413
  31 from lib import topotest
  32 from lib.topogen import Topogen, TopoRouter, get_topogen
  33 from lib.common_config import step
  34
  35 pytestmark = [pytest.mark.bgpd]
  36
  37
  38 def build_topo(tgen):
  39     for routern in range(1, 3):
  40         tgen.add_router("r{}".format(routern))
  41
  42     switch = tgen.add_switch("s1")
  43     switch.add_link(tgen.gears["r1"])
  44     switch.add_link(tgen.gears["r2"])
  45
  46
  47 def setup_module(mod):
  48     tgen = Topogen(build_topo, mod.__name__)
  49     tgen.start_topology()
  50
  51     router_list = tgen.routers()
  52
  53     for i, (rname, router) in enumerate(router_list.items(), 1):
  54         router.load_config(
  55             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  56         )
  57         router.load_config(
  58             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
  59         )
  60
  61     tgen.start_router()
  62
  63
  64 def teardown_module(mod):
  65     tgen = get_topogen()
  66     tgen.stop_topology()
  67
  68
  69 def test_bgp_hold_timer_expired_gr():
  70     # TC1
  71     tgen = get_topogen()
  72
  73     if tgen.routers_have_failure():
  74         pytest.skip(tgen.errors)
  75
  76     r1 = tgen.gears["r1"]
  77     r2 = tgen.gears["r2"]
  78
  79     def _bgp_converge():
  80         output = json.loads(r2.vtysh_cmd("show ip bgp neighbor 192.168.255.1 json"))
  81         expected = {
  82             "192.168.255.1": {
  83                 "bgpState": "Established",
  84                 "addressFamilyInfo": {"ipv4Unicast": {"acceptedPrefixCounter": 2}},
  85             }
  86         }
  87         return topotest.json_cmp(output, expected)
  88
  89     def _disable_link_r1_r2():
  90         r1.cmd_raises("ip link set down dev r1-eth0")
  91
  92     def _enable_link_r1_r2():
  93         r1.cmd_raises("ip link set up dev r1-eth0")
  94
  95     def _bgp_check_hold_timer_expire_reason():
  96         output = json.loads(r2.vtysh_cmd("show ip bgp neighbor 192.168.255.1 json"))
  97         expected = {
  98             "192.168.255.1": {
  99                 "lastNotificationReason": "Hold Timer Expired",
 100             }
 101         }
 102         return topotest.json_cmp(output, expected)
 103
 104     def _bgp_check_hold_timer_expire_stale():
 105         output = json.loads(r2.vtysh_cmd("show bgp ipv4 unicast 172.16.255.1/32 json"))
 106         expected = {
 107             "paths": [
 108                 {
 109                     "stale": True,
 110                     "valid": True,
 111                 }
 112             ]
 113         }
 114         return topotest.json_cmp(output, expected)
 115
 116     step("Initial BGP converge")
 117     test_func = functools.partial(_bgp_converge)
 118     _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 119     assert result is None, "Failed to see BGP convergence on R2"
 120
 121     step("Disable the link between R1-R2")
 122     _disable_link_r1_r2()
 123
 124     step("Check if R2 sent HoldTimerExpire notification to R1")
 125     test_func = functools.partial(_bgp_check_hold_timer_expire_reason)
 126     _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 127     assert result is None, "Failed to see Hold Timer Expired notification from R2 on R1"
 128
 129     step("Check if the routes are retained at R2")
 130     test_func = functools.partial(_bgp_check_hold_timer_expire_stale)
 131     _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 132     assert result is None, "Failed to see retained stale routes on R2"
 133
 134     step("Enable the link between R1-R2")
 135     _enable_link_r1_r2()
 136
 137
 138 def test_bgp_administrative_reset_gr():
 139     # TC2
 140     tgen = get_topogen()
 141
 142     if tgen.routers_have_failure():
 143         pytest.skip(tgen.errors)
 144
 145     r1 = tgen.gears["r1"]
 146     r2 = tgen.gears["r2"]
 147
 148     def _bgp_converge():
 149         output = json.loads(r2.vtysh_cmd("show ip bgp neighbor 192.168.255.1 json"))
 150         expected = {
 151             "192.168.255.1": {
 152                 "bgpState": "Established",
 153                 "addressFamilyInfo": {"ipv4Unicast": {"acceptedPrefixCounter": 2}},
 154             }
 155         }
 156         return topotest.json_cmp(output, expected)
 157
 158     def _bgp_check_hard_reset():
 159         output = json.loads(r1.vtysh_cmd("show ip bgp neighbor 192.168.255.2 json"))
 160         expected = {
 161             "192.168.255.2": {
 162                 "lastNotificationReason": "Cease/Administrative Reset",
 163                 "lastNotificationHardReset": False,
 164             }
 165         }
 166         return topotest.json_cmp(output, expected)
 167
 168     def _bgp_check_gr_notification_stale():
 169         output = json.loads(r1.vtysh_cmd("show bgp ipv4 unicast 172.16.255.2/32 json"))
 170         expected = {
 171             "paths": [
 172                 {
 173                     "stale": True,
 174                     "valid": True,
 175                 }
 176             ]
 177         }
 178         return topotest.json_cmp(output, expected)
 179
 180     def _bgp_clear_r1_and_shutdown():
 181         r2.vtysh_cmd(
 182             """
 183             clear ip bgp 192.168.255.1
 184             configure terminal
 185              router bgp
 186               neighbor 192.168.255.1 shutdown
 187             """
 188         )
 189
 190     step("Initial BGP converge")
 191     test_func = functools.partial(_bgp_converge)
 192     _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 193     assert result is None, "Failed to see BGP convergence on R2"
 194
 195     step("Reset and shutdown R1")
 196     _bgp_clear_r1_and_shutdown()
 197
 198     step("Check if Hard Reset notification wasn't sent from R2")
 199     test_func = functools.partial(_bgp_check_hard_reset)
 200     _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 201     assert result is None, "Failed to send Administrative Reset notification from R2"
 202
 203     step("Check if stale routes are retained on R1")
 204     test_func = functools.partial(_bgp_check_gr_notification_stale)
 205     _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 206     assert result is None, "Failed to see retained stale routes on R1"
 207
 208
 209 if __name__ == "__main__":
 210     args = ["-s"] + sys.argv[1:]
 211     sys.exit(pytest.main(args))