from vpp_papi import VppEnum
-class NAT44EDTestCase(VppTestCase):
+class TestNAT44ED(VppTestCase):
+ """ NAT44ED Test Case """
nat_addr = '10.0.0.3'
max_sessions = 100
def setUp(self):
- super(NAT44EDTestCase, self).setUp()
+ super().setUp()
self.plugin_enable()
def tearDown(self):
- super(NAT44EDTestCase, self).tearDown()
+ super().tearDown()
if not self.vpp_dead:
self.plugin_disable()
@classmethod
def setUpClass(cls):
- super(NAT44EDTestCase, cls).setUpClass()
+ super().setUpClass()
cls.create_pg_interfaces(range(12))
cls.interfaces = list(cls.pg_interfaces[:4])
self.assertEqual(sd_params.get('XDPORT'),
"%d" % self.tcp_external_port)
-
-class TestNAT44ED(NAT44EDTestCase):
- """ NAT44ED Test Case """
-
def test_icmp_error(self):
""" NAT44ED test ICMP error message with inner header"""
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg1.get_capture(pkt_count * 3)
+ capture = self.pg1.get_capture(pkt_count * 3, timeout=5)
if_idx = self.pg0.sw_if_index
tc2 = self.statistics['/nat44-ed/in2out/slowpath/tcp']
capture = self.pg8.get_capture(len(pkts))
self.verify_capture_out(capture, ignore_port=True)
- if_idx = self.pg7.sw_if_index
+ if_idx = self.pg8.sw_if_index
cnt = self.statistics['/nat44-ed/in2out/slowpath/tcp']
self.assertEqual(cnt[:, if_idx].sum() - tcpn[:, if_idx].sum(), 2)
cnt = self.statistics['/nat44-ed/in2out/slowpath/udp']
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
+ def test_icmp_error_fwd_outbound(self):
+ """ NAT44ED ICMP error outbound with forwarding enabled """
+
+ # Ensure that an outbound ICMP error message is properly associated
+ # with the inbound forward bypass session it is related to.
+ payload = "H" * 10
+
+ self.nat_add_address(self.nat_addr)
+ self.nat_add_inside_interface(self.pg0)
+ self.nat_add_outside_interface(self.pg1)
+
+ # enable forwarding and initiate connection out2in
+ self.vapi.nat44_forwarding_enable_disable(enable=1)
+ p1 = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
+ IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4) /
+ UDP(sport=21, dport=20) / payload)
+
+ self.pg1.add_stream(p1)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(1)[0]
+
+ self.logger.info(self.vapi.cli("show nat44 sessions"))
+
+ # reply with ICMP error message in2out
+ # We cannot reliably retrieve forward bypass sessions via the API.
+ # session dumps for a user will only look on the worker that the
+ # user is supposed to be mapped to in2out. The forward bypass session
+ # is not necessarily created on that worker.
+ p2 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
+ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
+ ICMP(type='dest-unreach', code='port-unreachable') /
+ capture[IP:])
+
+ self.pg0.add_stream(p2)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(1)[0]
+
+ self.logger.info(self.vapi.cli("show nat44 sessions"))
+
+ self.logger.info(ppp("p1 packet:", p1))
+ self.logger.info(ppp("p2 packet:", p2))
+ self.logger.info(ppp("capture packet:", capture))
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)