Project 'savoirfairelinux/ring-daemon' was moved to 'savoirfairelinux/jami-daemon'. Please update any links and bookmarks that may still have the old path.
Select Git revision
test_network.py
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
test_network.py 3.05 KiB
import subprocess
import time
import logging
import psutil
import re
import matplotlib.pyplot as plt
import argparse
def _remove_rate(interface):
cmd = ["sudo",
"tc", "qdisc", "del", "dev", interface, "root"]
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
def _change_bitrate(logger, interface, bitrate):
logger.warning("setting bitrate %s", bitrate)
cmd = ["sudo",
"tc", "qdisc", "add", "dev", interface, "root", "tbf", "rate",
bitrate+"kbit", "latency", "50ms", "burst", bitrate+"b"]
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
def _init_bitrate_list(first_val, last_val, inc):
return range(first_val, last_val, inc)
def test_network(logger, interface, time_between_test, path_image):
logger.warning("running test_case_network_going_down on %s", interface)
max = 5000
min = 2000
inc = 200
list_total_dropped = []
list_bits_sent_per_sec = []
list_bitrate = _init_bitrate_list(max, min, -inc)
bits_sent_offset = _add_eth_stats(logger, interface, 0, 0, list_bits_sent_per_sec)
for bitrate in list_bitrate:
_change_bitrate(logger, interface, str(bitrate))
time.sleep(time_between_test)
_add_tc_stats(logger, interface, list_total_dropped)
bits_sent_offset = _add_eth_stats(logger, interface, bits_sent_offset, time_between_test, list_bits_sent_per_sec)
_remove_rate(interface)
fig, ax = plt.subplots()
plt.plot(list_bitrate, ':b', label='bitrate')
plt.plot(list_total_dropped, 'r-', label='pkt dropped')
plt.plot(list_bits_sent_per_sec, 'g-', label='bits sent')
plt.grid(True)
legend = plt.legend()
if path_image:
plt.savefig(path_image)
else:
plt.show()
def _add_tc_stats(logger, interface, list_total_dropped):
cmd = ["tc", "-s", "qdisc", "show", "dev", interface]
stdout = subprocess.check_output(cmd)
m = re.search('dropped (.+?),', stdout)
if m:
list_total_dropped.append(int(m.group(1)))
else:
list_total_dropped.append(0)
def _add_eth_stats(logger, interface, offset, period, list_bits_sent_per_sec):
ret = psutil.net_io_counters(pernic=True)[interface][0]
if period != 0:
rate = 8 * (ret - offset) / (period * 1000.0)
list_bits_sent_per_sec.append(rate)
return ret
def init_logger():
logger = logging.getLogger('test_network')
logger.setLevel(logging.WARNING)
steam_handler = logging.StreamHandler()
steam_handler.setLevel(logging.WARNING)
logger.addHandler(steam_handler)
return logger
def main():
logger = init_logger()
parser = argparse.ArgumentParser()
parser.add_argument('-i', '--interface', help='interface', required=True)
parser.add_argument('-d', '--delay', help='time between test', required=True)
parser.add_argument('-w', '--write', help='path to save images')
args = parser.parse_args()
_remove_rate(args.interface)
test_network(logger, args.interface, int(args.delay), args.write)
if __name__ == '__main__':
main()