tests/topotests/ospf_basic_functionality/test_ospf_lan.py

   1 #!/usr/bin/python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # Copyright (c) 2020 by VMware, Inc. ("VMware")
   6 # Used Copyright (c) 2018 by Network Device Education Foundation, Inc.
   7 # ("NetDEF") in this file.
   8 #
   9
  10
  11 """OSPF Basic Functionality Automation."""
  12 import os
  13 import sys
  14 import time
  15 import pytest
  16 from copy import deepcopy
  17 from lib.topotest import frr_unicode
  18
  19 # Save the Current Working Directory to find configuration files.
  20 CWD = os.path.dirname(os.path.realpath(__file__))
  21 sys.path.append(os.path.join(CWD, "../"))
  22 sys.path.append(os.path.join(CWD, "../lib/"))
  23
  24 # pylint: disable=C0413
  25 # Import topogen and topotest helpers
  26 from lib.topogen import Topogen, get_topogen
  27
  28 # Import topoJson from lib, to create topology and initial configuration
  29 from lib.common_config import (
  30     start_topology,
  31     write_test_header,
  32     create_interfaces_cfg,
  33     write_test_footer,
  34     reset_config_on_routers,
  35     step,
  36     shutdown_bringup_interface,
  37     stop_router,
  38     start_router,
  39 )
  40 from lib.topolog import logger
  41 from lib.topojson import build_config_from_json
  42 from lib.ospf import (
  43     verify_ospf_neighbor,
  44     clear_ospf,
  45     create_router_ospf,
  46     verify_ospf_interface,
  47 )
  48 from ipaddress import IPv4Address
  49
  50 pytestmark = [pytest.mark.ospfd, pytest.mark.staticd]
  51
  52
  53 # Global variables
  54 topo = None
  55
  56 NETWORK = {
  57     "ipv4": [
  58         "11.0.20.1/32",
  59         "11.0.20.2/32",
  60         "11.0.20.3/32",
  61         "11.0.20.4/32",
  62         "11.0.20.5/32",
  63     ]
  64 }
  65
  66 """
  67 Topology:
  68
  69       Please view in a fixed-width font such as Courier.
  70       Topo : Broadcast Networks
  71         +---+       +---+          +---+           +---+
  72         |R0 +       +R1 +          +R2 +           +R3 |
  73         +-+-+       +-+-+          +-+-+           +-+-+
  74           |           |              |               |
  75           |           |              |               |
  76         --+-----------+--------------+---------------+-----
  77                          Ethernet Segment
  78
  79 Testcases:
  80 1.     OSPF Hello protocol - Verify DR BDR Elections
  81 2.     OSPF IFSM -Verify state change events on DR / BDR / DR Other
  82
  83  """
  84
  85
  86 def setup_module(mod):
  87     """
  88     Sets up the pytest environment
  89
  90     * `mod`: module name
  91     """
  92     testsuite_run_time = time.asctime(time.localtime(time.time()))
  93     logger.info("Testsuite start time: {}".format(testsuite_run_time))
  94     logger.info("=" * 40)
  95
  96     logger.info("Running setup_module to create topology")
  97
  98     # This function initiates the topology build with Topogen...
  99     json_file = "{}/ospf_lan.json".format(CWD)
 100     tgen = Topogen(json_file, mod.__name__)
 101     global topo
 102     topo = tgen.json_topo
 103     # ... and here it calls Mininet initialization functions.
 104
 105     # Starting topology, create tmp files which are loaded to routers
 106     #  to start daemons and then start routers
 107     start_topology(tgen)
 108
 109     # Creating configuration from JSON
 110     build_config_from_json(tgen, topo)
 111
 112     # Don't run this test if we have any failure.
 113     if tgen.routers_have_failure():
 114         pytest.skip(tgen.errors)
 115     # Api call verify whether OSPF is converged
 116     ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True)
 117     assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
 118         ospf_covergence
 119     )
 120
 121     logger.info("Running setup_module() done")
 122
 123
 124 def teardown_module():
 125     """Teardown the pytest environment"""
 126
 127     logger.info("Running teardown_module to delete topology")
 128
 129     tgen = get_topogen()
 130
 131     try:
 132         # Stop toplogy and Remove tmp files
 133         tgen.stop_topology()
 134
 135     except OSError:
 136         # OSError exception is raised when mininet tries to stop switch
 137         # though switch is stopped once but mininet tries to stop same
 138         # switch again, where it ended up with exception
 139         pass
 140
 141
 142 # ##################################
 143 # Test cases start here.
 144 # ##################################
 145
 146
 147 def test_ospf_lan_tc1_p0(request):
 148     """
 149     OSPF Hello protocol - Verify DR BDR Elections
 150
 151     """
 152     tc_name = request.node.name
 153     write_test_header(tc_name)
 154     tgen = get_topogen()
 155
 156     # Don't run this test if we have any failure.
 157     if tgen.routers_have_failure():
 158         pytest.skip(tgen.errors)
 159
 160     global topo
 161     step("Bring up the base config as per the topology")
 162     reset_config_on_routers(tgen)
 163     step("Verify that DR BDR DRother are elected in the LAN.")
 164     input_dict = {
 165         "r0": {
 166             "ospf": {
 167                 "neighbors": {
 168                     "r1": {"state": "Full", "role": "DR"},
 169                     "r2": {"state": "Full", "role": "DROther"},
 170                     "r3": {"state": "Full", "role": "DROther"},
 171                 }
 172             }
 173         }
 174     }
 175     dut = "r0"
 176     result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
 177     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 178
 179     step(
 180         "Verify that all the routers are in FULL state with DR and BDR "
 181         "in the topology"
 182     )
 183
 184     input_dict = {
 185         "r1": {
 186             "ospf": {
 187                 "neighbors": {
 188                     "r0": {"state": "Full", "role": "Backup"},
 189                     "r2": {"state": "Full", "role": "DROther"},
 190                     "r3": {"state": "Full", "role": "DROther"},
 191                 }
 192             }
 193         }
 194     }
 195     dut = "r1"
 196     result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
 197     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 198
 199     step(
 200         "Configure DR pririty 100 on R0 and clear ospf neighbors " "on all the routers."
 201     )
 202
 203     input_dict = {
 204         "r0": {
 205             "links": {
 206                 "s1": {
 207                     "interface": topo["routers"]["r0"]["links"]["s1"]["interface"],
 208                     "ospf": {"priority": 100},
 209                 }
 210             }
 211         }
 212     }
 213
 214     result = create_interfaces_cfg(tgen, input_dict)
 215     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 216
 217     step("Clear ospf neighbours in all routers")
 218     for rtr in ["r0", "r1", "r2", "r3"]:
 219         clear_ospf(tgen, rtr)
 220
 221     step("Verify that DR election is triggered and R0 is elected as DR")
 222     input_dict = {
 223         "r0": {
 224             "ospf": {
 225                 "neighbors": {
 226                     "r1": {"state": "Full", "role": "Backup"},
 227                     "r2": {"state": "Full", "role": "DROther"},
 228                     "r3": {"state": "Full", "role": "DROther"},
 229                 }
 230             }
 231         }
 232     }
 233     dut = "r0"
 234     result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
 235     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 236
 237     step(
 238         "Configure DR pririty 150 on R0 and clear ospf neighbors " "on all the routers."
 239     )
 240
 241     input_dict = {
 242         "r0": {
 243             "links": {
 244                 "s1": {
 245                     "interface": topo["routers"]["r0"]["links"]["s1"]["interface"],
 246                     "ospf": {"priority": 150},
 247                 }
 248             }
 249         }
 250     }
 251
 252     result = create_interfaces_cfg(tgen, input_dict)
 253     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 254
 255     step("Clear ospf neighbours in all routers")
 256     for rtr in ["r0", "r1"]:
 257         clear_ospf(tgen, rtr)
 258
 259     step("Verify that DR election is triggered and R0 is elected as DR")
 260     input_dict = {
 261         "r0": {
 262             "ospf": {
 263                 "neighbors": {
 264                     "r1": {"state": "Full", "role": "Backup"},
 265                     "r2": {"state": "Full", "role": "DROther"},
 266                     "r3": {"state": "Full", "role": "DROther"},
 267                 }
 268             }
 269         }
 270     }
 271     dut = "r0"
 272     result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
 273     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 274
 275     step("Configure DR priority 0 on R0 & Clear ospf nbrs on all the routers")
 276
 277     input_dict = {
 278         "r0": {
 279             "links": {
 280                 "s1": {
 281                     "interface": topo["routers"]["r0"]["links"]["s1"]["interface"],
 282                     "ospf": {"priority": 0},
 283                 }
 284             }
 285         }
 286     }
 287
 288     result = create_interfaces_cfg(tgen, input_dict)
 289     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 290
 291     step("Clear ospf neighbours in all routers")
 292     for rtr in ["r1"]:
 293         clear_ospf(tgen, rtr)
 294
 295     step("Verify that DR election is triggered and R0 is elected as DRother")
 296     input_dict = {
 297         "r0": {
 298             "ospf": {
 299                 "neighbors": {
 300                     "r1": {"state": "Full", "role": "DR"},
 301                     "r2": {"state": "2-Way", "role": "DROther"},
 302                     "r3": {"state": "2-Way", "role": "DROther"},
 303                 }
 304             }
 305         }
 306     }
 307     dut = "r0"
 308     result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
 309     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 310
 311     step(
 312         "Configure DR priority to default on R0 and Clear ospf neighbors"
 313         " on all the routers"
 314     )
 315
 316     input_dict = {
 317         "r0": {
 318             "links": {
 319                 "s1": {
 320                     "interface": topo["routers"]["r0"]["links"]["s1"]["interface"],
 321                     "ospf": {"priority": 100},
 322                 }
 323             }
 324         }
 325     }
 326
 327     result = create_interfaces_cfg(tgen, input_dict)
 328     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 329
 330     step("Clear ospf neighbours in all routers")
 331     for rtr in ["r0", "r1"]:
 332         clear_ospf(tgen, rtr)
 333
 334     step("Verify that DR election is triggered and R0 is elected as DR")
 335     input_dict = {
 336         "r0": {
 337             "ospf": {
 338                 "neighbors": {
 339                     "r1": {"state": "Full", "role": "Backup"},
 340                     "r2": {"state": "Full", "role": "DROther"},
 341                     "r3": {"state": "Full", "role": "DROther"},
 342                 }
 343             }
 344         }
 345     }
 346     dut = "r0"
 347     result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
 348     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 349
 350     step("Shut interface on R0")
 351     dut = "r0"
 352     intf = topo["routers"]["r0"]["links"]["s1"]["interface"]
 353     shutdown_bringup_interface(tgen, dut, intf, False)
 354
 355     result = verify_ospf_neighbor(tgen, topo, dut, lan=True, expected=False)
 356     assert (
 357         result is not True
 358     ), "Testcase {} : Failed \n " "r0: OSPF neighbors-hip is up \n Error: {}".format(
 359         tc_name, result
 360     )
 361
 362     step("No Shut interface on R0")
 363     dut = "r0"
 364     intf = topo["routers"]["r0"]["links"]["s1"]["interface"]
 365     shutdown_bringup_interface(tgen, dut, intf, True)
 366
 367     input_dict = {
 368         "r0": {
 369             "ospf": {
 370                 "neighbors": {
 371                     "r1": {"state": "Full", "role": "DR"},
 372                     "r2": {"state": "Full", "role": "DROther"},
 373                     "r3": {"state": "Full", "role": "DROther"},
 374                 }
 375             }
 376         }
 377     }
 378     step("Verify that after no shut ospf neighbours are full on R0.")
 379     result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
 380     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 381
 382     step("Clear ospf on DR router in the topology.")
 383     clear_ospf(tgen, "r0")
 384
 385     step("Verify that BDR is getting promoted to DR after clear.")
 386     step("Verify that all the nbrs are in FULL state with the elected DR.")
 387     result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
 388     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 389
 390     step("Change the ip on LAN intf on R0 to other ip from the same subnet.")
 391     topo_modify_change_ip = deepcopy(topo)
 392     intf_ip = topo_modify_change_ip["routers"]["r0"]["links"]["s1"]["ipv4"]
 393     topo_modify_change_ip["routers"]["r0"]["links"]["s1"]["ipv4"] = str(
 394         IPv4Address(frr_unicode(intf_ip.split("/")[0])) + 4
 395     ) + "/{}".format(intf_ip.split("/")[1])
 396
 397     build_config_from_json(tgen, topo_modify_change_ip, save_bkup=False)
 398
 399     clear_ospf(tgen, "r0")
 400     step(
 401         "Verify that OSPF is in FULL state with other routers with "
 402         "newly configured IP."
 403     )
 404     result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
 405     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 406
 407     step("Change the ospf router id on the R0 and clear ip ospf interface.")
 408     change_rid = {"r0": {"ospf": {"router_id": "100.1.1.100"}}}
 409
 410     result = create_router_ospf(tgen, topo, change_rid)
 411     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 412     topo["routers"]["r0"]["ospf"]["router_id"] = "100.1.1.100"
 413     step("Reload the FRR router")
 414
 415     stop_router(tgen, "r0")
 416     start_router(tgen, "r0")
 417
 418     step(
 419         "Verify that OSPF is in FULL state with other routers with"
 420         " newly configured router id."
 421     )
 422     input_dict = {
 423         "r1": {
 424             "ospf": {
 425                 "neighbors": {
 426                     "r0": {"state": "Full", "role": "Backup"},
 427                     "r2": {"state": "Full", "role": "DROther"},
 428                     "r3": {"state": "Full", "role": "DROther"},
 429                 }
 430             }
 431         }
 432     }
 433     dut = "r1"
 434     result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
 435     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 436
 437     step("Reconfigure the original router id and clear ip ospf interface.")
 438     change_rid = {"r0": {"ospf": {"router_id": "100.1.1.0"}}}
 439     result = create_router_ospf(tgen, topo, change_rid)
 440     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 441     topo["routers"]["r0"]["ospf"]["router_id"] = "100.1.1.0"
 442     step("Reload the FRR router")
 443     # stop/start -> restart FRR router and verify
 444     stop_router(tgen, "r0")
 445     start_router(tgen, "r0")
 446
 447     step("Verify that OSPF is enabled with router id previously configured.")
 448     input_dict = {
 449         "r1": {
 450             "ospf": {
 451                 "neighbors": {
 452                     "r0": {"state": "Full", "role": "Backup"},
 453                     "r2": {"state": "Full", "role": "DROther"},
 454                     "r3": {"state": "Full", "role": "DROther"},
 455                 }
 456             }
 457         }
 458     }
 459     dut = "r1"
 460     result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
 461     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 462
 463     write_test_footer(tc_name)
 464
 465
 466 def test_ospf_lan_tc2_p0(request):
 467     """
 468     OSPF IFSM -Verify state change events on DR / BDR / DR Other
 469
 470     """
 471     tc_name = request.node.name
 472     write_test_header(tc_name)
 473     tgen = get_topogen()
 474
 475     # Don't run this test if we have any failure.
 476     if tgen.routers_have_failure():
 477         pytest.skip(tgen.errors)
 478
 479     global topo
 480     step("Bring up the base config as per the topology")
 481     reset_config_on_routers(tgen)
 482     step(
 483         "Verify that OSPF is subscribed to multi cast services "
 484         "(All SPF, all DR Routers)."
 485     )
 486     step("Verify that interface is enabled in ospf.")
 487     dut = "r0"
 488     input_dict = {
 489         "r0": {
 490             "links": {
 491                 "s1": {
 492                     "ospf": {
 493                         "priority": 98,
 494                         "timerDeadSecs": 4,
 495                         "area": "0.0.0.3",
 496                         "mcastMemberOspfDesignatedRouters": True,
 497                         "mcastMemberOspfAllRouters": True,
 498                         "ospfEnabled": True,
 499                     }
 500                 }
 501             }
 502         }
 503     }
 504     result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
 505     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 506
 507     step("Delete the ip address")
 508     topo1 = {
 509         "r0": {
 510             "links": {
 511                 "r3": {
 512                     "ipv4": topo["routers"]["r0"]["links"]["s1"]["ipv4"],
 513                     "interface": topo["routers"]["r0"]["links"]["s1"]["interface"],
 514                     "delete": True,
 515                 }
 516             }
 517         }
 518     }
 519
 520     result = create_interfaces_cfg(tgen, topo1)
 521     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 522
 523     step("Change the ip on the R0 interface")
 524
 525     topo_modify_change_ip = deepcopy(topo)
 526     intf_ip = topo_modify_change_ip["routers"]["r0"]["links"]["s1"]["ipv4"]
 527     topo_modify_change_ip["routers"]["r0"]["links"]["s1"]["ipv4"] = str(
 528         IPv4Address(frr_unicode(intf_ip.split("/")[0])) + 3
 529     ) + "/{}".format(intf_ip.split("/")[1])
 530
 531     build_config_from_json(tgen, topo_modify_change_ip, save_bkup=False)
 532     step("Verify that interface is enabled in ospf.")
 533     dut = "r0"
 534     input_dict = {
 535         "r0": {
 536             "links": {
 537                 "s1": {
 538                     "ospf": {
 539                         "ipAddress": topo_modify_change_ip["routers"]["r0"]["links"][
 540                             "s1"
 541                         ]["ipv4"].split("/")[0],
 542                         "ipAddressPrefixlen": int(
 543                             topo_modify_change_ip["routers"]["r0"]["links"]["s1"][
 544                                 "ipv4"
 545                             ].split("/")[1]
 546                         ),
 547                     }
 548                 }
 549             }
 550         }
 551     }
 552     result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
 553     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 554
 555     step("Modify the mask on the R0 interface")
 556     ip_addr = topo_modify_change_ip["routers"]["r0"]["links"]["s1"]["ipv4"]
 557     mask = topo_modify_change_ip["routers"]["r0"]["links"]["s1"]["ipv4"]
 558     step("Delete the ip address")
 559     topo1 = {
 560         "r0": {
 561             "links": {
 562                 "r3": {
 563                     "ipv4": ip_addr,
 564                     "interface": topo["routers"]["r0"]["links"]["s1"]["interface"],
 565                     "delete": True,
 566                 }
 567             }
 568         }
 569     }
 570
 571     result = create_interfaces_cfg(tgen, topo1)
 572     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 573
 574     step("Change the ip on the R0 interface")
 575
 576     topo_modify_change_ip = deepcopy(topo)
 577     intf_ip = topo_modify_change_ip["routers"]["r0"]["links"]["s1"]["ipv4"]
 578     topo_modify_change_ip["routers"]["r0"]["links"]["s1"]["ipv4"] = str(
 579         IPv4Address(frr_unicode(intf_ip.split("/")[0])) + 3
 580     ) + "/{}".format(int(intf_ip.split("/")[1]) + 1)
 581
 582     build_config_from_json(tgen, topo_modify_change_ip, save_bkup=False)
 583     step("Verify that interface is enabled in ospf.")
 584     dut = "r0"
 585     input_dict = {
 586         "r0": {
 587             "links": {
 588                 "s1": {
 589                     "ospf": {
 590                         "ipAddress": topo_modify_change_ip["routers"]["r0"]["links"][
 591                             "s1"
 592                         ]["ipv4"].split("/")[0],
 593                         "ipAddressPrefixlen": int(
 594                             topo_modify_change_ip["routers"]["r0"]["links"]["s1"][
 595                                 "ipv4"
 596                             ].split("/")[1]
 597                         ),
 598                     }
 599                 }
 600             }
 601         }
 602     }
 603     result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
 604     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 605
 606     step("Change the area id on the interface")
 607     input_dict = {
 608         "r0": {
 609             "links": {
 610                 "s1": {
 611                     "interface": topo["routers"]["r0"]["links"]["s1"]["interface"],
 612                     "ospf": {"area": "0.0.0.3"},
 613                     "delete": True,
 614                 }
 615             }
 616         }
 617     }
 618
 619     result = create_interfaces_cfg(tgen, input_dict)
 620     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 621
 622     input_dict = {
 623         "r0": {
 624             "links": {
 625                 "s1": {
 626                     "interface": topo["routers"]["r0"]["links"]["s1"]["interface"],
 627                     "ospf": {"area": "0.0.0.2"},
 628                 }
 629             }
 630         }
 631     }
 632
 633     result = create_interfaces_cfg(tgen, input_dict)
 634     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 635     step("Verify that interface is enabled in ospf.")
 636     dut = "r0"
 637     input_dict = {
 638         "r0": {"links": {"s1": {"ospf": {"area": "0.0.0.2", "ospfEnabled": True}}}}
 639     }
 640     result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
 641     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 642
 643     write_test_footer(tc_name)
 644
 645
 646 if __name__ == "__main__":
 647     args = ["-s"] + sys.argv[1:]
 648     sys.exit(pytest.main(args))