tests/topotests/ospf_topo1/test_ospf_topo1.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_ospf_topo1.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2017 by
   8 # Network Device Education Foundation, Inc. ("NetDEF")
   9 #
  10 # Permission to use, copy, modify, and/or distribute this software
  11 # for any purpose with or without fee is hereby granted, provided
  12 # that the above copyright notice and this permission notice appear
  13 # in all copies.
  14 #
  15 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  16 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  17 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  18 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  19 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  20 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  21 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  22 # OF THIS SOFTWARE.
  23 #
  24
  25 """
  26 test_ospf_topo1.py: Test the FRR OSPF routing daemon.
  27 """
  28
  29 import os
  30 import re
  31 import sys
  32 from functools import partial
  33 import pytest
  34
  35 # Save the Current Working Directory to find configuration files.
  36 CWD = os.path.dirname(os.path.realpath(__file__))
  37 sys.path.append(os.path.join(CWD, "../"))
  38
  39 # pylint: disable=C0413
  40 # Import topogen and topotest helpers
  41 from lib import topotest
  42 from lib.topogen import Topogen, TopoRouter, get_topogen
  43 from lib.topolog import logger
  44
  45 # Required to instantiate the topology builder class.
  46
  47 pytestmark = [pytest.mark.ospfd]
  48
  49
  50 def build_topo(tgen):
  51     "Build function"
  52
  53     # Create 4 routers
  54     for routern in range(1, 5):
  55         tgen.add_router("r{}".format(routern))
  56
  57     # Create a empty network for router 1
  58     switch = tgen.add_switch("s1")
  59     switch.add_link(tgen.gears["r1"])
  60
  61     # Create a empty network for router 2
  62     switch = tgen.add_switch("s2")
  63     switch.add_link(tgen.gears["r2"])
  64
  65     # Interconect router 1, 2 and 3
  66     switch = tgen.add_switch("s3")
  67     switch.add_link(tgen.gears["r1"])
  68     switch.add_link(tgen.gears["r2"])
  69     switch.add_link(tgen.gears["r3"])
  70
  71     # Create empty netowrk for router3
  72     switch = tgen.add_switch("s4")
  73     switch.add_link(tgen.gears["r3"])
  74
  75     # Interconect router 3 and 4
  76     switch = tgen.add_switch("s5")
  77     switch.add_link(tgen.gears["r3"])
  78     switch.add_link(tgen.gears["r4"])
  79
  80     # Create a empty network for router 4
  81     switch = tgen.add_switch("s6")
  82     switch.add_link(tgen.gears["r4"])
  83
  84
  85 def setup_module(mod):
  86     "Sets up the pytest environment"
  87     tgen = Topogen(build_topo, mod.__name__)
  88     tgen.start_topology()
  89
  90     ospf6_config = "ospf6d.conf"
  91
  92     router_list = tgen.routers()
  93     for rname, router in router_list.items():
  94         router.load_config(
  95             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  96         )
  97         router.load_config(
  98             TopoRouter.RD_OSPF, os.path.join(CWD, "{}/ospfd.conf".format(rname))
  99         )
 100         router.load_config(
 101             TopoRouter.RD_OSPF6, os.path.join(CWD, "{}/{}".format(rname, ospf6_config))
 102         )
 103
 104     # Initialize all routers.
 105     tgen.start_router()
 106
 107
 108 def teardown_module(mod):
 109     "Teardown the pytest environment"
 110     tgen = get_topogen()
 111     tgen.stop_topology()
 112
 113
 114 def test_wait_protocol_convergence():
 115     "Wait for OSPFv2/OSPFv3 to converge"
 116     tgen = get_topogen()
 117     if tgen.routers_have_failure():
 118         pytest.skip(tgen.errors)
 119
 120     logger.info("waiting for protocols to converge")
 121
 122     def expect_ospfv2_neighbor_full(router, neighbor):
 123         "Wait until OSPFv2 convergence."
 124         logger.info("waiting OSPFv2 router '{}'".format(router))
 125
 126         def run_command_and_expect():
 127             """
 128             Function that runs command and expect the following outcomes:
 129              * Full/DR
 130              * Full/DROther
 131              * Full/Backup
 132             """
 133             result = tgen.gears[router].vtysh_cmd(
 134                 "show ip ospf neighbor json", isjson=True
 135             )
 136             if (
 137                 topotest.json_cmp(
 138                     result, {"neighbors": {neighbor: [{"converged": "Full"}]}}
 139                 )
 140                 is None
 141             ):
 142                 return None
 143
 144             if (
 145                 topotest.json_cmp(
 146                     result, {"neighbors": {neighbor: [{"converged": "Full"}]}}
 147                 )
 148                 is None
 149             ):
 150                 return None
 151
 152             return topotest.json_cmp(
 153                 result, {"neighbors": {neighbor: [{"converged": "Full"}]}}
 154             )
 155
 156         _, result = topotest.run_and_expect(
 157             run_command_and_expect, None, count=130, wait=1
 158         )
 159         assertmsg = '"{}" convergence failure'.format(router)
 160         assert result is None, assertmsg
 161
 162     def expect_ospfv3_neighbor_full(router, neighbor):
 163         "Wait until OSPFv3 convergence."
 164         logger.info("waiting OSPFv3 router '{}'".format(router))
 165         test_func = partial(
 166             topotest.router_json_cmp,
 167             tgen.gears[router],
 168             "show ipv6 ospf6 neighbor json",
 169             {"neighbors": [{"neighborId": neighbor, "state": "Full"}]},
 170         )
 171         _, result = topotest.run_and_expect(test_func, None, count=130, wait=1)
 172         assertmsg = '"{}" convergence failure'.format(router)
 173         assert result is None, assertmsg
 174
 175     # Wait for OSPFv2 convergence
 176     expect_ospfv2_neighbor_full("r1", "10.0.255.2")
 177     expect_ospfv2_neighbor_full("r1", "10.0.255.3")
 178     expect_ospfv2_neighbor_full("r2", "10.0.255.1")
 179     expect_ospfv2_neighbor_full("r2", "10.0.255.3")
 180     expect_ospfv2_neighbor_full("r3", "10.0.255.1")
 181     expect_ospfv2_neighbor_full("r3", "10.0.255.2")
 182     expect_ospfv2_neighbor_full("r3", "10.0.255.4")
 183     expect_ospfv2_neighbor_full("r4", "10.0.255.3")
 184
 185     # Wait for OSPFv3 convergence
 186     expect_ospfv3_neighbor_full("r1", "10.0.255.2")
 187     expect_ospfv3_neighbor_full("r1", "10.0.255.3")
 188     expect_ospfv3_neighbor_full("r2", "10.0.255.1")
 189     expect_ospfv3_neighbor_full("r2", "10.0.255.3")
 190     expect_ospfv3_neighbor_full("r3", "10.0.255.1")
 191     expect_ospfv3_neighbor_full("r3", "10.0.255.2")
 192     expect_ospfv3_neighbor_full("r3", "10.0.255.4")
 193     expect_ospfv3_neighbor_full("r4", "10.0.255.3")
 194
 195
 196 def compare_show_ipv6_ospf6(rname, expected):
 197     """
 198     Calls 'show ipv6 ospf6 route' for router `rname` and compare the obtained
 199     result with the expected output.
 200     """
 201     tgen = get_topogen()
 202     current = tgen.gears[rname].vtysh_cmd("show ipv6 ospf6 route")
 203
 204     # Remove the link addresses
 205     current = re.sub(r"fe80::[^ ]+", "fe80::xxxx:xxxx:xxxx:xxxx", current)
 206     expected = re.sub(r"fe80::[^ ]+", "fe80::xxxx:xxxx:xxxx:xxxx", expected)
 207
 208     # Remove the time
 209     current = re.sub(r"\d+:\d{2}:\d{2}", "", current)
 210     expected = re.sub(r"\d+:\d{2}:\d{2}", "", expected)
 211
 212     return topotest.difflines(
 213         topotest.normalize_text(current),
 214         topotest.normalize_text(expected),
 215         title1="Current output",
 216         title2="Expected output",
 217     )
 218
 219
 220 def test_ospf_convergence():
 221     "Test OSPF daemon convergence"
 222     tgen = get_topogen()
 223     if tgen.routers_have_failure():
 224         pytest.skip("skipped because of router(s) failure")
 225
 226     for router, rnode in tgen.routers().items():
 227         logger.info('Waiting for router "%s" convergence', router)
 228
 229         # Load expected results from the command
 230         reffile = os.path.join(CWD, "{}/ospfroute.txt".format(router))
 231         expected = open(reffile).read()
 232
 233         # Run test function until we get an result. Wait at most 80 seconds.
 234         test_func = partial(
 235             topotest.router_output_cmp, rnode, "show ip ospf route", expected
 236         )
 237         result, diff = topotest.run_and_expect(test_func, "", count=160, wait=0.5)
 238         assert result, "OSPF did not converge on {}:\n{}".format(router, diff)
 239
 240
 241 def test_ospf_kernel_route():
 242     "Test OSPF kernel route installation"
 243     tgen = get_topogen()
 244     if tgen.routers_have_failure():
 245         pytest.skip("skipped because of router(s) failure")
 246
 247     rlist = tgen.routers().values()
 248     for router in rlist:
 249         logger.info('Checking OSPF IPv4 kernel routes in "%s"', router.name)
 250
 251         routes = topotest.ip4_route(router)
 252         expected = {
 253             "10.0.1.0/24": {},
 254             "10.0.2.0/24": {},
 255             "10.0.3.0/24": {},
 256             "10.0.10.0/24": {},
 257             "172.16.0.0/24": {},
 258             "172.16.1.0/24": {},
 259         }
 260         assertmsg = 'OSPF IPv4 route mismatch in router "{}"'.format(router.name)
 261         assert topotest.json_cmp(routes, expected) is None, assertmsg
 262
 263
 264 def test_ospf6_convergence():
 265     "Test OSPF6 daemon convergence"
 266     tgen = get_topogen()
 267     if tgen.routers_have_failure():
 268         pytest.skip("skipped because of router(s) failure")
 269
 270     ospf6route_file = "{}/ospf6route_ecmp.txt"
 271     for rnum in range(1, 5):
 272         router = "r{}".format(rnum)
 273
 274         logger.info('Waiting for router "%s" IPv6 OSPF convergence', router)
 275
 276         # Load expected results from the command
 277         reffile = os.path.join(CWD, ospf6route_file.format(router))
 278         expected = open(reffile).read()
 279
 280         # Run test function until we get an result. Wait at most 60 seconds.
 281         test_func = partial(compare_show_ipv6_ospf6, router, expected)
 282         result, diff = topotest.run_and_expect(test_func, "", count=25, wait=3)
 283         if (not result) and (rnum == 1):
 284             # Didn't match the new ECMP version - try the old pre-ECMP format
 285             ospf6route_file = "{}/ospf6route.txt"
 286
 287             # Load expected results from the command
 288             reffile = os.path.join(CWD, ospf6route_file.format(router))
 289             expected = open(reffile).read()
 290
 291             test_func = partial(compare_show_ipv6_ospf6, router, expected)
 292             result, diff = topotest.run_and_expect(test_func, "", count=1, wait=3)
 293             if not result:
 294                 # Didn't match the old version - switch back to new ECMP version
 295                 # and fail
 296                 ospf6route_file = "{}/ospf6route_ecmp.txt"
 297
 298                 # Load expected results from the command
 299                 reffile = os.path.join(CWD, ospf6route_file.format(router))
 300                 expected = open(reffile).read()
 301
 302                 test_func = partial(compare_show_ipv6_ospf6, router, expected)
 303                 result, diff = topotest.run_and_expect(test_func, "", count=1, wait=3)
 304
 305         assert result, "OSPF6 did not converge on {}:\n{}".format(router, diff)
 306
 307
 308 def test_ospf6_kernel_route():
 309     "Test OSPF kernel route installation"
 310     tgen = get_topogen()
 311     if tgen.routers_have_failure():
 312         pytest.skip("skipped because of router(s) failure")
 313
 314     rlist = tgen.routers().values()
 315     for router in rlist:
 316         logger.info('Checking OSPF IPv6 kernel routes in "%s"', router.name)
 317
 318         routes = topotest.ip6_route(router)
 319         expected = {
 320             "2001:db8:1::/64": {},
 321             "2001:db8:2::/64": {},
 322             "2001:db8:3::/64": {},
 323             "2001:db8:100::/64": {},
 324             "2001:db8:200::/64": {},
 325             "2001:db8:300::/64": {},
 326         }
 327         assertmsg = 'OSPF IPv6 route mismatch in router "{}"'.format(router.name)
 328         assert topotest.json_cmp(routes, expected) is None, assertmsg
 329
 330
 331 def test_ospf_json():
 332     "Test 'show ip ospf json' output for coherency."
 333     tgen = get_topogen()
 334     if tgen.routers_have_failure():
 335         pytest.skip("skipped because of router(s) failure")
 336
 337     for rnum in range(1, 5):
 338         router = tgen.gears["r{}".format(rnum)]
 339         logger.info('Comparing router "%s" "show ip ospf json" output', router.name)
 340         expected = {
 341             "routerId": "10.0.255.{}".format(rnum),
 342             "tosRoutesOnly": True,
 343             "rfc2328Conform": True,
 344             "spfScheduleDelayMsecs": 0,
 345             "holdtimeMinMsecs": 50,
 346             "holdtimeMaxMsecs": 5000,
 347             "lsaMinIntervalMsecs": 5000,
 348             "lsaMinArrivalMsecs": 1000,
 349             "writeMultiplier": 20,
 350             "refreshTimerMsecs": 10000,
 351             "asbrRouter": "injectingExternalRoutingInformation",
 352             "attachedAreaCounter": 1,
 353             "areas": {},
 354         }
 355         # Area specific additional checks
 356         if router.name == "r1" or router.name == "r2" or router.name == "r3":
 357             expected["areas"]["0.0.0.0"] = {
 358                 "areaIfActiveCounter": 2,
 359                 "areaIfTotalCounter": 2,
 360                 "authentication": "authenticationNone",
 361                 "backbone": True,
 362                 "lsaAsbrNumber": 1,
 363                 "lsaNetworkNumber": 1,
 364                 "lsaNssaNumber": 0,
 365                 "lsaNumber": 7,
 366                 "lsaOpaqueAreaNumber": 0,
 367                 "lsaOpaqueLinkNumber": 0,
 368                 "lsaRouterNumber": 3,
 369                 "lsaSummaryNumber": 2,
 370                 "nbrFullAdjacentCounter": 2,
 371             }
 372         if router.name == "r3" or router.name == "r4":
 373             expected["areas"]["0.0.0.1"] = {
 374                 "areaIfActiveCounter": 1,
 375                 "areaIfTotalCounter": 1,
 376                 "authentication": "authenticationNone",
 377                 "lsaAsbrNumber": 2,
 378                 "lsaNetworkNumber": 1,
 379                 "lsaNssaNumber": 0,
 380                 "lsaNumber": 9,
 381                 "lsaOpaqueAreaNumber": 0,
 382                 "lsaOpaqueLinkNumber": 0,
 383                 "lsaRouterNumber": 2,
 384                 "lsaSummaryNumber": 4,
 385                 "nbrFullAdjacentCounter": 1,
 386             }
 387             # r4 has more interfaces for area 0.0.0.1
 388             if router.name == "r4":
 389                 expected["areas"]["0.0.0.1"].update(
 390                     {
 391                         "areaIfActiveCounter": 2,
 392                         "areaIfTotalCounter": 2,
 393                     }
 394                 )
 395
 396         # router 3 has an additional area
 397         if router.name == "r3":
 398             expected["attachedAreaCounter"] = 2
 399
 400         output = router.vtysh_cmd("show ip ospf json", isjson=True)
 401         result = topotest.json_cmp(output, expected)
 402         assert result is None, '"{}" JSON output mismatches the expected result'.format(
 403             router.name
 404         )
 405
 406
 407 def test_ospf_link_down():
 408     "Test OSPF convergence after a link goes down"
 409     tgen = get_topogen()
 410     if tgen.routers_have_failure():
 411         pytest.skip("skipped because of router(s) failure")
 412
 413     # Simulate a network down event on router3 switch3 interface.
 414     router3 = tgen.gears["r3"]
 415     router3.peer_link_enable("r3-eth0", False)
 416
 417     # Expect convergence on all routers
 418     for router, rnode in tgen.routers().items():
 419         logger.info('Waiting for router "%s" convergence after link failure', router)
 420         # Load expected results from the command
 421         reffile = os.path.join(CWD, "{}/ospfroute_down.txt".format(router))
 422         expected = open(reffile).read()
 423
 424         # Run test function until we get an result. Wait at most 80 seconds.
 425         test_func = partial(
 426             topotest.router_output_cmp, rnode, "show ip ospf route", expected
 427         )
 428         result, diff = topotest.run_and_expect(test_func, "", count=140, wait=0.5)
 429         assert result, "OSPF did not converge on {}:\n{}".format(router, diff)
 430
 431
 432 def test_ospf_link_down_kernel_route():
 433     "Test OSPF kernel route installation"
 434     tgen = get_topogen()
 435     if tgen.routers_have_failure():
 436         pytest.skip("skipped because of router(s) failure")
 437
 438     rlist = tgen.routers().values()
 439     for router in rlist:
 440         logger.info(
 441             'Checking OSPF IPv4 kernel routes in "%s" after link down', router.name
 442         )
 443
 444         routes = topotest.ip4_route(router)
 445         expected = {
 446             "10.0.1.0/24": {},
 447             "10.0.2.0/24": {},
 448             "10.0.3.0/24": {},
 449             "10.0.10.0/24": {},
 450             "172.16.0.0/24": {},
 451             "172.16.1.0/24": {},
 452         }
 453         if router.name == "r1" or router.name == "r2":
 454             expected.update(
 455                 {
 456                     "10.0.10.0/24": None,
 457                     "172.16.0.0/24": None,
 458                     "172.16.1.0/24": None,
 459                 }
 460             )
 461         elif router.name == "r3" or router.name == "r4":
 462             expected.update(
 463                 {
 464                     "10.0.1.0/24": None,
 465                     "10.0.2.0/24": None,
 466                 }
 467             )
 468         # Route '10.0.3.0' is no longer available for r4 since it is down.
 469         if router.name == "r4":
 470             expected.update(
 471                 {
 472                     "10.0.3.0/24": None,
 473                 }
 474             )
 475         assertmsg = 'OSPF IPv4 route mismatch in router "{}" after link down'.format(
 476             router.name
 477         )
 478         assert topotest.json_cmp(routes, expected) is None, assertmsg
 479
 480
 481 def test_ospf6_link_down():
 482     "Test OSPF6 daemon convergence after link goes down"
 483     tgen = get_topogen()
 484     if tgen.routers_have_failure():
 485         pytest.skip("skipped because of router(s) failure")
 486
 487     for rnum in range(1, 5):
 488         router = "r{}".format(rnum)
 489
 490         logger.info(
 491             'Waiting for router "%s" IPv6 OSPF convergence after link down', router
 492         )
 493
 494         # Load expected results from the command
 495         reffile = os.path.join(CWD, "{}/ospf6route_down.txt".format(router))
 496         expected = open(reffile).read()
 497
 498         # Run test function until we get an result. Wait at most 60 seconds.
 499         test_func = partial(compare_show_ipv6_ospf6, router, expected)
 500         result, diff = topotest.run_and_expect(test_func, "", count=25, wait=3)
 501         assert result, "OSPF6 did not converge on {}:\n{}".format(router, diff)
 502
 503
 504 def test_ospf6_link_down_kernel_route():
 505     "Test OSPF kernel route installation"
 506     tgen = get_topogen()
 507     if tgen.routers_have_failure():
 508         pytest.skip("skipped because of router(s) failure")
 509
 510     rlist = tgen.routers().values()
 511     for router in rlist:
 512         logger.info(
 513             'Checking OSPF IPv6 kernel routes in "%s" after link down', router.name
 514         )
 515
 516         routes = topotest.ip6_route(router)
 517         expected = {
 518             "2001:db8:1::/64": {},
 519             "2001:db8:2::/64": {},
 520             "2001:db8:3::/64": {},
 521             "2001:db8:100::/64": {},
 522             "2001:db8:200::/64": {},
 523             "2001:db8:300::/64": {},
 524         }
 525         if router.name == "r1" or router.name == "r2":
 526             expected.update(
 527                 {
 528                     "2001:db8:100::/64": None,
 529                     "2001:db8:200::/64": None,
 530                     "2001:db8:300::/64": None,
 531                 }
 532             )
 533         elif router.name == "r3" or router.name == "r4":
 534             expected.update(
 535                 {
 536                     "2001:db8:1::/64": None,
 537                     "2001:db8:2::/64": None,
 538                 }
 539             )
 540         # Route '2001:db8:3::/64' is no longer available for r4 since it is down.
 541         if router.name == "r4":
 542             expected.update(
 543                 {
 544                     "2001:db8:3::/64": None,
 545                 }
 546             )
 547         assertmsg = 'OSPF IPv6 route mismatch in router "{}" after link down'.format(
 548             router.name
 549         )
 550         assert topotest.json_cmp(routes, expected) is None, assertmsg
 551
 552
 553 def test_memory_leak():
 554     "Run the memory leak test and report results."
 555     tgen = get_topogen()
 556     if not tgen.is_memleak_enabled():
 557         pytest.skip("Memory leak test/report is disabled")
 558
 559     tgen.report_memory_leaks()
 560
 561
 562 if __name__ == "__main__":
 563     args = ["-s"] + sys.argv[1:]
 564     sys.exit(pytest.main(args))