X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=src%2Fplugins%2Fnat%2Ftest%2Ftest_nat.py;h=1cddc405f076bc675f009dc9960823ef5df89166;hb=c611f36bbc75a7157bbec26a78178872ddc5441f;hp=2c0fa1017d6ffc6e10e67e2aebb24a42f61b7852;hpb=e4deacc4220511c5ee93eca6b059d2a64ab1d36c;p=vpp.git

diff --git a/src/plugins/nat/test/test_nat.py b/src/plugins/nat/test/test_nat.py
index 2c0fa1017d6..1cddc405f07 100644
--- a/src/plugins/nat/test/test_nat.py
+++ b/src/plugins/nat/test/test_nat.py
@@ -1442,6 +1442,38 @@ class TestNAT44(MethodHolder):
     def tearDownClass(cls):
         super(TestNAT44, cls).tearDownClass()
 
+    def test_clear_sessions(self):
+        """ NAT44 session clearing test """
+
+        self.nat44_add_address(self.nat_addr)
+        flags = self.config_flags.NAT_IS_INSIDE
+        self.vapi.nat44_interface_add_del_feature(
+            sw_if_index=self.pg0.sw_if_index,
+            flags=flags, is_add=1)
+        self.vapi.nat44_interface_add_del_feature(
+            sw_if_index=self.pg1.sw_if_index,
+            is_add=1)
+
+        nat_config = self.vapi.nat_show_config()
+        self.assertEqual(0, nat_config.endpoint_dependent)
+
+        pkts = self.create_stream_in(self.pg0, self.pg1)
+        self.pg0.add_stream(pkts)
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pg_start()
+        capture = self.pg1.get_capture(len(pkts))
+        self.verify_capture_out(capture)
+
+        sessions = self.statistics.get_counter('/nat44/total-sessions')
+        self.assertTrue(sessions[0][0] > 0)
+        self.logger.info("sessions before clearing: %s" % sessions[0][0])
+
+        self.vapi.cli("clear nat44 sessions")
+
+        sessions = self.statistics.get_counter('/nat44/total-sessions')
+        self.assertEqual(sessions[0][0], 0)
+        self.logger.info("sessions after clearing: %s" % sessions[0][0])
+
     def test_dynamic(self):
         """ NAT44 dynamic translation test """
         self.nat44_add_address(self.nat_addr)
@@ -4565,6 +4597,43 @@ class TestNAT44EndpointDependent(MethodHolder):
         self.reass_hairpinning(proto=IP_PROTOS.udp)
         self.reass_hairpinning(proto=IP_PROTOS.icmp)
 
+    def test_clear_sessions(self):
+        """ NAT44 ED session clearing test """
+
+        self.nat44_add_address(self.nat_addr)
+        flags = self.config_flags.NAT_IS_INSIDE
+        self.vapi.nat44_interface_add_del_feature(
+            sw_if_index=self.pg0.sw_if_index,
+            flags=flags, is_add=1)
+        self.vapi.nat44_interface_add_del_feature(
+            sw_if_index=self.pg1.sw_if_index,
+            is_add=1)
+
+        nat_config = self.vapi.nat_show_config()
+        self.assertEqual(1, nat_config.endpoint_dependent)
+
+        pkts = self.create_stream_in(self.pg0, self.pg1)
+        self.pg0.add_stream(pkts)
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pg_start()
+        capture = self.pg1.get_capture(len(pkts))
+        self.verify_capture_out(capture)
+
+        sessions = self.statistics.get_counter('/nat44/total-sessions')
+        self.assertTrue(sessions[0][0] > 0)
+        self.logger.info("sessions before clearing: %s" % sessions[0][0])
+
+        # just for testing purposes
+        self.logger.info(self.vapi.cli("show nat44 summary"))
+
+        self.vapi.cli("clear nat44 sessions")
+
+        self.logger.info(self.vapi.cli("show nat44 summary"))
+
+        sessions = self.statistics.get_counter('/nat44/total-sessions')
+        self.assertEqual(sessions[0][0], 0)
+        self.logger.info("sessions after clearing: %s" % sessions[0][0])
+
     def test_dynamic(self):
         """ NAT44 dynamic translation test """