diff options
Diffstat (limited to 'test')
| -rwxr-xr-x | test/framework.py | 2 | ||||
| -rw-r--r-- | test/test_nat44_ed.py | 18 | ||||
| -rw-r--r-- | test/test_nat44_ed_output.py | 229 | ||||
| -rw-r--r-- | test/test_nat44_ei.py | 2 |
4 files changed, 11 insertions, 240 deletions
diff --git a/test/framework.py b/test/framework.py index 486553befa1..aa533f7b641 100755 --- a/test/framework.py +++ b/test/framework.py @@ -1304,8 +1304,6 @@ class VppTestCase(CPUInterface, unittest.TestCase): n_rx = len(pkts) self.pg_send(intf, pkts, worker=worker, trace=trace) rx = output.get_capture(n_rx) - if trace: - self.logger.debug(self.vapi.cli("show trace")) return rx def send_and_expect_only(self, intf, pkts, output, timeout=None): diff --git a/test/test_nat44_ed.py b/test/test_nat44_ed.py index 2fb955afa74..ec8d7c8d159 100644 --- a/test/test_nat44_ed.py +++ b/test/test_nat44_ed.py @@ -19,8 +19,7 @@ from vpp_ip_route import VppIpRoute, VppRoutePath from vpp_papi import VppEnum -class TestNAT44ED(VppTestCase): - """ NAT44ED Test Case """ +class NAT44EDTestCase(VppTestCase): nat_addr = '10.0.0.3' @@ -38,18 +37,17 @@ class TestNAT44ED(VppTestCase): max_sessions = 100 def setUp(self): - super().setUp() + super(NAT44EDTestCase, self).setUp() self.plugin_enable() def tearDown(self): - super().tearDown() + super(NAT44EDTestCase, self).tearDown() if not self.vpp_dead: self.plugin_disable() def plugin_enable(self): self.vapi.nat44_ed_plugin_enable_disable( - sessions=self.max_sessions, enable=1, - inside_vrf=0, outside_vrf=1) + sessions=self.max_sessions, enable=1) def plugin_disable(self): self.vapi.nat44_ed_plugin_enable_disable(enable=0) @@ -148,7 +146,7 @@ class TestNAT44ED(VppTestCase): @classmethod def setUpClass(cls): - super().setUpClass() + super(NAT44EDTestCase, cls).setUpClass() cls.create_pg_interfaces(range(12)) cls.interfaces = list(cls.pg_interfaces[:4]) @@ -912,6 +910,10 @@ class TestNAT44ED(VppTestCase): 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""" @@ -2256,7 +2258,7 @@ class TestNAT44EDMW(TestNAT44ED): self.pg_enable_capture(self.pg_interfaces) self.pg_start() - capture = self.pg1.get_capture(pkt_count * 3, timeout=5) + capture = self.pg1.get_capture(pkt_count * 3) if_idx = self.pg0.sw_if_index tc2 = self.statistics['/nat44-ed/in2out/slowpath/tcp'] diff --git a/test/test_nat44_ed_output.py b/test/test_nat44_ed_output.py deleted file mode 100644 index ea5c14e7064..00000000000 --- a/test/test_nat44_ed_output.py +++ /dev/null @@ -1,229 +0,0 @@ -#!/usr/bin/env python3 -"""NAT44 ED output-feature tests""" - -import random -import unittest -from scapy.layers.inet import ICMP, Ether, IP, TCP -from scapy.packet import Raw -from scapy.data import IP_PROTOS -from framework import VppTestCase, VppTestRunner -from vpp_papi import VppEnum - - -def get_nat44_ed_in2out_worker_index(ip, vpp_worker_count): - if 0 == vpp_worker_count: - return 0 - numeric = socket.inet_aton(ip) - numeric = struct.unpack("!L", numeric)[0] - numeric = socket.htonl(numeric) - h = numeric + (numeric >> 8) + (numeric >> 16) + (numeric >> 24) - return 1 + h % vpp_worker_count - - -class TestNAT44EDOutput(VppTestCase): - """ NAT44 ED output feature Test Case """ - max_sessions = 1024 - - @classmethod - def setUpClass(cls): - super().setUpClass() - cls.create_pg_interfaces(range(2)) - cls.interfaces = list(cls.pg_interfaces) - - @classmethod - def tearDownClass(cls): - super().tearDownClass() - - def setUp(self): - super().setUp() - for i in self.interfaces: - i.admin_up() - i.config_ip4() - i.resolve_arp() - self.vapi.nat44_ed_plugin_enable_disable(sessions=self.max_sessions, - enable=1) - - def tearDown(self): - if not self.vpp_dead: - self.logger.debug(self.vapi.cli("show nat44 sessions")) - super().tearDown() - if not self.vpp_dead: - for i in self.pg_interfaces: - i.unconfig_ip4() - i.admin_down() - self.vapi.nat44_ed_plugin_enable_disable(enable=0) - - def test_static_dynamic(self): - """ Create static mapping which matches existing dynamic mapping """ - - old_timeouts = self.vapi.nat_get_timeouts() - new_transitory = 2 - self.vapi.nat_set_timeouts( - udp=old_timeouts.udp, - tcp_established=old_timeouts.tcp_established, - icmp=old_timeouts.icmp, - tcp_transitory=new_transitory) - - local_host = self.pg0.remote_ip4 - remote_host = self.pg1.remote_ip4 - nat_intf = self.pg1 - outside_addr = nat_intf.local_ip4 - - self.vapi.nat44_add_del_address_range(first_ip_address=outside_addr, - last_ip_address=outside_addr, - vrf_id=0xffffffff, - is_add=1, - flags=0) - self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg0.sw_if_index, - is_add=1) - self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg0.sw_if_index, - flags=VppEnum.vl_api_nat_config_flags_t.NAT_IS_INSIDE, is_add=1) - self.vapi.nat44_interface_add_del_output_feature( - is_add=1, - sw_if_index=self.pg1.sw_if_index) - - thread_index = get_nat44_ed_in2out_worker_index( - local_host, self.vpp_worker_count) - port_per_thread = int((0xffff-1024) / max(1, self.vpp_worker_count)) - local_sport = 1024 + random.randint(1, port_per_thread) - if self.vpp_worker_count > 0: - local_sport += port_per_thread * (thread_index - 1) - - remote_dport = 10000 - - pg0 = self.pg0 - pg1 = self.pg1 - - # first setup a dynamic TCP session - - # SYN packet in->out - p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) / - IP(src=local_host, dst=remote_host) / - TCP(sport=local_sport, dport=remote_dport, flags="S")) - p = self.send_and_expect(pg0, [p], pg1)[0] - - self.assertEqual(p[IP].src, outside_addr) - self.assertEqual(p[TCP].sport, local_sport) - outside_port = p[TCP].sport - - # SYN+ACK packet out->in - p = (Ether(src=pg1.remote_mac, dst=pg1.local_mac) / - IP(src=remote_host, dst=outside_addr) / - TCP(sport=remote_dport, dport=outside_port, flags="SA")) - self.send_and_expect(pg1, [p], pg0) - - # ACK packet in->out - p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) / - IP(src=local_host, dst=remote_host) / - TCP(sport=local_sport, dport=remote_dport, flags="A")) - self.send_and_expect(pg0, [p], pg1) - - # now we have a session up, create a conflicting static mapping - self.vapi.nat44_add_del_static_mapping( - is_add=1, - local_ip_address=local_host, - external_ip_address=outside_addr, - external_sw_if_index=0xffffffff, - local_port=local_sport, - external_port=outside_port, - protocol=IP_PROTOS.tcp, - flags=VppEnum.vl_api_nat_config_flags_t.NAT_IS_OUT2IN_ONLY) - - sessions = self.vapi.nat44_user_session_dump(local_host, 0) - self.assertEqual(1, len(sessions)) - - # now send some more data over existing session - it should pass - - # in->out - p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) / - IP(src=local_host, dst=remote_host) / - TCP(sport=local_sport, dport=remote_dport) / - Raw("zippity zap")) - self.send_and_expect(pg0, [p], pg1) - - # out->in - p = (Ether(src=pg1.remote_mac, dst=pg1.local_mac) / - IP(src=remote_host, dst=outside_addr) / - TCP(sport=remote_dport, dport=outside_port) / - Raw("flippity flop")) - self.send_and_expect(pg1, [p], pg0) - - # now close the session - - # FIN packet in -> out - p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) / - IP(src=local_host, dst=remote_host) / - TCP(sport=local_sport, dport=remote_dport, flags="FA", seq=100, - ack=300)) - self.send_and_expect(pg0, [p], pg1) - - # FIN+ACK packet out -> in - p = (Ether(src=pg1.remote_mac, dst=pg1.local_mac) / - IP(src=remote_host, dst=outside_addr) / - TCP(sport=remote_dport, dport=outside_port, flags="FA", seq=300, - ack=101)) - self.send_and_expect(pg1, [p], pg0) - - # ACK packet in -> out - p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) / - IP(src=local_host, dst=remote_host) / - TCP(sport=local_sport, dport=remote_dport, flags="A", seq=101, - ack=301)) - self.send_and_expect(pg0, [p], pg1) - - # session now in transitory timeout - # try SYN packet in->out - should be dropped - p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) / - IP(src=local_host, dst=remote_host) / - TCP(sport=local_sport, dport=remote_dport, flags="S")) - pg0.add_stream(p) - self.pg_enable_capture() - self.pg_start() - - self.sleep(new_transitory, "wait for transitory timeout") - pg0.assert_nothing_captured(0) - - # session should still exist - sessions = self.vapi.nat44_user_session_dump(pg0.remote_ip4, 0) - self.assertEqual(1, len(sessions)) - - # send FIN+ACK packet in->out - will cause session to be wiped - # but won't create a new session - p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) / - IP(src=local_host, dst=remote_host) / - TCP(sport=local_sport, dport=remote_dport, flags="FA", seq=300, - ack=101)) - pg1.add_stream(p) - self.pg_enable_capture() - self.pg_start() - pg0.assert_nothing_captured(0) - - sessions = self.vapi.nat44_user_session_dump(pg0.remote_ip4, 0) - self.assertEqual(0, len(sessions)) - - # create a new session and make sure the outside port is remapped - # SYN packet in->out - - p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) / - IP(src=local_host, dst=remote_host) / - TCP(sport=local_sport, dport=remote_dport, flags="S")) - p = self.send_and_expect(pg0, [p], pg1)[0] - - self.assertEqual(p[IP].src, outside_addr) - self.assertNotEqual(p[TCP].sport, local_sport) - - # make sure static mapping works and creates a new session - # SYN packet out->in - p = (Ether(src=pg1.remote_mac, dst=pg1.local_mac) / - IP(src=remote_host, dst=outside_addr) / - TCP(sport=remote_dport, dport=outside_port, flags="S")) - self.send_and_expect(pg1, [p], pg0) - - sessions = self.vapi.nat44_user_session_dump(pg0.remote_ip4, 0) - self.assertEqual(2, len(sessions)) - - -if __name__ == '__main__': - unittest.main(testRunner=VppTestRunner) diff --git a/test/test_nat44_ei.py b/test/test_nat44_ei.py index 5fdcf3fa3c7..74a082eee0a 100644 --- a/test/test_nat44_ei.py +++ b/test/test_nat44_ei.py @@ -605,7 +605,7 @@ class MethodHolder(VppTestCase): self.assertEqual(struct.pack("!H", self.udp_port_out), record[227]) else: - self.fail(f"Invalid protocol {scapy.compat.orb(record[4])}") + self.fail("Invalid protocol") self.assertEqual(3, nat44_ses_create_num) self.assertEqual(3, nat44_ses_delete_num) |