tests/topotests/bgp_snmp_bgp4v2mib/test_bgp_snmp_bgp4v2mib.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # Copyright (c) 2022 Donatas Abraitis <donatas@opensourcerouting.org>
   5 #
   6 # Permission to use, copy, modify, and/or distribute this software
   7 # for any purpose with or without fee is hereby granted, provided
   8 # that the above copyright notice and this permission notice appear
   9 # in all copies.
  10 #
  11 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  12 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  13 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  14 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  15 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  16 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  17 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  18 # OF THIS SOFTWARE.
  19 #
  20
  21 """
  22 Test some of the BGP4V2-MIB entries.
  23 """
  24
  25 import os
  26 import sys
  27 import json
  28 from time import sleep
  29 import pytest
  30 import functools
  31
  32 CWD = os.path.dirname(os.path.realpath(__file__))
  33 sys.path.append(os.path.join(CWD, "../"))
  34
  35 # pylint: disable=C0413
  36 # Import topogen and topotest helpers
  37 from lib.topogen import Topogen, TopoRouter, get_topogen
  38 from lib.snmptest import SnmpTester
  39 from lib import topotest
  40
  41 pytestmark = [pytest.mark.bgpd, pytest.mark.snmp]
  42
  43
  44 def build_topo(tgen):
  45     tgen.add_router("r1")
  46     tgen.add_router("r2")
  47
  48     switch = tgen.add_switch("s1")
  49     switch.add_link(tgen.gears["r1"])
  50     switch.add_link(tgen.gears["r2"])
  51
  52
  53 def setup_module(mod):
  54     snmpd = os.system("which snmpd")
  55     if snmpd:
  56         error_msg = "SNMP not installed - skipping"
  57         pytest.skip(error_msg)
  58
  59     tgen = Topogen(build_topo, mod.__name__)
  60     tgen.start_topology()
  61
  62     for rname, router in tgen.routers().items():
  63         router.load_config(
  64             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  65         )
  66         router.load_config(
  67             TopoRouter.RD_BGP,
  68             os.path.join(CWD, "{}/bgpd.conf".format(rname)),
  69             "-M snmp",
  70         )
  71         router.load_config(
  72             TopoRouter.RD_SNMP,
  73             os.path.join(CWD, "{}/snmpd.conf".format(rname)),
  74             "-Le -Ivacm_conf,usmConf,iquery -V -DAgentX",
  75         )
  76
  77     tgen.start_router()
  78
  79
  80 def teardown_module(mod):
  81     tgen = get_topogen()
  82     tgen.stop_topology()
  83
  84
  85 def test_bgp_snmp_bgp4v2():
  86     tgen = get_topogen()
  87
  88     r2 = tgen.gears["r2"]
  89
  90     def _bgp_converge_summary():
  91         output = json.loads(r2.vtysh_cmd("show bgp summary json"))
  92         expected = {
  93             "ipv4Unicast": {
  94                 "peers": {
  95                     "192.168.12.1": {
  96                         "state": "Established",
  97                         "pfxRcd": 2,
  98                     }
  99                 }
 100             },
 101             "ipv6Unicast": {
 102                 "peers": {
 103                     "2001:db8::12:1": {
 104                         "state": "Established",
 105                         "pfxRcd": 2,
 106                     }
 107                 }
 108             },
 109         }
 110         return topotest.json_cmp(output, expected)
 111
 112     test_func = functools.partial(_bgp_converge_summary)
 113     _, result = topotest.run_and_expect(test_func, None, count=30, wait=1)
 114     assert result is None, "Can't see connections established"
 115
 116     def _bgp_converge_prefixes():
 117         output = json.loads(r2.vtysh_cmd("show bgp all json"))
 118         expected = {
 119             "ipv4Unicast": {
 120                 "routes": {
 121                     "10.0.0.0/31": [
 122                         {
 123                             "metric": 1,
 124                             "origin": "IGP",
 125                         }
 126                     ],
 127                     "10.0.0.2/32": [
 128                         {
 129                             "metric": 2,
 130                             "origin": "incomplete",
 131                         }
 132                     ],
 133                 }
 134             },
 135             "ipv6Unicast": {
 136                 "routes": {
 137                     "2001:db8::1/128": [
 138                         {
 139                             "metric": 1,
 140                             "origin": "IGP",
 141                         }
 142                     ],
 143                     "2001:db8:1::/56": [
 144                         {
 145                             "metric": 2,
 146                             "origin": "incomplete",
 147                         }
 148                     ],
 149                 }
 150             },
 151         }
 152         return topotest.json_cmp(output, expected)
 153
 154     test_func = functools.partial(_bgp_converge_prefixes)
 155     _, result = topotest.run_and_expect(test_func, None, count=30, wait=1)
 156     assert result is None, "Can't see prefixes from R1"
 157
 158     snmp = SnmpTester(r2, "localhost", "public", "2c", "-Ln -On")
 159
 160     def _snmpwalk_remote_addr():
 161         expected = {
 162             "1.3.6.1.3.5.1.1.2.1.5.1.4.192.168.12.1": "C0 A8 0C 01",
 163             "1.3.6.1.3.5.1.1.2.1.5.2.16.32.1.13.184.0.0.0.0.0.0.0.0.0.18.0.1": "20 01 0D B8 00 00 00 00 00 00 00 00 00 12 00 01",
 164         }
 165
 166         # bgp4V2PeerRemoteAddr
 167         output, _ = snmp.walk(".1.3.6.1.3.5.1.1.2.1.5")
 168         return output == expected
 169
 170     _, result = topotest.run_and_expect(_snmpwalk_remote_addr, True, count=10, wait=1)
 171     assertmsg = "Can't fetch SNMP for bgp4V2PeerRemoteAddr"
 172     assert result, assertmsg
 173
 174     def _snmpwalk_peer_state():
 175         expected = {
 176             "1.3.6.1.3.5.1.1.2.1.13.1.4.192.168.12.1": "6",
 177             "1.3.6.1.3.5.1.1.2.1.13.2.16.32.1.13.184.0.0.0.0.0.0.0.0.0.18.0.1": "6",
 178         }
 179
 180         # bgp4V2PeerState
 181         output, _ = snmp.walk(".1.3.6.1.3.5.1.1.2.1.13")
 182         return output == expected
 183
 184     _, result = topotest.run_and_expect(_snmpwalk_peer_state, True, count=10, wait=1)
 185     assertmsg = "Can't fetch SNMP for bgp4V2PeerState"
 186     assert result, assertmsg
 187
 188     def _snmpwalk_peer_last_error_code_received():
 189         expected = {
 190             "1.3.6.1.3.5.1.1.3.1.1.1.4.192.168.12.1": "0",
 191             "1.3.6.1.3.5.1.1.3.1.1.2.16.32.1.13.184.0.0.0.0.0.0.0.0.0.18.0.1": "0",
 192         }
 193
 194         # bgp4V2PeerLastErrorCodeReceived
 195         output, _ = snmp.walk(".1.3.6.1.3.5.1.1.3.1.1")
 196         return output == expected
 197
 198     _, result = topotest.run_and_expect(
 199         _snmpwalk_peer_last_error_code_received, True, count=10, wait=1
 200     )
 201     assertmsg = "Can't fetch SNMP for bgp4V2PeerLastErrorCodeReceived"
 202     assert result, assertmsg
 203
 204     def _snmpwalk_origin():
 205         expected = {
 206             "1.3.6.1.3.5.1.1.9.1.9.1.4.10.0.0.0.31.192.168.12.1": "1",
 207             "1.3.6.1.3.5.1.1.9.1.9.1.4.10.0.0.2.32.192.168.12.1": "3",
 208             "1.3.6.1.3.5.1.1.9.1.9.2.16.32.1.13.184.0.0.0.0.0.0.0.0.0.0.0.1.128.32.1.13.184.0.0.0.0.0.0.0.0.0.18.0.1": "1",
 209             "1.3.6.1.3.5.1.1.9.1.9.2.16.32.1.13.184.0.1.0.0.0.0.0.0.0.0.0.0.56.32.1.13.184.0.0.0.0.0.0.0.0.0.18.0.1": "3",
 210         }
 211
 212         # bgp4V2NlriOrigin
 213         output, _ = snmp.walk(".1.3.6.1.3.5.1.1.9.1.9")
 214         return output == expected
 215
 216     _, result = topotest.run_and_expect(_snmpwalk_origin, True, count=10, wait=1)
 217     assertmsg = "Can't fetch SNMP for bgp4V2NlriOrigin"
 218     assert result, assertmsg
 219
 220     def _snmpwalk_med():
 221         expected = {
 222             "1.3.6.1.3.5.1.1.9.1.17.1.4.10.0.0.0.31.192.168.12.1": "1",
 223             "1.3.6.1.3.5.1.1.9.1.17.1.4.10.0.0.2.32.192.168.12.1": "2",
 224             "1.3.6.1.3.5.1.1.9.1.17.2.16.32.1.13.184.0.0.0.0.0.0.0.0.0.0.0.1.128.32.1.13.184.0.0.0.0.0.0.0.0.0.18.0.1": "1",
 225             "1.3.6.1.3.5.1.1.9.1.17.2.16.32.1.13.184.0.1.0.0.0.0.0.0.0.0.0.0.56.32.1.13.184.0.0.0.0.0.0.0.0.0.18.0.1": "2",
 226         }
 227
 228         # bgp4V2NlriMed
 229         output, _ = snmp.walk(".1.3.6.1.3.5.1.1.9.1.17")
 230         return output == expected
 231
 232     _, result = topotest.run_and_expect(_snmpwalk_med, True, count=10, wait=1)
 233     assertmsg = "Can't fetch SNMP for bgp4V2NlriMed"
 234     assert result, assertmsg
 235
 236
 237 def test_memory_leak():
 238     "Run the memory leak test and report results."
 239     tgen = get_topogen()
 240     if not tgen.is_memleak_enabled():
 241         pytest.skip("Memory leak test/report is disabled")
 242
 243     tgen.report_memory_leaks()
 244
 245
 246 if __name__ == "__main__":
 247     args = ["-s"] + sys.argv[1:]
 248     sys.exit(pytest.main(args))