tests/topotests/bgp_labeled_unicast_addpath/test_bgp_labeled_unicast_addpath.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # Copyright (c) 2022 by
   5 # Donatas Abraitis <donatas@opensourcerouting.org>
   6 #
   7 # Permission to use, copy, modify, and/or distribute this software
   8 # for any purpose with or without fee is hereby granted, provided
   9 # that the above copyright notice and this permission notice appear
  10 # in all copies.
  11 #
  12 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  13 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  14 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  15 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  16 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  17 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  18 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  19 # OF THIS SOFTWARE.
  20 #
  21
  22 """
  23 Check if labeled-unicast works correctly with addpath capability.
  24 Initially R3 MUST announce 10.0.0.1/32 multipath(2) from R1 + R2.
  25 Later, we enable R5 and 10.0.0.1/32 multipath(3) MUST be announced,
  26 R1 + R2 + R5.
  27 """
  28
  29 import os
  30 import sys
  31 import json
  32 import pytest
  33 import functools
  34
  35 CWD = os.path.dirname(os.path.realpath(__file__))
  36 sys.path.append(os.path.join(CWD, "../"))
  37
  38 # pylint: disable=C0413
  39 from lib import topotest
  40 from lib.topogen import Topogen, TopoRouter, get_topogen
  41 from lib.common_config import step
  42
  43 pytestmark = [pytest.mark.bgpd]
  44
  45
  46 def build_topo(tgen):
  47     for routern in range(1, 6):
  48         tgen.add_router("r{}".format(routern))
  49
  50     switch = tgen.add_switch("s1")
  51     switch.add_link(tgen.gears["r1"])
  52     switch.add_link(tgen.gears["r3"])
  53
  54     switch = tgen.add_switch("s2")
  55     switch.add_link(tgen.gears["r2"])
  56     switch.add_link(tgen.gears["r3"])
  57
  58     switch = tgen.add_switch("s3")
  59     switch.add_link(tgen.gears["r3"])
  60     switch.add_link(tgen.gears["r4"])
  61
  62     switch = tgen.add_switch("s4")
  63     switch.add_link(tgen.gears["r3"])
  64     switch.add_link(tgen.gears["r5"])
  65
  66
  67 def setup_module(mod):
  68     tgen = Topogen(build_topo, mod.__name__)
  69     tgen.start_topology()
  70
  71     router_list = tgen.routers()
  72
  73     for i, (rname, router) in enumerate(router_list.items(), 1):
  74         router.load_config(
  75             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  76         )
  77         router.load_config(
  78             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
  79         )
  80
  81     tgen.start_router()
  82
  83
  84 def teardown_module(mod):
  85     tgen = get_topogen()
  86     tgen.stop_topology()
  87
  88
  89 def test_bgp_addpath_labeled_unicast():
  90     tgen = get_topogen()
  91
  92     if tgen.routers_have_failure():
  93         pytest.skip(tgen.errors)
  94
  95     r3 = tgen.gears["r3"]
  96     r4 = tgen.gears["r4"]
  97
  98     def _bgp_check_advertised_routes(prefix_num):
  99         output = json.loads(
 100             r3.vtysh_cmd(
 101                 "show bgp ipv4 labeled-unicast neighbors 192.168.34.4 advertised-routes json"
 102             )
 103         )
 104         expected = {
 105             "advertisedRoutes": {
 106                 "10.0.0.1/32": {
 107                     "appliedStatusSymbols": {
 108                         "*": True,
 109                         ">": True,
 110                         "=": True,
 111                     }
 112                 }
 113             },
 114             "totalPrefixCounter": prefix_num,
 115         }
 116         return topotest.json_cmp(output, expected)
 117
 118     test_func = functools.partial(_bgp_check_advertised_routes, 2)
 119     _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 120     assert (
 121         result is None
 122     ), "Failed to advertise labeled-unicast with addpath (multipath)"
 123
 124     def _bgp_check_received_routes():
 125         output = json.loads(r4.vtysh_cmd("show bgp ipv4 labeled-unicast json"))
 126         expected = {
 127             "routes": {
 128                 "10.0.0.1/32": [
 129                     {
 130                         "valid": True,
 131                         "path": "65003 65001",
 132                     },
 133                     {
 134                         "valid": True,
 135                         "path": "65003 65002",
 136                     },
 137                 ]
 138             }
 139         }
 140         return topotest.json_cmp(output, expected)
 141
 142     test_func = functools.partial(_bgp_check_received_routes)
 143     _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 144     assert result is None, "Failed to receive labeled-unicast with addpath (multipath)"
 145
 146     step("Enable BGP session for R5")
 147     r3.vtysh_cmd(
 148         """
 149           configure terminal
 150             router bgp 65003
 151               no neighbor 192.168.35.5 shutdown
 152         """
 153     )
 154
 155     test_func = functools.partial(_bgp_check_advertised_routes, 3)
 156     _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 157     assert (
 158         result is None
 159     ), "Failed to advertise labeled-unicast with addpath (multipath)"
 160
 161     step("Disable BGP session for R5")
 162     r3.vtysh_cmd(
 163         """
 164           configure terminal
 165             router bgp 65003
 166               neighbor 192.168.35.5 shutdown
 167         """
 168     )
 169
 170     test_func = functools.partial(_bgp_check_advertised_routes, 2)
 171     _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 172     assert (
 173         result is None
 174     ), "Failed to advertise labeled-unicast with addpath (multipath)"
 175
 176
 177 if __name__ == "__main__":
 178     args = ["-s"] + sys.argv[1:]
 179     sys.exit(pytest.main(args))