tests/topotests/bgp_bfd_down_cease_notification/test_bgp_bfd_down_cease_notification.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # bgp_bfd_down_cease_notification.py
   6 #
   7 # Copyright (c) 2022 by
   8 # Donatas Abraitis <donatas@opensourcerouting.org>
   9 #
  10
  11 """
  12 Check if Cease/BFD Down notification message is sent/received
  13 when the BFD is down.
  14 """
  15
  16 import os
  17 import sys
  18 import json
  19 import pytest
  20 import functools
  21
  22 CWD = os.path.dirname(os.path.realpath(__file__))
  23 sys.path.append(os.path.join(CWD, "../"))
  24
  25 # pylint: disable=C0413
  26 from lib import topotest
  27 from lib.topogen import Topogen, TopoRouter, get_topogen
  28 from lib.common_config import kill_router_daemons, step
  29
  30 pytestmark = [pytest.mark.bfdd, pytest.mark.bgpd]
  31
  32
  33 def build_topo(tgen):
  34     for routern in range(1, 3):
  35         tgen.add_router("r{}".format(routern))
  36
  37     switch = tgen.add_switch("s1")
  38     switch.add_link(tgen.gears["r1"])
  39     switch.add_link(tgen.gears["r2"])
  40
  41
  42 def setup_module(mod):
  43     tgen = Topogen(build_topo, mod.__name__)
  44     tgen.start_topology()
  45
  46     router_list = tgen.routers()
  47
  48     for i, (rname, router) in enumerate(router_list.items(), 1):
  49         router.load_config(
  50             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  51         )
  52         router.load_config(
  53             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
  54         )
  55         router.load_config(
  56             TopoRouter.RD_BFD, os.path.join(CWD, "{}/bfdd.conf".format(rname))
  57         )
  58
  59     tgen.start_router()
  60
  61
  62 def teardown_module(mod):
  63     tgen = get_topogen()
  64     tgen.stop_topology()
  65
  66
  67 def test_bgp_bfd_down_notification():
  68     tgen = get_topogen()
  69
  70     if tgen.routers_have_failure():
  71         pytest.skip(tgen.errors)
  72
  73     r2 = tgen.gears["r2"]
  74
  75     def _bgp_converge():
  76         output = json.loads(r2.vtysh_cmd("show ip bgp neighbor 192.168.255.1 json"))
  77         expected = {
  78             "192.168.255.1": {
  79                 "bgpState": "Established",
  80                 "addressFamilyInfo": {"ipv4Unicast": {"acceptedPrefixCounter": 2}},
  81                 "peerBfdInfo": {"status": "Up"},
  82             }
  83         }
  84         return topotest.json_cmp(output, expected)
  85
  86     def _bgp_bfd_down_notification():
  87         output = json.loads(r2.vtysh_cmd("show ip bgp neighbor 192.168.255.1 json"))
  88         expected = {
  89             "192.168.255.1": {
  90                 "lastNotificationReason": "Cease/BFD Down",
  91             }
  92         }
  93         return topotest.json_cmp(output, expected)
  94
  95     step("Initial BGP converge")
  96     test_func = functools.partial(_bgp_converge)
  97     _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
  98     assert result is None, "Failed to see BGP convergence on R2"
  99
 100     step("Kill bfdd on R2")
 101     kill_router_daemons(tgen, "r2", ["bfdd"])
 102
 103     step("Check if we received Cease/BFD Down notification message")
 104     test_func = functools.partial(_bgp_bfd_down_notification)
 105     _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 106     assert result is None, "Failed to see BGP Cease/BFD Down notification message on R2"
 107
 108
 109 if __name__ == "__main__":
 110     args = ["-s"] + sys.argv[1:]
 111     sys.exit(pytest.main(args))