tests/topotests/bgp_ebgp_requires_policy/test_bgp_ebgp_requires_policy.py

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