tests/topotests/bgp_roles_filtering/test_bgp_roles_filtering.py

   1 #!/usr/bin/python
   2 # SPDX-License-Identifier: ISC
   3 #
   4 # test_bgp_roles_filtering.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2022 by Eugene Bogomazov <eb@qrator.net>
   8 # Copyright (c) 2017 by
   9 # Network Device Education Foundation, Inc. ("NetDEF")
  10 #
  11
  12 """
  13 test_bgp_roles_filtering: test leaks prevention and mitigation with roles
  14 """
  15
  16 import json
  17 import os
  18 import sys
  19 import functools
  20 import pytest
  21
  22 CWD = os.path.dirname(os.path.realpath(__file__))
  23 sys.path.append(os.path.join(CWD, "../"))
  24
  25 # pylint: disable=C0413
  26 from lib import topotest
  27 from lib.bgp import verify_bgp_convergence_from_running_config
  28 from lib.topogen import Topogen, TopoRouter, get_topogen
  29 from lib.topolog import logger
  30
  31 pytestmark = [pytest.mark.bgpd]
  32
  33
  34 topodef = {f"s{i}": (f"r{i}", "r10") for i in range(1, 8)}
  35
  36
  37 @pytest.fixture(scope="module")
  38 def tgen(request):
  39     tgen = Topogen(topodef, request.module.__name__)
  40     tgen.start_topology()
  41     router_list = tgen.routers()
  42     for rname, router in router_list.items():
  43         router.load_config(TopoRouter.RD_ZEBRA, "zebra.conf")
  44         router.load_config(TopoRouter.RD_BGP, "bgpd.conf")
  45     tgen.start_router()
  46     yield tgen
  47     tgen.stop_topology()
  48
  49
  50 @pytest.fixture(autouse=True)
  51 def skip_on_failure(tgen):
  52     if tgen.routers_have_failure():
  53         pytest.skip("skipped because of previous test failure")
  54
  55
  56 def test_r10_routes(tgen):
  57     # provider-undefine pair bur strict-mode was set
  58     def _routes_half_converged():
  59         routes = json.loads(tgen.gears["r10"].vtysh_cmd("show bgp ipv4 json"))["routes"]
  60         output = sorted(routes.keys())
  61         expected = [
  62             "192.0.2.1/32",
  63             "192.0.2.2/32",
  64             "192.0.2.3/32",
  65             "192.0.2.4/32",
  66             "192.0.2.5/32",
  67             "192.0.2.6/32",
  68             "192.0.2.7/32",
  69         ]
  70         return output == expected
  71
  72     success, result = topotest.run_and_expect(
  73         _routes_half_converged, True, count=20, wait=3
  74     )
  75     assert success, "Routes did not converged"
  76
  77     routes_with_otc = list()
  78     for number in range(1, 8):
  79         prefix = f"192.0.2.{number}/32"
  80         route_details = json.loads(
  81             tgen.gears["r10"].vtysh_cmd(f"show bgp ipv4 {prefix} json")
  82         )
  83         if route_details["paths"][0].get("otc") is not None:
  84             routes_with_otc.append(prefix)
  85     assert routes_with_otc == [
  86         "192.0.2.1/32",
  87         "192.0.2.2/32",
  88         "192.0.2.6/32",
  89         "192.0.2.7/32",
  90     ]
  91
  92
  93 def test_r1_routes(tgen):
  94     routes = json.loads(tgen.gears["r1"].vtysh_cmd("show bgp ipv4 json"))["routes"]
  95     routes_list = sorted(routes.keys())
  96     assert routes_list == [
  97         "192.0.2.1/32",  # own
  98         "192.0.2.3/32",
  99         "192.0.2.4/32",
 100         "192.0.2.5/32",
 101     ]
 102
 103
 104 def test_r6_routes(tgen):
 105     routes = json.loads(tgen.gears["r6"].vtysh_cmd("show bgp ipv4 json"))["routes"]
 106     routes_list = sorted(routes.keys())
 107     assert routes_list == [
 108         "192.0.2.3/32",
 109         "192.0.2.4/32",
 110         "192.0.2.5/32",
 111         "192.0.2.6/32",  # own
 112     ]
 113
 114
 115 def test_r4_routes(tgen):
 116     routes = json.loads(tgen.gears["r4"].vtysh_cmd("show bgp ipv4 json"))["routes"]
 117     routes_list = sorted(routes.keys())
 118     assert routes_list == [
 119         "192.0.2.1/32",
 120         "192.0.2.2/32",
 121         "192.0.2.3/32",
 122         "192.0.2.4/32",
 123         "192.0.2.5/32",
 124         "192.0.2.6/32",
 125         "192.0.2.7/32",
 126     ]
 127
 128
 129 if __name__ == "__main__":
 130     args = ["-s"] + sys.argv[1:]
 131     sys.exit(pytest.main(args))