#!/usr/bin/env python3 # # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # # iperf3 integration test # Description: Test for ipertf3 # Author: Susant Sahani # # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # # Copyright (c) 2018 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing # to use, modify, copy, or redistribute it subject to the terms # and conditions of the GNU General Public License version 2. # # This program is distributed in the hope that it will be # useful, but WITHOUT ANY WARRANTY; without even the implied # warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR # PURPOSE. See the GNU General Public License for more details. # # You should have received a copy of the GNU General Public # License along with this program; if not, write to the Free # Software Foundation, Inc., 51 Franklin Street, Fifth Floor, # Boston, MA 02110-1301, USA. # # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ import errno import os import sys import time import unittest import subprocess import signal import shutil import re import psutil IPERF3_SERVER_LOGS='/var/run/iperf3-test-log' IPERF3_PID_FILE='/var/run/iperf3-test-pid' TEST_COMPLETE='Test Complete. Summary Results:' SERVER_TEST_PROTOCOL_TCP_MSG='Starting Test: protocol: TCP' SERVER_TEST_PROTOCOL_UDP_MSG='Starting Test: protocol: UDP' SERVER_LISTENING_PORT_5202_MSG='Server listening on 5202' SERVER_LISTENING_PORT_5201_MSG='Server listening on 5201' SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_MSG='local 127.0.0.1 port 5202 connected to 127.0.0.1 port \d+' SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_IPV6_MSG='local ::1 port 5202 connected to ::1 port \d+' SERVER_CONNECTED_LOCAL_PORT_5201_WITH_REMOTE_PORT_MSG='local 127.0.0.1 port 5201 connected to 127.0.0.1 port \d+' SERVER_INCOMING_CONNECTION_MSG='Accepted connection from 127.0.0.1, port \d+' SERVER_INCOMING_CONNECTION_IPV6_MSG='Accepted connection from ::1, port \d+' CLIENT_CONNECTED_TO_SERVER_PORT_5202_MSG='local 127.0.0.1 port \d+ connected to 127.0.0.1 port 5202' CLIENT_CONNECTED_TO_SERVER_PORT_5201_MSG='local 127.0.0.1 port \d+ connected to 127.0.0.1 port 5201' CLIENT_CONNECTED_TO_SERVER_PORT_5202_IPV6_MSG='local ::1 port \d+ connected to ::1 port 5202' CLIENT_TEST_COMPLETE_MSG='iperf Done' def setUpModule(): """Initialize the environment, and perform sanity checks on it.""" if shutil.which('iperf3') is None: raise OSError(errno.ENOENT, 'iperf3 not found') for pid in psutil.pids(): p = psutil.Process(pid) if p.name() == "iperf3": raise unittest.SkipTest('iperf3 is already active') class Iperf3ServerLogUtilities(): """Provide a set of utility functions to facilitate iperf3 tests. This class must be inherited along with unittest.TestCase to define some required methods. """ def findTextInServerLog(self, **kwargs): """Look various success lines from iperf3 server logs.""" if kwargs is not None: with open (IPERF3_SERVER_LOGS, 'rt') as in_file: contents = in_file.read() for key in kwargs: self.assertRegex(contents, kwargs[key]) def KillIperf3Server(self): time.sleep(3) try: with open(IPERF3_PID_FILE, 'r') as f: pid = f.read().rstrip(' \t\r\n\0') os.kill(int(pid), signal.SIGTERM) os.remove(IPERF3_SERVER_LOGS) os.remove(IPERF3_PID_FILE) except IOError: pass def VerifyPortIsOpenAuto(self, port): """ Tests whether iperf3 is listening on given port """ netstat_output = subprocess.check_output(['netstat', '-antpl']).rstrip().decode('utf-8') self.assertIsNotNone(netstat_output) match='tcp.*' + str(port) +'.*LISTEN.*/iperf3' self.assertRegex(netstat_output, match) class Iperf3TestsBindToInterface(unittest.TestCase, Iperf3ServerLogUtilities): def setUp(self): """Setup veth interface""" subprocess.check_output(['ip', 'link', 'add', 'iperf-veth', 'type', 'veth', 'peer', 'name', 'iperf-veth-peer']) subprocess.check_output(['ip', 'addr', 'add', '192.168.50.5', 'dev', 'iperf-veth']) subprocess.check_output(['ip', 'addr', 'add', '192.168.50.6', 'dev', 'iperf-veth-peer']) subprocess.check_output(['ip', 'link', 'set', 'iperf-veth', 'up']) subprocess.check_output(['ip', 'link', 'set', 'iperf-veth-peer', 'up']) """Start iperf3.""" subprocess.check_output(['iperf3', '-s', '--daemon', '--verbose' , '--logfile', '/var/run/iperf3-test-log', '--pidfile', '/var/run/iperf3-test-pid', '--bind', '192.168.50.5']) time.sleep(3) self.VerifyPortIsOpenAuto(5201) def tearDown(self): self.KillIperf3Server() subprocess.check_output(['ip', 'link', 'del', 'iperf-veth']) def test_tcp_client_bind_to_interface_default_port(self): client_output = subprocess.check_output(['iperf3', '-c', '192.168.50.5', '--bind', '192.168.50.6']).rstrip().decode('utf-8') self.assertRegex(client_output, 'Connecting to host 192.168.50.5, port 5201') self.assertRegex(client_output, 'local 192.168.50.6 port \d+ connected to 192.168.50.5 port 5201') self.assertRegex(client_output, CLIENT_TEST_COMPLETE_MSG) self.findTextInServerLog(test1='Accepted connection from 192.168.50.6, port \d+', test2='local 192.168.50.5 port 5201 connected to 192.168.50.6 port \d+', test3=SERVER_TEST_PROTOCOL_TCP_MSG, test4=TEST_COMPLETE) class Iperf3Tests(unittest.TestCase, Iperf3ServerLogUtilities): def setUp(self): """Start iperf3.""" subprocess.check_output(['iperf3', '-s', '-p', '5202', '--daemon', '--verbose' , '--logfile', '/var/run/iperf3-test-log', '--pidfile', '/var/run/iperf3-test-pid']) time.sleep(3) self.VerifyPortIsOpenAuto(5202) def tearDown(self): self.KillIperf3Server() def test_tcp_client_custom_port_5202(self): client_output = subprocess.check_output(['iperf3', '-c', '127.0.0.1', '-p', '5202']).rstrip().decode('utf-8') self.assertRegex(client_output, CLIENT_CONNECTED_TO_SERVER_PORT_5202_MSG) self.assertRegex(client_output, CLIENT_TEST_COMPLETE_MSG) self.findTextInServerLog(test1=SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_MSG, test2=SERVER_INCOMING_CONNECTION_MSG, test3=SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_MSG, test4=SERVER_TEST_PROTOCOL_TCP_MSG, test5=TEST_COMPLETE) def test_udp_client_custom_port_5202(self): client_output = subprocess.check_output(['iperf3', '-c', '127.0.0.1', '-u', '-p', '5202']).rstrip().decode('utf-8') self.assertRegex(client_output, CLIENT_CONNECTED_TO_SERVER_PORT_5202_MSG) self.assertRegex(client_output, "Total Datagrams") self.assertRegex(client_output, CLIENT_TEST_COMPLETE_MSG) self.findTextInServerLog(test1=SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_MSG, test2=SERVER_INCOMING_CONNECTION_MSG, test3=SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_MSG, test4=SERVER_TEST_PROTOCOL_UDP_MSG, test5=TEST_COMPLETE) def test_tcp_client_custom_port_5202_time_15sec(self): client_output = subprocess.check_output(['iperf3', '-c', '127.0.0.1', '-p', '5202', '-t','15']).rstrip().decode('utf-8') self.assertRegex(client_output, CLIENT_CONNECTED_TO_SERVER_PORT_5202_MSG) self.assertRegex(client_output, CLIENT_TEST_COMPLETE_MSG) self.findTextInServerLog(test1=SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_MSG, test2=SERVER_INCOMING_CONNECTION_MSG, test3=SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_MSG, test4=SERVER_TEST_PROTOCOL_TCP_MSG, test5='15 second test', test6=TEST_COMPLETE) def test_tcp_client_port_5202_ipv6(self): client_output = subprocess.check_output(['iperf3', '-c', '::1', '-p', '5202']).rstrip().decode('utf-8') self.assertRegex(client_output, CLIENT_CONNECTED_TO_SERVER_PORT_5202_IPV6_MSG) self.assertRegex(client_output, CLIENT_TEST_COMPLETE_MSG) self.findTextInServerLog(test1=SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_IPV6_MSG, test2=SERVER_INCOMING_CONNECTION_IPV6_MSG, test3=SERVER_TEST_PROTOCOL_TCP_MSG, test4=TEST_COMPLETE) def test_udp_client_port_5202_ipv6(self): client_output = subprocess.check_output(['iperf3', '-c', '::1', '-u', '-p', '5202']).rstrip().decode('utf-8') self.assertRegex(client_output, CLIENT_CONNECTED_TO_SERVER_PORT_5202_IPV6_MSG) self.assertRegex(client_output, CLIENT_TEST_COMPLETE_MSG) self.findTextInServerLog(test1=SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_IPV6_MSG, test2=SERVER_INCOMING_CONNECTION_IPV6_MSG, test3=SERVER_TEST_PROTOCOL_UDP_MSG, test4=TEST_COMPLETE) def test_tcp_client_custom_port_5202_parallel(self): client_output = subprocess.check_output(['iperf3', '-c', '127.0.0.1', '-p', '5202', '--parallel', '10']).rstrip().decode('utf-8') self.assertRegex(client_output, CLIENT_CONNECTED_TO_SERVER_PORT_5202_MSG) self.assertRegex(client_output, CLIENT_TEST_COMPLETE_MSG) self.findTextInServerLog(test1=SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_MSG, test2=SERVER_INCOMING_CONNECTION_MSG, test3=SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_MSG, test4='Starting Test: protocol: TCP, 10 streams', test5=TEST_COMPLETE) def test_tcp_client_custom_port_5202_reverse(self): client_output = subprocess.check_output(['iperf3', '-c', '127.0.0.1', '-p', '5202', '-R']).rstrip().decode('utf-8') self.assertRegex(client_output, CLIENT_CONNECTED_TO_SERVER_PORT_5202_MSG) self.assertRegex(client_output, 'Reverse mode, remote host 127.0.0.1 is sending') self.assertRegex(client_output, CLIENT_TEST_COMPLETE_MSG) self.findTextInServerLog(test1=SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_MSG, test2=SERVER_INCOMING_CONNECTION_MSG, test3=SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_MSG, test4=TEST_COMPLETE) def test_tcp_client_custom_port_5202_dscp(self): client_output = subprocess.check_output(['iperf3', '-c', '127.0.0.1', '-p', '5202', '--dscp', '22']).rstrip().decode('utf-8') self.assertRegex(client_output, CLIENT_CONNECTED_TO_SERVER_PORT_5202_MSG) self.assertRegex(client_output, CLIENT_TEST_COMPLETE_MSG) self.findTextInServerLog(test1=SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_MSG, test2=SERVER_INCOMING_CONNECTION_MSG, test3=SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_MSG, test4='tos 22', test5=TEST_COMPLETE) def test_tcp_client_custom_port_5202_mss(self): client_output = subprocess.check_output(['iperf3', '-c', '127.0.0.1', '-p', '5202', '--set-mss', '400']).rstrip().decode('utf-8') self.assertRegex(client_output, CLIENT_CONNECTED_TO_SERVER_PORT_5202_MSG) self.assertRegex(client_output, CLIENT_TEST_COMPLETE_MSG) self.findTextInServerLog(test1=SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_MSG, test2=SERVER_INCOMING_CONNECTION_MSG, test3=SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_MSG, test4='TCP MSS: 400', test5=TEST_COMPLETE) def test_tcp_client_custom_port_5202_title(self): client_output = subprocess.check_output(['iperf3', '-c', '127.0.0.1', '-p', '5202', '--title', 'Susant']).rstrip().decode('utf-8') self.assertRegex(client_output, CLIENT_CONNECTED_TO_SERVER_PORT_5202_MSG) self.assertRegex(client_output, "Susant") self.assertRegex(client_output, CLIENT_TEST_COMPLETE_MSG) self.findTextInServerLog(test1=SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_MSG, test2=SERVER_INCOMING_CONNECTION_MSG, test3=SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_MSG, test4=TEST_COMPLETE) def test_tcp_client_custom_port_5202_get_server_output(self): client_output = subprocess.check_output(['iperf3', '-c', '127.0.0.1', '-p', '5202', '--get-server-output']).rstrip().decode('utf-8') self.assertRegex(client_output, "Server output:") self.assertRegex(client_output, CLIENT_TEST_COMPLETE_MSG) self.findTextInServerLog(test1=SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_MSG, test2=SERVER_INCOMING_CONNECTION_MSG, test5=TEST_COMPLETE) def test_tcp_client_custom_port_5202_custom_client_port(self): client_output = subprocess.check_output(['iperf3', '-c', '127.0.0.1', '-p', '5202', '--bind', '127.0.0.1', '--cport', '5203']).rstrip().decode('utf-8') self.assertRegex(client_output, CLIENT_CONNECTED_TO_SERVER_PORT_5202_MSG) self.assertRegex(client_output, CLIENT_TEST_COMPLETE_MSG) self.findTextInServerLog(test1=SERVER_CONNECTED_LOCAL_PORT_5202_WITH_REMOTE_PORT_MSG, test2='port 5203', test3=TEST_COMPLETE) class Iperf3SanityTests(unittest.TestCase, Iperf3ServerLogUtilities): def setUp(self): """Start iperf3 default TCP mode.""" subprocess.check_output(['iperf3', '--daemon', '--server', '--verbose' , '--logfile', '/var/run/iperf3-test-log', '--pidfile', '/var/run/iperf3-test-pid']) time.sleep(3) self.VerifyPortIsOpenAuto(5201) def tearDown(self): self.KillIperf3Server() def test_basic_tcp_client_default_port_5201(self): client_output = subprocess.check_output(['iperf3', '-c', '127.0.0.1',]).rstrip().decode('utf-8') self.assertRegex(client_output, CLIENT_CONNECTED_TO_SERVER_PORT_5201_MSG) self.assertRegex(client_output, CLIENT_TEST_COMPLETE_MSG) self.findTextInServerLog(test1=SERVER_LISTENING_PORT_5201_MSG, test2=SERVER_INCOMING_CONNECTION_MSG, test3=SERVER_CONNECTED_LOCAL_PORT_5201_WITH_REMOTE_PORT_MSG, test4=SERVER_TEST_PROTOCOL_TCP_MSG, test5=TEST_COMPLETE) def test_server_log_file_exists(self): self.assertTrue(os.path.exists(IPERF3_SERVER_LOGS)) def test_server_pid_file_exists(self): self.assertTrue(os.path.exists(IPERF3_PID_FILE)) def test_server_daemon_pid_equal_to_pid_file(self): with open(IPERF3_PID_FILE, 'r') as f: pid = f.read().rstrip(' \t\r\n\0') proc_path = os.path.join('/proc/', pid) self.assertTrue(os.path.exists(proc_path)) process = psutil.Process(int(pid)) self.assertEqual(process.name(), 'iperf3') f.close() def test_basic_udp_client_default_port_5201(self): client_output = subprocess.check_output(['iperf3', '-c', '127.0.0.1', '-u']).rstrip().decode('utf-8') self.assertRegex(client_output, CLIENT_CONNECTED_TO_SERVER_PORT_5201_MSG) self.assertRegex(client_output, 'Total Datagrams') self.assertRegex(client_output, CLIENT_TEST_COMPLETE_MSG) self.findTextInServerLog(test1=SERVER_LISTENING_PORT_5201_MSG, test2=SERVER_INCOMING_CONNECTION_MSG, test3=SERVER_CONNECTED_LOCAL_PORT_5201_WITH_REMOTE_PORT_MSG, test4=SERVER_TEST_PROTOCOL_UDP_MSG, test5=TEST_COMPLETE) if __name__ == '__main__': unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout, verbosity=2))