|
#!/usr/local/bin/python2.7 |
|
# -*- coding: UTF-8 -*- |
|
# |
|
# Interface test with iperf3. |
|
# |
|
|
|
import os |
|
import re |
|
import sys |
|
from pprint import pprint |
|
from subprocess import check_output |
|
from argparse import ArgumentParser |
|
|
|
def set_parser(options="args"): |
|
|
|
parser = ArgumentParser() |
|
parser.add_argument( |
|
"-s", dest="server", action="store", |
|
help="IP of server to connect") |
|
parser.add_argument( |
|
"-t", dest="time", action="store", |
|
help="Choose time to execute the test in seconds") |
|
parser.add_argument( |
|
"-d", dest="data", action="store", |
|
help="Choose data to execute the test in MegaBytes") |
|
parser.add_argument( |
|
"-np", dest="n_ports", action="store", |
|
help="\ |
|
Choose number of ports to use for tests with more \ |
|
than one number os ports specified it separated by \ |
|
a colon(,). To test a range of ports use 1..10. \ |
|
To test a range of ports with a interval use 1..100..10, \ |
|
where the last sequence is the interval of ports.\ |
|
Range can't start with 0. \ |
|
NUMBER MAX OF PORTS IS 128", |
|
default="1") |
|
|
|
if options != "help": |
|
return parser.parse_args() |
|
else: |
|
parser.print_help() |
|
sys.exit(0) |
|
|
|
opts = set_parser() |
|
|
|
def get_data(ports): |
|
|
|
if opts.time: |
|
command = "iperf3 -c {} -t {} -P {}; exit 0".format( |
|
opts.server, opts.time, ports) |
|
elif opts.data: |
|
command = "iperf3 -c {} -n {}M -P {}; exit 0".format( |
|
opts.server, opts.data, ports) |
|
else: |
|
set_parser("help") |
|
|
|
data_pkt = check_output([command], shell=True) |
|
|
|
if int(ports) == 1: |
|
p = re.compile(r'\[\ +[0-9]+\].*bits\/sec') |
|
else: |
|
p = re.compile(r'\[SUM\].*bits\/sec') |
|
|
|
return p.findall(data_pkt) |
|
|
|
def main(): |
|
|
|
if not opts.server: |
|
set_parser("help") |
|
|
|
get_n_ports = opts.n_ports.split(',') |
|
if len(get_n_ports) == 1 and ".." in get_n_ports[0]: |
|
ranges = get_n_ports[0].split('..') |
|
range_i = int(ranges[0]) |
|
if (len(ranges) == 2): |
|
n_ports = range(range_i, int(ranges[1])) |
|
elif len(ranges) == 3: |
|
n_ports = range(range_i, int(ranges[1]), int(ranges[2])) |
|
else: |
|
n_ports = get_n_ports |
|
|
|
try: |
|
print('') |
|
for port in n_ports: |
|
|
|
if int(port) > 128 or int(port) < 1: |
|
set_parser("help") |
|
|
|
sum_data = get_data(port) |
|
if int(port) == 1: |
|
f_data = 6 |
|
else: |
|
f_data = 5 |
|
print("##### Test using {} ports #####".format(port)) |
|
print("Bandwidth min: {}".format(min([int(x.split()[f_data]) for x in sum_data]))) |
|
print("Bandwidth max: {}".format(max([int(x.split()[f_data]) for x in sum_data]))) |
|
print("Bandwidth media: {}".format(" ".join([sum_data[-1].split()[f_data], sum_data[-1].split()[f_data+1]]))) |
|
if opts.data: |
|
print("Time to transfer {} Mb: {} seconds".format(opts.data, sum_data[-1].split()[-6].split('-')[1])) |
|
|
|
print('') |
|
except ValueError, error: |
|
print ("Please, check how to use list ans range of ports...") |
|
print error, "\n" |
|
set_parser("help") |
|
except Exception, error: |
|
print sys.exc_info() |
|
|
|
if __name__ == "__main__": |
|
main() |