import threading
import time

# 方式一：繼承 Thread 類
class APITestThread(threading.Thread):
    def __init__(self, test_name):
        super().__init__()
        self.test_name = test_name
    
    def run(self):
        print(f"開(kāi)始執(zhí)行測(cè)試: {self.test_name}")
        time.sleep(2)  # 模擬API請(qǐng)求耗時(shí)
        print(f"測(cè)試完成: {self.test_name}")

# 方式二：使用 Thread 的 target 參數(shù)（更推薦）
def run_test(test_name):
    print(f"開(kāi)始執(zhí)行測(cè)試: {self.test_name}")
    time.sleep(2)
    print(f"測(cè)試完成: {self.test_name}")

# 創(chuàng)建并啟動(dòng)線程
threads = []
for i in range(5):
    t = threading.Thread(target=run_test, args=(f"test_{i}",))
    threads.append(t)
    t.start()

# 等待所有線程完成
for t in threads:
    t.join()

import threading
import queue
from typing import List, Callable

class TestExecutor:
    def __init__(self, max_workers: int = 5):
        self.max_workers = max_workers
        self.task_queue = queue.Queue()
        self.results = []
        self.lock = threading.Lock()
    
    def add_test(self, test_func: Callable, *args):
        """添加測(cè)試任務(wù)"""
        self.task_queue.put((test_func, args))
    
    def worker(self):
        """工作線程函數(shù)"""
        while True:
            try:
                # 從隊(duì)列獲取任務(wù)，設(shè)置超時(shí)避免死鎖
                test_func, args = self.task_queue.get(timeout=1)
                try:
                    result = test_func(*args)
                    with self.lock:
                        self.results.append({
                            'test': args[0],
                            'status': 'PASS',
                            'result': result
                        })
                except Exception as e:
                    with self.lock:
                        self.results.append({
                            'test': args[0],
                            'status': 'FAIL',
                            'error': str(e)
                        })
                finally:
                    self.task_queue.task_done()
            except queue.Empty:
                break
    
    def run(self):
        """啟動(dòng)所有工作線程并等待完成"""
        threads = []
        for _ in range(self.max_workers):
            t = threading.Thread(target=self.worker)
            t.start()
            threads.append(t)
        
        # 等待所有任務(wù)完成
        self.task_queue.join()
        
        # 等待所有線程結(jié)束
        for t in threads:
            t.join()
        
        return self.results

import threading
import json

class TestReporter:
    def __init__(self, report_file: str):
        self.report_file = report_file
        self.lock = threading.Lock()
        self.test_results = []
    
    def add_result(self, test_name: str, status: str, duration: float):
        """線程安全地添加測(cè)試結(jié)果"""
        with self.lock:  # 使用上下文管理器自動(dòng)加鎖解鎖
            self.test_results.append({
                'test': test_name,
                'status': status,
                'duration': duration,
                'timestamp': time.time()
            })
    
    def save_report(self):
        """保存測(cè)試報(bào)告"""
        with self.lock:
            with open(self.report_file, 'w', encoding='utf-8') as f:
                json.dump(self.test_results, f, indent=2, ensure_ascii=False)

import threading
from concurrent.futures import ThreadPoolExecutor

# 模擬數(shù)據(jù)庫(kù)連接池測(cè)試
def test_db_connection(pool_id: int):
    """測(cè)試特定連接池的連接"""
    import random
    time.sleep(random.uniform(0.1, 0.5))  # 模擬網(wǎng)絡(luò)延遲
    return {
        'pool_id': pool_id,
        'status': 'connected',
        'latency': random.randint(50, 200)
    }

# 使用 ThreadPoolExecutor 簡(jiǎn)化并發(fā)操作
def run_concurrent_db_tests():
    with ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(test_db_connection, i) for i in range(100)]
        
        results = []
        for future in futures:
            try:
                result = future.result(timeout=5)
                results.append(result)
            except Exception as e:
                results.append({'error': str(e)})
    
    return results

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

def api_test_endpoint(url: str, timeout: int = 10):
    """測(cè)試單個(gè)API端點(diǎn)"""
    try:
        response = requests.get(url, timeout=timeout)
        return {
            'url': url,
            'status_code': response.status_code,
            'success': response.status_code == 200,
            'response_time': response.elapsed.total_seconds()
        }
    except Exception as e:
        return {
            'url': url,
            'success': False,
            'error': str(e)
        }

def run_api_test_suite(urls: List[str], max_workers: int = 10):
    """
    并發(fā)測(cè)試多個(gè)API端點(diǎn)
    
    Args:
        urls: 要測(cè)試的URL列表
        max_workers: 最大并發(fā)數(shù)
    
    Returns:
        測(cè)試結(jié)果列表
    """
    results = []
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有任務(wù)
        future_to_url = {
            executor.submit(api_test_endpoint, url): url 
            for url in urls
        }
        
        # 實(shí)時(shí)獲取完成的任務(wù)結(jié)果
        for future in as_completed(future_to_url):
            url = future_to_url[future]
            try:
                result = future.result()
                results.append(result)
                print(f"測(cè)試完成: {url} - {'?' if result['success'] else '?'}")
            except Exception as e:
                results.append({
                    'url': url,
                    'success': False,
                    'error': str(e)
                })
    
    return results

# 實(shí)際使用示例
if __name__ == "__main__":
    test_urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/status/200",
        "https://httpbin.org/status/500",
        # ... 更多測(cè)試URL
    ]
    
    results = run_api_test_suite(test_urls, max_workers=5)
    
    # 統(tǒng)計(jì)結(jié)果
    success_count = sum(1 for r in results if r.get('success'))
    print(f"\n測(cè)試完成: {success_count}/{len(results)} 通過(guò)")

import threading
import queue
import time
from enum import Enum

class TestStatus(Enum):
    PENDING = "待執(zhí)行"
    RUNNING = "執(zhí)行中"
    COMPLETED = "已完成"
    FAILED = "失敗"

class TestOrchestrator:
    """測(cè)試編排器，管理復(fù)雜的測(cè)試流程"""
    
    def __init__(self):
        self.task_queue = queue.Queue()
        self.result_queue = queue.Queue()
        self.status_map = {}
        self.lock = threading.Lock()
    
    def producer(self, test_cases: list):
        """生產(chǎn)者：將測(cè)試用例放入隊(duì)列"""
        for case in test_cases:
            self.task_queue.put(case)
            with self.lock:
                self.status_map[case['id']] = TestStatus.PENDING
        
        # 發(fā)送結(jié)束信號(hào)
        for _ in range(3):  # 3個(gè)工作線程
            self.task_queue.put(None)
    
    def consumer(self, worker_id: int):
        """消費(fèi)者：執(zhí)行測(cè)試并返回結(jié)果"""
        while True:
            try:
                task = self.task_queue.get(timeout=1)
                if task is None:
                    break
                
                # 更新?tīng)顟B(tài)為運(yùn)行中
                with self.lock:
                    self.status_map[task['id']] = TestStatus.RUNNING
                
                # 執(zhí)行測(cè)試
                result = self.execute_test(task)
                
                # 更新?tīng)顟B(tài)并返回結(jié)果
                with self.lock:
                    self.status_map[task['id']] = (
                        TestStatus.COMPLETED if result['success'] 
                        else TestStatus.FAILED
                    )
                
                self.result_queue.put(result)
                self.task_queue.task_done()
                
            except queue.Empty:
                break
    
    def execute_test(self, task: dict) -> dict:
        """實(shí)際執(zhí)行測(cè)試邏輯"""
        # 模擬測(cè)試執(zhí)行
        time.sleep(task.get('duration', 1))
        return {
            'task_id': task['id'],
            'name': task['name'],
            'success': task.get('should_pass', True),
            'worker_id': threading.current_thread().ident
        }
    
    def run(self, test_cases: list):
        """啟動(dòng)整個(gè)測(cè)試流程"""
        # 啟動(dòng)生產(chǎn)者線程
        producer_thread = threading.Thread(
            target=self.producer, 
            args=(test_cases,)
        )
        producer_thread.start()
        
        # 啟動(dòng)消費(fèi)者線程
        consumers = []
        for i in range(3):
            t = threading.Thread(target=self.consumer, args=(i,))
            t.start()
            consumers.append(t)
        
        # 收集結(jié)果
        results = []
        while len(results) < len(test_cases):
            try:
                result = self.result_queue.get(timeout=10)
                results.append(result)
                print(f"[Worker {result['worker_id']}] {result['name']}: {'?' if result['success'] else '?'}")
            except queue.Empty:
                break
        
        # 等待所有線程結(jié)束
        producer_thread.join()
        for t in consumers:
            t.join()
        
        return results

def safe_worker(task_queue, stop_event):
    """使用 Event 對(duì)象優(yōu)雅停止線程"""
    while not stop_event.is_set():
        try:
            task = task_queue.get(timeout=0.5)
            # 處理任務(wù)...
        except queue.Empty:
            continue
    
    print("線程優(yōu)雅退出")

# 使用方式
stop_event = threading.Event()
worker_thread = threading.Thread(target=safe_worker, args=(queue, stop_event))
worker_thread.start()

# 需要停止時(shí)
stop_event.set()
worker_thread.join(timeout=5)  # 等待線程結(jié)束

# 確保結(jié)果有序
results = sorted(results, key=lambda x: x['test_id'])