simple-load-test

login.py

import os

import faker
import locust
import pyquery

fake = faker.Faker()

username, password = os.getenv('AUTH_USER'), os.getenv('AUTH_PASS')
auth = (username, password) if username and password else ()

class UserBehavior(locust.TaskSet):

    @locust.task
    def signup(self):
        resp = self.client.get('users/sign_up', auth=auth)
        resp.raise_for_status()
        dom = pyquery.PyQuery(resp.content)
        auth_token = dom.find('input[name="authenticity_token"]')[0].attrib['value']
        data = {
            'user[email]': fake.email(),
            'authenticity_token': auth_token,
            'commit': 'Sign up',
        }
        self.client.post('users', data=data, auth=auth)

class WebsiteUser(locust.HttpLocust):
    task_set = UserBehavior
    min_wait = 5000
    max_wait = 10000

Minor edit (update faker lib, add a few more requests that a typical user would make en route to acc creation, send email to single gmail account using + char to differentiate accs.)

import os

from faker import Factory
import locust
import pyquery

fake = Factory.create()

username, password = os.getenv('AUTH_USER'), os.getenv('AUTH_PASS')
auth = (username, password) if username and password else ()

class UserBehavior(locust.TaskSet):

    @locust.task
    def signup(self):
        resp = self.client.get('/', auth=auth)
        resp.raise_for_status()

        resp = self.client.get('/start', auth=auth)
        resp.raise_for_status()

        resp = self.client.get('/users/sign_up', auth=auth)
        resp.raise_for_status()

        dom = pyquery.PyQuery(resp.content)
        auth_token = dom.find('input[name="authenticity_token"]')[0].attrib['value']
        data = {
            'user[email]': 'jgrevich.gsa+' + fake.md5() + '@gmail.com',
            'authenticity_token': auth_token,
            'commit': 'Sign up',
        }
        self.client.post('/users', data=data, auth=auth)

class WebsiteUser(locust.HttpLocust):
    task_set = UserBehavior
    min_wait = 5000
    max_wait = 10000

An issue I found with using a single gmail account is that when sending 25k emails ( 😁 ) gmail batches/throttles incoming emails and you do not necessarily receive them in order. I received emails for ~6hrs after the perf test.

With confirmation via gmail:

import os
import re
import time
import uuid
import base64
import httplib2

from apiclient import discovery
import oauth2client
from oauth2client import client
from oauth2client import tools

import locust
import pyquery

username, password = os.getenv('AUTH_USER'), os.getenv('AUTH_PASS')
auth = (username, password) if username and password else ()

email_template = os.getenv('EMAIL_TEMPLATE')

SCOPES = 'https://www.googleapis.com/auth/gmail.readonly'
CLIENT_SECRET_FILE = 'client-secret.json'
APPLICATION_NAME = 'login-test'

# https://developers.google.com/gmail/api/quickstart/python
def get_credentials():
    """Gets valid user credentials from storage.

    If nothing has been stored, or if the stored credentials are invalid,
    the OAuth2 flow is completed to obtain the new credentials.

    Returns:
        Credentials, the obtained credential.
    """
    home_dir = os.path.expanduser('~')
    credential_dir = os.path.join(home_dir, '.credentials')
    os.makedirs(credential_dir, exist_ok=True)
    credential_path = os.path.join(credential_dir,
                                   'gmail-python-quickstart.json')

    store = oauth2client.file.Storage(credential_path)
    credentials = store.get()
    if not credentials or credentials.invalid:
        flow = client.flow_from_clientsecrets(CLIENT_SECRET_FILE, SCOPES)
        flow.user_agent = APPLICATION_NAME
        credentials = tools.run_flow(flow, store)
        print('Storing credentials to ' + credential_path)
    return credentials

token_pattern = re.compile(r'confirmation_token=([\w-]+)')

def confirm_email(username):
    credentials = get_credentials()
    http = credentials.authorize(httplib2.Http())
    service = discovery.build('gmail', 'v1', http=http)
    messages = service.users().messages()
    msgs = messages.list(userId='me', q='to:{}'.format(username)).execute()
    if not msgs['resultSizeEstimate']:
        return None
    msg = msgs['messages'][0]
    msg = messages.get(userId='me', id=msg['id'], format='full').execute()
    parsed = base64.urlsafe_b64decode(msg['payload']['parts'][0]['body']['data'])
    return token_pattern.search(parsed.decode('utf-8')).groups()[0]

class UserBehavior(locust.TaskSet):

    @locust.task
    def signup(self):
        resp = self.client.get('users/sign_up', auth=auth)
        resp.raise_for_status()

        dom = pyquery.PyQuery(resp.content)
        auth_token = dom.find('input[name="authenticity_token"]')[0].attrib['value']
        email = email_template.format(uuid.uuid4())
        data = {
            'user[email]': email,
            'authenticity_token': auth_token,
            'commit': 'Sign up',
        }
        resp = self.client.post('users', data=data, auth=auth)
        resp.raise_for_status()

        confirm_token = None
        for tries in range(5):
            confirm_token = confirm_email(email)
            if confirm_token is not None:
                break
            time.sleep(3)
        print('confirmation_token: {}'.format(confirm_token))
        if confirm_token is None:
            return

        resp = self.client.get(
            'users/confirmation',
            params={'confirmation_token': confirm_token},
            name='/users/confirmation?confirmation_token=',
            auth=auth,
        )
        resp.raise_for_status()

class WebsiteUser(locust.HttpLocust):
    task_set = UserBehavior
    min_wait = 5000
    max_wait = 10000

I'm getting No module named faker when I try to run this. What are the prerequisites for running this script? I tried pip install faker but that didn't help.