tests/topotests/bgp_roles_capability/test_bgp_roles_capability.py

   1 #!/usr/bin/python
   2 #
   3 # test_bgp_roles_capability.py
   4 # Part of NetDEF Topology Tests
   5 #
   6 # Copyright (c) 2022 by Eugene Bogomazov <eb@qrator.net>
   7 # Copyright (c) 2017 by
   8 # Network Device Education Foundation, Inc. ("NetDEF")
   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 VMWARE 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 test_bgp_roles_capability: test bgp roles negotiation
  27 """
  28
  29 import json
  30 import os
  31 import sys
  32 import functools
  33 import pytest
  34
  35 CWD = os.path.dirname(os.path.realpath(__file__))
  36 sys.path.append(os.path.join(CWD, "../"))
  37
  38 # pylint: disable=C0413
  39 from lib import topotest
  40 from lib.topogen import Topogen, TopoRouter, get_topogen
  41 from lib.topolog import logger
  42
  43 pytestmark = [pytest.mark.bgpd]
  44
  45
  46 topodef = {f"s{i}": ("r1", f"r{i}") for i in range(2, 7)}
  47
  48
  49 @pytest.fixture(scope="module")
  50 def tgen(request):
  51     tgen = Topogen(topodef, request.module.__name__)
  52     tgen.start_topology()
  53     router_list = tgen.routers()
  54     for rname, router in router_list.items():
  55         router.load_config(TopoRouter.RD_ZEBRA, "zebra.conf")
  56         router.load_config(TopoRouter.RD_BGP, "bgpd.conf")
  57     tgen.start_router()
  58     yield tgen
  59     tgen.stop_topology()
  60
  61
  62 @pytest.fixture(autouse=True)
  63 def skip_on_failure(tgen):
  64     if tgen.routers_have_failure():
  65         pytest.skip("skipped because of previous test failure")
  66
  67
  68 def find_neighbor_status(router, neighbor_ip):
  69     return json.loads(router.vtysh_cmd(f"show bgp neighbors {neighbor_ip} json"))[
  70         neighbor_ip
  71     ]
  72
  73
  74 def check_role_mismatch(router, neighbor_ip):
  75     return is_role_mismatch(find_neighbor_status(router, neighbor_ip))
  76
  77
  78 def is_role_mismatch(neighbor_status):
  79     return (
  80         neighbor_status["bgpState"] != "Established"
  81         and neighbor_status.get("lastErrorCodeSubcode") == "020B"  # <2, 11>
  82         and "Role Mismatch" in neighbor_status.get("lastNotificationReason", "")
  83     )
  84
  85
  86 def check_session_established(router, neighbor_ip):
  87     neighbor_status = find_neighbor_status(router, neighbor_ip)
  88     return neighbor_status["bgpState"] == "Established"
  89
  90
  91 def test_correct_pair(tgen):
  92     # provider-customer pair
  93     router = tgen.gears["r1"]
  94     neighbor_ip = "192.168.2.2"
  95     check_r2_established = functools.partial(
  96         check_session_established, router, neighbor_ip
  97     )
  98     success, result = topotest.run_and_expect(
  99         check_r2_established, True, count=20, wait=3
 100     )
 101     assert success, "Session with r2 is not Established"
 102
 103     neighbor_status = find_neighbor_status(router, neighbor_ip)
 104     assert neighbor_status["localRole"] == "provider"
 105     assert neighbor_status["remoteRole"] == "customer"
 106     assert (
 107         neighbor_status["neighborCapabilities"].get("role") == "advertisedAndReceived"
 108     )
 109
 110
 111 def test_role_pair_mismatch(tgen):
 112     # provider-peer mistmatch
 113     router = tgen.gears["r3"]
 114     neighbor_ip = "192.168.3.1"
 115     check_r3_mismatch = functools.partial(check_role_mismatch, router, neighbor_ip)
 116     success, result = topotest.run_and_expect(check_r3_mismatch, True, count=20, wait=3)
 117     assert success, "Session between r1 and r3 was not correctly closed"
 118
 119
 120 def test_single_role_advertising(tgen):
 121     # provider-undefined pair; we set role
 122     router = tgen.gears["r1"]
 123     neighbor_ip = "192.168.4.2"
 124     check_r4_established = functools.partial(
 125         check_session_established, router, neighbor_ip
 126     )
 127     success, result = topotest.run_and_expect(
 128         check_r4_established, True, count=20, wait=3
 129     )
 130     assert success, "Session with r4 is not Established"
 131
 132     neighbor_status = find_neighbor_status(router, neighbor_ip)
 133     assert neighbor_status["localRole"] == "provider"
 134     assert neighbor_status["remoteRole"] == "undefined"
 135     assert neighbor_status["neighborCapabilities"].get("role") == "advertised"
 136
 137
 138 def test_single_role_receiving(tgen):
 139     # provider-undefined pair; we receive role
 140     router = tgen.gears["r4"]
 141     neighbor_ip = "192.168.4.1"
 142     check_r1_established = functools.partial(
 143         check_session_established, router, neighbor_ip
 144     )
 145     success, result = topotest.run_and_expect(
 146         check_r1_established, True, count=20, wait=3
 147     )
 148     assert success, "Session with r1 is not Established"
 149
 150     neighbor_status = find_neighbor_status(router, neighbor_ip)
 151     assert neighbor_status["localRole"] == "undefined"
 152     assert neighbor_status["remoteRole"] == "provider"
 153     assert neighbor_status["neighborCapabilities"].get("role") == "received"
 154
 155
 156 def test_role_strict_mode(tgen):
 157     # provider-undefined pair with strict-mode
 158     router = tgen.gears["r5"]
 159     neighbor_ip = "192.168.5.1"
 160     check_r5_mismatch = functools.partial(check_role_mismatch, router, neighbor_ip)
 161     success, result = topotest.run_and_expect(check_r5_mismatch, True, count=20, wait=3)
 162     assert success, "Session between r1 and r5 was not correctly closed"
 163
 164
 165 def test_correct_pair_peer_group(tgen):
 166     # provider-customer pair (using peer-groups)
 167     router = tgen.gears["r1"]
 168     neighbor_ip = "192.168.6.2"
 169     check_r6_established = functools.partial(
 170         check_session_established, router, neighbor_ip
 171     )
 172     success, _ = topotest.run_and_expect(check_r6_established, True, count=20, wait=3)
 173     assert success, "Session with r6 is not Established"
 174
 175     neighbor_status = find_neighbor_status(router, neighbor_ip)
 176     assert neighbor_status["localRole"] == "provider"
 177     assert neighbor_status["remoteRole"] == "customer"
 178     assert (
 179         neighbor_status["neighborCapabilities"].get("role") == "advertisedAndReceived"
 180     )
 181
 182
 183 if __name__ == "__main__":
 184     args = ["-s"] + sys.argv[1:]
 185     sys.exit(pytest.main(args))