tests/topotests/ripng_topo1/test_ripng_topo1.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_ripng_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_ripng_topo1.py: Test of RIPng Topology
  27
  28 """
  29
  30 import os
  31 import re
  32 import sys
  33 import pytest
  34 import unicodedata
  35 from time import sleep
  36
  37 from mininet.topo import Topo
  38 from mininet.net import Mininet
  39 from mininet.node import Node, OVSSwitch, Host
  40 from mininet.log import setLogLevel, info
  41 from mininet.cli import CLI
  42 from mininet.link import Intf
  43
  44 from functools import partial
  45
  46 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
  47 from lib import topotest
  48
  49 fatal_error = ""
  50
  51 pytestmark = [pytest.mark.ripd]
  52
  53 #####################################################
  54 ##
  55 ##   Network Topology Definition
  56 ##
  57 #####################################################
  58
  59
  60 class NetworkTopo(Topo):
  61     "RIPng Topology 1"
  62
  63     def build(self, **_opts):
  64
  65         # Setup Routers
  66         router = {}
  67         #
  68         # Setup Main Router
  69         router[1] = topotest.addRouter(self, "r1")
  70         #
  71         # Setup RIPng Routers
  72         for i in range(2, 4):
  73             router[i] = topotest.addRouter(self, "r%s" % i)
  74
  75         # Setup Switches
  76         switch = {}
  77         #
  78         # On main router
  79         # First switch is for a dummy interface (for local network)
  80         switch[1] = self.addSwitch("sw1", cls=topotest.LegacySwitch)
  81         self.addLink(switch[1], router[1], intfName2="r1-eth0")
  82         #
  83         # Switches for RIPng
  84         # switch 2 switch is for connection to RIP router
  85         switch[2] = self.addSwitch("sw2", cls=topotest.LegacySwitch)
  86         self.addLink(switch[2], router[1], intfName2="r1-eth1")
  87         self.addLink(switch[2], router[2], intfName2="r2-eth0")
  88         # switch 3 is between RIP routers
  89         switch[3] = self.addSwitch("sw3", cls=topotest.LegacySwitch)
  90         self.addLink(switch[3], router[2], intfName2="r2-eth1")
  91         self.addLink(switch[3], router[3], intfName2="r3-eth1")
  92         # switch 4 is stub on remote RIP router
  93         switch[4] = self.addSwitch("sw4", cls=topotest.LegacySwitch)
  94         self.addLink(switch[4], router[3], intfName2="r3-eth0")
  95
  96         switch[5] = self.addSwitch("sw5", cls=topotest.LegacySwitch)
  97         self.addLink(switch[5], router[1], intfName2="r1-eth2")
  98         switch[6] = self.addSwitch("sw6", cls=topotest.LegacySwitch)
  99         self.addLink(switch[6], router[1], intfName2="r1-eth3")
 100
 101
 102 #####################################################
 103 ##
 104 ##   Tests starting
 105 ##
 106 #####################################################
 107
 108
 109 def setup_module(module):
 110     global topo, net
 111
 112     print("\n\n** %s: Setup Topology" % module.__name__)
 113     print("******************************************\n")
 114
 115     print("Cleanup old Mininet runs")
 116     os.system("sudo mn -c > /dev/null 2>&1")
 117
 118     thisDir = os.path.dirname(os.path.realpath(__file__))
 119     topo = NetworkTopo()
 120
 121     net = Mininet(controller=None, topo=topo)
 122     net.start()
 123
 124     # Starting Routers
 125     #
 126     for i in range(1, 4):
 127         net["r%s" % i].loadConf("zebra", "%s/r%s/zebra.conf" % (thisDir, i))
 128         net["r%s" % i].loadConf("ripngd", "%s/r%s/ripngd.conf" % (thisDir, i))
 129         net["r%s" % i].startRouter()
 130
 131     # For debugging after starting FRR daemons, uncomment the next line
 132     # CLI(net)
 133
 134
 135 def teardown_module(module):
 136     global net
 137
 138     print("\n\n** %s: Shutdown Topology" % module.__name__)
 139     print("******************************************\n")
 140
 141     # End - Shutdown network
 142     net.stop()
 143
 144
 145 def test_router_running():
 146     global fatal_error
 147     global net
 148
 149     # Skip if previous fatal error condition is raised
 150     if fatal_error != "":
 151         pytest.skip(fatal_error)
 152
 153     print("\n\n** Check if FRR is running on each Router node")
 154     print("******************************************\n")
 155
 156     # Starting Routers
 157     for i in range(1, 4):
 158         fatal_error = net["r%s" % i].checkRouterRunning()
 159         assert fatal_error == "", fatal_error
 160
 161     # For debugging after starting FRR daemons, uncomment the next line
 162     # CLI(net)
 163
 164
 165 def test_converge_protocols():
 166     global fatal_error
 167     global net
 168
 169     # Skip if previous fatal error condition is raised
 170     if fatal_error != "":
 171         pytest.skip(fatal_error)
 172
 173     thisDir = os.path.dirname(os.path.realpath(__file__))
 174
 175     print("\n\n** Waiting for protocols convergence")
 176     print("******************************************\n")
 177
 178     # Not really implemented yet - just sleep 11 secs for now
 179     sleep(11)
 180
 181     # Make sure that all daemons are running
 182     for i in range(1, 4):
 183         fatal_error = net["r%s" % i].checkRouterRunning()
 184         assert fatal_error == "", fatal_error
 185
 186     # For debugging after starting FRR daemons, uncomment the next line
 187     # CLI(net)
 188
 189
 190 def test_ripng_status():
 191     global fatal_error
 192     global net
 193
 194     # Skip if previous fatal error condition is raised
 195     if fatal_error != "":
 196         pytest.skip(fatal_error)
 197
 198     thisDir = os.path.dirname(os.path.realpath(__file__))
 199
 200     # Verify RIP Status
 201     print("\n\n** Verifying RIPng status")
 202     print("******************************************\n")
 203     failures = 0
 204     for i in range(1, 4):
 205         refTableFile = "%s/r%s/ripng_status.ref" % (thisDir, i)
 206         if os.path.isfile(refTableFile):
 207             # Read expected result from file
 208             expected = open(refTableFile).read().rstrip()
 209             # Fix newlines (make them all the same)
 210             expected = ("\n".join(expected.splitlines()) + "\n").splitlines(1)
 211
 212             # Actual output from router
 213             actual = (
 214                 net["r%s" % i]
 215                 .cmd('vtysh -c "show ipv6 ripng status" 2> /dev/null')
 216                 .rstrip()
 217             )
 218             # Mask out Link-Local mac address portion. They are random...
 219             actual = re.sub(r" fe80::[0-9a-f:]+", " fe80::XXXX:XXXX:XXXX:XXXX", actual)
 220             # Drop time in next due
 221             actual = re.sub(r"in [0-9]+ seconds", "in XX seconds", actual)
 222             # Drop time in last update
 223             actual = re.sub(r" [0-2][0-9]:[0-5][0-9]:[0-5][0-9]", " XX:XX:XX", actual)
 224             # Fix newlines (make them all the same)
 225             actual = ("\n".join(actual.splitlines()) + "\n").splitlines(1)
 226
 227             # Generate Diff
 228             diff = topotest.get_textdiff(
 229                 actual,
 230                 expected,
 231                 title1="actual IPv6 RIPng status",
 232                 title2="expected IPv6 RIPng status",
 233             )
 234
 235             # Empty string if it matches, otherwise diff contains unified diff
 236             if diff:
 237                 sys.stderr.write(
 238                     "r%s failed IPv6 RIPng status check:\n%s\n" % (i, diff)
 239                 )
 240                 failures += 1
 241             else:
 242                 print("r%s ok" % i)
 243
 244             assert failures == 0, "IPv6 RIPng status failed for router r%s:\n%s" % (
 245                 i,
 246                 diff,
 247             )
 248
 249     # Make sure that all daemons are running
 250     for i in range(1, 4):
 251         fatal_error = net["r%s" % i].checkRouterRunning()
 252         assert fatal_error == "", fatal_error
 253
 254     # For debugging after starting FRR daemons, uncomment the next line
 255     # CLI(net)
 256
 257
 258 def test_ripng_routes():
 259     global fatal_error
 260     global net
 261
 262     # Skip if previous fatal error condition is raised
 263     if fatal_error != "":
 264         pytest.skip(fatal_error)
 265
 266     thisDir = os.path.dirname(os.path.realpath(__file__))
 267
 268     # Verify RIPng Status
 269     print("\n\n** Verifying RIPng routes")
 270     print("******************************************\n")
 271     failures = 0
 272     for i in range(1, 4):
 273         refTableFile = "%s/r%s/show_ipv6_ripng.ref" % (thisDir, i)
 274         if os.path.isfile(refTableFile):
 275             # Read expected result from file
 276             expected = open(refTableFile).read().rstrip()
 277             # Fix newlines (make them all the same)
 278             expected = ("\n".join(expected.splitlines()) + "\n").splitlines(1)
 279
 280             # Actual output from router
 281             actual = (
 282                 net["r%s" % i].cmd('vtysh -c "show ipv6 ripng" 2> /dev/null').rstrip()
 283             )
 284             # Drop Time
 285             actual = re.sub(r" [0-9][0-9]:[0-5][0-9]", " XX:XX", actual)
 286             # Mask out Link-Local mac address portion. They are random...
 287             actual = re.sub(
 288                 r" fe80::[0-9a-f: ]+", " fe80::XXXX:XXXX:XXXX:XXXX   ", actual
 289             )
 290             # Remove trailing spaces on all lines
 291             actual = "\n".join([line.rstrip() for line in actual.splitlines()])
 292
 293             # Fix newlines (make them all the same)
 294             actual = ("\n".join(actual.splitlines()) + "\n").splitlines(1)
 295
 296             # Generate Diff
 297             diff = topotest.get_textdiff(
 298                 actual,
 299                 expected,
 300                 title1="actual SHOW IPv6 RIPng",
 301                 title2="expected SHOW IPv6 RIPng",
 302             )
 303
 304             # Empty string if it matches, otherwise diff contains unified diff
 305             if diff:
 306                 sys.stderr.write("r%s failed SHOW IPv6 RIPng check:\n%s\n" % (i, diff))
 307                 failures += 1
 308             else:
 309                 print("r%s ok" % i)
 310
 311             assert failures == 0, "SHOW IPv6 RIPng failed for router r%s:\n%s" % (
 312                 i,
 313                 diff,
 314             )
 315
 316     # Make sure that all daemons are running
 317     for i in range(1, 4):
 318         fatal_error = net["r%s" % i].checkRouterRunning()
 319         assert fatal_error == "", fatal_error
 320
 321     # For debugging after starting FRR daemons, uncomment the next line
 322     # CLI(net)
 323
 324
 325 def test_zebra_ipv6_routingTable():
 326     global fatal_error
 327     global net
 328
 329     # Skip if previous fatal error condition is raised
 330     if fatal_error != "":
 331         pytest.skip(fatal_error)
 332
 333     thisDir = os.path.dirname(os.path.realpath(__file__))
 334
 335     # Verify OSPFv3 Routing Table
 336     print("\n\n** Verifying Zebra IPv6 Routing Table")
 337     print("******************************************\n")
 338     failures = 0
 339     for i in range(1, 4):
 340         refTableFile = "%s/r%s/show_ipv6_route.ref" % (thisDir, i)
 341         if os.path.isfile(refTableFile):
 342             # Read expected result from file
 343             expected = open(refTableFile).read().rstrip()
 344             # Fix newlines (make them all the same)
 345             expected = ("\n".join(expected.splitlines()) + "\n").splitlines(1)
 346
 347             # Actual output from router
 348             actual = (
 349                 net["r%s" % i]
 350                 .cmd('vtysh -c "show ipv6 route" 2> /dev/null | grep "^R"')
 351                 .rstrip()
 352             )
 353             # Mask out Link-Local mac address portion. They are random...
 354             actual = re.sub(r" fe80::[0-9a-f:]+", " fe80::XXXX:XXXX:XXXX:XXXX", actual)
 355             # Drop timers on end of line
 356             actual = re.sub(r", [0-2][0-9]:[0-5][0-9]:[0-5][0-9]", "", actual)
 357             # Fix newlines (make them all the same)
 358             actual = ("\n".join(actual.splitlines()) + "\n").splitlines(1)
 359
 360             # Generate Diff
 361             diff = topotest.get_textdiff(
 362                 actual,
 363                 expected,
 364                 title1="actual Zebra IPv6 routing table",
 365                 title2="expected Zebra IPv6 routing table",
 366             )
 367
 368             # Empty string if it matches, otherwise diff contains unified diff
 369             if diff:
 370                 sys.stderr.write(
 371                     "r%s failed Zebra IPv6 Routing Table Check:\n%s\n" % (i, diff)
 372                 )
 373                 failures += 1
 374             else:
 375                 print("r%s ok" % i)
 376
 377             assert (
 378                 failures == 0
 379             ), "Zebra IPv6 Routing Table verification failed for router r%s:\n%s" % (
 380                 i,
 381                 diff,
 382             )
 383
 384     # Make sure that all daemons are running
 385     for i in range(1, 4):
 386         fatal_error = net["r%s" % i].checkRouterRunning()
 387         assert fatal_error == "", fatal_error
 388
 389     # For debugging after starting FRR daemons, uncomment the next line
 390     # CLI(net)
 391
 392
 393 def test_shutdown_check_stderr():
 394     global fatal_error
 395     global net
 396
 397     # Skip if previous fatal error condition is raised
 398     if fatal_error != "":
 399         pytest.skip(fatal_error)
 400
 401     if os.environ.get("TOPOTESTS_CHECK_STDERR") is None:
 402         print(
 403             "SKIPPED final check on StdErr output: Disabled (TOPOTESTS_CHECK_STDERR undefined)\n"
 404         )
 405         pytest.skip("Skipping test for Stderr output")
 406
 407     thisDir = os.path.dirname(os.path.realpath(__file__))
 408
 409     print("\n\n** Verifying unexpected STDERR output from daemons")
 410     print("******************************************\n")
 411
 412     net["r1"].stopRouter()
 413
 414     log = net["r1"].getStdErr("ripngd")
 415     if log:
 416         print("\nRIPngd StdErr Log:\n" + log)
 417     log = net["r1"].getStdErr("zebra")
 418     if log:
 419         print("\nZebra StdErr Log:\n" + log)
 420
 421
 422 def test_shutdown_check_memleak():
 423     global fatal_error
 424     global net
 425
 426     # Skip if previous fatal error condition is raised
 427     if fatal_error != "":
 428         pytest.skip(fatal_error)
 429
 430     if os.environ.get("TOPOTESTS_CHECK_MEMLEAK") is None:
 431         print(
 432             "SKIPPED final check on Memory leaks: Disabled (TOPOTESTS_CHECK_MEMLEAK undefined)\n"
 433         )
 434         pytest.skip("Skipping test for memory leaks")
 435
 436     thisDir = os.path.dirname(os.path.realpath(__file__))
 437
 438     net["r1"].stopRouter()
 439     net["r1"].report_memory_leaks(
 440         os.environ.get("TOPOTESTS_CHECK_MEMLEAK"), os.path.basename(__file__)
 441     )
 442
 443
 444 if __name__ == "__main__":
 445
 446     setLogLevel("info")
 447     # To suppress tracebacks, either use the following pytest call or add "--tb=no" to cli
 448     # retval = pytest.main(["-s", "--tb=no"])
 449     retval = pytest.main(["-s"])
 450     sys.exit(retval)