rx = rx[0]
self.validate_ns(rx_intf, rx, tgt_ip)
- def send_and_assert_no_replies(self, intf, pkts, remark):
- intf.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- for i in self.pg_interfaces:
- i.get_capture(0)
- i.assert_nothing_captured(remark=remark)
-
def verify_ip(self, rx, smac, dmac, sip, dip):
ether = rx[Ether]
self.assertEqual(ether.dst, dmac)
i.unconfig_ip4()
i.admin_down()
- def send_and_assert_no_replies(self, intf, pkts, remark):
- intf.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- for i in self.pg_interfaces:
- i.get_capture(0)
- i.assert_nothing_captured(remark=remark)
-
def test_ip_disabled(self):
""" IP Disabled """
super(TestIP6LoadBalance, self).tearDown()
def send_and_expect_load_balancing(self, input, pkts, outputs):
+ self.vapi.cli("clear trace")
input.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.assertNotEqual(0, len(rx))
def send_and_expect_one_itf(self, input, pkts, itf):
+ self.vapi.cli("clear trace")
input.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
i.unconfig_ip6()
i.admin_down()
- def send_and_expect(self, input, pkts, output):
- input.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- rx = output.get_capture(len(pkts))
- return rx
-
- def send_and_assert_no_replies(self, intf, pkts, remark):
- self.vapi.cli("clear trace")
- intf.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- for i in self.pg_interfaces:
- i.get_capture(0)
- i.assert_nothing_captured(remark=remark)
-
def test_ip_punt(self):
""" IP6 punt police and redirect """
i.unconfig_ip6()
i.admin_down()
- def send_and_expect(self, input, pkts, output):
- self.vapi.cli("clear trace")
- input.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- rx = output.get_capture(len(pkts))
- return rx
-
- def send_and_assert_no_replies(self, intf, pkts, remark):
- self.vapi.cli("clear trace")
- intf.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- for i in self.pg_interfaces:
- i.get_capture(0)
- i.assert_nothing_captured(remark=remark)
-
def test_ip_input(self):
""" IP6 Input Exceptions """
# 0: "hop limit exceeded in transit",
self.assertEqual(icmp.code, 0)
- self.logger.error(self.vapi.cli("sh error"))
-
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)