tests: Add support for getting corefile patterns on FreeBSD

[vpp.git] / test / test_nat44_ei.py

diff --git a/test/test_nat44_ei.py b/test/test_nat44_ei.py
index 259c445..ae9194b 100644 (file)
--- a/test/test_nat44_ei.py
+++ b/test/test_nat44_ei.py
@@ -8,8 +8,8 @@ import unittest
 from io import BytesIO
 
 import scapy.compat
-from framework import tag_fixme_debian11, is_distro_debian11
-from framework import VppTestCase, VppTestRunner
+from framework import VppTestCase, VppLoInterface
+from asfframework import VppTestRunner, tag_fixme_debian11, is_distro_debian11
 from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
 from scapy.all import (
     bind_layers,
@@ -663,6 +663,7 @@ class MethodHolder(VppTestCase):
         self.assertEqual(scapy.compat.orb(record[230]), 3)
         # natPoolID
         self.assertEqual(struct.pack("!I", 0), record[283])
+        return len(data)
 
     def verify_ipfix_max_sessions(self, data, limit):
         self.assertEqual(1, len(data))
@@ -673,6 +674,7 @@ class MethodHolder(VppTestCase):
         self.assertEqual(struct.pack("!I", 1), record[466])
         # maxSessionEntries
         self.assertEqual(struct.pack("!I", limit), record[471])
+        return len(data)
 
     def verify_no_nat44_user(self):
         """Verify that there is no NAT44EI user"""
@@ -801,7 +803,6 @@ class MethodHolder(VppTestCase):
         proto=IP_PROTOS.tcp,
         ignore_port=False,
     ):
-
         layer = self.proto2layer(proto)
 
         if proto == IP_PROTOS.tcp:
@@ -2464,7 +2465,7 @@ class TestNAT44EI(MethodHolder):
             Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
             / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
             / TCP(sport=3025)
-        )
+        ) * 3
         self.pg0.add_stream(p)
         self.pg_enable_capture(self.pg_interfaces)
         self.pg_start()
@@ -2483,10 +2484,12 @@ class TestNAT44EI(MethodHolder):
             if p.haslayer(Template):
                 ipfix.add_template(p.getlayer(Template))
         # verify events in data set
+        event_count = 0
         for p in capture:
             if p.haslayer(Data):
                 data = ipfix.decode_data_set(p.getlayer(Set))
-                self.verify_ipfix_addr_exhausted(data)
+                event_count += self.verify_ipfix_addr_exhausted(data)
+        self.assertEqual(event_count, 1)
 
     def test_ipfix_max_sessions(self):
         """NAT44EI IPFIX logging maximum session entries exceeded"""
@@ -2530,7 +2533,7 @@ class TestNAT44EI(MethodHolder):
             Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
             / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
             / TCP(sport=1025)
-        )
+        ) * 3
         self.pg0.add_stream(p)
         self.pg_enable_capture(self.pg_interfaces)
         self.pg_start()
@@ -2549,10 +2552,14 @@ class TestNAT44EI(MethodHolder):
             if p.haslayer(Template):
                 ipfix.add_template(p.getlayer(Template))
         # verify events in data set
+        event_count = 0
         for p in capture:
             if p.haslayer(Data):
                 data = ipfix.decode_data_set(p.getlayer(Set))
-                self.verify_ipfix_max_sessions(data, max_sessions_per_thread)
+                event_count += self.verify_ipfix_max_sessions(
+                    data, max_sessions_per_thread
+                )
+        self.assertEqual(event_count, 1)
 
     def test_syslog_apmap(self):
         """NAT44EI syslog address and port mapping creation and deletion"""
@@ -4061,6 +4068,59 @@ class TestNAT44EI(MethodHolder):
         # a nonzero default should be reported for user_sessions
         self.assertNotEqual(nat44_ei_config.user_sessions, 0)
 
+    def test_delete_interface(self):
+        """NAT44EI delete nat interface"""
+
+        self.nat44_add_address(self.nat_addr)
+
+        interfaces = self.create_loopback_interfaces(4)
+
+        self.vapi.nat44_ei_interface_add_del_feature(
+            sw_if_index=interfaces[0].sw_if_index, is_add=1
+        )
+        flags = self.config_flags.NAT44_EI_IF_INSIDE
+        self.vapi.nat44_ei_interface_add_del_feature(
+            sw_if_index=interfaces[1].sw_if_index, flags=flags, is_add=1
+        )
+        flags |= self.config_flags.NAT44_EI_IF_OUTSIDE
+        self.vapi.nat44_ei_interface_add_del_feature(
+            sw_if_index=interfaces[2].sw_if_index, flags=flags, is_add=1
+        )
+        self.vapi.nat44_ei_add_del_output_interface(
+            sw_if_index=interfaces[3].sw_if_index, is_add=1
+        )
+
+        nat_sw_if_indices = [
+            i.sw_if_index
+            for i in self.vapi.nat44_ei_interface_dump()
+            + list(self.vapi.vpp.details_iter(self.vapi.nat44_ei_output_interface_get))
+        ]
+        self.assertEqual(len(nat_sw_if_indices), len(interfaces))
+
+        loopbacks = []
+        for i in interfaces:
+            # delete nat-enabled interface
+            self.assertIn(i.sw_if_index, nat_sw_if_indices)
+            i.remove_vpp_config()
+
+            # create interface with the same index
+            lo = VppLoInterface(self)
+            loopbacks.append(lo)
+            self.assertEqual(lo.sw_if_index, i.sw_if_index)
+
+            # check interface is not nat-enabled
+            nat_sw_if_indices = [
+                i.sw_if_index
+                for i in self.vapi.nat44_ei_interface_dump()
+                + list(
+                    self.vapi.vpp.details_iter(self.vapi.nat44_ei_output_interface_get)
+                )
+            ]
+            self.assertNotIn(lo.sw_if_index, nat_sw_if_indices)
+
+        for i in loopbacks:
+            i.remove_vpp_config()
+
 
 class TestNAT44Out2InDPO(MethodHolder):
     """NAT44EI Test Cases using out2in DPO"""