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