Whenever you want to quickly bombard a URL with some concurrent traffic, you can use this:
import random
import time
import requests
import concurrent.futures
def _get_size(url):
sleep = random.random() / 10
# print("sleep", sleep)
time.sleep(sleep)
r = requests.get(url)
# print(r.status_code)
assert len(r.text)
return len(r.text)
def run(url, times=10):
sizes = []
futures = []
with concurrent.futures.ThreadPoolExecutor() as executor:
for _ in range(times):
futures.append(executor.submit(_get_size, url))
for future in concurrent.futures.as_completed(futures):
sizes.append(future.result())
return sizes
if __name__ == "__main__":
import sys
print(run(sys.argv[1]))
It's really basic but it works wonderfully. It starts 10 concurrent threads that all hit the same URL at almost the same time.
I've been using this stress test a local Django server to test some atomicity writes with the file system.
Comments