summaryrefslogtreecommitdiff
path: root/service/test/load/locustfile.py
blob: 2969a2189f52d6fbed09c6add13e65daf0245c29 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
import os
import json
from random import randint

from leap.auth import SRPAuth
from leap.exceptions import SRPAuthenticationError
from locust import HttpLocust, TaskSet, task
from pixelated.resources.login_resource import LoginResource

import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

LEAP_PROVIDER = os.environ.get('LEAP_PROVIDER', 'dev.pixelated-project.org')
LEAP_SERVER_HOST = os.environ.get(
    'LEAP_SERVER_HOST',
    'https://api.%s:4430' % LEAP_PROVIDER)
LEAP_VERIFY_CERTIFICATE = os.environ.get(
    'LEAP_VERIFY_CERTIFICATE',
    '~/.leap/ca.crt')
MAX_NUMBER_USER = os.environ.get('MAX_NUMBER_USER', 80)
INVITES_FILENAME = os.environ.get('INVITES_FILENAME', '/tmp/invite_codes.txt')
INVITES_ENABLED = os.environ.get('INVITES_ENABLED', 'true') == 'true'


def load_invite_from_number(number):
    with open(INVITES_FILENAME) as invites_file:
        lines = invites_file.readlines()
        return lines[number].strip()


class UserBehavior(TaskSet):
    def __init__(self, *args, **kwargs):
        super(UserBehavior, self).__init__(*args, **kwargs)
        self.cookies = {}

    def on_start(self):
        self.login()

    def _get_or_create_user(self, number):
        srp_auth = SRPAuth(
            LEAP_SERVER_HOST,
            os.path.expanduser(LEAP_VERIFY_CERTIFICATE))
        username, password = ('loadtest%d' % number), ('password_%d' % number)
        try:
            srp_auth.authenticate(username, password)
        except SRPAuthenticationError:
            invite_code = None
            if INVITES_ENABLED:
                invite_code = load_invite_from_number(number)

            srp_auth.register(username, password, invite_code)
        return username, password

    def login(self):
        number = randint(1, int(MAX_NUMBER_USER))
        username, password = self._get_or_create_user(number)
        response = self.client.post(
            "/%s" % LoginResource.BASE_URL,
            {"username": username, "password": password},
            verify=False,
            cookies=self.cookies)
        self.cookies.update(response.cookies.get_dict())
        self.username = username

    @task(1)
    def index(self):
        self.client.get("/", verify=False)

    @task(2)
    def mail_box(self):
        self.client.get("/mails?q=tag:'inbox'&p=1&w=25", verify=False)

    @task(3)
    def send_mail(self):
        payload = {
            "tags": ["drafts"],
            "body": "some text lorem ipsum",
            "attachments": [],
            "ident": "",
            "header": {
                "to": ["%s@%s" % (self.username, LEAP_PROVIDER)],
                "cc": [],
                "bcc": [],
                "subject": "load testing"}}

        self.cookies.update(
            self.client.get("/", verify=False).cookies.get_dict())
        with self.client.post(
            '/mails',
            json=payload,
            catch_response=True,
            cookies=self.cookies,
            headers={
                'X-Requested-With': 'XMLHttpRequest',
                'X-XSRF-TOKEN': self.cookies.get('XSRF-TOKEN', '')}) as email_response:
            if email_response.status_code == 201:
                email_id = json.loads(email_response.content)['ident']
                print email_id
                self.delete_mail(email_id)
            else:
                email_response.failure(
                    'Error: email not Sent, status code: %s' % (
                        email_response.status_code))

    def delete_mail(self, ident):
        payload = {"idents": [ident]}
        self.client.post(
            '/mails/delete',
            json=payload,
            cookies=self.cookies,
            verify=False,
            headers={
                'X-Requested-With': 'XMLHttpRequest',
                'X-XSRF-TOKEN': self.cookies.get('XSRF-TOKEN', '')})


class WebsiteUser(HttpLocust):
    task_set = UserBehavior
    min_wait = 1000
    max_wait = 5000