]>
git.proxmox.com Git - mirror_frr.git/blob - tests/topotests/ldp_snmp/test_ldp_snmp_topo1.py
4 # test_ldp_isis_topo1.py
5 # Part of NetDEF Topology Tests
7 # Copyright (c) 2020 by Volta Networks
9 # Permission to use, copy, modify, and/or distribute this software
10 # for any purpose with or without fee is hereby granted, provided
11 # that the above copyright notice and this permission notice appear
14 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
15 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
16 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
17 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
18 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
19 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
20 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
25 test_ldp_vpls_topo1.py:
27 +---------+ +---------+
31 +---------+ +---------+
32 ce1-eth0 (172.16.1.1/24)| |ce2-eth0 (172.16.1.2/24)
36 +---------+ 10.0.1.0/24 +---------+
38 | RT1 +----------------+ RT2 |
39 | 1.1.1.1 | rt2-eth1| 2.2.2.2 |
41 +---------+ +---------+
45 10.0.2.0/24| +---------+ |10.0.3.0/24
48 +--------+ 3.3.3.3 +-------+
54 ce3-eth0 (172.16.1.3/24)|
66 from functools
import partial
68 # Save the Current Working Directory to find configuration files.
69 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
70 sys
.path
.append(os
.path
.join(CWD
, "../"))
72 # pylint: disable=C0413
73 # Import topogen and topotest helpers
74 from lib
import topotest
75 from lib
.topogen
import Topogen
, TopoRouter
, get_topogen
76 from lib
.topolog
import logger
77 from lib
.snmptest
import SnmpTester
79 # Required to instantiate the topology builder class.
81 pytestmark
= [pytest
.mark
.ldpd
, pytest
.mark
.isisd
, pytest
.mark
.snmp
]
90 for router
in ["ce1", "ce2", "ce3", "r1", "r2", "r3"]:
91 tgen
.add_router(router
)
96 switch
= tgen
.add_switch("s1")
97 switch
.add_link(tgen
.gears
["ce1"])
98 switch
.add_link(tgen
.gears
["r1"])
100 switch
= tgen
.add_switch("s2")
101 switch
.add_link(tgen
.gears
["ce2"])
102 switch
.add_link(tgen
.gears
["r2"])
104 switch
= tgen
.add_switch("s3")
105 switch
.add_link(tgen
.gears
["ce3"])
106 switch
.add_link(tgen
.gears
["r3"])
108 switch
= tgen
.add_switch("s4")
109 switch
.add_link(tgen
.gears
["r1"])
110 switch
.add_link(tgen
.gears
["r2"])
112 switch
= tgen
.add_switch("s5")
113 switch
.add_link(tgen
.gears
["r1"])
114 switch
.add_link(tgen
.gears
["r3"])
116 switch
= tgen
.add_switch("s6")
117 switch
.add_link(tgen
.gears
["r2"])
118 switch
.add_link(tgen
.gears
["r3"])
121 def setup_module(mod
):
122 "Sets up the pytest environment"
123 tgen
= Topogen(build_topo
, mod
.__name
__)
124 tgen
.start_topology()
126 router_list
= tgen
.routers()
128 # For all registered routers, load the zebra configuration file
129 for rname
, router
in router_list
.items():
131 TopoRouter
.RD_ZEBRA
, os
.path
.join(CWD
, "{}/zebra.conf".format(rname
))
133 # Don't start isisd and ldpd in the CE nodes
134 if router
.name
[0] == "r":
136 TopoRouter
.RD_ISIS
, os
.path
.join(CWD
, "{}/isisd.conf".format(rname
))
140 os
.path
.join(CWD
, "{}/ldpd.conf".format(rname
)),
145 os
.path
.join(CWD
, "{}/snmpd.conf".format(rname
)),
146 "-Le -Ivacm_conf,usmConf,iquery -V -DAgentX,trap",
152 def teardown_module(mod
):
153 "Teardown the pytest environment"
156 # This function tears down the whole topology.
160 def router_compare_json_output(rname
, command
, reference
):
161 "Compare router JSON output"
163 logger
.info('Comparing router "%s" "%s" output', rname
, command
)
166 filename
= "{}/{}/{}".format(CWD
, rname
, reference
)
167 expected
= json
.loads(open(filename
).read())
169 # Run test function until we get an result.
170 test_func
= partial(topotest
.router_json_cmp
, tgen
.gears
[rname
], command
, expected
)
171 _
, diff
= topotest
.run_and_expect(test_func
, None, count
=320, wait
=0.5)
172 assertmsg
= '"{}" JSON output mismatches the expected result'.format(rname
)
173 assert diff
is None, assertmsg
176 def test_isis_convergence():
177 logger
.info("Test: check ISIS adjacencies")
180 # Skip if previous fatal error condition is raised
181 if tgen
.routers_have_failure():
182 pytest
.skip(tgen
.errors
)
184 for rname
in ["r1", "r2", "r3"]:
185 router_compare_json_output(
187 "show yang operational-data /frr-interface:lib isisd",
188 "show_yang_interface_isis_adjacencies.ref",
193 logger
.info("Test: verify RIB")
196 # Skip if previous fatal error condition is raised
197 # TODO: disabling this check to avoid 'snmpd not running' errors
198 # if tgen.routers_have_failure():
199 # pytest.skip(tgen.errors)
201 for rname
in ["r1", "r2", "r3"]:
202 router_compare_json_output(rname
, "show ip route json", "show_ip_route.ref")
205 def test_ldp_adjacencies():
206 logger
.info("Test: verify LDP adjacencies")
209 # Skip if previous fatal error condition is raised
210 # TODO: disabling this check to avoid 'snmpd not running' errors
211 # if tgen.routers_have_failure():
212 # pytest.skip(tgen.errors)
214 for rname
in ["r1", "r2", "r3"]:
215 router_compare_json_output(
216 rname
, "show mpls ldp discovery json", "show_ldp_discovery.ref"
220 def test_ldp_neighbors():
221 logger
.info("Test: verify LDP neighbors")
224 # Skip if previous fatal error condition is raised
225 # if tgen.routers_have_failure():
226 # pytest.skip(tgen.errors)
228 for rname
in ["r1", "r2", "r3"]:
229 router_compare_json_output(
230 rname
, "show mpls ldp neighbor json", "show_ldp_neighbor.ref"
234 def test_r1_ldp_lsr_objects():
235 "Test mplsLdpLsrObjects objects"
238 r1
= tgen
.gears
["r1"]
239 r1_snmp
= SnmpTester(r1
, "1.1.1.1", "public", "2c")
241 assert r1_snmp
.test_oid("mplsLdpLsrId", "01 01 01 01")
242 assert r1_snmp
.test_oid("mplsLdpLsrLoopDetectionCapable", "none(1)")
245 def test_r1_ldp_entity_table():
246 "Test mplsLdpEntityTable"
249 r1
= tgen
.gears
["r1"]
250 r1_snmp
= SnmpTester(r1
, "1.1.1.1", "public", "2c")
252 assert r1_snmp
.test_oid_walk("mplsLdpEntityLdpId", ["1.1.1.1:0"])
253 assert r1_snmp
.test_oid_walk("mplsLdpEntityIndex", ["1"])
254 assert r1_snmp
.test_oid_walk("mplsLdpEntityProtocolVersion", ["1"])
255 assert r1_snmp
.test_oid_walk("mplsLdpEntityAdminStatus", ["enable(1)"])
256 assert r1_snmp
.test_oid_walk("mplsLdpEntityOperStatus", ["enabled(2)"])
257 assert r1_snmp
.test_oid_walk("mplsLdpEntityTcpPort", ["646"])
258 assert r1_snmp
.test_oid_walk("mplsLdpEntityUdpDscPort", ["646"])
259 assert r1_snmp
.test_oid_walk("mplsLdpEntityMaxPduLength", ["4096 octets"])
260 assert r1_snmp
.test_oid_walk("mplsLdpEntityKeepAliveHoldTimer", ["180 seconds"])
261 assert r1_snmp
.test_oid_walk("mplsLdpEntityHelloHoldTimer", ["0 seconds"])
262 assert r1_snmp
.test_oid_walk("mplsLdpEntityInitSessionThreshold", ["0"])
263 assert r1_snmp
.test_oid_walk(
264 "mplsLdpEntityLabelDistMethod", ["downstreamUnsolicited(2)"]
266 assert r1_snmp
.test_oid_walk("mplsLdpEntityLabelRetentionMode", ["liberal(2)"])
267 assert r1_snmp
.test_oid_walk("mplsLdpEntityPathVectorLimit", ["0"])
268 assert r1_snmp
.test_oid_walk("mplsLdpEntityHopCountLimit", ["0"])
269 assert r1_snmp
.test_oid_walk("mplsLdpEntityTransportAddrKind", ["loopback(2)"])
270 assert r1_snmp
.test_oid_walk("mplsLdpEntityTargetPeer", ["true(1)"])
271 assert r1_snmp
.test_oid_walk("mplsLdpEntityTargetPeerAddrType", ["ipv4(1)"])
272 assert r1_snmp
.test_oid_walk("mplsLdpEntityTargetPeerAddr", ["01 01 01 01"])
273 assert r1_snmp
.test_oid_walk("mplsLdpEntityLabelType", ["generic(1)"])
274 assert r1_snmp
.test_oid_walk("mplsLdpEntityDiscontinuityTime", ["(0) 0:00:00.00"])
275 assert r1_snmp
.test_oid_walk("mplsLdpEntityStorageType", ["nonVolatile(3)"])
276 assert r1_snmp
.test_oid_walk("mplsLdpEntityRowStatus", ["createAndGo(4)"])
279 def test_r1_ldp_entity_stats_table():
280 "Test mplsLdpEntityStatsTable"
283 r1
= tgen
.gears
["r1"]
284 r1_snmp
= SnmpTester(r1
, "1.1.1.1", "public", "2c")
286 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsSessionAttempts", ["0"])
287 assert r1_snmp
.test_oid_walk(
288 "mplsLdpEntityStatsSessionRejectedNoHelloErrors", ["0"]
290 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsSessionRejectedAdErrors", ["0"])
291 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsSessionRejectedMaxPduErrors", ["0"])
292 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsSessionRejectedLRErrors", ["0"])
293 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsBadLdpIdentifierErrors", ["0"])
294 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsBadPduLengthErrors", ["0"])
295 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsBadMessageLengthErrors", ["0"])
296 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsBadTlvLengthErrors", ["0"])
297 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsMalformedTlvValueErrors", ["0"])
298 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsKeepAliveTimerExpErrors", ["0"])
299 assert r1_snmp
.test_oid_walk(
300 "mplsLdpEntityStatsShutdownReceivedNotifications", ["0"]
302 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsShutdownSentNotifications", ["0"])
305 def test_r1_ldp_peer_table():
306 "Test mplsLdpPeerTable"
309 r1
= tgen
.gears
["r1"]
310 r1_snmp
= SnmpTester(r1
, "1.1.1.1", "public", "2c")
312 assert r1_snmp
.test_oid_walk("mplsLdpPeerLdpId", ["2.2.2.2:0", "3.3.3.3:0"])
313 assert r1_snmp
.test_oid_walk(
314 "mplsLdpPeerLabelDistMethod",
315 ["downstreamUnsolicited(2)", "downstreamUnsolicited(2)"],
317 assert r1_snmp
.test_oid_walk("mplsLdpPeerPathVectorLimit", ["0", "0"])
318 assert r1_snmp
.test_oid_walk("mplsLdpPeerTransportAddrType", ["ipv4(1)", "ipv4(1)"])
319 assert r1_snmp
.test_oid_walk(
320 "mplsLdpPeerTransportAddr", ["02 02 02 02", "03 03 03 03"]
324 def test_r1_ldp_session_table():
325 "Test mplsLdpSessionTable"
328 r1
= tgen
.gears
["r1"]
329 r1_snmp
= SnmpTester(r1
, "1.1.1.1", "public", "2c")
331 assert r1_snmp
.test_oid_walk(
332 "mplsLdpSessionState", ["operational(5)", "operational(5)"]
334 assert r1_snmp
.test_oid_walk("mplsLdpSessionRole", ["passive(3)", "passive(3)"])
335 assert r1_snmp
.test_oid_walk("mplsLdpSessionProtocolVersion", ["1", "1"])
336 assert r1_snmp
.test_oid_walk(
337 "mplsLdpSessionKeepAliveTime", ["180 seconds", "180 seconds"]
339 assert r1_snmp
.test_oid_walk(
340 "mplsLdpSessionMaxPduLength", ["4096 octets", "4096 octets"]
342 assert r1_snmp
.test_oid_walk(
343 "mplsLdpSessionDiscontinuityTime", ["(0) 0:00:00.00", "(0) 0:00:00.00"]
347 def test_r1_ldp_session_stats_table():
348 "Test mplsLdpSessionStatsTable"
351 r1
= tgen
.gears
["r1"]
352 r1_snmp
= SnmpTester(r1
, "1.1.1.1", "public", "2c")
354 assert r1_snmp
.test_oid_walk("mplsLdpSessionStatsUnknownMesTypeErrors", ["0", "0"])
355 assert r1_snmp
.test_oid_walk("mplsLdpSessionStatsUnknownTlvErrors", ["0", "0"])
358 def test_r1_ldp_hello_adjacency_table():
359 "Test mplsLdpHelloAdjacencyTable"
362 r1
= tgen
.gears
["r1"]
363 r1_snmp
= SnmpTester(r1
, "1.1.1.1", "public", "2c")
365 assert r1_snmp
.test_oid_walk("mplsLdpHelloAdjacencyIndex", ["1", "2", "1"])
366 assert r1_snmp
.test_oid_walk("mplsLdpHelloAdjacencyHoldTime", ["15", "45", "15"])
367 assert r1_snmp
.test_oid_walk(
368 "mplsLdpHelloAdjacencyType", ["link(1)", "targeted(2)", "link(1)"]
372 # Memory leak test template
373 # disabling memory leak
374 def test_memory_leak():
375 "Run the memory leak test and report results."
377 if not tgen
.is_memleak_enabled():
378 pytest
.skip("Memory leak test/report is disabled")
380 tgen
.report_memory_leaks()
383 if __name__
== "__main__":
384 args
= ["-s"] + sys
.argv
[1:]
385 sys
.exit(pytest
.main(args
))