202 lines
7.2 KiB
Python
202 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
import json
|
|
import random
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
|
|
# ==================== CONFIGURATION ====================
|
|
API_KEY = "" # Kept your actual API key
|
|
BASE_URL = "https://map-saas.intaleqapp.com/api/maps/route"
|
|
CONCURRENCY = 1000 # Number of concurrent threads
|
|
TOTAL_REQUESTS = 10000 # Total number of requests to send
|
|
TIMEOUT_SECONDS = 10 # Request timeout
|
|
|
|
# Bounding boxes for heavily populated regions inside Jordan and Syria (excluding deserts and Egypt since it is not in the map database)
|
|
REGIONS = {
|
|
"Amman (Jordan)": {
|
|
"lat_min": 31.85, "lat_max": 32.15,
|
|
"lng_min": 35.80, "lng_max": 36.00
|
|
},
|
|
"Irbid (Jordan)": {
|
|
"lat_min": 32.45, "lat_max": 32.60,
|
|
"lng_min": 35.80, "lng_max": 35.95
|
|
},
|
|
"Damascus (Syria)": {
|
|
"lat_min": 33.45, "lat_max": 33.55,
|
|
"lng_min": 36.25, "lng_max": 36.35
|
|
},
|
|
"Aleppo (Syria)": {
|
|
"lat_min": 36.15, "lat_max": 36.25,
|
|
"lng_min": 37.10, "lng_max": 37.20
|
|
}
|
|
}
|
|
# =======================================================
|
|
|
|
def generate_random_route():
|
|
"""Generates random starting and ending points inside Jordan and Syria populated cities."""
|
|
region_name = random.choice(list(REGIONS.keys()))
|
|
bbox = REGIONS[region_name]
|
|
|
|
# Pick a random starting point in the selected region
|
|
from_lat = random.uniform(bbox["lat_min"], bbox["lat_max"])
|
|
from_lng = random.uniform(bbox["lng_min"], bbox["lng_max"])
|
|
|
|
# Pick a destination within the same region (~10-15km max) to ensure a quick and valid route
|
|
to_lat = from_lat + random.uniform(-0.08, 0.08)
|
|
to_lng = from_lng + random.uniform(-0.08, 0.08)
|
|
|
|
# Clip coordinates to region bounds
|
|
to_lat = max(bbox["lat_min"], min(to_lat, bbox["lat_max"]))
|
|
to_lng = max(bbox["lng_min"], min(to_lng, bbox["lng_max"]))
|
|
|
|
return region_name, from_lat, from_lng, to_lat, to_lng
|
|
|
|
def send_request(request_id):
|
|
region_name, from_lat, from_lng, to_lat, to_lng = generate_random_route()
|
|
|
|
# Construct dynamic URL with random coordinates
|
|
url = (
|
|
f"{BASE_URL}?fromLat={from_lat:.5f}&fromLng={from_lng:.5f}"
|
|
f"&toLat={to_lat:.5f}&toLng={to_lng:.5f}&locale=ar"
|
|
)
|
|
|
|
req = urllib.request.Request(
|
|
url,
|
|
headers={
|
|
"x-api-key": API_KEY,
|
|
"User-Agent": "Benchmark-Client/1.0"
|
|
}
|
|
)
|
|
|
|
start_time = time.perf_counter()
|
|
status_code = 0
|
|
error_message = None
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=TIMEOUT_SECONDS) as response:
|
|
status_code = response.status
|
|
response.read()
|
|
except urllib.error.HTTPError as e:
|
|
status_code = e.code
|
|
try:
|
|
err_body = e.read().decode('utf-8')
|
|
error_message = f"HTTP {e.code}: {json.loads(err_body).get('message', e.reason)}"
|
|
except Exception:
|
|
error_message = f"HTTP Error {e.code}: {e.reason}"
|
|
except urllib.error.URLError as e:
|
|
status_code = 0
|
|
error_message = f"URL Error: {e.reason}"
|
|
except Exception as e:
|
|
status_code = 0
|
|
error_message = f"Generic Error: {str(e)}"
|
|
|
|
end_time = time.perf_counter()
|
|
latency = (end_time - start_time) * 1000.0 # Convert to milliseconds
|
|
|
|
return {
|
|
"id": request_id,
|
|
"region": region_name,
|
|
"success": 200 <= status_code < 300,
|
|
"status_code": status_code,
|
|
"latency": latency,
|
|
"error": error_message
|
|
}
|
|
|
|
def print_report(results, elapsed_time):
|
|
latencies = [r["latency"] for r in results]
|
|
successes = [r for r in results if r["success"]]
|
|
failures = [r for r in results if not r["success"]]
|
|
|
|
latencies.sort()
|
|
|
|
total_reqs = len(results)
|
|
success_count = len(successes)
|
|
failure_count = len(failures)
|
|
|
|
avg_latency = sum(latencies) / total_reqs if total_reqs > 0 else 0
|
|
min_latency = latencies[0] if latencies else 0
|
|
max_latency = latencies[-1] if latencies else 0
|
|
|
|
def percentile(p):
|
|
if not latencies:
|
|
return 0
|
|
idx = int(len(latencies) * p)
|
|
return latencies[min(idx, len(latencies) - 1)]
|
|
|
|
rps = total_reqs / elapsed_time if elapsed_time > 0 else 0
|
|
|
|
# Calculate stats per region
|
|
region_stats = {}
|
|
for r in results:
|
|
reg = r["region"]
|
|
if reg not in region_stats:
|
|
region_stats[reg] = {"total": 0, "success": 0, "latencies": []}
|
|
region_stats[reg]["total"] += 1
|
|
if r["success"]:
|
|
region_stats[reg]["success"] += 1
|
|
region_stats[reg]["latencies"].append(r["latency"])
|
|
|
|
print("\n" + "="*50)
|
|
print(" API LOAD TESTING REPORT ")
|
|
print("="*50)
|
|
print(f"Target URL: {BASE_URL}")
|
|
print(f"Concurrency Level: {CONCURRENCY} threads")
|
|
print(f"Total Requests: {total_reqs}")
|
|
print(f"Time Taken: {elapsed_time:.3f} seconds")
|
|
print(f"Successful Requests: {success_count} ({success_count/total_reqs*100:.1f}%)")
|
|
print(f"Failed Requests: {failure_count} ({failure_count/total_reqs*100:.1f}%)")
|
|
print(f"Requests per Second: {rps:.2f} RPS")
|
|
print("-"*50)
|
|
print("PER-REGION SUMMARY:")
|
|
for region, stats in region_stats.items():
|
|
r_avg = sum(stats["latencies"]) / stats["total"] if stats["total"] > 0 else 0
|
|
print(f" {region:17}: {stats['total']} reqs, Avg: {r_avg:.1f}ms, Success: {stats['success']}/{stats['total']}")
|
|
print("-"*50)
|
|
print("LATENCY STATISTICS (ms):")
|
|
print(f" Min: {min_latency:.2f} ms")
|
|
print(f" Max: {max_latency:.2f} ms")
|
|
print(f" Average: {avg_latency:.2f} ms")
|
|
print(f" Median (50%): {percentile(0.50):.2f} ms")
|
|
print(f" 90th Percentile: {percentile(0.90):.2f} ms")
|
|
print(f" 95th Percentile: {percentile(0.95):.2f} ms")
|
|
print(f" 99th Percentile: {percentile(0.99):.2f} ms")
|
|
print("="*50)
|
|
|
|
if failure_count > 0:
|
|
print("\nERROR SUMMARY:")
|
|
errors = {}
|
|
for f in failures:
|
|
err = f["error"] or f"HTTP {f['status_code']}"
|
|
errors[err] = errors.get(err, 0) + 1
|
|
for err, count in errors.items():
|
|
print(f" - {err}: {count} occurrence(s)")
|
|
print("="*50)
|
|
|
|
def main():
|
|
print(f"Starting dynamic benchmark of: {BASE_URL}")
|
|
print(f"Sending {TOTAL_REQUESTS} randomized requests (Amman, Irbid, Damascus, Aleppo)...")
|
|
print(f"Concurrency Level: {CONCURRENCY} concurrent threads")
|
|
|
|
results = []
|
|
start_time = time.perf_counter()
|
|
|
|
with ThreadPoolExecutor(max_workers=CONCURRENCY) as executor:
|
|
futures = {executor.submit(send_request, i): i for i in range(TOTAL_REQUESTS)}
|
|
|
|
completed = 0
|
|
for future in as_completed(futures):
|
|
results.append(future.result())
|
|
completed += 1
|
|
if completed % (TOTAL_REQUESTS // 10 or 1) == 0 or completed == TOTAL_REQUESTS:
|
|
print(f"Progress: {completed}/{TOTAL_REQUESTS} requests completed...")
|
|
|
|
end_time = time.perf_counter()
|
|
elapsed_time = end_time - start_time
|
|
|
|
print_report(results, elapsed_time)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|