Stress test Home Assistant via the HTTP api

release_the_kraken.py

from multiprocessing.dummy import Pool
import sys
import time

from homeassistant import remote

# Docs: remote.API(host, [password], [port], [use_ssl])
api = remote.API('127.0.0.1')

start_time = time.time()

if len(sys.argv) < 2:
    amount = 1000
else:
    amount = int(sys.argv[1])


def job(i):
    remote.set_state(api, 'kraken.test', 'Run {}'.format(i))

    if i % 100 == 0:
        print('.', end='', flush=True)

pool = Pool(10)

try:
    pool.map(job, range(amount))
except KeyboardInterrupt:
    pass

print()
print("--- Set {} states in {:.2f} seconds ---".format(
    amount, time.time() - start_time))