relay-throughput-testing/src/throughput_client.py

#!/usr/bin/python3
#
import throughput_protocols
import useful
import os
import argparse
import logging
#
if __name__ == '__main__':
	logging.basicConfig(level=logging.DEBUG)
	#
	parser = argparse.ArgumentParser(description='Test the network throughput (optionally through a proxy).')
	parser.add_argument('ip', type=str, help='destination ip address')
	parser.add_argument('port', type=int, help='destination port')
	parser.add_argument('num_bytes', type=useful.parse_bytes,
	                    help='number of bytes to send (can also end with \'B\', \'KiB\', \'MiB\', or \'GiB\')', metavar='num-bytes')
	parser.add_argument('--proxy', type=str, help='proxy ip address and port', metavar=('ip','port'), nargs=2)
	parser.add_argument('--wait', type=int,
	                    help='wait until the given time before pushing data (time in seconds since epoch)', metavar='time')
	parser.add_argument('--buffer-len', type=useful.parse_bytes,
	                    help='size of the send and receive buffers (can also end with \'B\', \'KiB\', \'MiB\', or \'GiB\')', metavar='bytes')
	parser.add_argument('--no-accel', action='store_true', help='don\'t use C acceleration (use pure Python)')
	args = parser.parse_args()
	#
	endpoint = (args.ip, args.port)
	proxy = None
	#
	if args.proxy is not None:
		proxy = (args.proxy[0], int(args.proxy[1]))
	#
	username = bytes([x for x in os.urandom(12) if x != 0])
	#username = None
	#
	client = throughput_protocols.ClientProtocol(endpoint, args.num_bytes, proxy=proxy,
	                                             username=username, wait_until=args.wait,
	                                             send_buffer_len=args.buffer_len,
	                                             use_acceleration=(not args.no_accel))
	client.run()
#