tests/topotests/rip_bfd_topo1/test_rip_bfd_topo1.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # test_rip_bfd_topo1.py
   6 # Part of NetDEF Topology Tests
   7 #
   8 # Copyright (c) 2023 by
   9 # Network Device Education Foundation, Inc. ("NetDEF")
  10 #
  11
  12 """
  13 test_rip_bfd_topo1.py: Test RIP BFD integration.
  14 """
  15
  16 import sys
  17 import re
  18 import pytest
  19
  20 from functools import partial
  21 from lib import topotest
  22 from lib.topogen import Topogen, TopoRouter
  23 from lib.topolog import logger
  24
  25 pytestmark = [
  26     pytest.mark.bfdd,
  27     pytest.mark.ripd,
  28 ]
  29
  30
  31 @pytest.fixture(scope="module")
  32 def tgen(request):
  33     "Setup/Teardown the environment and provide tgen argument to tests"
  34
  35     topodef = {
  36         "s1": ("r1", "r2"),
  37         "s2": ("r1", "r3")
  38     }
  39     tgen = Topogen(topodef, request.module.__name__)
  40     tgen.start_topology()
  41
  42     router_list = tgen.routers()
  43
  44     for router_name, router in router_list.items():
  45         router.load_config(TopoRouter.RD_BFD, "bfdd.conf")
  46         router.load_config(TopoRouter.RD_RIP, "ripd.conf")
  47         router.load_config(TopoRouter.RD_ZEBRA, "zebra.conf")
  48         if router_name in ["r2", "r3"]:
  49             router.load_config(TopoRouter.RD_STATIC, "staticd.conf")
  50
  51     tgen.start_router()
  52     yield tgen
  53     tgen.stop_topology()
  54
  55
  56 @pytest.fixture(autouse=True)
  57 def skip_on_failure(tgen):
  58     "Test if routers is still running otherwise skip tests"
  59     if tgen.routers_have_failure():
  60         pytest.skip("skipped because of previous test failure")
  61
  62
  63 def show_rip_json(router):
  64     "Get router 'show ip rip' JSON output"
  65     output = router.vtysh_cmd("show ip rip")
  66     routes = output.splitlines()[6:]
  67     json = {}
  68
  69     for route in routes:
  70         match = re.match(
  71             r"(.)\((.)\)\s+([^\s]+)\s+([^\s]+)\s+([^\s]+)\s+([^\s]+)", route)
  72         if match is None:
  73             continue
  74
  75         route_entry = {
  76                 "code": match[1],
  77                 "subCode": match[2],
  78                 "nextHop": match[4],
  79                 "metric": int(match[5]),
  80                 "from": match[6],
  81         }
  82
  83         if json.get(match[3]) is None:
  84             json[match[3]] = []
  85
  86         json[match[3]].append(route_entry)
  87
  88     return json
  89
  90
  91 def expect_routes(router, routes, time_amount):
  92     "Expect 'routes' in 'router'."
  93
  94     def test_function():
  95         "Internal test function."
  96         return topotest.json_cmp(show_rip_json(router), routes)
  97
  98     _, result = topotest.run_and_expect(test_function,
  99                                         None,
 100                                         count=time_amount,
 101                                         wait=1)
 102     assert result is None, "Unexpected routing table in {}".format(
 103         router.name)
 104
 105
 106 def expect_bfd_peers(router, peers):
 107     "Expect 'peers' in 'router' BFD status."
 108     test_func = partial(
 109         topotest.router_json_cmp,
 110         router,
 111         "show bfd peers json",
 112         peers,
 113     )
 114     _, result = topotest.run_and_expect(test_func, None, count=20, wait=1)
 115     assert result is None, "{} BFD peer status mismatch".format(router)
 116
 117
 118 def test_rip_convergence(tgen):
 119     "Test that RIP learns the neighbor routes."
 120
 121     expect_routes(
 122         tgen.gears["r1"], {
 123             "10.254.254.2/32": [{
 124                 "code": "R",
 125                 "subCode": "n",
 126                 "from": "192.168.0.2"
 127             }],
 128             "10.254.254.3/32": [{
 129                 "code": "R",
 130                 "subCode": "n",
 131                 "from": "192.168.1.2"
 132             }],
 133             "10.254.254.100/32": [{
 134                 "code": "R",
 135                 "subCode": "n",
 136                 "from": "192.168.0.2",
 137             }, {
 138                 "code": "R",
 139                 "subCode": "n",
 140                 "from": "192.168.1.2",
 141             }]
 142         }, 40)
 143
 144     expect_bfd_peers(tgen.gears["r1"], [{
 145         "peer": "192.168.0.2",
 146         "status": "up",
 147         "receive-interval": 1000,
 148         "transmit-interval": 1000,
 149     }, {
 150         "peer": "192.168.1.2",
 151         "status": "up",
 152         "receive-interval": 1000,
 153         "transmit-interval": 1000,
 154     }])
 155
 156     expect_routes(
 157         tgen.gears["r2"], {
 158             "10.254.254.1/32": [{
 159                 "code": "R",
 160                 "subCode": "n",
 161                 "from": "192.168.0.1"
 162             }],
 163             "10.254.254.3/32": [{
 164                 "code": "R",
 165                 "subCode": "n",
 166                 "from": "192.168.0.1"
 167             }],
 168             "10.254.254.100/32": [{
 169                 "code": "S",
 170                 "subCode": "r",
 171                 "from": "self"
 172             }]
 173         }, 40)
 174
 175     expect_bfd_peers(tgen.gears["r2"], [{
 176         "peer": "192.168.0.1",
 177         "status": "up",
 178         "receive-interval": 1000,
 179         "transmit-interval": 1000,
 180     }])
 181
 182     expect_routes(
 183         tgen.gears["r3"], {
 184             "10.254.254.1/32": [{
 185                 "code": "R",
 186                 "subCode": "n",
 187                 "from": "192.168.1.1"
 188             }],
 189             "10.254.254.2/32": [{
 190                 "code": "R",
 191                 "subCode": "n",
 192                 "from": "192.168.1.1"
 193             }],
 194             "10.254.254.100/32": [{
 195                 "code": "S",
 196                 "subCode": "r",
 197                 "from": "self"
 198             }]
 199         }, 40)
 200
 201     expect_bfd_peers(tgen.gears["r3"], [{
 202         "peer": "192.168.1.1",
 203         "status": "up",
 204         "receive-interval": 1000,
 205         "transmit-interval": 1000,
 206     }])
 207
 208
 209 def test_rip_bfd_convergence(tgen):
 210     "Test that RIP drop the gone neighbor routes."
 211
 212     tgen.gears["r3"].link_enable("r3-eth0", False)
 213
 214     expect_routes(
 215         tgen.gears["r1"], {
 216             "10.254.254.2/32": [{
 217                 "code": "R",
 218                 "subCode": "n",
 219                 "from": "192.168.0.2"
 220             }],
 221             "10.254.254.3/32": None,
 222             "10.254.254.100/32": [{
 223                 "code": "R",
 224                 "subCode": "n",
 225                 "from": "192.168.0.2",
 226             }]
 227         }, 6)
 228
 229     expect_routes(
 230         tgen.gears["r3"], {
 231             "10.254.254.1/32": None,
 232             "10.254.254.2/32": None,
 233             "10.254.254.100/32": [{
 234                 "code": "S",
 235                 "subCode": "r",
 236                 "from": "self"
 237             }]
 238         }, 6)
 239
 240
 241 def test_memory_leak(tgen):
 242     "Run the memory leak test and report results."
 243
 244     if not tgen.is_memleak_enabled():
 245         pytest.skip("Memory leak test/report is disabled")
 246
 247     tgen.report_memory_leaks()
 248
 249
 250 if __name__ == "__main__":
 251     args = ["-s"] + sys.argv[1:]
 252     sys.exit(pytest.main(args))