]>
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)|
67 from time
import sleep
68 from functools
import partial
70 # Save the Current Working Directory to find configuration files.
71 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
72 sys
.path
.append(os
.path
.join(CWD
, "../"))
74 # pylint: disable=C0413
75 # Import topogen and topotest helpers
76 from lib
import topotest
77 from lib
.topogen
import Topogen
, TopoRouter
, get_topogen
78 from lib
.topolog
import logger
79 from lib
.snmptest
import SnmpTester
81 # Required to instantiate the topology builder class.
82 from mininet
.topo
import Topo
84 pytestmark
= [pytest
.mark
.ldpd
, pytest
.mark
.isisd
, pytest
.mark
.snmp
]
86 class TemplateTopo(Topo
):
87 "Test topology builder"
89 def build(self
, *_args
, **_opts
):
91 tgen
= get_topogen(self
)
96 for router
in ["ce1", "ce2", "ce3", "r1", "r2", "r3"]:
97 tgen
.add_router(router
)
102 switch
= tgen
.add_switch("s1")
103 switch
.add_link(tgen
.gears
["ce1"])
104 switch
.add_link(tgen
.gears
["r1"])
106 switch
= tgen
.add_switch("s2")
107 switch
.add_link(tgen
.gears
["ce2"])
108 switch
.add_link(tgen
.gears
["r2"])
110 switch
= tgen
.add_switch("s3")
111 switch
.add_link(tgen
.gears
["ce3"])
112 switch
.add_link(tgen
.gears
["r3"])
114 switch
= tgen
.add_switch("s4")
115 switch
.add_link(tgen
.gears
["r1"])
116 switch
.add_link(tgen
.gears
["r2"])
118 switch
= tgen
.add_switch("s5")
119 switch
.add_link(tgen
.gears
["r1"])
120 switch
.add_link(tgen
.gears
["r3"])
122 switch
= tgen
.add_switch("s6")
123 switch
.add_link(tgen
.gears
["r2"])
124 switch
.add_link(tgen
.gears
["r3"])
127 def setup_module(mod
):
128 "Sets up the pytest environment"
129 tgen
= Topogen(TemplateTopo
, mod
.__name
__)
130 tgen
.start_topology()
132 router_list
= tgen
.routers()
134 # For all registered routers, load the zebra configuration file
135 for rname
, router
in router_list
.items():
137 TopoRouter
.RD_ZEBRA
, os
.path
.join(CWD
, "{}/zebra.conf".format(rname
))
139 # Don't start isisd and ldpd in the CE nodes
140 if router
.name
[0] == "r":
142 TopoRouter
.RD_ISIS
, os
.path
.join(CWD
, "{}/isisd.conf".format(rname
))
146 os
.path
.join(CWD
, "{}/ldpd.conf".format(rname
)),
151 os
.path
.join(CWD
, "{}/snmpd.conf".format(rname
)),
152 "-Le -Ivacm_conf,usmConf,iquery -V -DAgentX,trap",
158 def teardown_module(mod
):
159 "Teardown the pytest environment"
162 # This function tears down the whole topology.
166 def router_compare_json_output(rname
, command
, reference
):
167 "Compare router JSON output"
169 logger
.info('Comparing router "%s" "%s" output', rname
, command
)
172 filename
= "{}/{}/{}".format(CWD
, rname
, reference
)
173 expected
= json
.loads(open(filename
).read())
175 # Run test function until we get an result.
176 test_func
= partial(topotest
.router_json_cmp
, tgen
.gears
[rname
], command
, expected
)
177 _
, diff
= topotest
.run_and_expect(test_func
, None, count
=320, wait
=0.5)
178 assertmsg
= '"{}" JSON output mismatches the expected result'.format(rname
)
179 assert diff
is None, assertmsg
182 def test_isis_convergence():
183 logger
.info("Test: check ISIS adjacencies")
186 # Skip if previous fatal error condition is raised
187 if tgen
.routers_have_failure():
188 pytest
.skip(tgen
.errors
)
190 for rname
in ["r1", "r2", "r3"]:
191 router_compare_json_output(
193 "show yang operational-data /frr-interface:lib isisd",
194 "show_yang_interface_isis_adjacencies.ref",
199 logger
.info("Test: verify RIB")
202 # Skip if previous fatal error condition is raised
203 # TODO: disabling this check to avoid 'snmpd not running' errors
204 # if tgen.routers_have_failure():
205 # pytest.skip(tgen.errors)
207 for rname
in ["r1", "r2", "r3"]:
208 router_compare_json_output(rname
, "show ip route json", "show_ip_route.ref")
211 def test_ldp_adjacencies():
212 logger
.info("Test: verify LDP adjacencies")
215 # Skip if previous fatal error condition is raised
216 # TODO: disabling this check to avoid 'snmpd not running' errors
217 # if tgen.routers_have_failure():
218 # pytest.skip(tgen.errors)
220 for rname
in ["r1", "r2", "r3"]:
221 router_compare_json_output(
222 rname
, "show mpls ldp discovery json", "show_ldp_discovery.ref"
226 def test_ldp_neighbors():
227 logger
.info("Test: verify LDP neighbors")
230 # Skip if previous fatal error condition is raised
231 # if tgen.routers_have_failure():
232 # pytest.skip(tgen.errors)
234 for rname
in ["r1", "r2", "r3"]:
235 router_compare_json_output(
236 rname
, "show mpls ldp neighbor json", "show_ldp_neighbor.ref"
240 def test_r1_ldp_lsr_objects():
241 "Test mplsLdpLsrObjects objects"
244 r1
= tgen
.net
.get("r1")
245 r1_snmp
= SnmpTester(r1
, "1.1.1.1", "public", "2c")
247 assert r1_snmp
.test_oid("mplsLdpLsrId", "01 01 01 01")
248 assert r1_snmp
.test_oid("mplsLdpLsrLoopDetectionCapable", "none(1)")
251 def test_r1_ldp_entity_table():
252 "Test mplsLdpEntityTable"
255 r1
= tgen
.net
.get("r1")
256 r1_snmp
= SnmpTester(r1
, "1.1.1.1", "public", "2c")
258 assert r1_snmp
.test_oid_walk("mplsLdpEntityLdpId", ["1.1.1.1:0"])
259 assert r1_snmp
.test_oid_walk("mplsLdpEntityIndex", ["1"])
260 assert r1_snmp
.test_oid_walk("mplsLdpEntityProtocolVersion", ["1"])
261 assert r1_snmp
.test_oid_walk("mplsLdpEntityAdminStatus", ["enable(1)"])
262 assert r1_snmp
.test_oid_walk("mplsLdpEntityOperStatus", ["enabled(2)"])
263 assert r1_snmp
.test_oid_walk("mplsLdpEntityTcpPort", ["646"])
264 assert r1_snmp
.test_oid_walk("mplsLdpEntityUdpDscPort", ["646"])
265 assert r1_snmp
.test_oid_walk("mplsLdpEntityMaxPduLength", ["4096 octets"])
266 assert r1_snmp
.test_oid_walk("mplsLdpEntityKeepAliveHoldTimer", ["180 seconds"])
267 assert r1_snmp
.test_oid_walk("mplsLdpEntityHelloHoldTimer", ["0 seconds"])
268 assert r1_snmp
.test_oid_walk("mplsLdpEntityInitSessionThreshold", ["0"])
269 assert r1_snmp
.test_oid_walk(
270 "mplsLdpEntityLabelDistMethod", ["downstreamUnsolicited(2)"]
272 assert r1_snmp
.test_oid_walk("mplsLdpEntityLabelRetentionMode", ["liberal(2)"])
273 assert r1_snmp
.test_oid_walk("mplsLdpEntityPathVectorLimit", ["0"])
274 assert r1_snmp
.test_oid_walk("mplsLdpEntityHopCountLimit", ["0"])
275 assert r1_snmp
.test_oid_walk("mplsLdpEntityTransportAddrKind", ["loopback(2)"])
276 assert r1_snmp
.test_oid_walk("mplsLdpEntityTargetPeer", ["true(1)"])
277 assert r1_snmp
.test_oid_walk("mplsLdpEntityTargetPeerAddrType", ["ipv4(1)"])
278 assert r1_snmp
.test_oid_walk("mplsLdpEntityTargetPeerAddr", ["01 01 01 01"])
279 assert r1_snmp
.test_oid_walk("mplsLdpEntityLabelType", ["generic(1)"])
280 assert r1_snmp
.test_oid_walk("mplsLdpEntityDiscontinuityTime", ["(0) 0:00:00.00"])
281 assert r1_snmp
.test_oid_walk("mplsLdpEntityStorageType", ["nonVolatile(3)"])
282 assert r1_snmp
.test_oid_walk("mplsLdpEntityRowStatus", ["createAndGo(4)"])
285 def test_r1_ldp_entity_stats_table():
286 "Test mplsLdpEntityStatsTable"
289 r1
= tgen
.net
.get("r1")
290 r1_snmp
= SnmpTester(r1
, "1.1.1.1", "public", "2c")
292 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsSessionAttempts", ["0"])
293 assert r1_snmp
.test_oid_walk(
294 "mplsLdpEntityStatsSessionRejectedNoHelloErrors", ["0"]
296 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsSessionRejectedAdErrors", ["0"])
297 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsSessionRejectedMaxPduErrors", ["0"])
298 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsSessionRejectedLRErrors", ["0"])
299 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsBadLdpIdentifierErrors", ["0"])
300 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsBadPduLengthErrors", ["0"])
301 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsBadMessageLengthErrors", ["0"])
302 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsBadTlvLengthErrors", ["0"])
303 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsMalformedTlvValueErrors", ["0"])
304 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsKeepAliveTimerExpErrors", ["0"])
305 assert r1_snmp
.test_oid_walk(
306 "mplsLdpEntityStatsShutdownReceivedNotifications", ["0"]
308 assert r1_snmp
.test_oid_walk("mplsLdpEntityStatsShutdownSentNotifications", ["0"])
311 def test_r1_ldp_peer_table():
312 "Test mplsLdpPeerTable"
315 r1
= tgen
.net
.get("r1")
316 r1_snmp
= SnmpTester(r1
, "1.1.1.1", "public", "2c")
318 assert r1_snmp
.test_oid_walk("mplsLdpPeerLdpId", ["2.2.2.2:0", "3.3.3.3:0"])
319 assert r1_snmp
.test_oid_walk(
320 "mplsLdpPeerLabelDistMethod",
321 ["downstreamUnsolicited(2)", "downstreamUnsolicited(2)"],
323 assert r1_snmp
.test_oid_walk("mplsLdpPeerPathVectorLimit", ["0", "0"])
324 assert r1_snmp
.test_oid_walk("mplsLdpPeerTransportAddrType", ["ipv4(1)", "ipv4(1)"])
325 assert r1_snmp
.test_oid_walk(
326 "mplsLdpPeerTransportAddr", ["02 02 02 02", "03 03 03 03"]
330 def test_r1_ldp_session_table():
331 "Test mplsLdpSessionTable"
334 r1
= tgen
.net
.get("r1")
335 r1_snmp
= SnmpTester(r1
, "1.1.1.1", "public", "2c")
337 assert r1_snmp
.test_oid_walk(
338 "mplsLdpSessionState", ["operational(5)", "operational(5)"]
340 assert r1_snmp
.test_oid_walk("mplsLdpSessionRole", ["passive(3)", "passive(3)"])
341 assert r1_snmp
.test_oid_walk("mplsLdpSessionProtocolVersion", ["1", "1"])
342 assert r1_snmp
.test_oid_walk(
343 "mplsLdpSessionKeepAliveTime", ["180 seconds", "180 seconds"]
345 assert r1_snmp
.test_oid_walk(
346 "mplsLdpSessionMaxPduLength", ["4096 octets", "4096 octets"]
348 assert r1_snmp
.test_oid_walk(
349 "mplsLdpSessionDiscontinuityTime", ["(0) 0:00:00.00", "(0) 0:00:00.00"]
353 def test_r1_ldp_session_stats_table():
354 "Test mplsLdpSessionStatsTable"
357 r1
= tgen
.net
.get("r1")
358 r1_snmp
= SnmpTester(r1
, "1.1.1.1", "public", "2c")
360 assert r1_snmp
.test_oid_walk("mplsLdpSessionStatsUnknownMesTypeErrors", ["0", "0"])
361 assert r1_snmp
.test_oid_walk("mplsLdpSessionStatsUnknownTlvErrors", ["0", "0"])
364 def test_r1_ldp_hello_adjacency_table():
365 "Test mplsLdpHelloAdjacencyTable"
368 r1
= tgen
.net
.get("r1")
369 r1_snmp
= SnmpTester(r1
, "1.1.1.1", "public", "2c")
371 assert r1_snmp
.test_oid_walk("mplsLdpHelloAdjacencyIndex", ["1", "2", "1"])
372 assert r1_snmp
.test_oid_walk("mplsLdpHelloAdjacencyHoldTime", ["15", "45", "15"])
373 assert r1_snmp
.test_oid_walk(
374 "mplsLdpHelloAdjacencyType", ["link(1)", "targeted(2)", "link(1)"]
378 # Memory leak test template
379 # disabling memory leak
380 def test_memory_leak():
381 "Run the memory leak test and report results."
383 if not tgen
.is_memleak_enabled():
384 pytest
.skip("Memory leak test/report is disabled")
386 tgen
.report_memory_leaks()
389 if __name__
== "__main__":
390 args
= ["-s"] + sys
.argv
[1:]
391 sys
.exit(pytest
.main(args
))