protocol=id_m.protocol,
is_add=0)
- adresses = self.vapi.nat44_address_dump()
- for addr in adresses:
+ addresses = self.vapi.nat44_address_dump()
+ for addr in addresses:
self.vapi.nat44_add_del_address_range(addr.ip_address,
addr.ip_address,
twice_nat=addr.twice_nat,
:param ip: IP address
:param is_add: 1 if add, 0 if delete (Default add)
- :param twice_nat: twice NAT address for extenal hosts
+ :param twice_nat: twice NAT address for external hosts
"""
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 same_port: Sorce port number is not translated (Default False)
+ :param same_port: Source port number is not translated (Default False)
:param dst_ip: Destination IP address (Default do not verify)
:param is_ip6: If L3 protocol is IPv6 (Default False)
"""
:param capture: Captured packets
:param nat_ip: Translated IP address
- :param same_port: Sorce port number is not translated (Default False)
+ :param same_port: Source port number is not translated (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 data: Payload data
:param proto: protocol (TCP, UDP, ICMP)
:param echo_reply: use echo_reply if protocol is ICMP
- :returns: Fragmets
+ :returns: Fragments
"""
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
- :returns: Fragmets
+ :returns: Fragments
"""
if pref is None:
dst_ip6 = ''.join(['64:ff9b::', dst])
super(TestNAT44, cls).tearDownClass()
raise
+ @classmethod
+ def tearDownClass(cls):
+ super(TestNAT44, cls).tearDownClass()
+
def test_dynamic(self):
""" NAT44 dynamic translation test """
self.nat44_add_address(self.nat_addr)
self.vapi.nat44_add_del_interface_addr(self.pg7.sw_if_index)
# no address in NAT pool
- adresses = self.vapi.nat44_address_dump()
- self.assertEqual(0, len(adresses))
+ addresses = self.vapi.nat44_address_dump()
+ self.assertEqual(0, len(addresses))
# configure interface address and check NAT address pool
self.pg7.config_ip4()
- 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)
# remove interface address and check NAT address pool
self.pg7.unconfig_ip4()
- adresses = self.vapi.nat44_address_dump()
- self.assertEqual(0, len(adresses))
+ addresses = self.vapi.nat44_address_dump()
+ self.assertEqual(0, len(addresses))
def test_interface_addr_static_mapping(self):
""" Static mapping with addresses from interface """
identity_mappings[0].sw_if_index)
def test_ipfix_nat44_sess(self):
- """ IPFIX logging NAT44 session created/delted """
+ """ IPFIX logging NAT44 session created/deleted """
self.ipfix_domain_id = 10
self.ipfix_src_port = 20202
- colector_port = 30303
+ collector_port = 30303
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)
src_address=self.pg3.local_ip4n,
path_mtu=512,
template_interval=10,
- collector_port=colector_port)
+ collector_port=collector_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[UDP].dport, colector_port)
+ self.assertEqual(p[UDP].dport, collector_port)
self.assertEqual(p[IPFIX].observationDomainID,
self.ipfix_domain_id)
if p.haslayer(Template):
def tearDown(self):
super(TestNAT44, self).tearDown()
- if not self.vpp_dead:
- self.logger.info(self.vapi.cli("show nat44 addresses"))
- self.logger.info(self.vapi.cli("show nat44 interfaces"))
- self.logger.info(self.vapi.cli("show nat44 static mappings"))
- self.logger.info(self.vapi.cli("show nat44 interface address"))
- self.logger.info(self.vapi.cli("show nat44 sessions detail"))
- self.logger.info(self.vapi.cli("show nat virtual-reassembly"))
- self.logger.info(self.vapi.cli("show nat44 hash tables detail"))
- self.logger.info(self.vapi.cli("show nat timeouts"))
- self.logger.info(
- self.vapi.cli("show nat addr-port-assignment-alg"))
- self.logger.info(self.vapi.cli("show nat ha"))
- self.clear_nat44()
- self.vapi.cli("clear logging")
+ self.clear_nat44()
+ self.vapi.cli("clear logging")
+
+ def show_commands_at_teardown(self):
+ self.logger.info(self.vapi.cli("show nat44 addresses"))
+ self.logger.info(self.vapi.cli("show nat44 interfaces"))
+ self.logger.info(self.vapi.cli("show nat44 static mappings"))
+ self.logger.info(self.vapi.cli("show nat44 interface address"))
+ self.logger.info(self.vapi.cli("show nat44 sessions detail"))
+ self.logger.info(self.vapi.cli("show nat virtual-reassembly"))
+ self.logger.info(self.vapi.cli("show nat44 hash tables detail"))
+ self.logger.info(self.vapi.cli("show nat timeouts"))
+ self.logger.info(
+ self.vapi.cli("show nat addr-port-assignment-alg"))
+ self.logger.info(self.vapi.cli("show nat ha"))
class TestNAT44EndpointDependent(MethodHolder):
super(TestNAT44EndpointDependent, cls).tearDownClass()
raise
+ @classmethod
+ def tearDownClass(cls):
+ super(TestNAT44EndpointDependent, cls).tearDownClass()
+
def test_frag_in_order(self):
""" NAT44 translate fragments arriving in order """
self.nat44_add_address(self.nat_addr)
self.vapi.nat44_interface_add_del_output_feature(self.pg1.sw_if_index,
is_inside=0)
- # session initiaded from service host - translate
+ # session initiated from service host - translate
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)
- # session initiaded from remote host - do not translate
+ # session initiated from remote host - do not translate
self.tcp_port_in = 60303
self.udp_port_in = 60304
self.icmp_id_in = 60305
def tearDown(self):
super(TestNAT44EndpointDependent, self).tearDown()
if not self.vpp_dead:
- self.logger.info(self.vapi.cli("show nat44 addresses"))
- self.logger.info(self.vapi.cli("show nat44 interfaces"))
- self.logger.info(self.vapi.cli("show nat44 static mappings"))
- self.logger.info(self.vapi.cli("show nat44 interface address"))
- self.logger.info(self.vapi.cli("show nat44 sessions detail"))
- self.logger.info(self.vapi.cli("show nat44 hash tables detail"))
- self.logger.info(self.vapi.cli("show nat timeouts"))
self.clear_nat44()
self.vapi.cli("clear logging")
+ def show_commands_at_teardown(self):
+ self.logger.info(self.vapi.cli("show nat44 addresses"))
+ self.logger.info(self.vapi.cli("show nat44 interfaces"))
+ self.logger.info(self.vapi.cli("show nat44 static mappings"))
+ self.logger.info(self.vapi.cli("show nat44 interface address"))
+ self.logger.info(self.vapi.cli("show nat44 sessions detail"))
+ self.logger.info(self.vapi.cli("show nat44 hash tables detail"))
+ self.logger.info(self.vapi.cli("show nat timeouts"))
+
class TestNAT44Out2InDPO(MethodHolder):
""" NAT44 Test Cases using out2in DPO """
super(TestNAT44Out2InDPO, cls).tearDownClass()
raise
+ @classmethod
+ def tearDownClass(cls):
+ super(TestNAT44Out2InDPO, cls).tearDownClass()
+
def configure_xlat(self):
self.dst_ip6_pfx = '1:2:3::'
self.dst_ip6_pfx_n = socket.inet_pton(socket.AF_INET6,
super(TestDeterministicNAT, cls).tearDownClass()
raise
+ @classmethod
+ def tearDownClass(cls):
+ super(TestDeterministicNAT, cls).tearDownClass()
+
def create_stream_in(self, in_if, out_if, ttl=64):
"""
Create packet stream for inside network
:param capture: Captured packets
:param nat_ip: Translated IP address (Default use global NAT address)
- :param same_port: Sorce port number is not translated (Default False)
+ :param same_port: Source port number is not translated (Default False)
"""
if nat_ip is None:
nat_ip = self.nat_addr
def tearDown(self):
super(TestDeterministicNAT, self).tearDown()
if not self.vpp_dead:
- self.logger.info(self.vapi.cli("show nat44 interfaces"))
- self.logger.info(self.vapi.cli("show nat timeouts"))
- self.logger.info(
- self.vapi.cli("show nat44 deterministic mappings"))
- self.logger.info(
- self.vapi.cli("show nat44 deterministic sessions"))
self.clear_nat_det()
+ def show_commands_at_teardown(self):
+ self.logger.info(self.vapi.cli("show nat44 interfaces"))
+ self.logger.info(self.vapi.cli("show nat timeouts"))
+ self.logger.info(
+ self.vapi.cli("show nat44 deterministic mappings"))
+ self.logger.info(
+ self.vapi.cli("show nat44 deterministic sessions"))
+
class TestNAT64(MethodHolder):
""" NAT64 Test Cases """
super(TestNAT64, cls).tearDownClass()
raise
+ @classmethod
+ def tearDownClass(cls):
+ super(TestNAT64, cls).tearDownClass()
+
def test_nat64_inside_interface_handles_neighbor_advertisement(self):
""" NAT64 inside interface handles Neighbor Advertisement """
self.vapi.nat64_add_del_interface_addr(self.pg4.sw_if_index)
# no address in NAT64 pool
- adresses = self.vapi.nat44_address_dump()
- self.assertEqual(0, len(adresses))
+ addresses = self.vapi.nat44_address_dump()
+ self.assertEqual(0, len(addresses))
# configure interface address and check NAT64 address pool
self.pg4.config_ip4()
# remove interface address and check NAT64 address pool
self.pg4.unconfig_ip4()
addresses = self.vapi.nat64_pool_addr_dump()
- self.assertEqual(0, len(adresses))
+ self.assertEqual(0, len(addresses))
@unittest.skipUnless(running_extended_tests, "part of extended tests")
def test_ipfix_max_bibs_sessions(self):
def tearDown(self):
super(TestNAT64, self).tearDown()
if not self.vpp_dead:
- self.logger.info(self.vapi.cli("show nat64 pool"))
- self.logger.info(self.vapi.cli("show nat64 interfaces"))
- self.logger.info(self.vapi.cli("show nat64 prefix"))
- self.logger.info(self.vapi.cli("show nat64 bib all"))
- self.logger.info(self.vapi.cli("show nat64 session table all"))
- self.logger.info(self.vapi.cli("show nat virtual-reassembly"))
self.clear_nat64()
+ def show_commands_at_teardown(self):
+ self.logger.info(self.vapi.cli("show nat64 pool"))
+ self.logger.info(self.vapi.cli("show nat64 interfaces"))
+ self.logger.info(self.vapi.cli("show nat64 prefix"))
+ self.logger.info(self.vapi.cli("show nat64 bib all"))
+ self.logger.info(self.vapi.cli("show nat64 session table all"))
+ self.logger.info(self.vapi.cli("show nat virtual-reassembly"))
+
class TestDSlite(MethodHolder):
""" DS-Lite Test Cases """
super(TestDSlite, cls).tearDownClass()
raise
+ @classmethod
+ def tearDownClass(cls):
+ super(TestDSlite, cls).tearDownClass()
+
def verify_syslog_apmadd(self, data, isaddr, isport, xsaddr, xsport,
sv6enc, proto):
message = data.decode('utf-8')
def tearDown(self):
super(TestDSlite, self).tearDown()
- if not self.vpp_dead:
- self.logger.info(self.vapi.cli("show dslite pool"))
- self.logger.info(
- self.vapi.cli("show dslite aftr-tunnel-endpoint-address"))
- self.logger.info(self.vapi.cli("show dslite sessions"))
+
+ def show_commands_at_teardown(self):
+ self.logger.info(self.vapi.cli("show dslite pool"))
+ self.logger.info(
+ self.vapi.cli("show dslite aftr-tunnel-endpoint-address"))
+ self.logger.info(self.vapi.cli("show dslite sessions"))
class TestDSliteCE(MethodHolder):
super(TestDSliteCE, cls).tearDownClass()
raise
+ @classmethod
+ def tearDownClass(cls):
+ super(TestDSliteCE, cls).tearDownClass()
+
def test_dslite_ce(self):
""" Test DS-Lite CE """
def tearDown(self):
super(TestDSliteCE, self).tearDown()
- if not self.vpp_dead:
- self.logger.info(
- self.vapi.cli("show dslite aftr-tunnel-endpoint-address"))
- self.logger.info(
- self.vapi.cli("show dslite b4-tunnel-endpoint-address"))
+
+ def show_commands_at_teardown(self):
+ self.logger.info(
+ self.vapi.cli("show dslite aftr-tunnel-endpoint-address"))
+ self.logger.info(
+ self.vapi.cli("show dslite b4-tunnel-endpoint-address"))
class TestNAT66(MethodHolder):
super(TestNAT66, cls).tearDownClass()
raise
+ @classmethod
+ def tearDownClass(cls):
+ super(TestNAT66, cls).tearDownClass()
+
def test_static(self):
""" 1:1 NAT66 test """
self.vapi.nat66_add_del_interface(self.pg0.sw_if_index)
def tearDown(self):
super(TestNAT66, self).tearDown()
- if not self.vpp_dead:
- self.logger.info(self.vapi.cli("show nat66 interfaces"))
- self.logger.info(self.vapi.cli("show nat66 static mappings"))
- self.clear_nat66()
+ self.clear_nat66()
+ def show_commands_at_teardown(self):
+ self.logger.info(self.vapi.cli("show nat66 interfaces"))
+ self.logger.info(self.vapi.cli("show nat66 static mappings"))
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)