#!/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()