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 time
  48 import pytest
  49 import functools
  50
  51 CWD = os.path.dirname(os.path.realpath(__file__))
  52 sys.path.append(os.path.join(CWD, "../"))
  53
  54 # pylint: disable=C0413
  55 from lib import topotest
  56 from lib.topogen import Topogen, TopoRouter, get_topogen
  57 from lib.topolog import logger
  58 from mininet.topo import Topo
  59
  60
  61 class TemplateTopo(Topo):
  62     def build(self, *_args, **_opts):
  63         tgen = get_topogen(self)
  64
  65         for routern in range(1, 7):
  66             tgen.add_router("r{}".format(routern))
  67
  68         # Scenario 1.
  69         switch = tgen.add_switch("s1")
  70         switch.add_link(tgen.gears["r1"])
  71         switch.add_link(tgen.gears["r2"])
  72
  73         # Scenario 2.
  74         switch = tgen.add_switch("s2")
  75         switch.add_link(tgen.gears["r3"])
  76         switch.add_link(tgen.gears["r4"])
  77
  78         # Scenario 3.
  79         switch = tgen.add_switch("s3")
  80         switch.add_link(tgen.gears["r5"])
  81         switch.add_link(tgen.gears["r6"])
  82
  83
  84 def setup_module(mod):
  85     tgen = Topogen(TemplateTopo, mod.__name__)
  86     tgen.start_topology()
  87
  88     router_list = tgen.routers()
  89
  90     for i, (rname, router) in enumerate(router_list.items(), 1):
  91         router.load_config(
  92             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  93         )
  94         router.load_config(
  95             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
  96         )
  97
  98     tgen.start_router()
  99
 100
 101 def teardown_module(mod):
 102     tgen = get_topogen()
 103     tgen.stop_topology()
 104
 105
 106 def test_ebgp_requires_policy():
 107     tgen = get_topogen()
 108
 109     if tgen.routers_have_failure():
 110         pytest.skip(tgen.errors)
 111
 112     def _bgp_converge(router):
 113         output = json.loads(
 114             tgen.gears[router].vtysh_cmd("show ip bgp neighbor 192.168.255.1 json")
 115         )
 116         expected = {"192.168.255.1": {"bgpState": "Established"}}
 117         return topotest.json_cmp(output, expected)
 118
 119     def _bgp_has_routes(router):
 120         output = json.loads(
 121             tgen.gears[router].vtysh_cmd(
 122                 "show ip bgp neighbor 192.168.255.1 routes json"
 123             )
 124         )
 125         expected = {"routes": {"172.16.255.254/32": [{"valid": True}]}}
 126         return topotest.json_cmp(output, expected)
 127
 128     def _bgp_advertised_routes(router):
 129         output = json.loads(
 130             tgen.gears[router].vtysh_cmd(
 131                 "show ip bgp neighbor 192.168.255.2 advertised-routes json"
 132             )
 133         )
 134         expected = {
 135             "advertisedRoutes": {},
 136             "totalPrefixCounter": 0,
 137             "filteredPrefixCounter": 0,
 138         }
 139         return topotest.json_cmp(output, expected)
 140
 141     # Scenario 1.
 142     logger.info("Scenario 1: r2 receives 192.168.255.1/32 from r1")
 143     test_func = functools.partial(_bgp_converge, "r2")
 144     success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
 145     assert success is True, "Failed bgp convergence (r2)"
 146
 147     test_func = functools.partial(_bgp_has_routes, "r2")
 148     success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
 149     assert success is True, "r2 does not receive 192.168.255.1/32"
 150
 151     # Scenario 2.
 152     logger.info("Scenario 2: r3 must not send 192.168.255.1/32 to r4")
 153     test_func = functools.partial(_bgp_converge, "r4")
 154     success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
 155     assert success is True, "Failed bgp convergence (r4)"
 156
 157     test_func = functools.partial(_bgp_advertised_routes, "r3")
 158     success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
 159     assert success is True, "r3 announced 192.168.255.1/32 to r4"
 160
 161     # Scenario 3.
 162     logger.info("Scenario 3: r6 receives 192.168.255.1/32 from r5 (iBGP)")
 163     test_func = functools.partial(_bgp_converge, "r6")
 164     success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
 165     assert success is True, "Failed bgp convergence (r6)"
 166
 167     test_func = functools.partial(_bgp_has_routes, "r6")
 168     success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
 169     assert success is True, "r6 does not receive 192.168.255.1/32"
 170
 171
 172 if __name__ == "__main__":
 173     args = ["-s"] + sys.argv[1:]
 174     sys.exit(pytest.main(args))