self.vapi.nat44_add_del_address_range(addr.ip_address,
addr.ip_address,
twice_nat=addr.twice_nat,
self.vapi.nat44_add_del_address_range(addr.ip_address,
addr.ip_address,
twice_nat=addr.twice_nat,
"""
nat_addr = socket.inet_pton(socket.AF_INET, ip)
self.vapi.nat44_add_del_address_range(nat_addr, nat_addr, is_add,
"""
nat_addr = socket.inet_pton(socket.AF_INET, ip)
self.vapi.nat44_add_del_address_range(nat_addr, nat_addr, is_add,
:param capture: Captured packets
:param nat_ip: Translated IP address (Default use global NAT address)
:param capture: Captured packets
:param nat_ip: Translated IP address (Default use global NAT address)
:param dst_ip: Destination IP address (Default do not verify)
:param is_ip6: If L3 protocol is IPv6 (Default False)
"""
:param dst_ip: Destination IP address (Default do not verify)
:param is_ip6: If L3 protocol is IPv6 (Default False)
"""
:param dst_ip: Destination IP address (Default do not verify)
"""
return self.verify_capture_out(capture, nat_ip, same_port, dst_ip,
:param dst_ip: Destination IP address (Default do not verify)
"""
return self.verify_capture_out(capture, nat_ip, same_port, dst_ip,
:param data: Payload data
:param proto: protocol (TCP, UDP, ICMP)
:param echo_reply: use echo_reply if protocol is ICMP
:param data: Payload data
:param proto: protocol (TCP, UDP, ICMP)
:param echo_reply: use echo_reply if protocol is ICMP
"""
if proto == IP_PROTOS.tcp:
p = (IP(src=src_if.remote_ip4, dst=dst) /
"""
if proto == IP_PROTOS.tcp:
p = (IP(src=src_if.remote_ip4, dst=dst) /
:param pref: NAT64 prefix
:param plen: NAT64 prefix length
:param fragsize: size of fragments
:param pref: NAT64 prefix
:param plen: NAT64 prefix length
:param fragsize: size of fragments
- adresses = self.vapi.nat44_address_dump()
- self.assertEqual(0, len(adresses))
+ addresses = self.vapi.nat44_address_dump()
+ self.assertEqual(0, len(addresses))
- adresses = self.vapi.nat44_address_dump()
- self.assertEqual(1, len(adresses))
- self.assertEqual(adresses[0].ip_address[0:4], self.pg7.local_ip4n)
+ addresses = self.vapi.nat44_address_dump()
+ self.assertEqual(1, len(addresses))
+ self.assertEqual(addresses[0].ip_address[0:4], self.pg7.local_ip4n)
- adresses = self.vapi.nat44_address_dump()
- self.assertEqual(0, len(adresses))
+ addresses = self.vapi.nat44_address_dump()
+ self.assertEqual(0, len(addresses))
bind_layers(UDP, IPFIX, dport=30303)
self.nat44_add_address(self.nat_addr)
self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
bind_layers(UDP, IPFIX, dport=30303)
self.nat44_add_address(self.nat_addr)
self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
self.vapi.nat_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
src_port=self.ipfix_src_port)
self.vapi.nat_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
src_port=self.ipfix_src_port)
self.assertEqual(p[IP].src, self.pg3.local_ip4)
self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
self.assertEqual(p[UDP].sport, self.ipfix_src_port)
self.assertEqual(p[IP].src, self.pg3.local_ip4)
self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
self.assertEqual(p[UDP].sport, self.ipfix_src_port)
pkts = self.create_stream_in(self.pg0, self.pg1)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
pkts = self.create_stream_in(self.pg0, self.pg1)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
capture = self.pg0.get_capture(len(pkts))
self.verify_capture_in(capture, self.pg0)
capture = self.pg0.get_capture(len(pkts))
self.verify_capture_in(capture, self.pg0)
:param capture: Captured packets
:param nat_ip: Translated IP address (Default use global NAT address)
:param capture: Captured packets
:param nat_ip: Translated IP address (Default use global NAT address)
- adresses = self.vapi.nat44_address_dump()
- self.assertEqual(0, len(adresses))
+ addresses = self.vapi.nat44_address_dump()
+ self.assertEqual(0, len(addresses))
@unittest.skipUnless(running_extended_tests, "part of extended tests")
def test_ipfix_max_bibs_sessions(self):
@unittest.skipUnless(running_extended_tests, "part of extended tests")
def test_ipfix_max_bibs_sessions(self):