1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
|
#!/usr/bin/env python
# SPDX-License-Identifier: ISC
#
# Copyright (c) 2022 Donatas Abraitis <donatas@opensourcerouting.org>
#
"""
Test some of the BGP4V2-MIB entries.
"""
import os
import sys
import json
from time import sleep
import pytest
import functools
CWD = os.path.dirname(os.path.realpath(__file__))
sys.path.append(os.path.join(CWD, "../"))
# pylint: disable=C0413
# Import topogen and topotest helpers
from lib.topogen import Topogen, TopoRouter, get_topogen
from lib.snmptest import SnmpTester
from lib import topotest
pytestmark = [pytest.mark.bgpd, pytest.mark.snmp]
def build_topo(tgen):
tgen.add_router("r1")
tgen.add_router("r2")
switch = tgen.add_switch("s1")
switch.add_link(tgen.gears["r1"])
switch.add_link(tgen.gears["r2"])
def setup_module(mod):
snmpd = os.system("which snmpd")
if snmpd:
error_msg = "SNMP not installed - skipping"
pytest.skip(error_msg)
tgen = Topogen(build_topo, mod.__name__)
tgen.start_topology()
for rname, router in tgen.routers().items():
router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
)
router.load_config(
TopoRouter.RD_BGP,
os.path.join(CWD, "{}/bgpd.conf".format(rname)),
"-M snmp",
)
router.load_config(
TopoRouter.RD_SNMP,
os.path.join(CWD, "{}/snmpd.conf".format(rname)),
"-Le -Ivacm_conf,usmConf,iquery -V -DAgentX",
)
tgen.start_router()
def teardown_module(mod):
tgen = get_topogen()
tgen.stop_topology()
def test_bgp_snmp_bgp4v2():
tgen = get_topogen()
r2 = tgen.gears["r2"]
def _bgp_converge_summary():
output = json.loads(r2.vtysh_cmd("show bgp summary json"))
expected = {
"ipv4Unicast": {
"peers": {
"192.168.12.1": {
"state": "Established",
"pfxRcd": 2,
}
}
},
"ipv6Unicast": {
"peers": {
"2001:db8::12:1": {
"state": "Established",
"pfxRcd": 2,
}
}
},
}
return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_converge_summary)
_, result = topotest.run_and_expect(test_func, None, count=30, wait=1)
assert result is None, "Can't see connections established"
def _bgp_converge_prefixes():
output = json.loads(r2.vtysh_cmd("show bgp all json"))
expected = {
"ipv4Unicast": {
"routes": {
"10.0.0.0/31": [
{
"metric": 1,
"origin": "IGP",
}
],
"10.0.0.2/32": [
{
"metric": 2,
"origin": "incomplete",
}
],
}
},
"ipv6Unicast": {
"routes": {
"2001:db8::1/128": [
{
"metric": 1,
"origin": "IGP",
}
],
"2001:db8:1::/56": [
{
"metric": 2,
"origin": "incomplete",
}
],
}
},
}
return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_converge_prefixes)
_, result = topotest.run_and_expect(test_func, None, count=30, wait=1)
assert result is None, "Can't see prefixes from R1"
snmp = SnmpTester(r2, "localhost", "public", "2c", "-Ln -On")
def _snmpwalk_remote_addr():
expected = {
"1.3.6.1.3.5.1.1.2.1.5.1.1.192.168.12.1": "C0 A8 0C 01",
"1.3.6.1.3.5.1.1.2.1.5.1.2.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",
}
# bgp4V2PeerRemoteAddr
output, _ = snmp.walk(".1.3.6.1.3.5.1.1.2.1.5")
return output == expected
_, result = topotest.run_and_expect(_snmpwalk_remote_addr, True, count=10, wait=1)
assertmsg = "Can't fetch SNMP for bgp4V2PeerRemoteAddr"
assert result, assertmsg
def _snmpwalk_peer_state():
expected = {
"1.3.6.1.3.5.1.1.2.1.13.1.1.192.168.12.1": "6",
"1.3.6.1.3.5.1.1.2.1.13.1.2.32.1.13.184.0.0.0.0.0.0.0.0.0.18.0.1": "6",
}
# bgp4V2PeerState
output, _ = snmp.walk(".1.3.6.1.3.5.1.1.2.1.13")
return output == expected
_, result = topotest.run_and_expect(_snmpwalk_peer_state, True, count=10, wait=1)
assertmsg = "Can't fetch SNMP for bgp4V2PeerState"
assert result, assertmsg
def _snmpwalk_peer_last_error_code_received():
expected = {
"1.3.6.1.3.5.1.1.3.1.1.1.1.192.168.12.1": "0",
"1.3.6.1.3.5.1.1.3.1.1.1.2.32.1.13.184.0.0.0.0.0.0.0.0.0.18.0.1": "0",
}
# bgp4V2PeerLastErrorCodeReceived
output, _ = snmp.walk(".1.3.6.1.3.5.1.1.3.1.1")
return output == expected
_, result = topotest.run_and_expect(
_snmpwalk_peer_last_error_code_received, True, count=10, wait=1
)
assertmsg = "Can't fetch SNMP for bgp4V2PeerLastErrorCodeReceived"
assert result, assertmsg
def _snmpwalk_origin():
expected = {
"1.3.6.1.3.5.1.1.9.1.9.1.1.1.1.10.0.0.0.31.1.192.168.12.1.1": "1",
"1.3.6.1.3.5.1.1.9.1.9.1.1.1.1.10.0.0.2.32.1.192.168.12.1.1": "3",
"1.3.6.1.3.5.1.1.9.1.9.1.2.1.2.32.1.13.184.0.0.0.0.0.0.0.0.0.0.0.1.128.2.32.1.13.184.0.0.0.0.0.0.0.0.0.18.0.1.1": "1",
"1.3.6.1.3.5.1.1.9.1.9.1.2.1.2.32.1.13.184.0.1.0.0.0.0.0.0.0.0.0.0.56.2.32.1.13.184.0.0.0.0.0.0.0.0.0.18.0.1.1": "3",
}
# bgp4V2NlriOrigin
output, _ = snmp.walk(".1.3.6.1.3.5.1.1.9.1.9")
return output == expected
_, result = topotest.run_and_expect(_snmpwalk_origin, True, count=10, wait=1)
assertmsg = "Can't fetch SNMP for bgp4V2NlriOrigin"
assert result, assertmsg
def _snmpwalk_med():
expected = {
"1.3.6.1.3.5.1.1.9.1.17.1.1.1.1.10.0.0.0.31.1.192.168.12.1.1": "1",
"1.3.6.1.3.5.1.1.9.1.17.1.1.1.1.10.0.0.2.32.1.192.168.12.1.1": "2",
"1.3.6.1.3.5.1.1.9.1.17.1.2.1.2.32.1.13.184.0.0.0.0.0.0.0.0.0.0.0.1.128.2.32.1.13.184.0.0.0.0.0.0.0.0.0.18.0.1.1": "1",
"1.3.6.1.3.5.1.1.9.1.17.1.2.1.2.32.1.13.184.0.1.0.0.0.0.0.0.0.0.0.0.56.2.32.1.13.184.0.0.0.0.0.0.0.0.0.18.0.1.1": "2",
}
# bgp4V2NlriMed
output, _ = snmp.walk(".1.3.6.1.3.5.1.1.9.1.17")
# tgen.mininet_cli()
return output == expected
_, result = topotest.run_and_expect(_snmpwalk_med, True, count=10, wait=1)
assertmsg = "Can't fetch SNMP for bgp4V2NlriMed"
assert result, assertmsg
def test_memory_leak():
"Run the memory leak test and report results."
tgen = get_topogen()
if not tgen.is_memleak_enabled():
pytest.skip("Memory leak test/report is disabled")
tgen.report_memory_leaks()
if __name__ == "__main__":
args = ["-s"] + sys.argv[1:]
sys.exit(pytest.main(args))
|
