tests/topotests/bgp_vpnv6_per_nexthop_label/test_bgp_vpnv6_per_nexthop_label.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3 #
   4 # test_bgp_vpnv6_per_nexthop_label.py
   5 #
   6 # Copyright 2023 6WIND S.A.
   7 #
   8
   9 """
  10  test_bgp_vpnv6_per_nexthop_label.py: Test the FRR BGP daemon using EBGP peering
  11  Let us exchange VPNv6 updates between both devices
  12  Updates from r1 will originate from the same RD, but will have separate
  13  label values.
  14
  15      +----------+
  16      |   r11    |
  17      |192::2:11 +---+
  18      |          |   |                   +----+--------+              +----------+
  19      +----------+   |         192::2::1 |vrf | r1     |192:168::/112 |    r2    |
  20                     +-------------------+    |       1+--------------+          |
  21      +----------+   |                   |VRF1|AS65500 |              | AS65501  |
  22      |   r12    |   |    +--------------+    |   VPNV4|              |VPNV4     |
  23      |192::2:12 +---+    |192:168::255:1+-+--+--------+              +----------+
  24      |          |        |
  25      +----------+        |
  26                          |
  27      +----------+        |
  28      |   r13    |        |
  29      |192:168:: +--------+
  30      | 255:13   |
  31      +----------+
  32 """
  33
  34 import os
  35 import sys
  36 import json
  37 from functools import partial
  38 import pytest
  39 import functools
  40
  41 # Save the Current Working Directory to find configuration files.
  42 CWD = os.path.dirname(os.path.realpath(__file__))
  43 sys.path.append(os.path.join(CWD, "../"))
  44
  45 # pylint: disable=C0413
  46 # Import topogen and topotest helpers
  47 from lib import topotest
  48 from lib.topogen import Topogen, TopoRouter, get_topogen
  49 from lib.topolog import logger
  50
  51
  52 pytestmark = [pytest.mark.bgpd]
  53
  54 PREFIXES_R11 = ["172:31::11/128", "172:31::20/128", "172:31::111/128"]
  55 PREFIXES_R12 = ["172:31::12/128"]
  56 PREFIXES_REDIST_R12 = ["172:31::15/128"]
  57 PREFIXES_R13 = ["172:31::13/128"]
  58 PREFIXES_REDIST_R14 = ["172:31::14/128"]
  59 PREFIXES_CONNECTED = ["192:168::255/112", "192:2::/64"]
  60
  61
  62 def build_topo(tgen):
  63     "Build function"
  64
  65     # Create 2 routers.
  66     tgen.add_router("r1")
  67     tgen.add_router("r2")
  68     tgen.add_router("r11")
  69     tgen.add_router("r12")
  70     tgen.add_router("r13")
  71     tgen.add_router("r14")
  72     tgen.add_router("rr")
  73
  74     switch = tgen.add_switch("s1")
  75     switch.add_link(tgen.gears["r1"])
  76     switch.add_link(tgen.gears["r2"])
  77
  78     switch = tgen.add_switch("s2")
  79     switch.add_link(tgen.gears["r1"])
  80     switch.add_link(tgen.gears["r11"])
  81     switch.add_link(tgen.gears["r12"])
  82     switch.add_link(tgen.gears["rr"])
  83
  84     switch = tgen.add_switch("s3")
  85     switch.add_link(tgen.gears["r2"])
  86
  87     switch = tgen.add_switch("s4")
  88     switch.add_link(tgen.gears["r1"])
  89     switch.add_link(tgen.gears["r13"])
  90
  91     switch = tgen.add_switch("s5")
  92     switch.add_link(tgen.gears["r1"])
  93     switch.add_link(tgen.gears["r14"])
  94
  95
  96 def _populate_iface():
  97     tgen = get_topogen()
  98     cmds_list = [
  99         "ip link add vrf1 type vrf table 10",
 100         "echo 100000 > /proc/sys/net/mpls/platform_labels",
 101         "ip link set dev vrf1 up",
 102         "ip link set dev {0}-eth1 master vrf1",
 103         "echo 1 > /proc/sys/net/mpls/conf/{0}-eth0/input",
 104     ]
 105     cmds_list_plus = [
 106         "ip link set dev {0}-eth2 master vrf1",
 107     ]
 108     cmd_no_ll = [
 109         "ip link set dev {0}-eth0 down",
 110         "echo 1 > /proc/sys/net/ipv6/conf/{0}-eth0/addr_gen_mode",
 111         "ip link set dev {0}-eth0 up",
 112     ]
 113
 114     for cmd in cmds_list:
 115         input = cmd.format("r1")
 116         logger.info("input: " + cmd)
 117         output = tgen.net["r1"].cmd(cmd.format("r1"))
 118         logger.info("output: " + output)
 119
 120     for cmd in cmds_list_plus:
 121         input = cmd.format("r1")
 122         logger.info("input: " + cmd)
 123         output = tgen.net["r1"].cmd(cmd.format("r1"))
 124         logger.info("output: " + output)
 125
 126     for cmd in cmds_list:
 127         input = cmd.format("r2")
 128         logger.info("input: " + cmd)
 129         output = tgen.net["r2"].cmd(cmd.format("r2"))
 130         logger.info("output: " + output)
 131
 132     for rtr in ("r11", "r13"):
 133         for cmd in cmd_no_ll:
 134             input = cmd.format(rtr)
 135             logger.info("input: " + cmd)
 136             output = tgen.net[rtr].cmd(cmd.format(rtr))
 137             logger.info("output: " + output)
 138
 139
 140 def setup_module(mod):
 141     "Sets up the pytest environment"
 142     tgen = Topogen(build_topo, mod.__name__)
 143     tgen.start_topology()
 144
 145     router_list = tgen.routers()
 146     _populate_iface()
 147
 148     for rname, router in router_list.items():
 149         router.load_config(
 150             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 151         )
 152         router.load_config(
 153             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
 154         )
 155
 156     # Initialize all routers.
 157     tgen.start_router()
 158
 159
 160 def teardown_module(_mod):
 161     "Teardown the pytest environment"
 162     tgen = get_topogen()
 163
 164     tgen.stop_topology()
 165
 166
 167 def bgp_vpnv6_table_check(router, group, label_list=None, label_value_expected=None):
 168     """
 169     Dump and check that vpnv6 entries have the same MPLS label value
 170     * 'router': the router to check
 171     * 'group': the list of prefixes to check. a single label value for the group has to be found
 172     * 'label_list': check that the label values are not present in the vpnv6 entries
 173     *              that list is updated with the present label value
 174     * 'label_value_expected': check that the mpls label read is the same as that value
 175     """
 176
 177     stored_label_inited = False
 178     for prefix in group:
 179         dump = router.vtysh_cmd("show bgp ipv6 vpn {} json".format(prefix), isjson=True)
 180         for rd, pathes in dump.items():
 181             for path in pathes["paths"]:
 182                 assert (
 183                     "remoteLabel" in path.keys()
 184                 ), "{0}, {1}, remoteLabel not present".format(router.name, prefix)
 185                 logger.info(
 186                     "{0}, {1}, label value is {2}".format(
 187                         router.name, prefix, path["remoteLabel"]
 188                     )
 189                 )
 190                 if stored_label_inited:
 191                     assert (
 192                         path["remoteLabel"] == stored_label
 193                     ), "{0}, {1}, label value not expected one (expected {2}, observed {3}".format(
 194                         router.name, prefix, stored_label, path["remoteLabel"]
 195                     )
 196                 else:
 197                     stored_label = path["remoteLabel"]
 198                     stored_label_inited = True
 199                     if label_list is not None:
 200                         assert (
 201                             stored_label not in label_list
 202                         ), "{0}, {1}, label already detected in a previous prefix".format(
 203                             router.name, prefix
 204                         )
 205                         label_list.add(stored_label)
 206
 207                 if label_value_expected:
 208                     assert (
 209                         path["remoteLabel"] == label_value_expected
 210                     ), "{0}, {1}, label value not expected (expected {2}, observed {3}".format(
 211                         router.name, prefix, label_value_expected, path["remoteLabel"]
 212                     )
 213
 214
 215 def bgp_vpnv6_table_check_all(router, label_list=None, same=False):
 216     """
 217     Dump and check that vpnv6 entries are correctly configured with specific label values
 218     * 'router': the router to check
 219     * 'label_list': check that the label values are not present in the vpnv6 entries
 220     *              that list is updated with the present label value found.
 221     * 'same': by default, set to False. Addresses groups are classified by addresses.
 222     *         if set to True, all entries of all groups should have a unique label value
 223     """
 224     if same:
 225         bgp_vpnv6_table_check(
 226             router,
 227             group=PREFIXES_R11
 228             + PREFIXES_R12
 229             + PREFIXES_REDIST_R12
 230             + PREFIXES_R13
 231             + PREFIXES_REDIST_R14
 232             + PREFIXES_CONNECTED,
 233             label_list=label_list,
 234         )
 235     else:
 236         for group in (
 237             PREFIXES_R11,
 238             PREFIXES_R12,
 239             PREFIXES_REDIST_R12,
 240             PREFIXES_R13,
 241             PREFIXES_REDIST_R14,
 242             PREFIXES_CONNECTED,
 243         ):
 244             bgp_vpnv6_table_check(router, group=group, label_list=label_list)
 245
 246
 247 def mpls_table_check(router, blacklist=None, label_list=None, whitelist=None):
 248     """
 249     Dump and check 'show mpls table json' output. An assert is triggered in case test fails
 250     * 'router': the router to check
 251     * 'blacklist': the list of nexthops (IP or interface) that should not be on output
 252     * 'label_list': the list of labels that should be in inLabel value
 253     * 'whitelist': the list of nexthops (IP or interface) that should be on output
 254     """
 255     nexthop_list = []
 256     if blacklist:
 257         nexthop_list.append(blacklist)
 258     logger.info("Checking MPLS labels on {}".format(router.name))
 259     dump = router.vtysh_cmd("show mpls table json", isjson=True)
 260     for in_label, label_info in dump.items():
 261         if label_list is not None:
 262             label_list.add(in_label)
 263         for nh in label_info["nexthops"]:
 264             assert (
 265                 nh["installed"] == True and nh["type"] == "BGP"
 266             ), "{}, show mpls table, nexthop is not installed".format(router.name)
 267             if "nexthop" in nh.keys():
 268                 assert (
 269                     nh["nexthop"] not in nexthop_list
 270                 ), "{}, show mpls table, duplicated or blacklisted nexthop address".format(
 271                     router.name
 272                 )
 273                 nexthop_list.append(nh["nexthop"])
 274             elif "interface" in nh.keys():
 275                 assert (
 276                     nh["interface"] not in nexthop_list
 277                 ), "{}, show mpls table, duplicated or blacklisted nexthop interface".format(
 278                     router.name
 279                 )
 280                 nexthop_list.append(nh["interface"])
 281             else:
 282                 assert (
 283                     0
 284                 ), "{}, show mpls table, entry with neither nexthop nor interface".format(
 285                     router.name
 286                 )
 287
 288     if whitelist:
 289         for entry in whitelist:
 290             assert (
 291                 entry in nexthop_list
 292             ), "{}, show mpls table, entry with nexthop {} not present in nexthop list".format(
 293                 router.name, entry
 294             )
 295
 296
 297 def check_show_bgp_vpn_prefix_not_found(router, ipversion, prefix, rd, label=None):
 298     output = json.loads(
 299         router.vtysh_cmd("show bgp {} vpn {} json".format(ipversion, prefix))
 300     )
 301     if label:
 302         expected = {rd: {"prefix": prefix, "paths": [{"remoteLabel": label}]}}
 303     else:
 304         expected = {rd: {"prefix": prefix}}
 305     ret = topotest.json_cmp(output, expected)
 306     if ret is None:
 307         return "not good"
 308     return None
 309
 310
 311 def check_show_bgp_vpn_prefix_found(router, ipversion, prefix, rd):
 312     output = json.loads(
 313         router.vtysh_cmd("show bgp {} vpn {} json".format(ipversion, prefix))
 314     )
 315     expected = {rd: {"prefix": prefix}}
 316     return topotest.json_cmp(output, expected)
 317
 318
 319 def check_show_mpls_table_entry_label_found(router, inlabel, interface):
 320     output = json.loads(router.vtysh_cmd("show mpls table {} json".format(inlabel)))
 321     expected = {
 322         "inLabel": inlabel,
 323         "installed": True,
 324         "nexthops": [{"interface": interface}],
 325     }
 326     return topotest.json_cmp(output, expected)
 327
 328
 329 def check_show_mpls_table_entry_label_not_found(router, inlabel):
 330     output = json.loads(router.vtysh_cmd("show mpls table {} json".format(inlabel)))
 331     expected = {"inlabel": inlabel, "installed": True}
 332     ret = topotest.json_cmp(output, expected)
 333     if ret is None:
 334         return "not good"
 335     return None
 336
 337
 338 def mpls_entry_get_interface(router, label):
 339     """
 340     Assert that the label is in MPLS table
 341     Assert an outgoing interface is programmed
 342     return the outgoing interface
 343     """
 344     outgoing_interface = None
 345
 346     logger.info("Checking MPLS labels on {}".format(router.name))
 347     dump = router.vtysh_cmd("show mpls table {} json".format(label), isjson=True)
 348     assert dump, "{}, show mpls table, inLabel {} not found".format(router.name, label)
 349
 350     for nh in dump["nexthops"]:
 351         assert (
 352             "interface" in nh.keys()
 353         ), "{}, show mpls table, nexthop interface not present for MPLS entry {}".format(
 354             router.name, label
 355         )
 356
 357         outgoing_interface = nh["interface"]
 358
 359     return outgoing_interface
 360
 361
 362 def test_protocols_convergence():
 363     """
 364     Assert that all protocols have converged
 365     statuses as they depend on it.
 366     """
 367     tgen = get_topogen()
 368     if tgen.routers_have_failure():
 369         pytest.skip(tgen.errors)
 370
 371     # Check BGP IPv6 routing tables on VRF1 of r1
 372     logger.info("Checking BGP IPv6 routes for convergence on r1 VRF1")
 373     router = tgen.gears["r1"]
 374     json_file = "{}/{}/bgp_ipv6_routes_vrf1.json".format(CWD, router.name)
 375
 376     expected = json.loads(open(json_file).read())
 377     test_func = partial(
 378         topotest.router_json_cmp,
 379         router,
 380         "show bgp vrf vrf1 ipv6 json",
 381         expected,
 382     )
 383     _, result = topotest.run_and_expect(test_func, None, count=20, wait=0.5)
 384     assertmsg = '"{}" JSON output mismatches'.format(router.name)
 385     assert result is None, assertmsg
 386
 387     logger.info("Checking BGP VPNv6 routes for convergence on r2")
 388     router = tgen.gears["r2"]
 389     json_file = "{}/{}/bgp_vpnv6_routes.json".format(CWD, router.name)
 390     expected = json.loads(open(json_file).read())
 391     test_func = partial(
 392         topotest.router_json_cmp,
 393         router,
 394         "show bgp ipv6 vpn json",
 395         expected,
 396     )
 397     _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 398     assertmsg = '"{}" JSON output mismatches'.format(router.name)
 399     assert result is None, assertmsg
 400
 401     # Check BGP labels received on r2
 402     logger.info("Checking BGP VPNv6 labels on r2")
 403     label_list = set()
 404     bgp_vpnv6_table_check_all(tgen.gears["r2"], label_list)
 405
 406     # Check MPLS labels received on r1
 407     mpls_table_check(tgen.gears["r1"], label_list)
 408
 409
 410 def test_flapping_bgp_vrf_down():
 411     """
 412     Turn down a remote BGP session
 413     """
 414     tgen = get_topogen()
 415     if tgen.routers_have_failure():
 416         pytest.skip(tgen.errors)
 417     logger.info("Unpeering BGP on r11")
 418     tgen.gears["r11"].vtysh_cmd(
 419         "configure terminal\nrouter bgp 65500\nno neighbor 192:2::100\n",
 420         isjson=False,
 421     )
 422
 423     def _bgp_prefix_not_found(router, vrf, ipversion, prefix):
 424         output = json.loads(
 425             router.vtysh_cmd(
 426                 "show bgp vrf {} {} {} json".format(vrf, ipversion, prefix)
 427             )
 428         )
 429         expected = {"prefix": prefix}
 430         ret = topotest.json_cmp(output, expected)
 431         if ret is None:
 432             return "not good"
 433         return None
 434
 435     # Check prefix from r11 is not present
 436     test_func = functools.partial(
 437         _bgp_prefix_not_found, tgen.gears["r1"], "vrf1", "ipv6", "172:31::11/128"
 438     )
 439     success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 440     assert (
 441         success
 442     ), "r1, prefix 172:31::11/128 from r11 did not disappear. r11 still connected to rr ?"
 443
 444     # Check BGP updated received on r2 are not from r11
 445     logger.info("Checking BGP VPNv6 labels on r2")
 446     for entry in PREFIXES_R11:
 447         dump = tgen.gears["r2"].vtysh_cmd(
 448             "show bgp ipv6 vpn {} json".format(entry), isjson=True
 449         )
 450         for rd in dump:
 451             assert False, "r2, {}, route distinguisher {} present".format(entry, rd)
 452
 453     mpls_table_check(tgen.gears["r1"], blacklist=["192:2::11"])
 454
 455
 456 def test_flapping_bgp_vrf_up():
 457     """
 458     Turn up a remote BGP session
 459     """
 460     tgen = get_topogen()
 461     if tgen.routers_have_failure():
 462         pytest.skip(tgen.errors)
 463     logger.info("Peering BGP on r11")
 464     tgen.gears["r11"].vtysh_cmd(
 465         "configure terminal\nrouter bgp 65500\nneighbor 192:2::100 remote-as 65500\n",
 466         isjson=False,
 467     )
 468     tgen.gears["r11"].vtysh_cmd(
 469         "configure terminal\nrouter bgp 65500\naddress-family ipv6 unicast\nneighbor 192:2::100 activate\n",
 470         isjson=False,
 471     )
 472
 473     # Check r2 gets prefix 172:31::11/128
 474     test_func = functools.partial(
 475         check_show_bgp_vpn_prefix_found,
 476         tgen.gears["r2"],
 477         "ipv6",
 478         "172:31::11/128",
 479         "444:1",
 480     )
 481     success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 482     assert (
 483         success
 484     ), "r2, prefix 172:31::11/128 from r11 not present. r11 still disconnected from rr ?"
 485     bgp_vpnv6_table_check_all(tgen.gears["r2"])
 486
 487
 488 def test_recursive_route():
 489     """
 490     Test static recursive route redistributed over BGP
 491     """
 492     tgen = get_topogen()
 493     if tgen.routers_have_failure():
 494         pytest.skip(tgen.errors)
 495
 496     logger.info("Enabling recursive static route")
 497     tgen.gears["r1"].vtysh_cmd(
 498         "configure terminal\nvrf vrf1\nipv6 route 172:31::30/128 172:31::20\n",
 499         isjson=False,
 500     )
 501     logger.info("Checking BGP VPNv6 labels on r2")
 502     # that route should be sent along with label for 192.0.2.11
 503
 504     def _prefix30_not_found(router):
 505         output = json.loads(router.vtysh_cmd("show bgp ipv6 vpn 172:31::30/128 json"))
 506         expected = {"444:1": {"prefix": "172:31::30/128"}}
 507         ret = topotest.json_cmp(output, expected)
 508         if ret is None:
 509             return "not good"
 510         return None
 511
 512     def _prefix30_found(router):
 513         output = json.loads(router.vtysh_cmd("show bgp ipv6 vpn 172:31::30/128 json"))
 514         expected = {"444:1": {"prefix": "172:31::30/128"}}
 515         return topotest.json_cmp(output, expected)
 516
 517     # Check r2 received vpnv6 update with 172:31::30
 518     test_func = functools.partial(_prefix30_found, tgen.gears["r2"])
 519     success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 520     assert success, "r2, VPNv6 update 172:31::30 not found"
 521
 522     # that route should be sent along with label for 192::2:11
 523     bgp_vpnv6_table_check(
 524         tgen.gears["r2"],
 525         group=PREFIXES_R11 + ["172:31::30/128"],
 526     )
 527
 528     # diagnostic
 529     logger.info("Dumping label nexthop table")
 530     tgen.gears["r1"].vtysh_cmd("show bgp vrf vrf1 label-nexthop detail", isjson=False)
 531     logger.info("Dumping nexthop table")
 532     tgen.gears["r1"].vtysh_cmd("show bgp vrf vrf1 nexthop detail", isjson=False)
 533
 534     logger.info("Disabling recursive static route")
 535     tgen.gears["r1"].vtysh_cmd(
 536         "configure terminal\nvrf vrf1\nno ipv6 route 172:31::30/128 172:31::20\n",
 537         isjson=False,
 538     )
 539
 540     # Check r2 removed 172:31::30 vpnv6 update
 541     test_func = functools.partial(_prefix30_not_found, tgen.gears["r2"])
 542     success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 543     assert success, "r2, VPNv6 update 172:31::30 still present"
 544
 545
 546 def test_prefix_changes_interface():
 547     """
 548     Test BGP update for a given prefix learnt on different interface
 549     """
 550     tgen = get_topogen()
 551     if tgen.routers_have_failure():
 552         pytest.skip(tgen.errors)
 553
 554     logger.info("Enabling a 172:31::50/128 prefix for r11")
 555     tgen.gears["r11"].vtysh_cmd(
 556         "configure terminal\nrouter bgp\naddress-family ipv6 unicast\nnetwork 172:31::50/128",
 557         isjson=False,
 558     )
 559
 560     # Check r2 received vpnv6 update with 172:31::50
 561     test_func = functools.partial(
 562         check_show_bgp_vpn_prefix_found,
 563         tgen.gears["r2"],
 564         "ipv6",
 565         "172:31::50/128",
 566         "444:1",
 567     )
 568     success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 569     assert success, "r2, VPNv6 update 172:31::50 not found"
 570
 571     # diagnostic
 572     logger.info("Dumping label nexthop table")
 573     tgen.gears["r1"].vtysh_cmd("show bgp vrf vrf1 label-nexthop detail", isjson=False)
 574
 575     label_list = set()
 576     bgp_vpnv6_table_check(
 577         tgen.gears["r2"],
 578         group=PREFIXES_R11 + ["172:31::50/128"],
 579         label_list=label_list,
 580     )
 581
 582     assert (
 583         len(label_list) == 1
 584     ), "Multiple Label values found for updates from r11 found"
 585
 586     oldlabel = label_list.pop()
 587     logger.info("r1, getting the outgoing interface used by label {}".format(oldlabel))
 588     old_outgoing_interface = mpls_entry_get_interface(tgen.gears["r1"], oldlabel)
 589     logger.info(
 590         "r1, outgoing interface used by label {} is {}".format(
 591             oldlabel, old_outgoing_interface
 592         )
 593     )
 594
 595     logger.info("Moving the 172:31::50/128 prefix from r11 to r13")
 596     tgen.gears["r11"].vtysh_cmd(
 597         "configure terminal\nrouter bgp\naddress-family ipv6 unicast\nno network 172:31::50/128",
 598         isjson=False,
 599     )
 600     tgen.gears["r13"].vtysh_cmd(
 601         "configure terminal\nrouter bgp\naddress-family ipv6 unicast\nnetwork 172:31::50/128",
 602         isjson=False,
 603     )
 604
 605     # Check r2 removed 172:31::50 vpnv6 update with old label
 606     test_func = functools.partial(
 607         check_show_bgp_vpn_prefix_not_found,
 608         tgen.gears["r2"],
 609         "ipv6",
 610         "172:31::50/128",
 611         "444:1",
 612         label=oldlabel,
 613     )
 614     success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 615     assert (
 616         success
 617     ), "r2, vpnv6 update 172:31::50 with old label {0} still present".format(oldlabel)
 618
 619     # diagnostic
 620     logger.info("Dumping label nexthop table")
 621     tgen.gears["r1"].vtysh_cmd("show bgp vrf vrf1 label-nexthop detail", isjson=False)
 622
 623     # Check r2 received new 172:31::50 vpnv6 update
 624     test_func = functools.partial(
 625         check_show_bgp_vpn_prefix_found,
 626         tgen.gears["r2"],
 627         "ipv6",
 628         "172:31::50/128",
 629         "444:1",
 630     )
 631     success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 632     assert success, "r2, vpnv6 update 172:31::50 not found"
 633
 634     label_list = set()
 635     bgp_vpnv6_table_check(
 636         tgen.gears["r2"],
 637         group=["172:31::13/128", "172:31::50/128"],
 638         label_list=label_list,
 639     )
 640     assert (
 641         len(label_list) == 1
 642     ), "Multiple Label values found for updates from r13 found"
 643
 644     newlabel = label_list.pop()
 645     logger.info("r1, getting the outgoing interface used by label {}".format(newlabel))
 646     new_outgoing_interface = mpls_entry_get_interface(tgen.gears["r1"], newlabel)
 647     logger.info(
 648         "r1, outgoing interface used by label {} is {}".format(
 649             newlabel, new_outgoing_interface
 650         )
 651     )
 652     if old_outgoing_interface == new_outgoing_interface:
 653         assert 0, "r1, outgoing interface did not change whereas BGP update moved"
 654
 655     logger.info("Restoring state by removing the 172:31::50/128 prefix from r13")
 656     tgen.gears["r13"].vtysh_cmd(
 657         "configure terminal\nrouter bgp\naddress-family ipv6 unicast\nno network 172:31::50/128",
 658         isjson=False,
 659     )
 660
 661
 662 def test_changing_default_label_value():
 663     """
 664     Change the MPLS default value
 665     Check that r1 VPNv6 entries have the 222 label value
 666     Check that MPLS entry with old label value is no more present
 667     Check that MPLS entry for local traffic has inLabel set to 222
 668     """
 669     tgen = get_topogen()
 670     if tgen.routers_have_failure():
 671         pytest.skip(tgen.errors)
 672
 673     router = tgen.gears["r1"]
 674
 675     # counting the number of labels used in the VPNv6 table
 676     label_list = set()
 677     logger.info("r1, VPNv6 table, check the number of labels used before modification")
 678     bgp_vpnv6_table_check_all(router, label_list)
 679     old_len = len(label_list)
 680     assert (
 681         old_len != 1
 682     ), "r1, number of labels used should be greater than 1, oberved {} ".format(old_len)
 683
 684     logger.info("r1, vrf1, changing the default MPLS label value to export to 222")
 685     router.vtysh_cmd(
 686         "configure terminal\nrouter bgp 65500 vrf vrf1\naddress-family ipv6 unicast\nlabel vpn export 222\n",
 687         isjson=False,
 688     )
 689
 690     # Check r1 updated the MPLS entry with the 222 label value
 691     logger.info(
 692         "r1, mpls table, check that MPLS entry with inLabel set to 222 has vrf1 interface"
 693     )
 694     test_func = functools.partial(
 695         check_show_mpls_table_entry_label_found, router, 222, "vrf1"
 696     )
 697     success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 698     assert success, "r1, mpls entry with label 222 not found"
 699
 700     # check label repartition is ok
 701     logger.info("r1, VPNv6 table, check the number of labels used after modification")
 702     label_list = set()
 703     bgp_vpnv6_table_check_all(router, label_list)
 704     new_len = len(label_list)
 705     assert (
 706         old_len == new_len
 707     ), "r1, number of labels after modification differ from previous, observed {}, expected {} ".format(
 708         new_len, old_len
 709     )
 710
 711     logger.info(
 712         "r1, VPNv6 table, check that prefixes that were using the vrf label have refreshed the label value to 222"
 713     )
 714     bgp_vpnv6_table_check(router, group=PREFIXES_CONNECTED, label_value_expected=222)
 715
 716
 717 def test_unconfigure_allocation_mode_nexthop():
 718     """
 719     Test unconfiguring allocation mode per nexthop
 720     Check on r2 that new MPLS label values have been propagated
 721     Check that show mpls table has no entry with label 17 (previously used)
 722     Check that all VPN updates on r1 should have label value moved to 222
 723     Check that show mpls table will only have 222 label value
 724     """
 725     tgen = get_topogen()
 726     if tgen.routers_have_failure():
 727         pytest.skip(tgen.errors)
 728
 729     logger.info("Unconfiguring allocation mode per nexthop")
 730     router = tgen.gears["r1"]
 731     dump = router.vtysh_cmd(
 732         "configure terminal\nrouter bgp 65500 vrf vrf1\naddress-family ipv6 unicast\nno label vpn export allocation-mode per-nexthop\n",
 733         isjson=False,
 734     )
 735
 736     # Check r1 updated the MPLS entry with the 222 label value
 737     logger.info(
 738         "r1, mpls table, check that MPLS entry with inLabel set to 17 is not present"
 739     )
 740     test_func = functools.partial(
 741         check_show_mpls_table_entry_label_not_found, router, 17
 742     )
 743     success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 744     assert success, "r1, mpls entry with label 17 still present"
 745
 746     # Check vpnv6 routes from r1
 747     logger.info("Checking VPNv6 routes on r1")
 748     label_list = set()
 749     bgp_vpnv6_table_check_all(router, label_list=label_list, same=True)
 750     assert len(label_list) == 1, "r1, multiple Label values found for VPNv6 updates"
 751
 752     new_label = label_list.pop()
 753     assert (
 754         new_label == 222
 755     ), "r1, wrong label value in VPNv6 table, expected 222, observed {}".format(
 756         new_label
 757     )
 758
 759     # Check mpls table with 222 value
 760     logger.info("Checking MPLS values on show mpls table of r1")
 761     label_list = set()
 762     label_list.add(222)
 763     mpls_table_check(router, label_list=label_list)
 764
 765
 766 def test_reconfigure_allocation_mode_nexthop():
 767     """
 768     Test re-configuring allocation mode per nexthop
 769     Check that show mpls table has no entry with label 17
 770     Check that all VPN updates on r1 should have multiple label values and not only 222
 771     Check that show mpls table will have multiple label values and not only 222
 772     """
 773     tgen = get_topogen()
 774     if tgen.routers_have_failure():
 775         pytest.skip(tgen.errors)
 776
 777     logger.info("Reconfiguring allocation mode per nexthop")
 778     router = tgen.gears["r1"]
 779     dump = router.vtysh_cmd(
 780         "configure terminal\nrouter bgp 65500 vrf vrf1\naddress-family ipv6 unicast\nlabel vpn export allocation-mode per-nexthop\n",
 781         isjson=False,
 782     )
 783
 784     # Check that show mpls table has no entry with label 17
 785     logger.info(
 786         "r1, mpls table, check that MPLS entry with inLabel set to 17 is present"
 787     )
 788     test_func = functools.partial(
 789         check_show_mpls_table_entry_label_not_found, router, 17
 790     )
 791     success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 792     assert success, "r1, mpls entry with label 17 still present"
 793
 794     # Check vpnv6 routes from r1
 795     logger.info("Checking VPNv6 routes on r1")
 796     label_list = set()
 797     bgp_vpnv6_table_check_all(router, label_list=label_list)
 798     assert len(label_list) != 1, "r1, only 1 label values found for VPNv6 updates"
 799
 800     # Check mpls table with all values
 801     logger.info("Checking MPLS values on show mpls table of r1")
 802     mpls_table_check(router, label_list=label_list)
 803
 804
 805 def test_memory_leak():
 806     "Run the memory leak test and report results."
 807     tgen = get_topogen()
 808     if not tgen.is_memleak_enabled():
 809         pytest.skip("Memory leak test/report is disabled")
 810
 811     tgen.report_memory_leaks()
 812
 813
 814 if __name__ == "__main__":
 815     args = ["-s"] + sys.argv[1:]
 816     sys.exit(pytest.main(args))