NAT: update nat_show_config_reply API (VPP-1403)

[vpp.git] / test / test_nat.py

diff --git a/test/test_nat.py b/test/test_nat.py
index 4bae701..286c74a 100644 (file)
--- a/test/test_nat.py
+++ b/test/test_nat.py
@@ -136,6 +136,7 @@ class MethodHolder(VppTestCase):
 
         self.vapi.nat_set_reass()
         self.vapi.nat_set_reass(is_ip6=1)
+        self.verify_no_nat44_user()
 
     def nat44_add_static_mapping(self, local_ip, external_ip='0.0.0.0',
                                  local_port=0, external_port=0, vrf_id=0,
@@ -987,6 +988,11 @@ class MethodHolder(VppTestCase):
         # postNAPTDestinationTransportPort
         self.assertEqual(struct.pack("!H", dst_port), record[228])
 
+    def verify_no_nat44_user(self):
+        """ Verify that there is no NAT44 user """
+        users = self.vapi.nat44_user_dump()
+        self.assertEqual(len(users), 0)
+
 
 class TestNAT44(MethodHolder):
     """ NAT44 Test Cases """
@@ -2909,6 +2915,12 @@ class TestNAT44(MethodHolder):
         sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, 0)
         self.assertEqual(nsessions - len(sessions), 2)
 
+        self.vapi.nat44_del_session(sessions[0].inside_ip_address,
+                                    sessions[0].inside_port,
+                                    sessions[0].protocol)
+
+        self.verify_no_nat44_user()
+
     def test_set_get_reass(self):
         """ NAT44 set/get virtual fragmentation reassembly """
         reas_cfg1 = self.vapi.nat_get_reass()
@@ -3333,6 +3345,9 @@ class TestNAT44EndpointDependent(MethodHolder):
         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
                                                   is_inside=0)
 
+        nat_config = self.vapi.nat_show_config()
+        self.assertEqual(1, nat_config.endpoint_dependent)
+
         # in2out
         pkts = self.create_stream_in(self.pg0, self.pg1)
         self.pg0.add_stream(pkts)
@@ -3960,6 +3975,63 @@ class TestNAT44EndpointDependent(MethodHolder):
             self.logger.error(ppp("Unexpected or invalid packet:", p))
             raise
 
+    def test_next_src_nat(self):
+        """ On way back forward packet to nat44-in2out node. """
+        twice_nat_addr = '10.0.1.3'
+        external_port = 80
+        local_port = 8080
+        post_twice_nat_port = 0
+
+        self.vapi.nat44_forwarding_enable_disable(1)
+        self.nat44_add_address(twice_nat_addr, twice_nat=1)
+        self.nat44_add_static_mapping(self.pg6.remote_ip4, self.pg1.remote_ip4,
+                                      local_port, external_port,
+                                      proto=IP_PROTOS.tcp, out2in_only=1,
+                                      self_twice_nat=1, vrf_id=1)
+        self.vapi.nat44_interface_add_del_feature(self.pg6.sw_if_index,
+                                                  is_inside=0)
+
+        p = (Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) /
+             IP(src=self.pg6.remote_ip4, dst=self.pg1.remote_ip4) /
+             TCP(sport=12345, dport=external_port))
+        self.pg6.add_stream(p)
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pg_start()
+        capture = self.pg6.get_capture(1)
+        p = capture[0]
+        try:
+            ip = p[IP]
+            tcp = p[TCP]
+            self.assertEqual(ip.src, twice_nat_addr)
+            self.assertNotEqual(tcp.sport, 12345)
+            post_twice_nat_port = tcp.sport
+            self.assertEqual(ip.dst, self.pg6.remote_ip4)
+            self.assertEqual(tcp.dport, local_port)
+            self.assert_packet_checksums_valid(p)
+        except:
+            self.logger.error(ppp("Unexpected or invalid packet:", p))
+            raise
+
+        p = (Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) /
+             IP(src=self.pg6.remote_ip4, dst=twice_nat_addr) /
+             TCP(sport=local_port, dport=post_twice_nat_port))
+        self.pg6.add_stream(p)
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pg_start()
+        capture = self.pg6.get_capture(1)
+        p = capture[0]
+        try:
+            ip = p[IP]
+            tcp = p[TCP]
+            self.assertEqual(ip.src, self.pg1.remote_ip4)
+            self.assertEqual(tcp.sport, external_port)
+            self.assertEqual(ip.dst, self.pg6.remote_ip4)
+            self.assertEqual(tcp.dport, 12345)
+            self.assert_packet_checksums_valid(p)
+        except:
+            self.logger.error(ppp("Unexpected or invalid packet:", p))
+            raise
+
     def twice_nat_common(self, self_twice_nat=False, same_pg=False, lb=False,
                          client_id=None):
         twice_nat_addr = '10.0.1.3'
@@ -4910,6 +4982,9 @@ class TestNAT44Out2InDPO(MethodHolder):
     def test_464xlat_ce(self):
         """ Test 464XLAT CE with NAT44 """
 
+        nat_config = self.vapi.nat_show_config()
+        self.assertEqual(1, nat_config.out2in_dpo)
+
         self.configure_xlat()
 
         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
@@ -5931,7 +6006,7 @@ class TestNAT64(MethodHolder):
                                                 self.nat_addr_n)
         self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
         self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
-        self.vapi.nat64_set_timeouts(icmp=5)
+        self.vapi.nat64_set_timeouts(icmp=5, tcp_trans=5, tcp_est=5)
 
         pkts = self.create_stream_in_ip6(self.pg0, self.pg1)
         self.pg0.add_stream(pkts)
@@ -5943,9 +6018,9 @@ class TestNAT64(MethodHolder):
 
         sleep(15)
 
-        # ICMP session after timeout
+        # ICMP and TCP session after timeout
         ses_num_after_timeout = self.nat64_get_ses_num()
-        self.assertNotEqual(ses_num_before_timeout, ses_num_after_timeout)
+        self.assertEqual(ses_num_before_timeout - ses_num_after_timeout, 2)
 
     def test_icmp_error(self):
         """ NAT64 ICMP Error message translation """
@@ -6892,6 +6967,9 @@ class TestDSlite(MethodHolder):
 
     def test_dslite(self):
         """ Test DS-Lite """
+        nat_config = self.vapi.nat_show_config()
+        self.assertEqual(0, nat_config.dslite_ce)
+
         self.vapi.dslite_add_del_pool_addr_range(self.nat_addr_n,
                                                  self.nat_addr_n)
         aftr_ip4 = '192.0.0.1'
@@ -7052,6 +7130,9 @@ class TestDSliteCE(MethodHolder):
     def test_dslite_ce(self):
         """ Test DS-Lite CE """
 
+        nat_config = self.vapi.nat_show_config()
+        self.assertEqual(1, nat_config.dslite_ce)
+
         b4_ip4 = '192.0.0.2'
         b4_ip4_n = socket.inet_pton(socket.AF_INET, b4_ip4)
         b4_ip6 = '2001:db8:62aa::375e:f4c1:1'