capture = self.pg1.get_capture(len(pkts))
self.verify_capture_out(capture)
self.nat44_add_address(self.nat_addr, is_add=0)
- self.vapi.cli("ipfix flush") # FIXME this should be an API call
+ self.vapi.ipfix_flush()
capture = self.pg3.get_capture(9)
ipfix = IPFIXDecoder()
# first load template
self.pg_start()
self.pg1.assert_nothing_captured()
sleep(1)
- self.vapi.cli("ipfix flush") # FIXME this should be an API call
+ self.vapi.ipfix_flush()
capture = self.pg3.get_capture(9)
ipfix = IPFIXDecoder()
# first load template
self.pg_start()
self.pg1.assert_nothing_captured()
sleep(1)
- self.vapi.cli("ipfix flush") # FIXME this should be an API call
+ self.vapi.ipfix_flush()
capture = self.pg3.get_capture(9)
ipfix = IPFIXDecoder()
# first load template
sw_if_index=self.pg1.sw_if_index,
is_add=1)
+ reas_cfg1 = self.vapi.nat_get_reass()
+ # this test was intermittently failing in some cases
+ # until we temporarily bump the reassembly timeouts
+ self.vapi.nat_set_reass(timeout=20, max_reass=1024, max_frag=5,
+ drop_frag=0)
+
self.frag_in_order(proto=IP_PROTOS.tcp)
self.frag_in_order(proto=IP_PROTOS.udp)
self.frag_in_order(proto=IP_PROTOS.icmp)
+ # restore the reassembly timeouts
+ self.vapi.nat_set_reass(timeout=reas_cfg1.ip4_timeout,
+ max_reass=reas_cfg1.ip4_max_reass,
+ max_frag=reas_cfg1.ip4_max_frag,
+ drop_frag=reas_cfg1.ip4_drop_frag)
+
def test_frag_forwarding(self):
""" NAT44 forwarding fragment test """
self.vapi.nat44_add_del_interface_addr(
self.pg_start()
self.pg1.assert_nothing_captured()
sleep(1)
- self.vapi.cli("ipfix flush") # FIXME this should be an API call
+ self.vapi.ipfix_flush()
capture = self.pg3.get_capture(9)
ipfix = IPFIXDecoder()
# first load template
sw_if_index=self.pg1.sw_if_index,
is_add=1)
self.vapi.nat44_forwarding_enable_disable(enable=True)
+ reas_cfg1 = self.vapi.nat_get_reass()
+ # this test was intermittently failing in some cases
+ # until we temporarily bump the reassembly timeouts
+ self.vapi.nat_set_reass(timeout=20, max_reass=1024, max_frag=5,
+ drop_frag=0)
self.frag_in_order(proto=IP_PROTOS.tcp, dont_translate=True)
+ # restore the reassembly timeouts
+ self.vapi.nat_set_reass(timeout=reas_cfg1.ip4_timeout,
+ max_reass=reas_cfg1.ip4_max_reass,
+ max_frag=reas_cfg1.ip4_max_frag,
+ drop_frag=reas_cfg1.ip4_drop_frag)
def test_frag_out_of_order(self):
""" NAT44 translate fragments arriving out of order """
self.config_flags.NAT_IS_EXT_HOST_VALID)
self.assertTrue(sessions[0].flags &
self.config_flags.NAT_IS_TWICE_NAT)
- self.logger.error(self.vapi.cli("show nat44 sessions detail"))
+ self.logger.info(self.vapi.cli("show nat44 sessions detail"))
self.vapi.nat44_del_session(
address=sessions[0].inside_ip_address,
port=sessions[0].inside_port,
capture = self.pg1.assert_nothing_captured()
# verify IPFIX logging
- self.vapi.cli("ipfix flush") # FIXME this should be an API call
+ self.vapi.ipfix_flush()
sleep(1)
capture = self.pg2.get_capture(10)
ipfix = IPFIXDecoder()
self.assertEqual(1000, dms[0].ses_num)
# verify IPFIX logging
- self.vapi.cli("ipfix flush") # FIXME this should be an API call
+ self.vapi.ipfix_flush()
sleep(1)
capture = self.pg2.get_capture(2)
ipfix = IPFIXDecoder()
self.pg_start()
self.pg1.assert_nothing_captured()
sleep(1)
- self.vapi.cli("ipfix flush") # FIXME this should be an API call
+ self.vapi.ipfix_flush()
capture = self.pg3.get_capture(9)
ipfix = IPFIXDecoder()
# first load template
self.pg_start()
self.pg1.assert_nothing_captured()
sleep(1)
- self.vapi.cli("ipfix flush") # FIXME this should be an API call
+ self.vapi.ipfix_flush()
capture = self.pg3.get_capture(1)
# verify events in data set
for p in capture:
self.pg_start()
self.pg1.assert_nothing_captured()
sleep(1)
- self.vapi.cli("ipfix flush") # FIXME this should be an API call
+ self.vapi.ipfix_flush()
capture = self.pg3.get_capture(9)
ipfix = IPFIXDecoder()
# first load template
self.pg_start()
p = self.pg1.get_capture(1)
self.tcp_port_out = p[0][TCP].sport
- self.vapi.cli("ipfix flush") # FIXME this should be an API call
+ self.vapi.ipfix_flush()
capture = self.pg3.get_capture(10)
ipfix = IPFIXDecoder()
# first load template
end_addr=self.nat_addr,
vrf_id=0xFFFFFFFF,
is_add=0)
- self.vapi.cli("ipfix flush") # FIXME this should be an API call
+ self.vapi.ipfix_flush()
capture = self.pg3.get_capture(2)
# verify events in data set
for p in capture: