tests/topotests/isis_sr_te_topo1/test_isis_sr_te_topo1.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # test_isis_sr_topo1.py
   6 # Part of NetDEF Topology Tests
   7 #
   8 # Copyright (c) 2019 by
   9 # Network Device Education Foundation, Inc. ("NetDEF")
  10 #
  11
  12 """
  13 test_isis_sr_te_topo1.py:
  14
  15                          +---------+
  16                          |         |
  17                          |   RT1   |
  18                          | 1.1.1.1 |
  19                          |         |
  20                          +---------+
  21                               |eth-sw1
  22                               |
  23                               |
  24                               |
  25          +---------+          |          +---------+
  26          |         |          |          |         |
  27          |   RT2   |eth-sw1   |   eth-sw1|   RT3   |
  28          | 2.2.2.2 +----------+----------+ 3.3.3.3 |
  29          |         |     10.0.1.0/24     |         |
  30          +---------+                     +---------+
  31     eth-rt4-1|  |eth-rt4-2          eth-rt5-1|  |eth-rt5-2
  32              |  |                            |  |
  33   10.0.2.0/24|  |10.0.3.0/24      10.0.4.0/24|  |10.0.5.0/24
  34              |  |                            |  |
  35     eth-rt2-1|  |eth-rt2-2          eth-rt3-1|  |eth-rt3-2
  36          +---------+                     +---------+
  37          |         |                     |         |
  38          |   RT4   |     10.0.6.0/24     |   RT5   |
  39          | 4.4.4.4 +---------------------+ 5.5.5.5 |
  40          |         |eth-rt5       eth-rt4|         |
  41          +---------+                     +---------+
  42        eth-rt6|                                |eth-rt6
  43               |                                |
  44    10.0.7.0/24|                                |10.0.8.0/24
  45               |          +---------+           |
  46               |          |         |           |
  47               |          |   RT6   |           |
  48               +----------+ 6.6.6.6 +-----------+
  49                   eth-rt4|         |eth-rt5
  50                          +---------+
  51                               |eth-dst (.1)
  52                               |
  53                               |10.0.11.0/24
  54                               |
  55                               |eth-rt6 (.2)
  56                          +---------+
  57                          |         |
  58                          |   DST   |
  59                          | 9.9.9.2 |
  60                          |         |
  61                          +---------+
  62
  63 """
  64
  65 import os
  66 import sys
  67 import pytest
  68 import json
  69 from functools import partial
  70
  71 # Save the Current Working Directory to find configuration files.
  72 CWD = os.path.dirname(os.path.realpath(__file__))
  73 sys.path.append(os.path.join(CWD, "../"))
  74
  75 # pylint: disable=C0413
  76 # Import topogen and topotest helpers
  77 from lib import topotest
  78 from lib.topogen import Topogen, TopoRouter, get_topogen
  79 from lib.topolog import logger
  80
  81 # Required to instantiate the topology builder class.
  82
  83 pytestmark = [pytest.mark.bgpd, pytest.mark.isisd, pytest.mark.pathd]
  84
  85
  86 def build_topo(tgen):
  87     "Build function"
  88
  89     #
  90     # Define FRR Routers
  91     #
  92     for router in ["rt1", "rt2", "rt3", "rt4", "rt5", "rt6", "dst"]:
  93         tgen.add_router(router)
  94
  95     #
  96     # Define connections
  97     #
  98     switch = tgen.add_switch("s1")
  99     switch.add_link(tgen.gears["rt1"], nodeif="eth-sw1")
 100     switch.add_link(tgen.gears["rt2"], nodeif="eth-sw1")
 101     switch.add_link(tgen.gears["rt3"], nodeif="eth-sw1")
 102
 103     switch = tgen.add_switch("s2")
 104     switch.add_link(tgen.gears["rt2"], nodeif="eth-rt4-1")
 105     switch.add_link(tgen.gears["rt4"], nodeif="eth-rt2-1")
 106
 107     switch = tgen.add_switch("s3")
 108     switch.add_link(tgen.gears["rt2"], nodeif="eth-rt4-2")
 109     switch.add_link(tgen.gears["rt4"], nodeif="eth-rt2-2")
 110
 111     switch = tgen.add_switch("s4")
 112     switch.add_link(tgen.gears["rt3"], nodeif="eth-rt5-1")
 113     switch.add_link(tgen.gears["rt5"], nodeif="eth-rt3-1")
 114
 115     switch = tgen.add_switch("s5")
 116     switch.add_link(tgen.gears["rt3"], nodeif="eth-rt5-2")
 117     switch.add_link(tgen.gears["rt5"], nodeif="eth-rt3-2")
 118
 119     switch = tgen.add_switch("s6")
 120     switch.add_link(tgen.gears["rt4"], nodeif="eth-rt5")
 121     switch.add_link(tgen.gears["rt5"], nodeif="eth-rt4")
 122
 123     switch = tgen.add_switch("s7")
 124     switch.add_link(tgen.gears["rt4"], nodeif="eth-rt6")
 125     switch.add_link(tgen.gears["rt6"], nodeif="eth-rt4")
 126
 127     switch = tgen.add_switch("s8")
 128     switch.add_link(tgen.gears["rt5"], nodeif="eth-rt6")
 129     switch.add_link(tgen.gears["rt6"], nodeif="eth-rt5")
 130
 131     switch = tgen.add_switch("s9")
 132     switch.add_link(tgen.gears["rt6"], nodeif="eth-dst")
 133     switch.add_link(tgen.gears["dst"], nodeif="eth-rt6")
 134
 135
 136 def setup_module(mod):
 137     "Sets up the pytest environment"
 138
 139     tgen = Topogen(build_topo, mod.__name__)
 140
 141     frrdir = tgen.config.get(tgen.CONFIG_SECTION, "frrdir")
 142     if not os.path.isfile(os.path.join(frrdir, "pathd")):
 143         pytest.skip("pathd daemon wasn't built")
 144
 145     tgen.start_topology()
 146
 147     router_list = tgen.routers()
 148
 149     # For all registered routers, load the zebra configuration file
 150     for rname, router in router_list.items():
 151         router.load_config(
 152             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 153         )
 154         router.load_config(
 155             TopoRouter.RD_ISIS, os.path.join(CWD, "{}/isisd.conf".format(rname))
 156         )
 157         router.load_config(
 158             TopoRouter.RD_PATH, os.path.join(CWD, "{}/pathd.conf".format(rname))
 159         )
 160         router.load_config(
 161             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
 162         )
 163
 164     tgen.start_router()
 165
 166
 167 def teardown_module(mod):
 168     "Teardown the pytest environment"
 169     tgen = get_topogen()
 170
 171     # This function tears down the whole topology.
 172     tgen.stop_topology()
 173
 174
 175 def setup_testcase(msg):
 176     logger.info(msg)
 177     tgen = get_topogen()
 178
 179     # Skip if previous fatal error condition is raised
 180     if tgen.routers_have_failure():
 181         pytest.skip(tgen.errors)
 182
 183     return tgen
 184
 185
 186 def print_cmd_result(rname, command):
 187     print(get_topogen().gears[rname].vtysh_cmd(command, isjson=False))
 188
 189
 190 def compare_json_test(router, command, reference, exact):
 191     output = router.vtysh_cmd(command, isjson=True)
 192     result = topotest.json_cmp(output, reference)
 193
 194     # Note: topotest.json_cmp() just checks on inclusion of keys.
 195     # For exact matching also compare the other way around.
 196     if not result and exact:
 197         return topotest.json_cmp(reference, output)
 198     else:
 199         return result
 200
 201
 202 def cmp_json_output(rname, command, reference, exact=False):
 203     "Compare router JSON output"
 204
 205     logger.info('Comparing router "%s" "%s" output', rname, command)
 206
 207     tgen = get_topogen()
 208     filename = "{}/{}/{}".format(CWD, rname, reference)
 209     expected = json.loads(open(filename).read())
 210
 211     # Run test function until we get an result. Wait at most 60 seconds.
 212     test_func = partial(compare_json_test, tgen.gears[rname], command, expected, exact)
 213     _, diff = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
 214     assertmsg = '"{}" JSON output mismatches the expected result'.format(rname)
 215     assert diff is None, assertmsg
 216
 217
 218 def cmp_json_output_exact(rname, command, reference):
 219     return cmp_json_output(rname, command, reference, True)
 220
 221
 222 def compare_json_test_inverted(router, command, reference, exact):
 223     "logically inverts result of compare_json_test"
 224
 225     # None vs something else
 226     result = compare_json_test(router, command, reference, exact)
 227     if result is None:
 228         return "Some"
 229     return None
 230
 231
 232 def cmp_json_output_doesnt(rname, command, reference):
 233     "Compare router JSON output, shouldn't include reference"
 234
 235     logger.info('Comparing (anti) router "%s" "%s" output', rname, command)
 236
 237     tgen = get_topogen()
 238     filename = "{}/{}/{}".format(CWD, rname, reference)
 239     expected = json.loads(open(filename).read())
 240
 241     # Run test function until we get an result. Wait at most 60 seconds.
 242     test_func = partial(
 243         compare_json_test_inverted, tgen.gears[rname], command, expected, exact=False
 244     )
 245     _, diff = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
 246     assertmsg = '"{}" JSON output mismatches the expected result'.format(rname)
 247     assert diff is None, assertmsg
 248
 249
 250 def dump_json(v):
 251     if isinstance(v, (dict, list)):
 252         return "\t" + "\t".join(
 253             json.dumps(v, indent=4, separators=(",", ": ")).splitlines(True)
 254         )
 255     else:
 256         return "'{}'".format(v)
 257
 258
 259 def add_candidate_path(rname, endpoint, pref, name, segment_list="default", color=1):
 260     get_topogen().net[rname].cmd(
 261         """ \
 262         vtysh -c "conf t" \
 263               -c "segment-routing" \
 264               -c "traffic-eng" \
 265               -c "policy color """
 266         + str(color)
 267         + " endpoint "
 268         + endpoint
 269         + """" \
 270               -c "candidate-path preference """
 271         + str(pref)
 272         + """ name """
 273         + name
 274         + """ explicit segment-list """
 275         + segment_list
 276         + '''"'''
 277     )
 278
 279
 280 def delete_candidate_path(rname, endpoint, pref, color=1):
 281     get_topogen().net[rname].cmd(
 282         """ \
 283         vtysh -c "conf t" \
 284               -c "segment-routing" \
 285               -c "traffic-eng" \
 286               -c "policy color """
 287         + str(color)
 288         + " endpoint "
 289         + endpoint
 290         + """" \
 291               -c "no candidate-path preference """
 292         + str(pref)
 293         + '''"'''
 294     )
 295
 296
 297 def add_segment(rname, name, index, label):
 298     get_topogen().net[rname].cmd(
 299         """ \
 300         vtysh -c "conf t" \
 301               -c "segment-routing" \
 302               -c "traffic-eng" \
 303               -c "segment-list """
 304         + name
 305         + """" \
 306               -c "index """
 307         + str(index)
 308         + """ mpls label """
 309         + str(label)
 310         + '''"'''
 311     )
 312
 313
 314 def delete_segment(rname, name, index):
 315     get_topogen().net[rname].cmd(
 316         """ \
 317         vtysh -c "conf t" \
 318               -c "segment-routing" \
 319               -c "traffic-eng" \
 320               -c "segment-list """
 321         + name
 322         + """" \
 323               -c "no index """
 324         + str(index)
 325         + '''"'''
 326     )
 327
 328
 329 def create_sr_policy(rname, endpoint, bsid):
 330     get_topogen().net[rname].cmd(
 331         """ \
 332         vtysh -c "conf t" \
 333               -c "segment-routing" \
 334               -c "traffic-eng" \
 335               -c "policy color 1 endpoint """
 336         + endpoint
 337         + """" \
 338               -c "name default" \
 339               -c "binding-sid """
 340         + str(bsid)
 341         + '''"'''
 342     )
 343
 344
 345 def delete_sr_policy(rname, endpoint):
 346     get_topogen().net[rname].cmd(
 347         """ \
 348         vtysh -c "conf t" \
 349               -c "segment-routing" \
 350               -c "traffic-eng" \
 351               -c "no policy color 1 endpoint """
 352         + endpoint
 353         + '''"'''
 354     )
 355
 356
 357 def create_prefix_sid(rname, prefix, sid):
 358     get_topogen().net[rname].cmd(
 359         """ \
 360         vtysh -c "conf t" \
 361               -c "router isis 1" \
 362               -c "segment-routing prefix """
 363         + prefix
 364         + " index "
 365         + str(sid)
 366         + '''"'''
 367     )
 368
 369
 370 def delete_prefix_sid(rname, prefix):
 371     get_topogen().net[rname].cmd(
 372         ''' \
 373         vtysh -c "conf t" \
 374               -c "router isis 1" \
 375               -c "no segment-routing prefix "'''
 376         + prefix
 377     )
 378
 379
 380 def set_route_map_color(rname, color):
 381     get_topogen().net[rname].cmd(
 382         ''' \
 383         vtysh -c "conf t" \
 384               -c "route-map SET_SR_POLICY permit 10" \
 385               -c "set sr-te color  "'''
 386         + str(color)
 387     )
 388
 389
 390 def router_bgp_shutdown_neighbor(rname, neighbor):
 391     get_topogen().net[rname].cmd(
 392         """ \
 393         vtysh -c "conf t" \
 394               -c "router bgp 1" \
 395               -c " neighbor """
 396         + neighbor
 397         + ' shutdown"'
 398     )
 399
 400
 401 def router_bgp_no_shutdown_neighbor(rname, neighbor):
 402     get_topogen().net[rname].cmd(
 403         """ \
 404         vtysh -c "conf t" \
 405               -c "router bgp 1" \
 406               -c " no neighbor """
 407         + neighbor
 408         + ' shutdown"'
 409     )
 410
 411
 412 def show_running_cfg(rname):
 413     output = (
 414         get_topogen()
 415         .net[rname]
 416         .cmd(
 417             """ \
 418         vtysh -c "show run" """
 419         )
 420     )
 421     logger.info(output)
 422
 423
 424 #
 425 # Step 1
 426 #
 427 # Checking the MPLS table using a single SR Policy and a single Candidate Path
 428 #
 429 def test_srte_init_step1():
 430     setup_testcase("Test (step 1): wait for IS-IS convergence / label distribution")
 431
 432     for rname in ["rt1", "rt6"]:
 433         cmp_json_output(
 434             rname, "show mpls table json", "step1/show_mpls_table_without_candidate.ref"
 435         )
 436
 437
 438 def test_srte_add_candidate_check_mpls_table_step1():
 439     setup_testcase("Test (step 1): check MPLS table regarding the added Candidate Path")
 440
 441     for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
 442         add_candidate_path(rname, endpoint, 100, "default")
 443         cmp_json_output(
 444             rname, "show mpls table json", "step1/show_mpls_table_with_candidate.ref"
 445         )
 446         delete_candidate_path(rname, endpoint, 100)
 447
 448
 449 def test_srte_reinstall_sr_policy_check_mpls_table_step1():
 450     setup_testcase(
 451         "Test (step 1): check MPLS table after the SR Policy was removed and reinstalled"
 452     )
 453
 454     for rname, endpoint, bsid in [("rt1", "6.6.6.6", 1111), ("rt6", "1.1.1.1", 6666)]:
 455         add_candidate_path(rname, endpoint, 100, "default")
 456         delete_sr_policy(rname, endpoint)
 457         cmp_json_output(
 458             rname, "show mpls table json", "step1/show_mpls_table_without_candidate.ref"
 459         )
 460         create_sr_policy(rname, endpoint, bsid)
 461         add_candidate_path(rname, endpoint, 100, "default")
 462         cmp_json_output(
 463             rname, "show mpls table json", "step1/show_mpls_table_with_candidate.ref"
 464         )
 465         delete_candidate_path(rname, endpoint, 100)
 466
 467
 468 #
 469 # Step 2
 470 #
 471 # Checking pathd operational data using a single SR Policy and a single Candidate Path
 472 #
 473 def test_srte_bare_policy_step2():
 474     setup_testcase("Test (step 2): bare SR Policy should not be operational")
 475
 476     for rname in ["rt1", "rt6"]:
 477         cmp_json_output_exact(
 478             rname,
 479             "show yang operational-data /frr-pathd:pathd pathd",
 480             "step2/show_operational_data.ref",
 481         )
 482
 483
 484 def test_srte_add_candidate_check_operational_data_step2():
 485     setup_testcase(
 486         "Test (step 2): add single Candidate Path, SR Policy should be operational"
 487     )
 488
 489     for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
 490         add_candidate_path(rname, endpoint, 100, "default")
 491         cmp_json_output(
 492             rname,
 493             "show yang operational-data /frr-pathd:pathd pathd",
 494             "step2/show_operational_data_with_candidate.ref",
 495         )
 496
 497
 498 def test_srte_config_remove_candidate_check_operational_data_step2():
 499     setup_testcase(
 500         "Test (step 2): remove single Candidate Path, SR Policy should not be operational anymore"
 501     )
 502
 503     for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
 504         delete_candidate_path(rname, endpoint, 100)
 505         cmp_json_output_exact(
 506             rname,
 507             "show yang operational-data /frr-pathd:pathd pathd",
 508             "step2/show_operational_data.ref",
 509         )
 510
 511
 512 #
 513 # Step 3
 514 #
 515 # Testing the Candidate Path selection
 516 #
 517 def test_srte_add_two_candidates_step3():
 518     setup_testcase("Test (step 3): second Candidate Path has higher Priority")
 519
 520     for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
 521         for pref, cand_name in [("100", "first"), ("200", "second")]:
 522             add_candidate_path(rname, endpoint, pref, cand_name)
 523         cmp_json_output(
 524             rname,
 525             "show yang operational-data /frr-pathd:pathd pathd",
 526             "step3/show_operational_data_with_two_candidates.ref",
 527         )
 528
 529     # cleanup
 530     for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
 531         for pref in ["100", "200"]:
 532             delete_candidate_path(rname, endpoint, pref)
 533
 534
 535 def test_srte_add_two_candidates_with_reverse_priority_step3():
 536     setup_testcase("Test (step 3): second Candidate Path has lower Priority")
 537
 538     # Use reversed priorities here
 539     for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
 540         for pref, cand_name in [("200", "first"), ("100", "second")]:
 541             add_candidate_path(rname, endpoint, pref, cand_name)
 542         cmp_json_output(
 543             rname,
 544             "show yang operational-data /frr-pathd:pathd pathd",
 545             "step3/show_operational_data_with_two_candidates.ref",
 546         )
 547
 548     # cleanup
 549     for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
 550         for pref in ["100", "200"]:
 551             delete_candidate_path(rname, endpoint, pref)
 552
 553
 554 def test_srte_remove_best_candidate_step3():
 555     setup_testcase("Test (step 3): delete the Candidate Path with higher priority")
 556
 557     for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
 558         for pref, cand_name in [("100", "first"), ("200", "second")]:
 559             add_candidate_path(rname, endpoint, pref, cand_name)
 560
 561     # Delete candidate with higher priority
 562     for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
 563         delete_candidate_path(rname, endpoint, 200)
 564
 565     # Candidate with lower priority should get active now
 566     for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
 567         cmp_json_output(
 568             rname,
 569             "show yang operational-data /frr-pathd:pathd pathd",
 570             "step3/show_operational_data_with_single_candidate.ref",
 571         )
 572         # cleanup
 573         delete_candidate_path(rname, endpoint, 100)
 574
 575
 576 #
 577 # Step 4
 578 #
 579 # Checking MPLS table with a single SR Policy and a Candidate Path with different Segment Lists and other modifications
 580 #
 581 def test_srte_change_segment_list_check_mpls_table_step4():
 582     setup_testcase("Test (step 4): check MPLS table for changed Segment List")
 583
 584     for rname, endpoint in [("rt1", "6.6.6.6"), ("rt6", "1.1.1.1")]:
 585         add_candidate_path(rname, endpoint, 100, "default")
 586         # now change the segment list name
 587         add_candidate_path(rname, endpoint, 100, "default", "test")
 588         cmp_json_output(rname, "show mpls table json", "step4/show_mpls_table.ref")
 589         delete_candidate_path(rname, endpoint, 100)
 590
 591
 592 def test_srte_segment_list_add_segment_check_mpls_table_step4():
 593     setup_testcase(
 594         "Test (step 4): check MPLS table for added (then changed and finally deleted) segment"
 595     )
 596
 597     add_candidate_path("rt1", "6.6.6.6", 100, "default", "test")
 598
 599     # first add a new segment
 600     add_segment("rt1", "test", 25, 16050)
 601     cmp_json_output(
 602         "rt1", "show mpls table json", "step4/show_mpls_table_add_segment.ref"
 603     )
 604
 605     # ... then change it ...
 606     add_segment("rt1", "test", 25, 16030)
 607     cmp_json_output(
 608         "rt1", "show mpls table json", "step4/show_mpls_table_change_segment.ref"
 609     )
 610
 611     # ... and finally delete it
 612     delete_segment("rt1", "test", 25)
 613     cmp_json_output("rt1", "show mpls table json", "step4/show_mpls_table.ref")
 614     delete_candidate_path("rt1", "6.6.6.6", 100)
 615
 616
 617 def save_rt(routername, filename):
 618     save_filename = routername + "/" + filename
 619     tgen = get_topogen()
 620     router = tgen.gears[routername]
 621
 622     config_output = router.vtysh_cmd("sh run")
 623
 624     route_output_json = json.loads(router.vtysh_cmd("show ip route bgp json"))
 625     route_output = dump_json(route_output_json)
 626
 627     f = open(save_filename, "w")
 628     f.write(config_output)
 629     f.write(route_output)
 630     f.close()
 631
 632
 633 #
 634 # Step 5
 635 #
 636 # Checking the nexthop using a single SR Policy and a Candidate Path with configured route-map
 637 #
 638 def test_srte_route_map_with_sr_policy_check_nextop_step5():
 639     setup_testcase(
 640         "Test (step 5): recursive nexthop learned through BGP neighbour should be aligned with SR Policy from route-map"
 641     )
 642
 643     # (re-)build the SR Policy two times to ensure that reinstalling still works
 644     for i in [1, 2]:
 645         cmp_json_output(
 646             "rt1", "show ip route bgp json", "step5/show_ip_route_bgp_inactive_srte.ref"
 647         )
 648
 649         delete_sr_policy("rt1", "6.6.6.6")
 650         cmp_json_output(
 651             "rt1", "show ip route bgp json", "step5/show_ip_route_bgp_inactive_srte.ref"
 652         )
 653
 654         create_sr_policy("rt1", "6.6.6.6", 1111)
 655         cmp_json_output(
 656             "rt1", "show ip route bgp json", "step5/show_ip_route_bgp_inactive_srte.ref"
 657         )
 658
 659         add_candidate_path("rt1", "6.6.6.6", 100, "default")
 660         cmp_json_output(
 661             "rt1", "show ip route bgp json", "step5/show_ip_route_bgp_active_srte.ref"
 662         )
 663
 664         delete_candidate_path("rt1", "6.6.6.6", 100)
 665
 666
 667 def test_srte_route_map_sr_policy_vs_route_order_step5():
 668     setup_testcase(
 669         "Test (step 5): Config policy first, add route after and check route validity"
 670     )
 671
 672     #
 673     # BGP route and route-map are already configured.
 674     # route-map sets color 1 on BGP routes
 675
 676     # Developer: to force pause here
 677     # tgen = get_topogen()
 678     # tgen.mininet_cli()
 679
 680     #
 681     # Configure policy/path
 682     #
 683     add_candidate_path("rt1", "6.6.6.6", 100, "default", "default", 1)
 684
 685     #
 686     # Route should be valid
 687     #
 688     logger.info(
 689         "BGP route and route-map are already configured, SR candidate path added after. Route should be valid"
 690     )
 691     cmp_json_output(
 692         "rt1", "show ip route bgp json", "step5/show_ip_route_bgp_active_srte.ref"
 693     )
 694
 695     #
 696     # shutdown/no-shutdown on BGP neighbor to delete/re-add BGP route
 697     #
 698     router_bgp_shutdown_neighbor("rt1", "6.6.6.6")
 699     router_bgp_no_shutdown_neighbor("rt1", "6.6.6.6")
 700
 701     #
 702     # Route should be valid (but isn't)
 703     #
 704     logger.info(
 705         "After shutdown + no-shutdown neighbor. Route should be valid, but isn't"
 706     )
 707     cmp_json_output(
 708         "rt1", "show ip route bgp json", "step5/show_ip_route_bgp_active_srte.ref"
 709     )
 710
 711     #
 712     # delete and re-add policy/path
 713     #
 714     delete_candidate_path("rt1", "6.6.6.6", 100, 1)
 715     add_candidate_path("rt1", "6.6.6.6", 100, "default", "default", 1)
 716
 717     #
 718     # Route should be valid
 719     #
 720     logger.info("After re-add candidate path. Route should be valid")
 721     cmp_json_output(
 722         "rt1", "show ip route bgp json", "step5/show_ip_route_bgp_active_srte.ref"
 723     )
 724
 725     # Developer: to force pause here
 726     # tgen = get_topogen()
 727     # tgen.mininet_cli()
 728
 729     # clean up
 730     delete_candidate_path("rt1", "6.6.6.6", 100, 2)
 731
 732
 733 def test_srte_route_map_sr_policy_vs_routemap_order_step5():
 734     setup_testcase(
 735         "Test (step 5): Config policy first, set route-map after and check route validity"
 736     )
 737
 738     #
 739     # BGP route and route-map are already configured.
 740     # route-map sets color 1 on BGP routes
 741     #
 742
 743     #
 744     # Configure policy/path
 745     #
 746     add_candidate_path("rt1", "6.6.6.6", 100, "default", "default", 1)
 747
 748     #
 749     # Route should be valid
 750     #
 751     logger.info("After add candidate path. Route should be valid")
 752     cmp_json_output(
 753         "rt1", "show ip route bgp json", "step5/show_ip_route_bgp_active_srte.ref"
 754     )
 755
 756     # Developer: to force pause here
 757     # tgen = get_topogen()
 758     # tgen.mininet_cli()
 759
 760     #
 761     # change route-map color to someting else and back again
 762     #
 763     set_route_map_color("rt1", 2)
 764     logger.info("route-map color was set to 2")
 765     # show_running_cfg("rt1")
 766     # 220625 nexthop no longer becomes empty. Colored routes without
 767     # matching SR policies now fall back to their non-colored equivalent
 768     # nexthops. So the route to 9.9.9.9/32 will now be valid, but with
 769     # different nexthop values.
 770     logger.info("now route table will lose policy-mapped route")
 771     cmp_json_output_doesnt(
 772         "rt1", "show ip route bgp json", "step5/show_ip_route_bgp_active_srte.ref"
 773     )
 774     set_route_map_color("rt1", 1)
 775     logger.info("route-map color was set to 1")
 776     # show_running_cfg("rt1")
 777
 778     #
 779     # Route should be valid (but isn't)
 780     #
 781     logger.info("After change route-map color. Route should be valid, but isn't")
 782     cmp_json_output(
 783         "rt1", "show ip route bgp json", "step5/show_ip_route_bgp_active_srte.ref"
 784     )
 785
 786     #
 787     # delete and re-add policy/path
 788     #
 789     delete_candidate_path("rt1", "6.6.6.6", 100, 1)
 790     add_candidate_path("rt1", "6.6.6.6", 100, "default", "default", 1)
 791
 792     #
 793     # Route should be valid
 794     #
 795     logger.info("After delete/re-add candidate path. Route should be valid")
 796     cmp_json_output(
 797         "rt1", "show ip route bgp json", "step5/show_ip_route_bgp_active_srte.ref"
 798     )
 799
 800     # Developer: force pause
 801     # tgen = get_topogen()
 802     # tgen.mininet_cli()
 803
 804     # clean up
 805     delete_candidate_path("rt1", "6.6.6.6", 100, 2)
 806
 807
 808 def test_srte_route_map_with_sr_policy_reinstall_prefix_sid_check_nextop_step5():
 809     setup_testcase(
 810         "Test (step 5): remove and re-install prefix SID on fist path element and check SR Policy activity"
 811     )
 812
 813     # first add a candidate path so the SR Policy is active
 814     add_candidate_path("rt1", "6.6.6.6", 100, "default")
 815     cmp_json_output(
 816         "rt1",
 817         "show yang operational-data /frr-pathd:pathd pathd",
 818         "step5/show_operational_data_active.ref",
 819     )
 820
 821     # delete prefix SID from first element of the configured path and check
 822     # if the SR Policy is inactive since the label can't be resolved anymore
 823     delete_prefix_sid("rt5", "5.5.5.5/32")
 824     cmp_json_output(
 825         "rt1",
 826         "show yang operational-data /frr-pathd:pathd pathd",
 827         "step5/show_operational_data_inactive.ref",
 828     )
 829     cmp_json_output(
 830         "rt1", "show ip route bgp json", "step5/show_ip_route_bgp_inactive_srte.ref"
 831     )
 832
 833     # re-create the prefix SID and check if the SR Policy is active
 834     create_prefix_sid("rt5", "5.5.5.5/32", 50)
 835     cmp_json_output(
 836         "rt1",
 837         "show yang operational-data /frr-pathd:pathd pathd",
 838         "step5/show_operational_data_active.ref",
 839     )
 840     cmp_json_output(
 841         "rt1", "show ip route bgp json", "step5/show_ip_route_bgp_active_srte.ref"
 842     )
 843
 844
 845 # Memory leak test template
 846 def test_memory_leak():
 847     "Run the memory leak test and report results."
 848     tgen = get_topogen()
 849     if not tgen.is_memleak_enabled():
 850         pytest.skip("Memory leak test/report is disabled")
 851
 852     tgen.report_memory_leaks()
 853
 854
 855 if __name__ == "__main__":
 856     args = ["-s"] + sys.argv[1:]
 857     sys.exit(pytest.main(args))