จากโค้ดที่ให้มา นี่คือโปรแกรม **SOAP Load Testing** สำหรับทดสอบประสิทธิภาพระบบที่เขียนด้วยภาษา Python มาอธิบายการทำงานให้ฟังครับ ## วัตถุประสงค์หลัก โปรแกรมนี้ใช้สำหรับทดสอบความสามารถในการรับโหลดของระบบ SOAP Web Service โดยจำลองการส่งคำขอจากผู้ใช้หลายคนพร้อมกัน ## ส่วนประกอบหลัก ### 1. คลาส SOAPLoadTester - **หน้าที่**: จัดการการทดสอบโหลดหลัก - **การทำงาน**: - สร้าง SOAP request โดยเปลี่ยน reference number และ timestamp ให้แตกต่างกันในแต่ละคำขอ - ส่งคำขอแบบ concurrent (พร้อมกัน) โดยใช้ ThreadPoolExecutor - บันทึกผลลัพธ์ทุกคำขอลงไฟล์ CSV - มีระบบ retry เมื่อเกิดข้อผิดพลาด ### 2. ฟีเจอร์พิเศษ **Multi-Process Support**: - แบ่งงานออกเป็นหลาย process เพื่อกระจายโหลด - แต่ละ process จะมี concurrent users และจำนวนคำขอแยกกัน - ช่วยหลีกเลี่ยงข้อจำกัดของ Python GIL (Global Interpreter Lock) **Load Balancing Techniques**: - สุ่ม source port เพื่อให้ F5 Load Balancer กระจายโหลดได้ดีขึ้น - ใส่ header พิเศษ (X-Request-ID, X-Process-ID, X-Thread-ID) - ใช้ "Connection: close" เพื่อไม่ให้ใช้ persistent connection **Error Handling**: - ระบบ retry อัตโนมัติเมื่อเกิดข้อผิดพลาด - จัดการข้อผิดพลาด SSL โดยการปิดการตรวจสอบใบรับรองอัตโนมัติ - บันทึกข้อผิดพลาดแยกไฟล์ ### 3. การวัดผลและรายงาน **สถิติที่คำนวณ**: - TPS (Transactions Per Second) - ทั้งหมดและที่สำเร็จ - Response Time: ต่ำสุด, สูงสุด, เฉลี่ย - Percentiles: 90th, 95th, 99th - อัตราความสำเร็จ/ล้มเหลว **ไฟล์ผลลัพธ์**: - `raw_response_times_*.csv`: ข้อมูลดิบของแต่ละคำขอ - `errors_*.csv`: รายการข้อผิดพลาด - `summary_*.csv`: สรุปผลการทดสอบ ### 4. การตรวจสอบตาม TOR (Terms of Reference) โปรแกรมจะตรวจสอบว่าผลการทดสอบผ่านเกณฑ์หรือไม่: - **TPS ≥ 700**: ระบบต้องรองรับได้อย่างน้อย 700 transactions ต่อวินาที - **Average Response Time ≤ 3 วินาที**: เวลาตอบสนองเฉลี่ยต้องไม่เกิน 3,000 ms ### 5. การใช้งาน ```bash python script.py --url https://example.com/soap --requests 1000 --concurrent 50 --processes 4 ``` **พารามิเตอร์สำคัญ**: - `--requests`: จำนวนคำขอทั้งหมด - `--concurrent`: จำนวนผู้ใช้พร้อมกัน - `--processes`: จำนวน process สำหรับกระจายโหลด - `--timeout`: เวลา timeout ของแต่ละคำขอ - `--retries`: จำนวนครั้งที่ retry เมื่อผิดพลาด ## จุดเด่นของโปรแกรม 1. **ทดสอบได้จริง**: จำลองสถานการณ์ผู้ใช้จริงด้วย concurrent requests 2. **รายงานละเอียด**: มีสถิติครบถ้วนและแยกประเภทข้อผิดพลาด 3. **ปรับแต่งได้**: มีพารามิเตอร์หลากหลายสำหรับการปรับแต่ง 4. **เสถียร**: มีระบบ retry และจัดการข้อผิดพลาดอย่างดี 5. **เหมาะสำหรับ Enterprise**: รองรับการทดสอบระดับสูงด้วย multi-process โปรแกรมนี้เหมาะสำหรับการทดสอบประสิทธิภาพระบบก่อนเข้าสู่การใช้งานจริง (Production) เพื่อให้มั่นใจว่าระบบรองรับโหลดงานได้ตามที่กำหนดไว้ครับ ```python import requests import time import concurrent.futures import csv import uuid import datetime import statistics import argparse import os import ssl import socket import multiprocessing import random from tqdm import tqdm import warnings warnings.filterwarnings("ignore", category=requests.packages.urllib3.exceptions.InsecureRequestWarning) class SOAPLoadTester: def __init__(self, url, soap_template, total_requests, concurrent_users, ramp_up=0, test_duration=None, output_dir="results", verify_ssl=True, delay_between_threads=0.05, timeout=30, retries=2, process_id=0): self.url = url self.soap_template = soap_template self.total_requests = total_requests self.concurrent_users = concurrent_users self.ramp_up = ramp_up self.test_duration = test_duration self.output_dir = output_dir self.results = [] self.verify_ssl = verify_ssl self.delay_between_threads = delay_between_threads self.timeout = timeout self.retries = retries self.process_id = process_id self.account_numbers = ["39860000396", "211750000216", "25850001646", "22910089746"] if not os.path.exists(output_dir): os.makedirs(output_dir) self.timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") self.raw_file = f"{output_dir}/raw_response_times_{self.timestamp}_p{self.process_id}.csv" self.error_file = f"{output_dir}/errors_{self.timestamp}_p{self.process_id}.csv" with open(self.raw_file, 'w', newline='') as f: writer = csv.writer(f) writer.writerow(['timestamp', 'request_id', 'response_time_ms', 'status_code', 'success', 'reference_no', 'account_number', 'process_id']) with open(self.error_file, 'w', newline='') as f: writer = csv.writer(f) writer.writerow(['timestamp', 'request_id', 'error_message', 'reference_no', 'account_number', 'process_id']) def get_soap_request(self, request_id): ref_no = f"REF{datetime.datetime.now().strftime('%Y%m%d%H%M%S')}{uuid.uuid4().hex[:3]}" timestamp = datetime.datetime.now().strftime("%Y-%m-%dT%H:%M:%S.000+07:00") account_number = self.account_numbers[request_id % len(self.account_numbers)] soap_request = self.soap_template.replace("REF20210331701", ref_no) soap_request = soap_request.replace("2021-03-31T14:54:00.000+07:00", timestamp) soap_request = soap_request.replace("1910000214", account_number) return soap_request, ref_no, account_number def send_request(self, request_id): if request_id > 0 and self.delay_between_threads > 0: time.sleep(self.delay_between_threads * (request_id % self.concurrent_users)) soap_request, ref_no, account_number = self.get_soap_request(request_id) for attempt in range(self.retries + 1): start_time = time.time() try: session = requests.Session() source_port = random.randint(10000, 60000) headers = { "Content-Type": "text/xml;charset=UTF-8", "SOAPAction": "", "User-Agent": f"GHB-LoadTest/{self.process_id}.{request_id}", "Connection": "close", "X-Request-ID": str(uuid.uuid4()), "X-Process-ID": str(self.process_id), "X-Thread-ID": str(request_id % self.concurrent_users), "X-Account-Number": account_number } response = session.post( self.url, data=soap_request, headers=headers, timeout=self.timeout, verify=self.verify_ssl ) end_time = time.time() duration_ms = (end_time - start_time) * 1000 success = 200 <= response.status_code < 300 if success: if "ResponseCode>AA<" in response.text: actual_success = True elif "true" in response.text: actual_success = True else: actual_success = True else: actual_success = False with open(self.raw_file, 'a', newline='') as f: writer = csv.writer(f) writer.writerow([ datetime.datetime.now().isoformat(), request_id, duration_ms, response.status_code, actual_success, ref_no, account_number, self.process_id ]) session.close() if success: return { 'request_id': request_id, 'response_time_ms': duration_ms, 'status_code': response.status_code, 'success': actual_success, 'ref_no': ref_no, 'account_number': account_number, 'process_id': self.process_id, 'response_text': response.text[:500] } else: error_msg = f"HTTP Error: {response.status_code}, Response: {response.text[:200]}" with open(self.error_file, 'a', newline='') as f: writer = csv.writer(f) writer.writerow([ datetime.datetime.now().isoformat(), request_id, error_msg, ref_no, account_number, self.process_id ]) if attempt < self.retries: time.sleep(0.5) continue return { 'request_id': request_id, 'response_time_ms': duration_ms, 'status_code': response.status_code, 'success': False, 'error': error_msg, 'ref_no': ref_no, 'account_number': account_number, 'process_id': self.process_id } except requests.exceptions.SSLError as e: end_time = time.time() duration_ms = (end_time - start_time) * 1000 error_msg = f"SSL Error: {str(e)}" with open(self.error_file, 'a', newline='') as f: writer = csv.writer(f) writer.writerow([ datetime.datetime.now().isoformat(), request_id, error_msg, ref_no, account_number, self.process_id ]) self.verify_ssl = False if attempt < self.retries: time.sleep(0.5) continue return { 'request_id': request_id, 'response_time_ms': duration_ms, 'status_code': 0, 'success': False, 'error': error_msg, 'ref_no': ref_no, 'account_number': account_number, 'process_id': self.process_id } except Exception as e: end_time = time.time() duration_ms = (end_time - start_time) * 1000 error_msg = f"Error: {str(e)}" with open(self.error_file, 'a', newline='') as f: writer = csv.writer(f) writer.writerow([ datetime.datetime.now().isoformat(), request_id, error_msg, ref_no, account_number, self.process_id ]) with open(self.raw_file, 'a', newline='') as f: writer = csv.writer(f) writer.writerow([ datetime.datetime.now().isoformat(), request_id, duration_ms, 0, False, ref_no, account_number, self.process_id ]) if attempt < self.retries: time.sleep(0.5) continue return { 'request_id': request_id, 'response_time_ms': duration_ms, 'status_code': 0, 'success': False, 'error': error_msg, 'ref_no': ref_no, 'account_number': account_number, 'process_id': self.process_id } def run_test(self): print(f"เริ่มการทดสอบ - Process ID: {self.process_id} - URL: {self.url}") print(f"จำนวนคำขอทั้งหมด: {self.total_requests}, ผู้ใช้งานพร้อมกัน: {self.concurrent_users}") print(f"เลขบัญชีที่ใช้ทดสอบ: {', '.join(self.account_numbers)}") print(f"จำนวนครั้งที่ลองใหม่: {self.retries}, Timeout: {self.timeout} วินาที") print(f"ตรวจสอบ SSL: {'เปิด' if self.verify_ssl else 'ปิด'}") print("กำลังทดสอบการเชื่อมต่อกับเซิร์ฟเวอร์...") try: test_request, test_ref, test_account = self.get_soap_request(0) session = requests.Session() headers = { "Content-Type": "text/xml;charset=UTF-8", "SOAPAction": "", "User-Agent": f"GHB-LoadTest/{self.process_id}.0", "Connection": "close", "X-Request-ID": str(uuid.uuid4()), "X-Process-ID": str(self.process_id), "X-Thread-ID": "0", "X-Account-Number": test_account } test_response = session.post( self.url, data=test_request, headers=headers, timeout=self.timeout, verify=self.verify_ssl ) print(f"การทดสอบการเชื่อมต่อสำเร็จ - รหัสการตอบกลับ: {test_response.status_code}") print(f"เลขบัญชีที่ทดสอบ: {test_account}") print("\nตัวอย่างการตอบกลับ (500 ตัวอักษรแรก):") print(test_response.text[:500]) print("...\n") session.close() except Exception as e: print(f"การทดสอบการเชื่อมต่อล้มเหลว: {str(e)}") print("กำลังดำเนินการทดสอบต่อไป...") start_time = time.time() with concurrent.futures.ThreadPoolExecutor(max_workers=self.concurrent_users) as executor: futures = [executor.submit(self.send_request, i) for i in range(self.total_requests)] for future in tqdm(concurrent.futures.as_completed(futures), total=self.total_requests, desc=f"Process {self.process_id}"): result = future.result() self.results.append(result) end_time = time.time() test_duration = end_time - start_time return self.summarize_results(test_duration) def summarize_results(self, test_duration): if not self.results: print(f"[Process {self.process_id}] ไม่มีผลลัพธ์สำหรับการสรุป") return None response_times = [r['response_time_ms'] for r in self.results] successful_requests = [r for r in self.results if r['success']] failed_requests = [r for r in self.results if not r['success']] tps = len(self.results) / test_duration successful_tps = len(successful_requests) / test_duration if successful_requests else 0 account_stats = {} for account in self.account_numbers: account_requests = [r for r in self.results if r.get('account_number') == account] account_successful = [r for r in account_requests if r['success']] account_stats[account] = { 'total': len(account_requests), 'successful': len(account_successful), 'success_rate': len(account_successful) / len(account_requests) * 100 if account_requests else 0 } error_types = {} for r in failed_requests: error_msg = r.get('error', 'Unknown error') if isinstance(error_msg, str) and len(error_msg) > 50: error_msg = error_msg[:50] + "..." error_types[error_msg] = error_types.get(error_msg, 0) + 1 summary = { 'total_requests': len(self.results), 'successful_requests': len(successful_requests), 'failed_requests': len(failed_requests), 'test_duration_sec': test_duration, 'transactions_per_second': tps, 'successful_tps': successful_tps, 'min_response_time_ms': min(response_times) if response_times else 0, 'max_response_time_ms': max(response_times) if response_times else 0, 'avg_response_time_ms': statistics.mean(response_times) if response_times else 0, 'process_id': self.process_id, 'account_stats': account_stats } if len(response_times) > 10: summary['percentile_90_ms'] = statistics.quantiles(response_times, n=10)[8] else: summary['percentile_90_ms'] = 0 if len(response_times) > 20: summary['percentile_95_ms'] = statistics.quantiles(response_times, n=20)[18] else: summary['percentile_95_ms'] = 0 if len(response_times) > 100: summary['percentile_99_ms'] = statistics.quantiles(response_times, n=100)[98] else: summary['percentile_99_ms'] = 0 print(f"\n[Process {self.process_id}] =============== สรุปผลการทดสอบ ===============") print(f"จำนวนคำขอทั้งหมด: {summary['total_requests']}") print(f"คำขอสำเร็จ: {summary['successful_requests']} ({(summary['successful_requests']/summary['total_requests']*100):.2f}%)") print(f"คำขอล้มเหลว: {summary['failed_requests']} ({(summary['failed_requests']/summary['total_requests']*100):.2f}%)") print(f"ระยะเวลาทดสอบ: {summary['test_duration_sec']:.2f} วินาที") print(f"Transactions Per Second (TPS) - ทั้งหมด: {summary['transactions_per_second']:.2f}") print(f"Transactions Per Second (TPS) - สำเร็จ: {summary['successful_tps']:.2f}") print("\n--- สถิติการใช้งานตามเลขบัญชี ---") for account, stats in account_stats.items(): print(f"บัญชี {account}: {stats['total']} คำขอ, สำเร็จ {stats['successful']} ({stats['success_rate']:.2f}%)") if error_types: print("\n--- ประเภทความผิดพลาด ---") for error, count in error_types.items(): print(f"{error}: {count} ครั้ง") print("\n--- Response Time ---") print(f"ต่ำสุด: {summary['min_response_time_ms']:.2f} ms") print(f"สูงสุด: {summary['max_response_time_ms']:.2f} ms") print(f"เฉลี่ย: {summary['avg_response_time_ms']:.2f} ms") print(f"90th Percentile: {summary['percentile_90_ms']:.2f} ms") print(f"95th Percentile: {summary['percentile_95_ms']:.2f} ms") print(f"99th Percentile: {summary['percentile_99_ms']:.2f} ms") summary_file = f"{self.output_dir}/summary_{self.timestamp}_p{self.process_id}.csv" with open(summary_file, 'w', newline='') as f: writer = csv.writer(f) writer.writerow(['metric', 'value']) for key, value in summary.items(): if key != 'account_stats': writer.writerow([key, value]) for account, stats in account_stats.items(): writer.writerow([f'account_{account}_total', stats['total']]) writer.writerow([f'account_{account}_successful', stats['successful']]) writer.writerow([f'account_{account}_success_rate', stats['success_rate']]) print(f"\n[Process {self.process_id}] ผลลัพธ์ถูกบันทึกไว้ที่:") print(f"- ข้อมูลดิบ: {self.raw_file}") print(f"- ข้อมูลความผิดพลาด: {self.error_file}") print(f"- สรุปผล: {summary_file}") return summary def run_process(args, process_id): soap_template = get_soap_template(args.template) requests_per_process = args.requests // args.processes if process_id < args.requests % args.processes: requests_per_process += 1 concurrent_per_process = args.concurrent // args.processes if process_id < args.concurrent % args.processes: concurrent_per_process += 1 process_output_dir = f"{args.output}/p{process_id}" if not os.path.exists(process_output_dir): os.makedirs(process_output_dir) tester = SOAPLoadTester( url=args.url, soap_template=soap_template, total_requests=requests_per_process, concurrent_users=concurrent_per_process, output_dir=process_output_dir, verify_ssl=not args.no_verify, delay_between_threads=args.delay, timeout=args.timeout, retries=args.retries, process_id=process_id ) return tester.run_test() def get_soap_template(template_file=None): if template_file and os.path.exists(template_file): with open(template_file, 'r') as f: return f.read() else: return """

REF20210331701 2021-03-31T14:54:00.000+07:00 GEN

1910000214 S """ def merge_results(result_list, output_dir): if not result_list or None in result_list: print("ไม่สามารถรวมผลลัพธ์ได้เนื่องจากมีบาง process ที่ไม่มีผลลัพธ์") return total_requests = sum(r['total_requests'] for r in result_list) successful_requests = sum(r['successful_requests'] for r in result_list) failed_requests = sum(r['failed_requests'] for r in result_list) weighted_avg_rt = sum(r['avg_response_time_ms'] * r['total_requests'] for r in result_list) / total_requests if total_requests > 0 else 0 min_rt = min(r['min_response_time_ms'] for r in result_list) max_rt = max(r['max_response_time_ms'] for r in result_list) avg_duration = sum(r['test_duration_sec'] for r in result_list) / len(result_list) total_tps = total_requests / avg_duration successful_tps = successful_requests / avg_duration combined_account_stats = {} account_numbers = ["39860000396", "211750000216", "25850001646", "22910089746"] for account in account_numbers: combined_account_stats[account] = { 'total': sum(r['account_stats'].get(account, {}).get('total', 0) for r in result_list), 'successful': sum(r['account_stats'].get(account, {}).get('successful', 0) for r in result_list) } if combined_account_stats[account]['total'] > 0: combined_account_stats[account]['success_rate'] = ( combined_account_stats[account]['successful'] / combined_account_stats[account]['total'] * 100 ) else: combined_account_stats[account]['success_rate'] = 0 p90_values = [r['percentile_90_ms'] for r in result_list if 'percentile_90_ms' in r and r['percentile_90_ms'] > 0] p95_values = [r['percentile_95_ms'] for r in result_list if 'percentile_95_ms' in r and r['percentile_95_ms'] > 0] p99_values = [r['percentile_99_ms'] for r in result_list if 'percentile_99_ms' in r and r['percentile_99_ms'] > 0] percentile_90 = statistics.mean(p90_values) if p90_values else 0 percentile_95 = statistics.mean(p95_values) if p95_values else 0 percentile_99 = statistics.mean(p99_values) if p99_values else 0 summary = { 'total_requests': total_requests, 'successful_requests': successful_requests, 'failed_requests': failed_requests, 'test_duration_sec': avg_duration, 'transactions_per_second': total_tps, 'successful_tps': successful_tps, 'min_response_time_ms': min_rt, 'max_response_time_ms': max_rt, 'avg_response_time_ms': weighted_avg_rt, 'percentile_90_ms': percentile_90, 'percentile_95_ms': percentile_95, 'percentile_99_ms': percentile_99, 'process_count': len(result_list), 'account_stats': combined_account_stats } print("\n=============== สรุปผลการทดสอบรวม ===============") print(f"จำนวน Processes: {len(result_list)}") print(f"จำนวนคำขอทั้งหมด: {summary['total_requests']}") print(f"คำขอสำเร็จ: {summary['successful_requests']} ({(summary['successful_requests']/summary['total_requests']*100):.2f}%)") print(f"คำขอล้มเหลว: {summary['failed_requests']} ({(summary['failed_requests']/summary['total_requests']*100):.2f}%)") print(f"ระยะเวลาทดสอบเฉลี่ย: {summary['test_duration_sec']:.2f} วินาที") print(f"Transactions Per Second (TPS) - ทั้งหมด: {summary['transactions_per_second']:.2f}") print(f"Transactions Per Second (TPS) - สำเร็จ: {summary['successful_tps']:.2f}") print("\n--- สถิติการใช้งานรวมตามเลขบัญชี ---") for account, stats in combined_account_stats.items(): print(f"บัญชี {account}: {stats['total']} คำขอ, สำเร็จ {stats['successful']} ({stats['success_rate']:.2f}%)") print("\n--- Response Time ---") print(f"ต่ำสุด: {summary['min_response_time_ms']:.2f} ms") print(f"สูงสุด: {summary['max_response_time_ms']:.2f} ms") print(f"เฉลี่ย: {summary['avg_response_time_ms']:.2f} ms") print(f"90th Percentile: {summary['percentile_90_ms']:.2f} ms") print(f"95th Percentile: {summary['percentile_95_ms']:.2f} ms") print(f"99th Percentile: {summary['percentile_99_ms']:.2f} ms") timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") summary_file = f"{output_dir}/summary_combined_{timestamp}.csv" with open(summary_file, 'w', newline='') as f: writer = csv.writer(f) writer.writerow(['metric', 'value']) for key, value in summary.items(): if key != 'account_stats': writer.writerow([key, value]) for account, stats in combined_account_stats.items(): writer.writerow([f'account_{account}_total', stats['total']]) writer.writerow([f'account_{account}_successful', stats['successful']]) writer.writerow([f'account_{account}_success_rate', stats['success_rate']]) print(f"\nผลลัพธ์รวมถูกบันทึกไว้ที่: {summary_file}") check_tor_compliance(summary) def check_tor_compliance(summary): print("\n=============== ผลการตรวจสอบตาม TOR ===============") tps_compliant = summary['successful_tps'] >= 700 print(f"TPS ต้องไม่น้อยกว่า 700: {'ผ่าน' if tps_compliant else 'ไม่ผ่าน'} (สำเร็จ: {summary['successful_tps']:.2f} TPS)") rt_compliant = summary['avg_response_time_ms'] <= 3000 print(f"Average Response Time ต้องไม่เกิน 3 วินาที: {'ผ่าน' if rt_compliant else 'ไม่ผ่าน'} ({summary['avg_response_time_ms']:.2f} ms)") if tps_compliant and rt_compliant: print("\n✅ ผลการทดสอบผ่านตามข้อกำหนดใน TOR") else: print("\n❌ ผลการทดสอบไม่ผ่านตามข้อกำหนดใน TOR") if not tps_compliant: print("\nคำแนะนำสำหรับการปรับปรุง TPS:") print("1. เพิ่มจำนวน processes") print("2. เพิ่มจำนวน concurrent users") print("3. ลดค่า delay_between_threads") print("4. ตรวจสอบข้อจำกัดด้านเครือข่ายและเซิร์ฟเวอร์") if not rt_compliant: print("\nคำแนะนำสำหรับการปรับปรุง Response Time:") print("1. ตรวจสอบการตั้งค่าการเชื่อมต่อและเครือข่าย") print("2. ตรวจสอบประสิทธิภาพของเซิร์ฟเวอร์") print("3. ลดจำนวน concurrent users") def main(): parser = argparse.ArgumentParser(description='SOAP Load Test script for GHB system testing with multi-process support and multiple account numbers') parser.add_argument('--url', type=str, default='https://gsysesb-uat.ghb.co.th/ws/GenAccountService', help='URL ของ SOAP endpoint') parser.add_argument('--template', type=str, help='ไฟล์ template SOAP (Optional)') parser.add_argument('--requests', type=int, default=1000, help='จำนวนคำขอทั้งหมดที่จะส่ง') parser.add_argument('--concurrent', type=int, default=50, help='จำนวนผู้ใช้งานพร้อมกัน (concurrent users)') parser.add_argument('--output', type=str, default='results', help='โฟลเดอร์สำหรับบันทึกผลลัพธ์') parser.add_argument('--no-verify', action='store_true', help='ไม่ตรวจสอบใบรับรอง SSL') parser.add_argument('--delay', type=float, default=0.02, help='ระยะเวลาหน่วงระหว่าง thread (วินาที)') parser.add_argument('--timeout', type=int, default=30, help='ระยะเวลา timeout ของคำขอ (วินาที)') parser.add_argument('--retries', type=int, default=2, help='จำนวนครั้งที่ลองใหม่เมื่อเกิดข้อผิดพลาด') parser.add_argument('--processes', type=int, default=4, help='จำนวน processes ที่จะรันพร้อมกัน (สำหรับกระจายโหลดไปหลาย server)') args = parser.parse_args() if not os.path.exists(args.output): os.makedirs(args.output) account_numbers = ["39860000396", "211750000216", "25850001646", "22910089746"] print(f"การทดสอบแบบ Multi-Process - จำนวน Processes: {args.processes}") print(f"URL: {args.url}") print(f"จำนวนคำขอทั้งหมด: {args.requests} ({args.requests // args.processes} ต่อ process)") print(f"จำนวนผู้ใช้งานพร้อมกัน: {args.concurrent} ({args.concurrent // args.processes} ต่อ process)") print(f"เลขบัญชีที่ใช้ทดสอบ: {', '.join(account_numbers)}") print(f"ตรวจสอบ SSL: {'ปิด' if args.no_verify else 'เปิด'}") with multiprocessing.Pool(processes=args.processes) as pool: results = pool.starmap(run_process, [(args, i) for i in range(args.processes)]) valid_results = [r for r in results if r is not None] if valid_results: merge_results(valid_results, args.output) else: print("ไม่มีผลลัพธ์จากการทดสอบที่สามารถนำมาสรุปได้") if __name__ == "__main__": main() ```

SECURE_ACCESS

SOAP Load Testing

CONFIRM_DELETE