2 # SPDX-License-Identifier: ISC
5 # Copyright (c) 2020 by VMware, Inc. ("VMware")
6 # Used Copyright (c) 2018 by Network Device Education Foundation,
7 # Inc. ("NetDEF") in this file.
11 Following tests are covered to test multicast pim sm:
14 - Create topology (setup module)
17 Following tests are covered:
19 1. TC:48 Verify mroute after configuring black-hole route for RP and source
20 2. TC:49 Verify mroute when RP is reachable using default route
21 3. TC:50 Verify mroute when LHR,FHR,RP and transit routers reachable
23 4. TC:52 Verify PIM nbr after changing interface ip
24 5. TC:53 Verify IGMP interface updated with correct detail after changing interface config
25 6. TC:54 Verify received and transmit hello stats are getting cleared after PIM nbr reset
35 pytestmark
= pytest
.mark
.pimd
37 # Save the Current Working Directory to find configuration files.
38 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
39 sys
.path
.append(os
.path
.join(CWD
, "../"))
40 sys
.path
.append(os
.path
.join(CWD
, "../lib/"))
42 # Required to instantiate the topology builder class.
44 # pylint: disable=C0413
45 # Import topogen and topotest helpers
46 from lib
.topogen
import Topogen
, get_topogen
48 from lib
.common_config
import (
53 reset_config_on_routers
,
54 shutdown_bringup_interface
,
57 required_linux_kernel_version
,
64 clear_pim_interface_traffic
,
68 get_pim_interface_traffic
,
71 from lib
.topolog
import logger
72 from lib
.topojson
import build_config_from_json
73 from time
import sleep
79 i4-----c1-------------c2---i5
82 i1-----l1------r2-----f1---i2
88 i1, i2, i3. i4, i5, i6, i7, i8 - FRR running iperf to send IGMP
99 GROUP_RANGE
= "224.0.0.0/4"
100 IGMP_GROUP
= "225.1.1.1/32"
101 IGMP_JOIN
= "225.1.1.1"
109 IGMP_JOIN_RANGE_1
= ["225.1.1.1", "225.1.1.2", "225.1.1.3", "225.1.1.4", "225.1.1.5"]
110 NEW_ADDRESS_1
= "192.168.20.1"
111 NEW_ADDRESS_2
= "192.168.20.2"
112 NEW_ADDRESS_1_SUBNET
= "192.168.20.1/24"
113 NEW_ADDRESS_2_SUBNET
= "192.168.20.2/24"
116 def setup_module(mod
):
118 Sets up the pytest environment
123 # Required linux kernel version for this suite to run.
124 result
= required_linux_kernel_version("4.19")
125 if result
is not True:
126 pytest
.skip("Kernel version should be >= 4.19")
128 testsuite_run_time
= time
.asctime(time
.localtime(time
.time()))
129 logger
.info("Testsuite start time: {}".format(testsuite_run_time
))
130 logger
.info("=" * 40)
131 logger
.info("Master Topology: \n {}".format(TOPOLOGY
))
133 logger
.info("Running setup_module to create topology")
135 json_file
= "{}/multicast_pim_sm_topo4.json".format(CWD
)
136 tgen
= Topogen(json_file
, mod
.__name
__)
138 topo
= tgen
.json_topo
139 # ... and here it calls Mininet initialization functions.
141 # Starting topology, create tmp files which are loaded to routers
142 # to start daemons and then start routers
145 # Don"t run this test if we have any failure.
146 if tgen
.routers_have_failure():
147 pytest
.skip(tgen
.errors
)
149 # Creating configuration from JSON
150 build_config_from_json(tgen
, topo
)
152 # XXX Replace this using "with McastTesterHelper()... " in each test if possible.
154 app_helper
= McastTesterHelper(tgen
)
156 logger
.info("Running setup_module() done")
159 def teardown_module():
160 """Teardown the pytest environment"""
162 logger
.info("Running teardown_module to delete topology")
168 # Stop toplogy and Remove tmp files
172 "Testsuite end time: {}".format(time
.asctime(time
.localtime(time
.time())))
174 logger
.info("=" * 40)
177 #####################################################
181 #####################################################
184 def reset_stats(stats
):
186 API to reset the stats
190 * `stats` : State dictionary holding helloRx and helloTx values
193 for router
, state_data
in stats
.items():
194 for state
, value
in state_data
.items():
195 stats
[router
][state
] = 0
197 "[DUT: %s]: stats %s value has reset" " reset, Current value: %s",
200 stats
[router
][state
],
206 def verify_state_incremented(state_before
, state_after
):
208 API to compare interface traffic state incrementing
212 * `state_before` : State dictionary for any particular instance
213 * `state_after` : State dictionary for any particular instance
216 for router
, state_data
in state_before
.items():
217 for state
, value
in state_data
.items():
218 if state_before
[router
][state
] >= state_after
[router
][state
]:
220 "[DUT: %s]: state %s value has not"
221 " incremented, Initial value: %s, "
222 "Current value: %s [FAILED!!]"
226 state_before
[router
][state
],
227 state_after
[router
][state
],
233 "[DUT: %s]: State %s value is "
234 "incremented, Initial value: %s, Current value: %s"
238 state_before
[router
][state
],
239 state_after
[router
][state
],
245 def test_mroute_when_RP_reachable_default_route_p2(request
):
247 TC_49 Verify mroute when and source RP is reachable using default route
251 tc_name
= request
.node
.name
252 write_test_header(tc_name
)
254 # Don"t run this test if we have any failure.
255 if tgen
.routers_have_failure():
256 pytest
.skip(tgen
.errors
)
258 # Creating configuration from JSON
259 app_helper
.stop_all_hosts()
261 reset_config_on_routers(tgen
)
262 clear_pim_interface_traffic(tgen
, topo
)
265 "Remove c1-c2 connected link to simulate topo "
266 "c1(FHR)---l1(RP)----r2---f1-----c2(LHR)"
269 intf_c1_c2
= topo
["routers"]["c1"]["links"]["c2"]["interface"]
270 intf_c2_c1
= topo
["routers"]["c2"]["links"]["c1"]["interface"]
271 shutdown_bringup_interface(tgen
, "c1", intf_c1_c2
, False)
272 shutdown_bringup_interface(tgen
, "c2", intf_c2_c1
, False)
274 step("Enable the PIM on all the interfaces of FRR1, FRR2, FRR3")
276 "Enable IGMP of FRR1 interface and send IGMP joins "
277 " from FRR1 node for group range (225.1.1.1-5)"
280 intf_c2_i5
= topo
["routers"]["c2"]["links"]["i5"]["interface"]
282 "c2": {"igmp": {"interfaces": {intf_c2_i5
: {"igmp": {"version": "2"}}}}}
284 result
= create_igmp_config(tgen
, topo
, input_dict
)
285 assert result
is True, "Testcase {}: Failed Error: {}".format(tc_name
, result
)
287 input_join
= {"i5": topo
["routers"]["i5"]["links"]["c2"]["interface"]}
289 for recvr
, recvr_intf
in input_join
.items():
290 result
= app_helper
.run_join(recvr
, IGMP_JOIN_RANGE_1
, join_intf
=recvr_intf
)
291 assert result
is True, "Testcase {}: Failed Error: {}".format(tc_name
, result
)
293 step("Configure static RP for (225.1.1.1-5) as R2")
300 "rp_addr": topo
["routers"]["l1"]["links"]["lo"]["ipv4"].split(
303 "group_addr_range": GROUP_RANGE
,
310 result
= create_pim_config(tgen
, topo
, input_dict
)
311 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
313 step("Send traffic from C1 to all the groups ( 225.1.1.1 to 225.1.1.5)")
315 input_src
= {"i4": topo
["routers"]["i4"]["links"]["c1"]["interface"]}
317 for src
, src_intf
in input_src
.items():
318 result
= app_helper
.run_traffic(src
, IGMP_JOIN_RANGE_1
, bind_intf
=src_intf
)
319 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
321 source_i4
= topo
["routers"]["i4"]["links"]["c1"]["ipv4"].split("/")[0]
327 "iif": topo
["routers"]["c2"]["links"]["f1"]["interface"],
328 "oil": topo
["routers"]["c2"]["links"]["i5"]["interface"],
335 "src_address": source_i4
,
336 "iif": topo
["routers"]["c2"]["links"]["f1"]["interface"],
337 "oil": topo
["routers"]["c2"]["links"]["i5"]["interface"],
341 step("Verify mroutes and iff upstream")
343 for data
in input_dict_sg
:
344 result
= verify_mroutes(
352 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
354 result
= verify_upstream_iif(
355 tgen
, data
["dut"], data
["iif"], data
["src_address"], IGMP_JOIN_RANGE_1
357 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
359 for data
in input_dict_starg
:
360 result
= verify_mroutes(
368 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
370 result
= verify_upstream_iif(
371 tgen
, data
["dut"], data
["iif"], data
["src_address"], IGMP_JOIN_RANGE_1
373 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
375 step("Delete static routes on c2")
380 "network": ["1.0.4.11/32", "10.0.2.1/24", "10.0.1.2/24"],
381 "next_hop": "10.0.3.2",
388 result
= create_static_routes(tgen
, input_dict
)
389 assert result
is True, "Testcase {} :Failed \n Error {}".format(tc_name
, result
)
391 step("Verify RP info unknown after removing static route from c2 ")
393 rp_address
= topo
["routers"]["l1"]["links"]["lo"]["ipv4"].split("/")[0]
395 result
= verify_pim_rp_info(
396 tgen
, topo
, dut
, GROUP_RANGE_1
, "Unknown", rp_address
, SOURCE
398 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
400 step("Verify mroute not present after Delete of static routes on c1")
402 for data
in input_dict_sg
:
403 result
= verify_mroutes(
412 assert result
is not True, (
413 "Testcase {} : Failed \n "
414 "Expected: [{}]: mroute (S, G) should not be present in mroute table \n "
415 "Found: {}".format(tc_name
, data
["dut"], result
)
418 result
= verify_upstream_iif(
426 assert result
is not True, (
427 "Testcase {} : Failed \n "
428 "Expected: [{}]: Upstream IIF interface {} should not be present\n "
429 "Found: {}".format(tc_name
, data
["dut"], data
["iif"], result
)
432 for data
in input_dict_starg
:
433 result
= verify_mroutes(
442 assert result
is not True, (
443 "Testcase {} : Failed \n "
444 "Expected: [{}]: mroute (S, G) should not be present in mroute table \n "
445 "Found: {}".format(tc_name
, data
["dut"], result
)
448 result
= verify_upstream_iif(
456 assert result
is not True, (
457 "Testcase {} : Failed \n "
458 "Expected: [{}]: Upstream IIF interface {} should not be present\n "
459 "Found: {}".format(tc_name
, data
["dut"], data
["iif"], result
)
462 step("Configure default routes on c2")
464 intf_f1_c2
= topo
["routers"]["f1"]["links"]["c2"]["ipv4"].split("/")[0]
467 "c2": {"static_routes": [{"network": "0.0.0.0/0", "next_hop": intf_f1_c2
}]}
469 result
= create_static_routes(tgen
, input_dict
)
470 assert result
is True, "Testcase {} :Failed \n Error {}".format(tc_name
, result
)
472 step("applying ip nht config on c2")
474 raw_config
= {"c2": {"raw_config": ["ip nht resolve-via-default"]}}
476 result
= apply_raw_config(tgen
, raw_config
)
477 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
479 step("Verify RP info is NOT unknown after removing static route from c2 ")
480 result
= verify_pim_rp_info(
481 tgen
, topo
, dut
, GROUP_RANGE_1
, "Unknown", rp_address
, SOURCE
, expected
=False
483 assert result
is not True, (
484 "Testcase {} : Failed \n "
485 "Expected: [{}]: RP Info should not be Unknown after removing static"
487 "Found: {}".format(tc_name
, data
["dut"], result
)
490 step("Verify (s,g) populated after adding default route ")
492 for data
in input_dict_sg
:
493 result
= verify_mroutes(
501 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
503 result
= verify_upstream_iif(
504 tgen
, data
["dut"], data
["iif"], data
["src_address"], IGMP_JOIN_RANGE_1
506 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
508 step("Verify (*,g) populated after adding default route ")
510 for data
in input_dict_starg
:
511 result
= verify_mroutes(
519 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
521 result
= verify_upstream_iif(
522 tgen
, data
["dut"], data
["iif"], data
["src_address"], IGMP_JOIN_RANGE_1
524 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
526 write_test_footer(tc_name
)
529 def test_mroute_with_RP_default_route_all_nodes_p2(request
):
531 TC_50 Verify mroute when LHR,FHR,RP and transit routers reachable
536 tc_name
= request
.node
.name
537 write_test_header(tc_name
)
539 # Don"t run this test if we have any failure.
540 if tgen
.routers_have_failure():
541 pytest
.skip(tgen
.errors
)
543 # Creating configuration from JSON
544 app_helper
.stop_all_hosts()
546 reset_config_on_routers(tgen
)
547 clear_pim_interface_traffic(tgen
, topo
)
550 "Remove c1-c2 connected link to simulate topo "
551 "c1(LHR)---l1(RP)----r2---f1-----c2(FHR)"
554 intf_c1_c2
= topo
["routers"]["c1"]["links"]["c2"]["interface"]
555 intf_c2_c1
= topo
["routers"]["c2"]["links"]["c1"]["interface"]
556 shutdown_bringup_interface(tgen
, "c1", intf_c1_c2
, False)
557 shutdown_bringup_interface(tgen
, "c2", intf_c2_c1
, False)
559 step("Enable the PIM on all the interfaces of FRR1, FRR2, FRR3")
561 "Enable IGMP of FRR1 interface and send IGMP joins "
562 " from FRR1 node for group range (225.1.1.1-5)"
565 intf_c1_i4
= topo
["routers"]["c1"]["links"]["i4"]["interface"]
567 "c1": {"igmp": {"interfaces": {intf_c1_i4
: {"igmp": {"version": "2"}}}}}
569 result
= create_igmp_config(tgen
, topo
, input_dict
)
570 assert result
is True, "Testcase {}: Failed Error: {}".format(tc_name
, result
)
572 input_join
= {"i4": topo
["routers"]["i4"]["links"]["c1"]["interface"]}
574 for recvr
, recvr_intf
in input_join
.items():
575 result
= app_helper
.run_join(recvr
, IGMP_JOIN_RANGE_1
, join_intf
=recvr_intf
)
576 assert result
is True, "Testcase {}: Failed Error: {}".format(tc_name
, result
)
578 step("Configure static RP for (225.1.1.1-5) as R2")
585 "rp_addr": topo
["routers"]["l1"]["links"]["lo"]["ipv4"].split(
588 "group_addr_range": GROUP_RANGE
,
595 result
= create_pim_config(tgen
, topo
, input_dict
)
596 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
598 step("Send traffic from C2 to all the groups ( 225.1.1.1 to 225.1.1.5)")
600 input_src
= {"i5": topo
["routers"]["i5"]["links"]["c2"]["interface"]}
602 for src
, src_intf
in input_src
.items():
603 result
= app_helper
.run_traffic(src
, IGMP_JOIN_RANGE_1
, bind_intf
=src_intf
)
604 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
606 source_i5
= topo
["routers"]["i5"]["links"]["c2"]["ipv4"].split("/")[0]
612 "iif": topo
["routers"]["c1"]["links"]["l1"]["interface"],
613 "oil": topo
["routers"]["c1"]["links"]["i4"]["interface"],
620 "src_address": source_i5
,
621 "iif": topo
["routers"]["c1"]["links"]["l1"]["interface"],
622 "oil": topo
["routers"]["c1"]["links"]["i4"]["interface"],
626 step("Verify mroutes and iff upstream")
628 for data
in input_dict_sg
:
629 result
= verify_mroutes(
637 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
639 result
= verify_upstream_iif(
640 tgen
, data
["dut"], data
["iif"], data
["src_address"], IGMP_JOIN_RANGE_1
642 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
644 for data
in input_dict_starg
:
645 result
= verify_mroutes(
653 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
655 result
= verify_upstream_iif(
656 tgen
, data
["dut"], data
["iif"], data
["src_address"], IGMP_JOIN_RANGE_1
658 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
660 step("Delete static routes RP on all the nodes")
664 {"network": ["1.0.4.11/32"], "next_hop": "10.0.3.2", "delete": True}
669 {"network": ["1.0.4.11/32"], "next_hop": "10.0.2.2", "delete": True}
674 {"network": ["1.0.4.11/32"], "next_hop": "10.0.12.1", "delete": True}
679 {"network": ["1.0.4.11/32"], "next_hop": "10.0.7.2", "delete": True}
684 result
= create_static_routes(tgen
, input_dict
)
685 assert result
is True, "Testcase {} :Failed \n Error {}".format(tc_name
, result
)
687 step("Verify RP info unknown after removing static route from c2 ")
689 rp_address
= topo
["routers"]["l1"]["links"]["lo"]["ipv4"].split("/")[0]
691 result
= verify_pim_rp_info(
692 tgen
, topo
, dut
, GROUP_RANGE_1
, "Unknown", rp_address
, SOURCE
694 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
696 for data
in input_dict_starg
:
697 result
= verify_mroutes(
706 assert result
is not True, (
707 "Testcase {} : Failed \n "
708 "Expected: [{}]: mroute (S, G) should not be present in mroute table \n "
709 "Found: {}".format(tc_name
, data
["dut"], result
)
712 result
= verify_upstream_iif(
720 assert result
is not True, (
721 "Testcase {} : Failed \n "
722 "Expected: [{}]: Upstream IIF interface {} should not be present\n "
723 "Found: {}".format(tc_name
, data
["dut"], data
["iif"], result
)
726 step("Configure default routes on all the nodes")
728 intf_f1_c2
= topo
["routers"]["f1"]["links"]["c2"]["ipv4"].split("/")[0]
729 intf_l1_c1
= topo
["routers"]["l1"]["links"]["c1"]["ipv4"].split("/")[0]
730 intf_l1_r2
= topo
["routers"]["l1"]["links"]["r2"]["ipv4"].split("/")[0]
731 intf_r2_f1
= topo
["routers"]["r2"]["links"]["f1"]["ipv4"].split("/")[0]
734 "c1": {"static_routes": [{"network": "0.0.0.0/0", "next_hop": intf_l1_c1
}]},
735 "c2": {"static_routes": [{"network": "0.0.0.0/0", "next_hop": intf_f1_c2
}]},
736 "r2": {"static_routes": [{"network": "0.0.0.0/0", "next_hop": intf_l1_r2
}]},
737 "f1": {"static_routes": [{"network": "0.0.0.0/0", "next_hop": intf_r2_f1
}]},
739 result
= create_static_routes(tgen
, input_dict
)
740 assert result
is True, "Testcase {} :Failed \n Error {}".format(tc_name
, result
)
742 step("applying ip nht config on c2")
745 "c1": {"raw_config": ["ip nht resolve-via-default"]},
746 "c2": {"raw_config": ["ip nht resolve-via-default"]},
747 "r2": {"raw_config": ["ip nht resolve-via-default"]},
748 "f1": {"raw_config": ["ip nht resolve-via-default"]},
749 "l1": {"raw_config": ["ip nht resolve-via-default"]},
752 result
= apply_raw_config(tgen
, raw_config
)
753 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
755 step("Verify RP info Not unknown after removing static route from c2 ")
757 step("Verify RP info is NOT unknown after removing static route from c2 ")
758 result
= verify_pim_rp_info(
759 tgen
, topo
, dut
, GROUP_RANGE_1
, "Unknown", rp_address
, SOURCE
, expected
=False
761 assert result
is not True, (
762 "Testcase {} : Failed \n "
763 "Expected: [{}]: RP Info should not be Unknown after removing static"
765 "Found: {}".format(tc_name
, data
["dut"], result
)
768 step("Verify (s,g) populated after adding default route ")
770 for data
in input_dict_sg
:
771 result
= verify_mroutes(
779 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
781 result
= verify_upstream_iif(
782 tgen
, data
["dut"], data
["iif"], data
["src_address"], IGMP_JOIN_RANGE_1
784 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
786 step("Verify (*,g) populated after adding default route ")
788 for data
in input_dict_starg
:
789 result
= verify_mroutes(
797 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
799 result
= verify_upstream_iif(
800 tgen
, data
["dut"], data
["iif"], data
["src_address"], IGMP_JOIN_RANGE_1
802 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
804 write_test_footer(tc_name
)
807 def test_PIM_hello_tx_rx_p1(request
):
809 TC_54 Verify received and transmit hello stats
810 are getting cleared after PIM nbr reset
814 tc_name
= request
.node
.name
815 write_test_header(tc_name
)
817 # Don"t run this test if we have any failure.
818 if tgen
.routers_have_failure():
819 pytest
.skip(tgen
.errors
)
821 # Creating configuration from JSON
822 app_helper
.stop_all_hosts()
824 reset_config_on_routers(tgen
)
825 clear_pim_interface_traffic(tgen
, topo
)
828 "Remove c1-c2 connected link to simulate topo "
829 "c1(LHR)---l1(RP)----r2---f1-----c2(FHR)"
832 intf_c1_c2
= topo
["routers"]["c1"]["links"]["c2"]["interface"]
833 intf_c2_c1
= topo
["routers"]["c2"]["links"]["c1"]["interface"]
834 shutdown_bringup_interface(tgen
, "c1", intf_c1_c2
, False)
835 shutdown_bringup_interface(tgen
, "c2", intf_c2_c1
, False)
837 step("Enable the PIM on all the interfaces of FRR1, FRR2, FRR3")
839 "Enable IGMP of FRR1 interface and send IGMP joins "
840 " from FRR1 node for group range (225.1.1.1-5)"
843 intf_c1_i4
= topo
["routers"]["c1"]["links"]["i4"]["interface"]
845 "c1": {"igmp": {"interfaces": {intf_c1_i4
: {"igmp": {"version": "2"}}}}}
847 result
= create_igmp_config(tgen
, topo
, input_dict
)
848 assert result
is True, "Testcase {}: Failed Error: {}".format(tc_name
, result
)
850 input_join
= {"i4": topo
["routers"]["i4"]["links"]["c1"]["interface"]}
852 for recvr
, recvr_intf
in input_join
.items():
853 result
= app_helper
.run_join(recvr
, IGMP_JOIN_RANGE_1
, join_intf
=recvr_intf
)
854 assert result
is True, "Testcase {}: Failed Error: {}".format(tc_name
, result
)
856 step("Configure static RP for (225.1.1.1-5) as R2")
863 "rp_addr": topo
["routers"]["l1"]["links"]["lo"]["ipv4"].split(
866 "group_addr_range": GROUP_RANGE
,
873 result
= create_pim_config(tgen
, topo
, input_dict
)
874 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
876 step("Send Mcast traffic from C2 to all the groups ( 225.1.1.1 to 225.1.1.5)")
878 input_src
= {"i5": topo
["routers"]["i5"]["links"]["c2"]["interface"]}
880 for src
, src_intf
in input_src
.items():
881 result
= app_helper
.run_traffic(src
, IGMP_JOIN_RANGE_1
, bind_intf
=src_intf
)
882 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
884 source_i5
= topo
["routers"]["i5"]["links"]["c2"]["ipv4"].split("/")[0]
890 "iif": topo
["routers"]["c1"]["links"]["l1"]["interface"],
891 "oil": topo
["routers"]["c1"]["links"]["i4"]["interface"],
898 "src_address": source_i5
,
899 "iif": topo
["routers"]["c1"]["links"]["l1"]["interface"],
900 "oil": topo
["routers"]["c1"]["links"]["i4"]["interface"],
904 step("(*,G) and (S,G) created on f1 and node verify using 'show ip mroute'")
905 for data
in input_dict_sg
:
906 result
= verify_mroutes(
914 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
916 for data
in input_dict_starg
:
917 result
= verify_mroutes(
925 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
927 intf_l1_c1
= topo
["routers"]["l1"]["links"]["c1"]["interface"]
928 intf_c1_l1
= topo
["routers"]["c1"]["links"]["l1"]["interface"]
932 intf_c1_l1
: ["helloTx", "helloRx"],
936 c1_state_before
= get_pim_interface_traffic(tgen
, state_dict
)
938 c1_state_before
, dict
939 ), "Testcase{} : Failed \n state_before is not dictionary \n Error: {}".format(
943 step("Flap PIM nbr while doing interface c1-l1 interface shut from f1 side")
944 shutdown_bringup_interface(tgen
, "c1", intf_c1_l1
, False)
946 """ Resetting the stats here since shutdown resets the stats.
948 reset_stats(c1_state_before
)
949 shutdown_bringup_interface(tgen
, "c1", intf_c1_l1
, True)
951 step("verify stats after no shutdown on c1 and that they are incremented")
955 while not done
and count
<= 7:
956 c1_state_after
= get_pim_interface_traffic(tgen
, state_dict
)
959 ), "Testcase{} : Failed \n state_before is not dictionary \n Error: {}".format(
963 result
= verify_state_incremented(c1_state_before
, c1_state_after
)
964 if result
is not True:
972 ), "Testcase{} : Failed Error: {}" "stats is not incremented".format(
976 step("verify before stats on l1")
979 intf_l1_c1
: ["helloTx", "helloRx"],
983 l1_state_before
= get_pim_interface_traffic(tgen
, l1_state_dict
)
985 l1_state_before
, dict
986 ), "Testcase{} : Failed \n state_before is not dictionary \n Error: {}".format(
990 step("Flap PIM nbr while doing interface r2-c1 shut from r2 side")
991 shutdown_bringup_interface(tgen
, "l1", intf_l1_c1
, False)
994 "After shut the interface from r2 side , verify r2 side rx and tx of hello"
995 "counters are resetted show ip pim interface traffic"
997 shutdown_bringup_interface(tgen
, "l1", intf_l1_c1
, True)
999 step("verify stats after on l1 are incremented")
1002 while not done
and count
<= 7:
1003 l1_state_after
= get_pim_interface_traffic(tgen
, l1_state_dict
)
1005 l1_state_after
, dict
1006 ), "Testcase{} : Failed \n state_before is not dictionary \n Error: {}".format(
1010 result
= verify_state_incremented(l1_state_before
, l1_state_after
)
1019 ), "Testcase{} : Failed Error: {}" "stats incremented".format(tc_name
, result
)
1021 step("Reinit the dict")
1022 c1_state_before
= {}
1023 l1_state_before
= {}
1027 step("verify before stats on C1")
1030 intf_c1_l1
: ["helloTx", "helloRx"],
1034 c1_state_before
= get_pim_interface_traffic(tgen
, state_dict
)
1036 c1_state_before
, dict
1037 ), "Testcase{} : Failed \n state_before is not dictionary \n Error: {}".format(
1041 step("Flap c1-r2 pim nbr while changing ip address from c1 side")
1042 c1_l1_ip_subnet
= topo
["routers"]["c1"]["links"]["l1"]["ipv4"]
1047 "interface {}".format(intf_c1_l1
),
1048 "no ip address {}".format(c1_l1_ip_subnet
),
1049 "ip address {}".format(NEW_ADDRESS_2_SUBNET
),
1054 result
= apply_raw_config(tgen
, raw_config
)
1055 assert result
is True, "Testcase {} : Failed Error: {}".format(tc_name
, result
)
1057 step("verify stats after on c1 are incremented")
1060 while not done
and count
<= 7:
1061 c1_state_after
= get_pim_interface_traffic(tgen
, state_dict
)
1063 c1_state_after
, dict
1064 ), "Testcase{} : Failed \n state_before is not dictionary \n Error: {}".format(
1068 result
= verify_state_incremented(c1_state_before
, c1_state_after
)
1069 if result
is not True:
1075 assert result
is True, "Testcase{} : Failed Error: {}" "stats incremented".format(
1079 write_test_footer(tc_name
)
1082 if __name__
== "__main__":
1083 args
= ["-s"] + sys
.argv
[1:]
1084 sys
.exit(pytest
.main(args
))