深入理解Python使用threading模块提升自动化测试效率

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"开始执行测试: {self.test_name}")
        time.sleep(2)  # 模拟API请求耗时
        print(f"测试完成: {self.test_name}")

# 方式二：使用 Thread 的 target 参数（更推荐）
def run_test(test_name):
    print(f"开始执行测试: {self.test_name}")
    time.sleep(2)
    print(f"测试完成: {self.test_name}")

# 创建并启动线程
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):
        """添加测试任务"""
        self.task_queue.put((test_func, args))
    
    def worker(self):
        """工作线程函数"""
        while True:
            try:
                # 从队列获取任务，设置超时避免死锁
                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):
        """启动所有工作线程并等待完成"""
        threads = []
        for _ in range(self.max_workers):
            t = threading.Thread(target=self.worker)
            t.start()
            threads.append(t)
        
        # 等待所有任务完成
        self.task_queue.join()
        
        # 等待所有线程结束
        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):
        """线程安全地添加测试结果"""
        with self.lock:  # 使用上下文管理器自动加锁解锁
            self.test_results.append({
                'test': test_name,
                'status': status,
                'duration': duration,
                'timestamp': time.time()
            })
    
    def save_report(self):
        """保存测试报告"""
        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

# 模拟数据库连接池测试
def test_db_connection(pool_id: int):
    """测试特定连接池的连接"""
    import random
    time.sleep(random.uniform(0.1, 0.5))  # 模拟网络延迟
    return {
        'pool_id': pool_id,
        'status': 'connected',
        'latency': random.randint(50, 200)
    }

# 使用 ThreadPoolExecutor 简化并发操作
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):
    """测试单个API端点"""
    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):
    """
    并发测试多个API端点
    
    Args:
        urls: 要测试的URL列表
        max_workers: 最大并发数
    
    Returns:
        测试结果列表
    """
    results = []
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有任务
        future_to_url = {
            executor.submit(api_test_endpoint, url): url 
            for url in urls
        }
        
        # 实时获取完成的任务结果
        for future in as_completed(future_to_url):
            url = future_to_url[future]
            try:
                result = future.result()
                results.append(result)
                print(f"测试完成: {url} - {'✓' if result['success'] else '✗'}")
            except Exception as e:
                results.append({
                    'url': url,
                    'success': False,
                    'error': str(e)
                })
    
    return results

# 实际使用示例
if __name__ == "__main__":
    test_urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/status/200",
        "https://httpbin.org/status/500",
        # ... 更多测试URL
    ]
    
    results = run_api_test_suite(test_urls, max_workers=5)
    
    # 统计结果
    success_count = sum(1 for r in results if r.get('success'))
    print(f"\n测试完成: {success_count}/{len(results)} 通过")

import threading
import queue
import time
from enum import Enum

class TestStatus(Enum):
    PENDING = "待执行"
    RUNNING = "执行中"
    COMPLETED = "已完成"
    FAILED = "失败"

class TestOrchestrator:
    """测试编排器，管理复杂的测试流程"""
    
    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):
        """生产者：将测试用例放入队列"""
        for case in test_cases:
            self.task_queue.put(case)
            with self.lock:
                self.status_map[case['id']] = TestStatus.PENDING
        
        # 发送结束信号
        for _ in range(3):  # 3个工作线程
            self.task_queue.put(None)
    
    def consumer(self, worker_id: int):
        """消费者：执行测试并返回结果"""
        while True:
            try:
                task = self.task_queue.get(timeout=1)
                if task is None:
                    break
                
                # 更新状态为运行中
                with self.lock:
                    self.status_map[task['id']] = TestStatus.RUNNING
                
                # 执行测试
                result = self.execute_test(task)
                
                # 更新状态并返回结果
                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:
        """实际执行测试逻辑"""
        # 模拟测试执行
        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):
        """启动整个测试流程"""
        # 启动生产者线程
        producer_thread = threading.Thread(
            target=self.producer, 
            args=(test_cases,)
        )
        producer_thread.start()
        
        # 启动消费者线程
        consumers = []
        for i in range(3):
            t = threading.Thread(target=self.consumer, args=(i,))
            t.start()
            consumers.append(t)
        
        # 收集结果
        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
        
        # 等待所有线程结束
        producer_thread.join()
        for t in consumers:
            t.join()
        
        return results

def safe_worker(task_queue, stop_event):
    """使用 Event 对象优雅停止线程"""
    while not stop_event.is_set():
        try:
            task = task_queue.get(timeout=0.5)
            # 处理任务...
        except queue.Empty:
            continue
    
    print("线程优雅退出")

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

# 需要停止时
stop_event.set()
worker_thread.join(timeout=5)  # 等待线程结束

# 确保结果有序
results = sorted(results, key=lambda x: x['test_id'])