tests/topotests/ldp_oc_acl_topo1/test_ldp_oc_acl_topo1.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # test_ldp_oc_acl_topo1.py
   6 # Part of NetDEF Topology Tests
   7 #
   8 # Copyright (c) 2020 by Volta Networks
   9 #
  10
  11 r"""
  12 test_ldp_oc_acl_topo1.py: Simple FRR LDP Test
  13
  14              +---------+
  15              |    r1   |
  16              | 1.1.1.1 |
  17              +----+----+
  18                   | .1  r1-eth0
  19                   |
  20             ~~~~~~~~~~~~~
  21           ~~     sw0     ~~
  22           ~~ 10.0.1.0/24 ~~
  23             ~~~~~~~~~~~~~
  24                   |10.0.1.0/24
  25                   |
  26                   | .2  r2-eth0
  27              +----+----+
  28              |    r2   |
  29              | 2.2.2.2 |
  30              +--+---+--+
  31     r2-eth2  .2 |   | .2  r2-eth1
  32          ______/     \______
  33         /                   \
  34   ~~~~~~~~~~~~~        ~~~~~~~~~~~~~
  35 ~~     sw2     ~~    ~~     sw1     ~~
  36 ~~ 10.0.3.0/24 ~~    ~~ 10.0.2.0/24 ~~
  37   ~~~~~~~~~~~~~        ~~~~~~~~~~~~~
  38         |                 /    |
  39          \      _________/     |
  40           \    /                \
  41 r3-eth1 .3 |  | .3  r3-eth0      | .4 r4-eth0
  42       +----+--+---+         +----+----+
  43       |     r3    |         |    r4   |
  44       |  3.3.3.3  |         | 4.4.4.4 |
  45       +-----------+         +---------+
  46 """
  47
  48 import os
  49 import sys
  50 import pytest
  51 import json
  52 from time import sleep
  53 from functools import partial
  54
  55 # Save the Current Working Directory to find configuration files.
  56 CWD = os.path.dirname(os.path.realpath(__file__))
  57 sys.path.append(os.path.join(CWD, "../"))
  58
  59 # pylint: disable=C0413
  60 # Import topogen and topotest helpers
  61 from lib import topotest
  62 from lib.topogen import Topogen, TopoRouter, get_topogen
  63 from lib.topolog import logger
  64
  65 # Required to instantiate the topology builder class.
  66
  67 pytestmark = [pytest.mark.ldpd, pytest.mark.ospfd]
  68
  69
  70 def build_topo(tgen):
  71     "Build function"
  72
  73     #
  74     # Define FRR Routers
  75     #
  76     for router in ["r1", "r2", "r3", "r4"]:
  77         tgen.add_router(router)
  78
  79     #
  80     # Define connections
  81     #
  82     switch = tgen.add_switch("s0")
  83     switch.add_link(tgen.gears["r1"])
  84     switch.add_link(tgen.gears["r2"])
  85
  86     switch = tgen.add_switch("s1")
  87     switch.add_link(tgen.gears["r2"])
  88     switch.add_link(tgen.gears["r3"])
  89     switch.add_link(tgen.gears["r4"])
  90
  91     switch = tgen.add_switch("s2")
  92     switch.add_link(tgen.gears["r2"])
  93     switch.add_link(tgen.gears["r3"])
  94
  95
  96 def setup_module(mod):
  97     "Sets up the pytest environment"
  98     tgen = Topogen(build_topo, mod.__name__)
  99     tgen.start_topology()
 100
 101     router_list = tgen.routers()
 102
 103     # For all registered routers, load the zebra configuration file
 104     for rname, router in router_list.items():
 105         router.load_config(
 106             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 107         )
 108         # Don't start ospfd and ldpd in the CE nodes
 109         if router.name[0] == "r":
 110             router.load_config(
 111                 TopoRouter.RD_OSPF, os.path.join(CWD, "{}/ospfd.conf".format(rname))
 112             )
 113             router.load_config(
 114                 TopoRouter.RD_LDP, os.path.join(CWD, "{}/ldpd.conf".format(rname))
 115             )
 116
 117     tgen.start_router()
 118
 119
 120 def teardown_module(mod):
 121     "Teardown the pytest environment"
 122     tgen = get_topogen()
 123
 124     # This function tears down the whole topology.
 125     tgen.stop_topology()
 126
 127
 128 def router_compare_json_output(rname, command, reference):
 129     "Compare router JSON output"
 130
 131     logger.info('Comparing router "%s" "%s" output', rname, command)
 132
 133     tgen = get_topogen()
 134     filename = "{}/{}/{}".format(CWD, rname, reference)
 135     expected = json.loads(open(filename).read())
 136
 137     # Run test function until we get an result. Wait at most 80 seconds.
 138     test_func = partial(topotest.router_json_cmp, tgen.gears[rname], command, expected)
 139     _, diff = topotest.run_and_expect(test_func, None, count=160, wait=0.5)
 140
 141     assertmsg = '"{}" JSON output mismatches the expected result'.format(rname)
 142     assert diff is None, assertmsg
 143
 144
 145 def test_ospf_convergence():
 146     logger.info("Test: check OSPF adjacencies")
 147
 148     tgen = get_topogen()
 149
 150     # Skip if previous fatal error condition is raised
 151     if tgen.routers_have_failure():
 152         pytest.skip(tgen.errors)
 153
 154     for rname in ["r1", "r2", "r3", "r4"]:
 155         router_compare_json_output(
 156             rname, "show ip ospf neighbor json", "show_ip_ospf_neighbor.json"
 157         )
 158
 159
 160 def test_rib():
 161     logger.info("Test: verify RIB")
 162     tgen = get_topogen()
 163
 164     # Skip if previous fatal error condition is raised
 165     if tgen.routers_have_failure():
 166         pytest.skip(tgen.errors)
 167
 168     for rname in ["r1", "r2", "r3", "r4"]:
 169         router_compare_json_output(rname, "show ip route json", "show_ip_route.ref")
 170
 171
 172 def test_ldp_adjacencies():
 173     logger.info("Test: verify LDP adjacencies")
 174     tgen = get_topogen()
 175
 176     # Skip if previous fatal error condition is raised
 177     if tgen.routers_have_failure():
 178         pytest.skip(tgen.errors)
 179
 180     for rname in ["r1", "r2", "r3", "r4"]:
 181         router_compare_json_output(
 182             rname, "show mpls ldp discovery json", "show_ldp_discovery.ref"
 183         )
 184
 185
 186 def test_ldp_neighbors():
 187     logger.info("Test: verify LDP neighbors")
 188     tgen = get_topogen()
 189
 190     # Skip if previous fatal error condition is raised
 191     if tgen.routers_have_failure():
 192         pytest.skip(tgen.errors)
 193
 194     for rname in ["r1", "r2", "r3", "r4"]:
 195         router_compare_json_output(
 196             rname, "show mpls ldp neighbor json", "show_ldp_neighbor.ref"
 197         )
 198
 199
 200 def test_ldp_bindings():
 201     logger.info("Test: verify LDP bindings")
 202     tgen = get_topogen()
 203
 204     # Skip if previous fatal error condition is raised
 205     if tgen.routers_have_failure():
 206         pytest.skip(tgen.errors)
 207
 208     for rname in ["r1", "r2", "r3", "r4"]:
 209         router_compare_json_output(
 210             rname, "show mpls ldp binding json", "show_ldp_binding.ref"
 211         )
 212
 213
 214 def test_ldp_bindings_all_routes():
 215     logger.info("Test: verify LDP bindings after host filter removed")
 216     tgen = get_topogen()
 217
 218     # Skip if previous fatal error condition is raised
 219     if tgen.routers_have_failure():
 220         pytest.skip(tgen.errors)
 221
 222     # remove ACL that blocks advertising everything but host routes */
 223     cmd = 'vtysh -c "configure terminal" -c "mpls ldp" -c "address-family ipv4" -c "no label local allocate host-routes"'
 224     tgen.net["r1"].cmd(cmd)
 225     sleep(2)
 226
 227     for rname in ["r1", "r2", "r3", "r4"]:
 228         router_compare_json_output(
 229             rname, "show mpls ldp binding json", "show_ldp_all_binding.ref"
 230         )
 231
 232
 233 # Memory leak test template
 234 def test_memory_leak():
 235     "Run the memory leak test and report results."
 236     tgen = get_topogen()
 237     if not tgen.is_memleak_enabled():
 238         pytest.skip("Memory leak test/report is disabled")
 239
 240     tgen.report_memory_leaks()
 241
 242
 243 if __name__ == "__main__":
 244     args = ["-s"] + sys.argv[1:]
 245     sys.exit(pytest.main(args))