371 lines
13 KiB
Python
371 lines
13 KiB
Python
"""
|
|
Locust load testing file for ATS (Applicant Tracking System)
|
|
|
|
This file contains comprehensive load testing scenarios for the ATS application,
|
|
including public access, authenticated user flows, and API endpoints.
|
|
"""
|
|
|
|
import random
|
|
import json
|
|
import time
|
|
from locust import HttpUser, task, between, events
|
|
from locust.exception import RescheduleTask
|
|
from faker import Faker
|
|
|
|
# Initialize Faker for generating realistic test data
|
|
fake = Faker()
|
|
|
|
class ATSUserBehavior(HttpUser):
|
|
"""
|
|
Base user behavior class for ATS load testing.
|
|
Simulates realistic user interactions with the system.
|
|
"""
|
|
|
|
# Wait time between tasks (1-5 seconds)
|
|
wait_time = between(1, 5)
|
|
|
|
def on_start(self):
|
|
"""Called when a simulated user starts."""
|
|
self.client.headers.update({
|
|
"User-Agent": "Locust-LoadTester/1.0",
|
|
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
|
"Accept-Language": "en-US,en;q=0.5",
|
|
"Accept-Encoding": "gzip, deflate",
|
|
"Connection": "keep-alive",
|
|
"Upgrade-Insecure-Requests": "1",
|
|
})
|
|
|
|
# Initialize user session data
|
|
self.is_logged_in = False
|
|
self.username = None
|
|
self.password = None
|
|
self.csrf_token = None
|
|
|
|
# Try to login if credentials are available
|
|
if hasattr(self.environment.parsed_options, 'login_credentials'):
|
|
self.try_login()
|
|
|
|
def try_login(self):
|
|
"""Attempt to login with provided credentials."""
|
|
if not self.is_logged_in and hasattr(self.environment.parsed_options, 'login_credentials'):
|
|
credentials = self.environment.parsed_options.login_credentials
|
|
if credentials:
|
|
# Use provided credentials or generate test ones
|
|
self.username = credentials.get('username', fake.user_name())
|
|
self.password = credentials.get('password', fake.password())
|
|
|
|
# Get login page to get CSRF token
|
|
response = self.client.get("/login/")
|
|
if response.status_code == 200:
|
|
# Extract CSRF token (simplified - in real implementation, parse HTML)
|
|
self.csrf_token = "test-csrf-token"
|
|
|
|
# Attempt login
|
|
login_data = {
|
|
'username': self.username,
|
|
'password': self.password,
|
|
'csrfmiddlewaretoken': self.csrf_token,
|
|
}
|
|
|
|
response = self.client.post("/login/", data=login_data)
|
|
if response.status_code in [200, 302]:
|
|
self.is_logged_in = True
|
|
print(f"User {self.username} logged in successfully")
|
|
else:
|
|
print(f"Login failed for user {self.username}: {response.status_code}")
|
|
|
|
class PublicUser(ATSUserBehavior):
|
|
"""
|
|
Simulates public/anonymous users browsing the ATS.
|
|
Focuses on job listings, career pages, and public information.
|
|
"""
|
|
|
|
weight = 3 # Higher weight as public users are more common
|
|
|
|
@task(3)
|
|
def view_job_listings(self):
|
|
"""Browse job listings page."""
|
|
with self.client.get("/jobs/", catch_response=True) as response:
|
|
if response.status_code == 200:
|
|
response.success()
|
|
else:
|
|
response.failure(f"Failed to load job listings: {response.status_code}")
|
|
|
|
@task(2)
|
|
def view_job_details(self):
|
|
"""View specific job details."""
|
|
# Try to view a job (assuming job slugs 1-100 exist)
|
|
job_id = random.randint(1, 100)
|
|
with self.client.get(f"/jobs/test-job-{job_id}/", catch_response=True) as response:
|
|
if response.status_code == 200:
|
|
response.success()
|
|
elif response.status_code == 404:
|
|
# Job doesn't exist, that's okay for testing
|
|
response.success()
|
|
else:
|
|
response.failure(f"Failed to load job details: {response.status_code}")
|
|
|
|
@task(1)
|
|
def view_careers_page(self):
|
|
"""View the main careers page."""
|
|
with self.client.get("/careers/", catch_response=True) as response:
|
|
if response.status_code == 200:
|
|
response.success()
|
|
else:
|
|
response.failure(f"Failed to load careers page: {response.status_code}")
|
|
|
|
@task(1)
|
|
def view_job_bank(self):
|
|
"""Browse job bank."""
|
|
with self.client.get("/jobs/bank/", catch_response=True) as response:
|
|
if response.status_code == 200:
|
|
response.success()
|
|
else:
|
|
response.failure(f"Failed to load job bank: {response.status_code}")
|
|
|
|
@task(1)
|
|
def access_application_form(self):
|
|
"""Access application form for a job."""
|
|
job_id = random.randint(1, 100)
|
|
with self.client.get(f"/application/test-job-{job_id}/", catch_response=True) as response:
|
|
if response.status_code == 200:
|
|
response.success()
|
|
elif response.status_code == 404:
|
|
response.success()
|
|
else:
|
|
response.failure(f"Failed to load application form: {response.status_code}")
|
|
|
|
class AuthenticatedUser(ATSUserBehavior):
|
|
"""
|
|
Simulates authenticated users (applicants, staff, admins).
|
|
Tests dashboard, application management, and user-specific features.
|
|
"""
|
|
|
|
weight = 2 # Medium weight for authenticated users
|
|
|
|
def on_start(self):
|
|
"""Ensure user is logged in."""
|
|
super().on_start()
|
|
if not self.is_logged_in:
|
|
# Skip authenticated tasks if not logged in
|
|
self.tasks = []
|
|
|
|
@task(3)
|
|
def view_dashboard(self):
|
|
"""View user dashboard."""
|
|
if not self.is_logged_in:
|
|
return
|
|
|
|
with self.client.get("/", catch_response=True) as response:
|
|
if response.status_code == 200:
|
|
response.success()
|
|
else:
|
|
response.failure(f"Failed to load dashboard: {response.status_code}")
|
|
|
|
@task(2)
|
|
def view_applications(self):
|
|
"""View user's applications."""
|
|
if not self.is_logged_in:
|
|
return
|
|
|
|
with self.client.get("/applications/", catch_response=True) as response:
|
|
if response.status_code == 200:
|
|
response.success()
|
|
else:
|
|
response.failure(f"Failed to load applications: {response.status_code}")
|
|
|
|
@task(2)
|
|
def browse_jobs_authenticated(self):
|
|
"""Browse jobs as authenticated user."""
|
|
if not self.is_logged_in:
|
|
return
|
|
|
|
with self.client.get("/jobs/", catch_response=True) as response:
|
|
if response.status_code == 200:
|
|
response.success()
|
|
else:
|
|
response.failure(f"Failed to load jobs: {response.status_code}")
|
|
|
|
@task(1)
|
|
def view_messages(self):
|
|
"""View user messages."""
|
|
if not self.is_logged_in:
|
|
return
|
|
|
|
with self.client.get("/messages/", catch_response=True) as response:
|
|
if response.status_code == 200:
|
|
response.success()
|
|
else:
|
|
response.failure(f"Failed to load messages: {response.status_code}")
|
|
|
|
@task(1)
|
|
def submit_application(self):
|
|
"""Submit a new application (simulated)."""
|
|
if not self.is_logged_in:
|
|
return
|
|
|
|
job_id = random.randint(1, 100)
|
|
application_data = {
|
|
'first_name': fake.first_name(),
|
|
'last_name': fake.last_name(),
|
|
'email': fake.email(),
|
|
'phone': fake.phone_number(),
|
|
'cover_letter': fake.text(max_nb_chars=500),
|
|
'csrfmiddlewaretoken': self.csrf_token or 'test-token',
|
|
}
|
|
|
|
with self.client.post(
|
|
f"/application/test-job-{job_id}/submit/",
|
|
data=application_data,
|
|
catch_response=True
|
|
) as response:
|
|
if response.status_code in [200, 302]:
|
|
response.success()
|
|
elif response.status_code == 404:
|
|
response.success() # Job doesn't exist
|
|
else:
|
|
response.failure(f"Failed to submit application: {response.status_code}")
|
|
|
|
class APIUser(ATSUserBehavior):
|
|
"""
|
|
Simulates API clients accessing the REST API endpoints.
|
|
Tests API performance under load.
|
|
"""
|
|
|
|
weight = 1 # Lower weight for API users
|
|
|
|
def on_start(self):
|
|
"""Setup API authentication."""
|
|
super().on_start()
|
|
self.client.headers.update({
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
})
|
|
|
|
# Try to get API token if credentials are available
|
|
if self.is_logged_in:
|
|
self.get_api_token()
|
|
|
|
def get_api_token(self):
|
|
"""Get API token for authenticated requests."""
|
|
# This would depend on your API authentication method
|
|
# For now, we'll simulate having a token
|
|
self.api_token = "test-api-token"
|
|
self.client.headers.update({
|
|
"Authorization": f"Bearer {self.api_token}"
|
|
})
|
|
|
|
@task(3)
|
|
def get_jobs_api(self):
|
|
"""Get jobs via API."""
|
|
with self.client.get("/api/v1/jobs/", catch_response=True) as response:
|
|
if response.status_code == 200:
|
|
response.success()
|
|
else:
|
|
response.failure(f"API jobs request failed: {response.status_code}")
|
|
|
|
@task(2)
|
|
def get_job_details_api(self):
|
|
"""Get specific job details via API."""
|
|
job_id = random.randint(1, 100)
|
|
with self.client.get(f"/api/v1/jobs/{job_id}/", catch_response=True) as response:
|
|
if response.status_code == 200:
|
|
response.success()
|
|
elif response.status_code == 404:
|
|
response.success()
|
|
else:
|
|
response.failure(f"API job details request failed: {response.status_code}")
|
|
|
|
@task(1)
|
|
def get_applications_api(self):
|
|
"""Get applications via API."""
|
|
if not self.is_logged_in:
|
|
return
|
|
|
|
with self.client.get("/api/v1/applications/", catch_response=True) as response:
|
|
if response.status_code == 200:
|
|
response.success()
|
|
else:
|
|
response.failure(f"API applications request failed: {response.status_code}")
|
|
|
|
@task(1)
|
|
def search_jobs_api(self):
|
|
"""Search jobs via API."""
|
|
search_params = {
|
|
'search': fake.job(),
|
|
'location': fake.city(),
|
|
'limit': random.randint(10, 50)
|
|
}
|
|
|
|
with self.client.get("/api/v1/jobs/", params=search_params, catch_response=True) as response:
|
|
if response.status_code == 200:
|
|
response.success()
|
|
else:
|
|
response.failure(f"API search request failed: {response.status_code}")
|
|
|
|
class FileUploadUser(ATSUserBehavior):
|
|
"""
|
|
Simulates users uploading files (resumes, documents).
|
|
Tests file upload performance and handling.
|
|
"""
|
|
|
|
weight = 1 # Lower weight for file upload operations
|
|
|
|
@task(1)
|
|
def upload_resume(self):
|
|
"""Simulate resume upload."""
|
|
if not self.is_logged_in:
|
|
return
|
|
|
|
# Create a fake file for upload
|
|
file_content = fake.text(max_nb_chars=1000).encode('utf-8')
|
|
files = {
|
|
'resume': ('resume.pdf', file_content, 'application/pdf')
|
|
}
|
|
|
|
job_id = random.randint(1, 100)
|
|
with self.client.post(
|
|
f"/applications/create/test-job-{job_id}/",
|
|
files=files,
|
|
catch_response=True
|
|
) as response:
|
|
if response.status_code in [200, 302]:
|
|
response.success()
|
|
elif response.status_code == 404:
|
|
response.success()
|
|
else:
|
|
response.failure(f"File upload failed: {response.status_code}")
|
|
|
|
# Event handlers for monitoring and logging
|
|
@events.request.add_listener
|
|
def on_request(request_type, name, response_time, response_length, response, **kwargs):
|
|
"""Log request details for analysis."""
|
|
if response and hasattr(response, 'status_code'):
|
|
status = response.status_code
|
|
else:
|
|
status = "unknown"
|
|
|
|
print(f"Request: {request_type} {name} - Status: {status} - Time: {response_time}ms")
|
|
|
|
@events.test_start.add_listener
|
|
def on_test_start(environment, **kwargs):
|
|
"""Called when test starts."""
|
|
print("=== ATS Load Test Started ===")
|
|
print(f"Target Host: {environment.host}")
|
|
print(f"Number of Users: {environment.parsed_options.num_users}")
|
|
print(f"Hatch Rate: {environment.parsed_options.hatch_rate}")
|
|
|
|
@events.test_stop.add_listener
|
|
def on_test_stop(environment, **kwargs):
|
|
"""Called when test stops."""
|
|
print("=== ATS Load Test Completed ===")
|
|
|
|
# Print summary statistics
|
|
stats = environment.stats
|
|
print(f"\nTotal Requests: {stats.total.num_requests}")
|
|
print(f"Total Failures: {stats.total.num_failures}")
|
|
print(f"Average Response Time: {stats.total.avg_response_time:.2f}ms")
|
|
print(f"Median Response Time: {stats.total.median_response_time:.2f}ms")
|
|
print(f"95th Percentile: {stats.total.get_response_time_percentile(0.95):.2f}ms")
|
|
print(f"Requests per Second: {stats.total.current_rps:.2f}")
|