locust: do not reuse users submitting/withdrawing apps
@@ -18,23 +18,45 @@ reMyApplicationsUrl = re.compile(
)
+def user_valid(user):
+ if all(
+ key in user
+ for key in (
+ "reuse",
+ "wpName",
+ "wpPassword",
+ )
+ ):
+ return True
+ else:
+ raise Exception("login failed: credentials required.")
+ return False
+
+
+def user_reusable(user):
+ # reuse = true if the user may have multiple simultaneous sessions.
+ if user["reuse"] == True:
+ return True
+ else:
+ return False
+
+
class LoggedInUser(HttpUser):
def on_start(self):
"""
Login a random user from wpUsers.
"""
+ self.user = {}
random.shuffle(wpUsers)
- # reuse = true if the user may have multiple simultaneous sessions.
- if "reuse" in wpUsers[0] and wpUsers[0]["reuse"] == True:
+ if user_valid(wpUsers[0]):
+ if user_reusable(wpUsers[0]):
self.user = wpUsers[0]
else:
self.user = wpUsers.pop(0)
- if self.user and "wpName" in self.user and "wpPassword" in self.user:
self.login()
- else:
- raise Exception("login failed: credentials required.")
def on_stop(self):
+ if user_valid(self.user):
self.logout()
def login(self):
@@ -82,17 +104,7 @@ class LoggedInUser(HttpUser):
"/accounts/logout/?next=/",
)
# if the user wasn't reusable, return them to the pool now that we're done.
- if not (
- all(
- key in self.user
- for key in (
- "reuse",
- "wpName",
- "wpPassword",
- )
- )
- and self.user["reuse"] == True
- ):
+ if not user_reusable(self.user):
wpUsers.append(self.user)
self.user = {}
@@ -149,6 +161,7 @@ class LoggedInUser(HttpUser):
@task(1)
def post_applications(self):
+ if user_valid(self.user) and not user_reusable(self.user):
name = "/applications/request/"
with self.client.get(
name,
@@ -242,7 +255,9 @@ class LoggedInUser(HttpUser):
url = urlparse(post_app_req.url)
this_host = str(url.scheme) + "://" + str(url.netloc)
if this_host != self.host:
- post_app_req.failure("unexpected host: " + this_host)
+ post_app_req.failure(
+ "unexpected host: " + this_host
+ )
with self.client.post(
post_app_req.url,
{
@@ -399,7 +414,9 @@ class LoggedInUser(HttpUser):
)
this_host = url.scheme + "://" + url.netloc
if this_host != self.host:
- post_app_apply.failure("unexpected host: " + this_host)
+ post_app_apply.failure(
+ "unexpected host: " + this_host
+ )
if "messages" in self.client.cookies:
match = reMyApplicationsUrl.search(
self.client.cookies["messages"]
