tests/topotests/bgp_ebgp_requires_policy/test_bgp_ebgp_requires_policy.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # bgp_ebgp_requires_policy.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2019 by
   8 # Donatas Abraitis <donatas.abraitis@gmail.com>
   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 bgp_ebgp_requires_policy.py:
  27
  28 Test if eBGP sender without a filter applied to the peer is allowed
  29 to send advertisements.
  30
  31 Scenario 1:
  32   r1 has a filter applied for outgoing direction,
  33   r2 receives 192.168.255.1/32.
  34
  35 Scenario 2:
  36   r3 hasn't a filter appied for outgoing direction,
  37   r4 does not receive 192.168.255.1/32.
  38
  39 Scenario 3:
  40   r5 and r6 establish iBGP session which in turn should ignore
  41   RFC8212. All routes for both directions MUST work.
  42 """
  43
  44 import os
  45 import sys
  46 import json
  47 import pytest
  48 import functools
  49
  50 CWD = os.path.dirname(os.path.realpath(__file__))
  51 sys.path.append(os.path.join(CWD, "../"))
  52
  53 # pylint: disable=C0413
  54 from lib import topotest
  55 from lib.topogen import Topogen, TopoRouter, get_topogen
  56 from lib.topolog import logger
  57
  58 pytestmark = [pytest.mark.bgpd]
  59
  60
  61 def build_topo(tgen):
  62     for routern in range(1, 7):
  63         tgen.add_router("r{}".format(routern))
  64
  65     # Scenario 1.
  66     switch = tgen.add_switch("s1")
  67     switch.add_link(tgen.gears["r1"])
  68     switch.add_link(tgen.gears["r2"])
  69
  70     # Scenario 2.
  71     switch = tgen.add_switch("s2")
  72     switch.add_link(tgen.gears["r3"])
  73     switch.add_link(tgen.gears["r4"])
  74
  75     # Scenario 3.
  76     switch = tgen.add_switch("s3")
  77     switch.add_link(tgen.gears["r5"])
  78     switch.add_link(tgen.gears["r6"])
  79
  80
  81 def setup_module(mod):
  82     tgen = Topogen(build_topo, mod.__name__)
  83     tgen.start_topology()
  84
  85     router_list = tgen.routers()
  86
  87     for i, (rname, router) in enumerate(router_list.items(), 1):
  88         router.load_config(
  89             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  90         )
  91         router.load_config(
  92             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
  93         )
  94
  95     tgen.start_router()
  96
  97
  98 def teardown_module(mod):
  99     tgen = get_topogen()
 100     tgen.stop_topology()
 101
 102
 103 def test_ebgp_requires_policy():
 104     tgen = get_topogen()
 105
 106     if tgen.routers_have_failure():
 107         pytest.skip(tgen.errors)
 108
 109     def _bgp_converge(router):
 110         output = json.loads(
 111             tgen.gears[router].vtysh_cmd("show ip bgp neighbor 192.168.255.1 json")
 112         )
 113         expected = {"192.168.255.1": {"bgpState": "Established"}}
 114         return topotest.json_cmp(output, expected)
 115
 116     def _bgp_has_routes(router):
 117         output = json.loads(
 118             tgen.gears[router].vtysh_cmd(
 119                 "show ip bgp neighbor 192.168.255.1 routes json"
 120             )
 121         )
 122         expected = {"routes": {"172.16.255.254/32": [{"valid": True}]}}
 123         return topotest.json_cmp(output, expected)
 124
 125     def _bgp_advertised_routes(router):
 126         output = json.loads(
 127             tgen.gears[router].vtysh_cmd(
 128                 "show ip bgp neighbor 192.168.255.2 advertised-routes json"
 129             )
 130         )
 131         expected = {
 132             "advertisedRoutes": {},
 133             "totalPrefixCounter": 0,
 134             "filteredPrefixCounter": 0,
 135         }
 136         return topotest.json_cmp(output, expected)
 137
 138     # Scenario 1.
 139     logger.info("Scenario 1: r2 receives 192.168.255.1/32 from r1")
 140     test_func = functools.partial(_bgp_converge, "r2")
 141     success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
 142     assert success is True, "Failed bgp convergence (r2)"
 143
 144     test_func = functools.partial(_bgp_has_routes, "r2")
 145     success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
 146     assert success is True, "r2 does not receive 192.168.255.1/32"
 147
 148     # Scenario 2.
 149     logger.info("Scenario 2: r3 must not send 192.168.255.1/32 to r4")
 150     test_func = functools.partial(_bgp_converge, "r4")
 151     success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
 152     assert success is True, "Failed bgp convergence (r4)"
 153
 154     test_func = functools.partial(_bgp_advertised_routes, "r3")
 155     success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
 156     assert success is True, "r3 announced 192.168.255.1/32 to r4"
 157
 158     # Scenario 3.
 159     logger.info("Scenario 3: r6 receives 192.168.255.1/32 from r5 (iBGP)")
 160     test_func = functools.partial(_bgp_converge, "r6")
 161     success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
 162     assert success is True, "Failed bgp convergence (r6)"
 163
 164     test_func = functools.partial(_bgp_has_routes, "r6")
 165     success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
 166     assert success is True, "r6 does not receive 192.168.255.1/32"
 167
 168
 169 if __name__ == "__main__":
 170     args = ["-s"] + sys.argv[1:]
 171     sys.exit(pytest.main(args))