{
return new Promise((resolve) => {
axe.run(config, (err, results) => {
if (err) {
resolve(null);
} else {
resolve(results);
}
});
});
}
""", axe_config)
return results
except Exception as e:
print(f"Error running axe scan: {str(e)}")
return None
def _process_axe_violations(self, page_url, violations):
"""Process and log axe violations"""
processed_violations = []
for violation in violations:
violation_data = {
'page': page_url,
'rule_id': violation.get('id', 'unknown'),
'impact': violation.get('impact', 'unknown'),
'description': violation.get('description', ''),
'help': violation.get('help', ''),
'help_url': violation.get('helpUrl', ''),
'nodes_count': len(violation.get('nodes', [])),
'tags': violation.get('tags', [])
}
processed_violations.append(violation_data)
self.accessibility_violations.append(violation_data)
# Log violation details
impact = violation_data['impact'].upper()
rule_id = violation_data['rule_id']
description = violation_data['description']
nodes_count = violation_data['nodes_count']
print(f"AXE VIOLATION [{impact}]: {rule_id} - {description} ({nodes_count} elements) on {page_url}")
return processed_violations
@task(1)
@pw
async def run_comprehensive_axe_scan(self, page: PageWithRetry):
"""Run comprehensive axe scan with all rules"""
async with event(self, "Comprehensive Axe Scan"):
await page.goto('/')
# Inject axe-core
await self._inject_axe_core(page)
# Run comprehensive scan with all rules
comprehensive_results = await page.evaluate("""
() => {
return new Promise((resolve) => {
// Run all axe rules
axe.run((err, results) => {
if (err) {
resolve(null);
} else {
resolve({
violations: results.violations.length,
passes: results.passes.length,
incomplete: results.incomplete.length,
inapplicable: results.inapplicable.length,
total_rules: results.violations.length + results.passes.length + results.incomplete.length + results.inapplicable.length
});
}
});
});
}
""")
if comprehensive_results:
print(f"COMPREHENSIVE AXE SCAN: {comprehensive_results['violations']} violations, "
f"{comprehensive_results['passes']} passes, "
f"{comprehensive_results['total_rules']} total rules checked")
def on_stop(self):
"""Final axe accessibility report"""
print("\n" + "="*50)
print("AXE ACCESSIBILITY TEST COMPLETE")
print("="*50)
print(f"Pages tested: {len(self.discovered_pages) + 1}")
print(f"Total axe violations: {len(self.accessibility_violations)}")
if self.accessibility_violations:
# Group violations by impact
impact_counts = {}
for violation in self.accessibility_violations:
impact = violation['impact']
impact_counts[impact] = impact_counts.get(impact, 0) + 1
print(f"\nVIOLATIONS BY IMPACT:")
for impact, count in sorted(impact_counts.items()):
print(f" {impact.upper()}: {count}")
print(f"\nTOP VIOLATIONS:")
for violation in self.accessibility_violations[:5]:
print(f"❌ [{violation['impact'].upper()}] {violation['rule_id']}: {violation['description']}")
print(f" Page: {violation['page']} ({violation['nodes_count']} elements)")
else:
print("✅ No accessibility violations found!")
```
## Comprehensive Axe-Core Site Scanning
```python theme={null}
from locust import task
from locust_plugins.users.playwright import PlaywrightUser, PageWithRetry, pw, event
import json
class ComprehensiveAxeSiteTester(PlaywrightUser):
def on_start(self):
"""Initialize comprehensive axe site testing"""
self.all_violations = []
self.pages_scanned = set()
self.pages_to_scan = []
self.scan_summary = {
'total_pages': 0,
'total_violations': 0,
'critical_violations': 0,
'serious_violations': 0,
'moderate_violations': 0,
'minor_violations': 0
}
@task(3)
@pw
async def comprehensive_site_scan(self, page: PageWithRetry):
"""Comprehensive axe scan across entire site"""
if not self.pages_to_scan:
await self._discover_all_pages(page)
if self.pages_to_scan:
page_url = self.pages_to_scan.pop(0)
if page_url not in self.pages_scanned:
await self._comprehensive_axe_test(page, page_url)
self.pages_scanned.add(page_url)
async def _discover_all_pages(self, page: PageWithRetry):
"""Discover all internal pages for comprehensive testing"""
async with event(self, "Site Discovery"):
await page.goto('/')
# Get all internal links
links = await page.locator('a[href]').all()
for link in links[:50]: # Limit to 50 pages for comprehensive testing
try:
href = await link.get_attribute('href')
if href and self._is_internal_page(href, page.url):
normalized_url = self._normalize_url(href, page.url)
if normalized_url and normalized_url not in self.pages_to_scan:
self.pages_to_scan.append(normalized_url)
except:
pass
print(f"Discovered {len(self.pages_to_scan)} pages for comprehensive axe scanning")
def _is_internal_page(self, href, base_url):
"""Check if URL is internal page"""
if not href:
return False
# Skip anchors, external protocols, and resources
skip_patterns = ['#', 'mailto:', 'tel:', 'javascript:', '.pdf', '.jpg', '.jpeg', '.png', '.gif', '.css', '.js', '.zip', '.doc', '.xml']
if any(pattern in href.lower() for pattern in skip_patterns):
return False
# Check if internal
if href.startswith('/'):
return True
elif href.startswith('http'):
base_domain = base_url.split('/')[2]
return base_domain in href
else:
return True # Relative links
def _normalize_url(self, href, base_url):
"""Normalize URL for consistency"""
if href.startswith('/'):
return href
elif href.startswith('http'):
base_domain = base_url.split('/')[2]
if base_domain in href:
return '/' + href.split(base_domain, 1)[1].lstrip('/')
else:
# Relative URL
return '/' + href.lstrip('./')
return None
async def _comprehensive_axe_test(self, page: PageWithRetry, page_url: str):
"""Comprehensive axe test on a single page"""
async with event(self, f"COMPREHENSIVE AXE: {page_url}"):
await page.goto(page_url)
# Inject axe-core
await self._inject_axe_core(page)
# Run comprehensive axe scan with all rule sets
axe_results = await self._run_comprehensive_axe_scan(page)
if axe_results:
self._process_comprehensive_results(page_url, axe_results)
violations = axe_results.get('violations', [])
if violations:
critical = len([v for v in violations if v.get('impact') == 'critical'])
serious = len([v for v in violations if v.get('impact') == 'serious'])
failure_msg = f"❌ {len(violations)} axe violations ({critical} critical, {serious} serious)"
raise Exception(failure_msg)
else:
print(f"✅ {page_url} - No violations found")
async def _inject_axe_core(self, page: PageWithRetry):
"""Inject axe-core library"""
await page.add_script_tag(url="https://cdnjs.cloudflare.com/ajax/libs/axe-core/4.8.2/axe.min.js")
await page.wait_for_function("typeof axe !== 'undefined'")
async def _run_comprehensive_axe_scan(self, page: PageWithRetry):
"""Run comprehensive axe scan with all rules"""
try:
# Configure comprehensive scan
axe_config = {
'runOnly': {
'type': 'tag',
'values': ['wcag2a', 'wcag2aa', 'wcag21aa', 'wcag22aa', 'best-practice', 'experimental']
}
}
results = await page.evaluate("""
(config) => {
return new Promise((resolve) => {
axe.run(config, (err, results) => {
if (err) {
resolve(null);
} else {
resolve(results);
}
});
});
}
""", axe_config)
return results
except Exception as e:
print(f"Error running comprehensive axe scan: {str(e)}")
return None
def _process_comprehensive_results(self, page_url, results):
"""Process comprehensive axe results"""
violations = results.get('violations', [])
self.scan_summary['total_pages'] += 1
self.scan_summary['total_violations'] += len(violations)
for violation in violations:
impact = violation.get('impact', 'unknown')
# Count by impact
if impact == 'critical':
self.scan_summary['critical_violations'] += 1
elif impact == 'serious':
self.scan_summary['serious_violations'] += 1
elif impact == 'moderate':
self.scan_summary['moderate_violations'] += 1
elif impact == 'minor':
self.scan_summary['minor_violations'] += 1
# Store detailed violation
violation_detail = {
'page': page_url,
'rule_id': violation.get('id'),
'impact': impact,
'description': violation.get('description'),
'help': violation.get('help'),
'help_url': violation.get('helpUrl'),
'tags': violation.get('tags', []),
'nodes': len(violation.get('nodes', []))
}
self.all_violations.append(violation_detail)
# Log violation
print(f"AXE [{impact.upper()}]: {violation_detail['rule_id']} - {violation_detail['description']} "
f"({violation_detail['nodes']} elements) on {page_url}")
@task(1)
@pw
async def generate_accessibility_audit_report(self, page: PageWithRetry):
"""Generate comprehensive accessibility audit report"""
if self.scan_summary['total_pages'] < 3:
return
async with event(self, "Accessibility Audit Report"):
# Calculate compliance score
total_possible_score = self.scan_summary['total_pages'] * 100
penalty_score = (
self.scan_summary['critical_violations'] * 10 +
self.scan_summary['serious_violations'] * 5 +
self.scan_summary['moderate_violations'] * 2 +
self.scan_summary['minor_violations'] * 1
)
compliance_score = max(0, 100 - (penalty_score / total_possible_score * 100))
print(f"\nACCESSIBILITY AUDIT REPORT:")
print(f"Pages scanned: {self.scan_summary['total_pages']}")
print(f"Total violations: {self.scan_summary['total_violations']}")
print(f"Critical: {self.scan_summary['critical_violations']}")
print(f"Serious: {self.scan_summary['serious_violations']}")
print(f"Moderate: {self.scan_summary['moderate_violations']}")
print(f"Minor: {self.scan_summary['minor_violations']}")
print(f"Compliance Score: {compliance_score:.1f}/100")
def on_stop(self):
"""Final comprehensive report"""
print("\n" + "="*60)
print("COMPREHENSIVE AXE ACCESSIBILITY AUDIT COMPLETE")
print("="*60)
print(f"Total pages scanned: {len(self.pages_scanned)}")
print(f"Total violations found: {len(self.all_violations)}")
if self.all_violations:
# Top violation types
rule_counts = {}
for violation in self.all_violations:
rule_id = violation['rule_id']
rule_counts[rule_id] = rule_counts.get(rule_id, 0) + 1
print("\nTOP VIOLATION TYPES:")
sorted_rules = sorted(rule_counts.items(), key=lambda x: x[1], reverse=True)
for rule_id, count in sorted_rules[:10]:
print(f" {rule_id}: {count} violations")
# Most problematic pages
page_counts = {}
for violation in self.all_violations:
page = violation['page']
page_counts[page] = page_counts.get(page, 0) + 1
print("\nMOST PROBLEMATIC PAGES:")
sorted_pages = sorted(page_counts.items(), key=lambda x: x[1], reverse=True)
for page, count in sorted_pages[:5]:
print(f" {page}: {count} violations")
else:
print("✅ No accessibility violations found across entire site!")
```
## Key Axe-Core Features
1. **Industry Standard**: Uses axe-core, the same engine used by accessibility professionals
2. **Comprehensive WCAG Coverage**: Tests WCAG 2.1 AA/AAA compliance automatically
3. **Detailed Violation Reports**: Specific rule violations with remediation guidance
4. **Real Browser Testing**: Accurate results from actual browser rendering
5. **Scalable Site Auditing**: Can scan entire websites automatically
6. **LoadForge Integration**: Violations show as failures in LoadForge reports
7. **Professional Reporting**: Impact levels (critical, serious, moderate, minor)
This axe-core integration provides enterprise-grade accessibility testing that matches tools used by accessibility consultants and compliance teams.
# Broken Links Checker
Source: https://docs.loadforge.com/examples/qa-testing/broken-links-checker
Crawl website and detect broken internal/external links with detailed reporting
This guide demonstrates how to create a comprehensive broken links checker with LoadForge, crawling through all pages and detecting various types of link issues.
## Use Cases
* Detecting 404 broken internal and external links
* Finding redirect chains and infinite loops
* Validating anchor links and fragments
* Monitoring external link health over time
* SEO link validation and reporting
## Simplified Internal Links Checker
```python theme={null}
from locust import HttpUser, task, between
import json
import time
import re
from urllib.parse import urljoin, urlparse
from collections import deque
import logging
class InternalLinksChecker(HttpUser):
wait_time = between(0.5, 1)
def on_start(self):
"""Initialize internal links checking"""
self.visited_pages = set()
self.pages_to_check = deque()
self.broken_links = []
self.total_links_checked = 0
self.base_domain = None
# Setup logging
logging.basicConfig(level=logging.INFO)
self.logger = logging.getLogger(__name__)
# Start with homepage
self._initialize_crawler()
def _initialize_crawler(self):
"""Initialize the crawler with the homepage"""
try:
homepage_response = self.client.get('/', name="INIT: Homepage Check")
if homepage_response.status_code == 200:
self.base_domain = urlparse(self.client.base_url).netloc
self.pages_to_check.append('/')
self.logger.info(f"Starting internal links check for domain: {self.base_domain}")
else:
self.logger.error(f"Failed to access homepage: {homepage_response.status_code}")
except Exception as e:
self.logger.error(f"Error initializing crawler: {str(e)}")
@task(10)
def crawl_internal_pages(self):
"""Main crawling task - only checks internal links"""
if not self.pages_to_check:
return
current_page = self.pages_to_check.popleft()
if current_page in self.visited_pages:
return
self.visited_pages.add(current_page)
try:
response = self.client.get(current_page, name=f"CRAWL: {current_page}")
if response.status_code == 200:
self._extract_and_validate_internal_links(current_page, response.text)
else:
self._log_broken_link(current_page, current_page, response.status_code,
f"Page not accessible: {response.status_code}")
except Exception as e:
self._log_broken_link(current_page, current_page, 0, f"Exception accessing page: {str(e)}")
def _extract_and_validate_internal_links(self, source_page, html_content):
"""Extract and validate only internal page links (URLs only, no resources)"""
# Only look for
tags (actual page links)
link_pattern = r']+href=["\']([^"\']+)["\'][^>]*>'
# Find all tag links
all_links = re.findall(link_pattern, html_content, re.IGNORECASE)
# Filter to only internal page links and validate them
for link in all_links:
if self._is_internal_link(link) and not self._should_skip_link(link):
# Only validate if it's a page link (not a resource)
if self._is_page_link(link):
self._validate_internal_link(source_page, link)
# Add to crawl queue
normalized_link = self._normalize_url(link)
if normalized_link and normalized_link not in self.visited_pages:
if normalized_link not in self.pages_to_check:
self.pages_to_check.append(normalized_link)
def _is_internal_link(self, link):
"""Check if a link is internal (same domain or relative)"""
# Skip anchors, external protocols
if link.startswith('#') or link.startswith('mailto:') or link.startswith('tel:'):
return False
# Relative paths are internal
if link.startswith('/') or not link.startswith('http'):
return True
# Check if absolute URL is same domain
if link.startswith('http'):
return urlparse(link).netloc == self.base_domain
return True
def _is_page_link(self, link):
"""Check if link is a page (not a resource like image/css/js)"""
# Skip common resource file extensions
resource_extensions = {'.css', '.js', '.jpg', '.jpeg', '.png', '.gif',
'.svg', '.ico', '.pdf', '.zip', '.mp4', '.mp3',
'.woff', '.woff2', '.ttf', '.eot', '.webp', '.avif'}
parsed_link = urlparse(link.lower())
path = parsed_link.path
# Check if it ends with a resource extension
for ext in resource_extensions:
if path.endswith(ext):
return False
# Always consider these as pages (common blog/article patterns)
page_patterns = [
r'/blog/',
r'/articles?/',
r'/posts?/',
r'/news/',
r'/(20\d{2})/', # Year pattern (2020, 2021, etc.)
r'/category/',
r'/tag/',
]
for pattern in page_patterns:
if re.search(pattern, path):
return True
return True
def _should_skip_link(self, link):
"""Determine if a link should be skipped entirely"""
skip_patterns = [
r'^mailto:',
r'^tel:',
r'^javascript:',
r'^data:',
r'^#$', # Empty anchor
r'^\s*$', # Empty or whitespace
]
for pattern in skip_patterns:
if re.match(pattern, link, re.IGNORECASE):
return True
return False
def _validate_internal_link(self, source_page, link):
"""Validate an internal link and log any issues"""
self.total_links_checked += 1
normalized_url = self._normalize_url(link)
if not normalized_url:
self._log_broken_link(source_page, link, 0, "Invalid URL format")
return
try:
with self.client.get(normalized_url,
name=f"LINK: {normalized_url}",
catch_response=True) as response:
if response.status_code == 404:
self._log_broken_link(source_page, link, 404, "Page not found")
response.failure("❌ 404 Not Found")
elif response.status_code >= 500:
self._log_broken_link(source_page, link, response.status_code, "Server error")
response.failure(f"❌ Server Error {response.status_code}")
elif response.status_code >= 400:
self._log_broken_link(source_page, link, response.status_code, "Client error")
response.failure(f"❌ Client Error {response.status_code}")
else:
response.success()
except Exception as e:
self._log_broken_link(source_page, link, 0, f"Request failed: {str(e)}")
def _log_broken_link(self, source_page, broken_link, status_code, reason):
"""Log broken link with detailed information"""
broken_link_info = {
'source_page': source_page,
'broken_link': broken_link,
'status_code': status_code,
'reason': reason,
'timestamp': time.time()
}
self.broken_links.append(broken_link_info)
# Log to console and LoadForge
error_msg = f"BROKEN LINK: {broken_link} (Status: {status_code}) found on page: {source_page} - {reason}"
self.logger.error(error_msg)
print(error_msg)
def _normalize_url(self, link):
"""Normalize URL for consistent checking"""
try:
# Handle relative paths
if link.startswith('/'):
return link
elif link.startswith('http'):
parsed = urlparse(link)
if parsed.netloc == self.base_domain:
return parsed.path + ('?' + parsed.query if parsed.query else '')
return None # External link, skip
else:
# Relative link - normalize to absolute path
return '/' + link.lstrip('./')
except Exception:
return None
@task(1)
def report_status(self):
"""Print status to console (no API submission needed)"""
if len(self.visited_pages) < 3: # Wait until we have some data
return
print(f"STATUS: {len(self.visited_pages)} pages crawled, "
f"{self.total_links_checked} links checked, "
f"{len(self.broken_links)} broken links found")
def on_stop(self):
"""Final summary when test completes"""
print("\n" + "="*50)
print("INTERNAL LINKS CHECK COMPLETE")
print("="*50)
print(f"Pages crawled: {len(self.visited_pages)}")
print(f"Total links checked: {self.total_links_checked}")
print(f"Broken links found: {len(self.broken_links)}")
if self.broken_links:
print(f"\nBROKEN LINKS FOUND ({len(self.broken_links)}):")
print("-" * 40)
for link_info in self.broken_links:
print(f"❌ {link_info['broken_link']} (HTTP {link_info['status_code']})")
print(f" Found on: {link_info['source_page']}")
print(f" Reason: {link_info['reason']}")
print()
else:
print("✅ No broken links found!")
```
## Comprehensive Broken Links Checker
```python theme={null}
from locust import HttpUser, task, between
import json
import time
import random
import re
from urllib.parse import urljoin, urlparse, urlunparse
from collections import defaultdict, deque
import requests
class BrokenLinksChecker(HttpUser):
wait_time = between(0.5, 2)
def on_start(self):
"""Initialize broken links checking"""
self.visited_pages = set()
self.pages_to_check = deque()
self.broken_links = []
self.redirect_chains = []
self.external_links = {}
self.anchor_links = {}
self.link_stats = defaultdict(int)
self.base_domain = None
# Start with homepage
self._initialize_crawler()
def _initialize_crawler(self):
"""Initialize the crawler with the homepage"""
homepage_response = self.client.get('/',
name="homepage_initial_check")
if homepage_response.status_code == 200:
self.base_domain = urlparse(self.client.base_url).netloc
self.pages_to_check.append('/')
print(f"Starting broken links check for domain: {self.base_domain}")
else:
print(f"Failed to access homepage: {homepage_response.status_code}")
@task(5)
def crawl_and_check_links(self):
"""Main crawling task to check links on pages"""
if not self.pages_to_check:
return
current_page = self.pages_to_check.popleft()
if current_page in self.visited_pages:
return
self.visited_pages.add(current_page)
# Get the page content
response = self.client.get(current_page,
name="crawl_page_for_links")
if response.status_code == 200:
self._extract_and_validate_links(current_page, response.text)
else:
self._record_broken_link(current_page, response.status_code, "Page not accessible")
def _extract_and_validate_links(self, page_url, html_content):
"""Extract all links from HTML and validate them"""
# Find all links in the HTML
link_patterns = [
r']+href=["\']([^"\']+)["\'][^>]*>', #
r']+href=["\']([^"\']+)["\'][^>]*>', #
r'
]+src=["\']([^"\']+)["\'][^>]*>', #
r'',
html_content, re.IGNORECASE | re.DOTALL)
image_in_structured_data = False
for script in json_ld_scripts:
if 'image' in script.lower():
image_in_structured_data = True
break
if image_in_structured_data:
print("✅ Images found in structured data")
else:
print("⚠️ No images in structured data (SEO opportunity)")
@task(1)
def generate_advanced_performance_report(self):
"""Generate advanced image performance report"""
advanced_report = {
'timestamp': time.time(),
'modern_format_support': {
'webp_tested': True,
'avif_tested': True,
'optimization_recommendations': []
},
'cdn_performance': {
'caching_effective': True,
'cdn_domains_checked': self.CDN_DOMAINS
},
'responsive_images': {
'implementation_detected': True,
'picture_elements_found': True
},
'accessibility_score': 85, # Would be calculated from actual checks
'seo_optimization': {
'structured_data': True,
'alt_text_quality': 'good'
}
}
self.client.post('/api/qa/advanced-image-performance-report',
json=advanced_report,
name="submit_advanced_performance_report")
print("ADVANCED REPORT: Performance and optimization analysis complete")
```
## Key Validation Features
1. **Simple Version**: Basic broken images, alt text, and configurable size limits
2. **Comprehensive Version**: Full validation including external images, responsive images, and detailed reporting
3. **Advanced Version**: Modern format support, CDN performance, SEO optimization, and accessibility testing
4. **Configurable Parameters**: Easy-to-modify size limits and validation settings at the top of each script
5. **Progressive Complexity**: Start simple, add features as needed
This guide provides three levels of image validation to match your testing needs and technical requirements.
# Simple Form Validation
Source: https://docs.loadforge.com/examples/qa-testing/simple-form-validation
Basic form validation testing to check form fields, validation rules, error handling, and submission flows
This guide shows how to test form validation logic and error handling. Perfect for validating form fields, submission flows, and user input validation.
## Use Cases
* Test form field validation rules
* Validate error messages and handling
* Test form submission flows
* Check input sanitization and security
## Simple Implementation
```python theme={null}
from locust import task, HttpUser
import random
import json
class FormValidationUser(HttpUser):
def on_start(self):
# Common form endpoints to test
self.forms = {
"/contact": {
"required_fields": ["name", "email", "message"],
"optional_fields": ["phone", "company"],
"validation_rules": {
"email": "email_format",
"phone": "phone_format"
}
},
"/signup": {
"required_fields": ["username", "email", "password"],
"optional_fields": ["first_name", "last_name"],
"validation_rules": {
"email": "email_format",
"password": "min_length_8",
"username": "alphanumeric"
}
},
"/profile": {
"required_fields": ["first_name", "last_name"],
"optional_fields": ["bio", "website", "location"],
"validation_rules": {
"website": "url_format",
"bio": "max_length_500"
}
}
}
# Test data for validation
self.test_data = {
"valid": {
"name": "John Doe",
"first_name": "John",
"last_name": "Doe",
"email": "john@example.com",
"username": "johndoe123",
"password": "SecurePass123!",
"phone": "+1-555-123-4567",
"message": "This is a test message",
"website": "https://example.com",
"bio": "Software developer with 5 years experience"
},
"invalid": {
"email": ["invalid-email", "@example.com", "test@", "plaintext"],
"password": ["123", "short", ""],
"phone": ["123", "invalid-phone", ""],
"website": ["not-a-url", "ftp://invalid", ""],
"username": ["", "user@name", "user name", "a"]
}
}
@task(4)
def test_valid_form_submission(self):
"""Test form submission with valid data"""
form_path = random.choice(list(self.forms.keys()))
form_config = self.forms[form_path]
# Build valid form data
form_data = {}
# Add required fields
for field in form_config["required_fields"]:
if field in self.test_data["valid"]:
form_data[field] = self.test_data["valid"][field]
# Add some optional fields
for field in form_config["optional_fields"]:
if field in self.test_data["valid"] and random.choice([True, False]):
form_data[field] = self.test_data["valid"][field]
with self.client.post(
form_path,
data=form_data,
name=f"Valid Form - {form_path}"
) as response:
if response.status_code in [200, 201, 302]:
print(f"Valid form submission {form_path}: SUCCESS ({response.status_code})")
# Check for success indicators
if response.status_code == 200:
success_indicators = ["success", "thank you", "submitted", "received"]
response_text = response.text.lower()
success_found = any(indicator in response_text for indicator in success_indicators)
if success_found:
print(f"Success message detected in {form_path}")
else:
print(f"No clear success message in {form_path}")
elif response.status_code == 400:
print(f"Valid form submission {form_path}: Validation error (unexpected)")
response.failure("Valid data rejected")
else:
response.failure(f"Unexpected response: {response.status_code}")
@task(3)
def test_missing_required_fields(self):
"""Test form submission with missing required fields"""
form_path = random.choice(list(self.forms.keys()))
form_config = self.forms[form_path]
if not form_config["required_fields"]:
return # Skip if no required fields
# Build form data missing one required field
form_data = {}
missing_field = random.choice(form_config["required_fields"])
for field in form_config["required_fields"]:
if field != missing_field and field in self.test_data["valid"]:
form_data[field] = self.test_data["valid"][field]
with self.client.post(
form_path,
data=form_data,
name=f"Missing Field - {form_path}"
) as response:
if response.status_code == 400:
print(f"Missing field test {form_path}: Correctly rejected ({missing_field})")
# Check for error message about missing field
if response.text:
error_indicators = ["required", "missing", "field", missing_field]
response_text = response.text.lower()
error_found = any(indicator in response_text for indicator in error_indicators)
if error_found:
print(f"Error message detected for missing {missing_field}")
else:
print(f"No clear error message for missing {missing_field}")
elif response.status_code in [200, 201, 302]:
print(f"Missing field test {form_path}: Unexpectedly accepted")
response.failure(f"Form accepted without required field: {missing_field}")
else:
response.failure(f"Unexpected response: {response.status_code}")
@task(3)
def test_invalid_field_formats(self):
"""Test form submission with invalid field formats"""
form_path = random.choice(list(self.forms.keys()))
form_config = self.forms[form_path]
# Find a field with validation rules
validation_fields = list(form_config.get("validation_rules", {}).keys())
if not validation_fields:
return # Skip if no validation rules
test_field = random.choice(validation_fields)
# Build form data with invalid value for test field
form_data = {}
# Add required fields (valid values)
for field in form_config["required_fields"]:
if field == test_field:
# Use invalid value for test field
if test_field in self.test_data["invalid"]:
form_data[field] = random.choice(self.test_data["invalid"][test_field])
else:
form_data[field] = "invalid_value"
elif field in self.test_data["valid"]:
form_data[field] = self.test_data["valid"][field]
with self.client.post(
form_path,
data=form_data,
name=f"Invalid Format - {form_path}"
) as response:
if response.status_code == 400:
print(f"Invalid format test {form_path}: Correctly rejected ({test_field})")
# Check for specific validation error
validation_keywords = ["invalid", "format", "valid", test_field]
response_text = response.text.lower()
error_found = any(keyword in response_text for keyword in validation_keywords)
if error_found:
print(f"Validation error message detected for {test_field}")
else:
print(f"No clear validation error for {test_field}")
elif response.status_code in [200, 201, 302]:
print(f"Invalid format test {form_path}: Unexpectedly accepted")
response.failure(f"Invalid {test_field} format accepted")
else:
response.failure(f"Unexpected response: {response.status_code}")
@task(2)
def test_empty_form_submission(self):
"""Test form submission with completely empty data"""
form_path = random.choice(list(self.forms.keys()))
with self.client.post(
form_path,
data={},
name=f"Empty Form - {form_path}"
) as response:
if response.status_code == 400:
print(f"Empty form test {form_path}: Correctly rejected")
# Check for multiple field errors
error_indicators = ["required", "missing", "field", "empty"]
response_text = response.text.lower()
error_count = sum(1 for indicator in error_indicators if indicator in response_text)
if error_count >= 2:
print(f"Multiple validation errors detected (good)")
else:
print(f"Limited error feedback for empty form")
elif response.status_code in [200, 201, 302]:
print(f"Empty form test {form_path}: Unexpectedly accepted")
response.failure("Empty form accepted")
else:
response.failure(f"Unexpected response: {response.status_code}")
@task(2)
def test_field_length_limits(self):
"""Test form fields with length limits"""
form_path = random.choice(list(self.forms.keys()))
form_config = self.forms[form_path]
# Test with very long values
long_values = {
"name": "A" * 1000,
"email": "a" * 500 + "@example.com",
"message": "This is a very long message. " * 100,
"bio": "Long bio content. " * 200,
"username": "a" * 500
}
# Build form data with one long field
form_data = {}
test_field = None
for field in form_config["required_fields"]:
if field in long_values:
form_data[field] = long_values[field]
test_field = field
break
elif field in self.test_data["valid"]:
form_data[field] = self.test_data["valid"][field]
if not test_field:
return # Skip if no testable field
with self.client.post(
form_path,
data=form_data,
name=f"Length Limit - {form_path}"
) as response:
if response.status_code == 400:
print(f"Length limit test {form_path}: Correctly rejected ({test_field})")
# Check for length-related error
length_keywords = ["long", "length", "limit", "maximum", "characters"]
response_text = response.text.lower()
length_error = any(keyword in response_text for keyword in length_keywords)
if length_error:
print(f"Length validation error detected for {test_field}")
else:
print(f"Generic error for length violation on {test_field}")
elif response.status_code in [200, 201, 302]:
print(f"Length limit test {form_path}: Accepted (no length limit)")
else:
response.failure(f"Unexpected response: {response.status_code}")
@task(1)
def test_special_characters(self):
"""Test form handling of special characters"""
form_path = random.choice(list(self.forms.keys()))
form_config = self.forms[form_path]
# Special characters to test
special_chars = {
"name": "John O'Connor-Smith",
"message": "Test with and special chars: àáâãäå",
"bio": "Bio with émojis 🚀 and spëcial chars",
"company": "Smith & Jones Co., Ltd."
}
# Build form data with special characters
form_data = {}
for field in form_config["required_fields"]:
if field in special_chars:
form_data[field] = special_chars[field]
elif field in self.test_data["valid"]:
form_data[field] = self.test_data["valid"][field]
with self.client.post(
form_path,
data=form_data,
name=f"Special Chars - {form_path}"
) as response:
if response.status_code in [200, 201, 302, 400]:
print(f"Special chars test {form_path}: {response.status_code}")
# Check that dangerous content is handled
if "