diff options
Diffstat (limited to 'src/plugins/nat/test/test_nat.py')
| -rw-r--r-- | src/plugins/nat/test/test_nat.py | 69 |
1 files changed, 69 insertions, 0 deletions
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 """ |