Skip to content
Snippets Groups Projects
Select Git revision
  • 03cc5544c2a9fa39e23483c2a1d8d07f8fb0dc24
  • master default protected
  • release/202005
  • release/202001
  • release/201912
  • release/201911
  • release/releaseWindowsTestOne
  • release/windowsReleaseTest
  • release/releaseTest
  • release/releaseWindowsTest
  • release/201910
  • release/qt/201910
  • release/windows-test/201910
  • release/201908
  • release/201906
  • release/201905
  • release/201904
  • release/201903
  • release/201902
  • release/201901
  • release/201812
  • 4.0.0
  • 2.2.0
  • 2.1.0
  • 2.0.1
  • 2.0.0
  • 1.4.1
  • 1.4.0
  • 1.3.0
  • 1.2.0
  • 1.1.0
31 results

alaw.cpp

Blame
  • 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()