diff options
Diffstat (limited to 'test/test_nat44_ed_output.py')
| -rw-r--r-- | test/test_nat44_ed_output.py | 151 |
1 files changed, 89 insertions, 62 deletions
diff --git a/test/test_nat44_ed_output.py b/test/test_nat44_ed_output.py index 4ea8a5b5eda..4d75241c321 100644 --- a/test/test_nat44_ed_output.py +++ b/test/test_nat44_ed_output.py @@ -21,7 +21,8 @@ def get_nat44_ed_in2out_worker_index(ip, vpp_worker_count): class TestNAT44EDOutput(VppTestCase): - """ NAT44 ED output feature Test Case """ + """NAT44 ED output feature Test Case""" + max_sessions = 1024 @classmethod @@ -40,8 +41,7 @@ class TestNAT44EDOutput(VppTestCase): i.admin_up() i.config_ip4() i.resolve_arp() - self.vapi.nat44_ed_plugin_enable_disable(sessions=self.max_sessions, - enable=1) + self.vapi.nat44_ed_plugin_enable_disable(sessions=self.max_sessions, enable=1) def tearDown(self): if not self.vpp_dead: @@ -54,7 +54,7 @@ class TestNAT44EDOutput(VppTestCase): self.vapi.nat44_ed_plugin_enable_disable(enable=0) def test_static_dynamic(self): - """ Create static mapping which matches existing dynamic mapping """ + """Create static mapping which matches existing dynamic mapping""" config = self.vapi.nat44_show_running_config() old_timeouts = config.timeouts @@ -63,31 +63,37 @@ class TestNAT44EDOutput(VppTestCase): udp=old_timeouts.udp, tcp_established=old_timeouts.tcp_established, icmp=old_timeouts.icmp, - tcp_transitory=new_transitory) + 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_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) + 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) + flags=VppEnum.vl_api_nat_config_flags_t.NAT_IS_INSIDE, + is_add=1, + ) self.vapi.nat44_ed_add_del_output_interface( - sw_if_index=self.pg1.sw_if_index, - is_add=1) + sw_if_index=self.pg1.sw_if_index, is_add=1 + ) 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_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) @@ -100,9 +106,11 @@ class TestNAT44EDOutput(VppTestCase): # 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 = ( + 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) @@ -110,15 +118,19 @@ class TestNAT44EDOutput(VppTestCase): 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")) + 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")) + 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 @@ -126,11 +138,12 @@ class TestNAT44EDOutput(VppTestCase): is_add=1, local_ip_address=local_host, external_ip_address=outside_addr, - external_sw_if_index=0xffffffff, + 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) + 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)) @@ -138,47 +151,56 @@ class TestNAT44EDOutput(VppTestCase): # 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")) + 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")) + 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)) + 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)) + 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)) + 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")) + 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() @@ -192,10 +214,11 @@ class TestNAT44EDOutput(VppTestCase): # 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)) + 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() @@ -207,9 +230,11 @@ class TestNAT44EDOutput(VppTestCase): # 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 = ( + 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) @@ -217,14 +242,16 @@ class TestNAT44EDOutput(VppTestCase): # 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")) + 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__': +if __name__ == "__main__": unittest.main(testRunner=VppTestRunner) |