inbound_load_tester.py

#!/usr/bin/python

import datetime
import random
import requests
import sys
import uuid

from multiprocessing import Pool


def text_me(timestamp):

    sender = '4471234' + ''.join([str(random.randint(0, 9)) for _ in range(5)])

    response = requests.post(
        'http://localhost:6011/notifications/sms/receive/mmg',
        json={
            'MSISDN': sender,
            'Number': '447900900999',
            'Message': str(uuid.uuid4()),
            'ID': str(uuid.uuid4()),
            'DateRecieved': timestamp,
        },
        auth=('username', 'testkey'),
    )
    return response


def time_response(n):
    timestamp = datetime.datetime.utcnow().isoformat()

    try:
        response = text_me(timestamp)
        print(
            timestamp,
            response.status_code,
            '{}ms'.format(int(response.elapsed.total_seconds() * 1000))
        )
    except requests.exceptions.ConnectionError:
        print(
            timestamp,
            "ERR",
        )


iterations = sys.argv[1]

# Needed to generate sufficient volume of requests when running on a single machine
pool_size = 16
pool = Pool(pool_size)
pool.map(time_response, range(int(iterations)))