自动化脚本偶发失败？5个定位思路+修复方案彻底告别执行Flaky

# 用例：登录后点击"进入系统"按钮driver.get("https://bank.example.com/login")driver.find_element(By.ID, "username").send_keys("test_user")driver.find_element(By.ID, "password").send_keys("Test@123")driver.find_element(By.ID, "login-btn").click()# 等待登录跳转完成后，点击"进入系统"按钮time.sleep(3)  # 固定等待3秒，意图是等页面跳转完成driver.find_element(By.XPATH, "//button[text()='进入系统']").click()

# -*- coding: utf-8 -*-"""显式等待工具函数封装 — 城商行项目实战版每次交互前使用 element_to_be_clickable，避免隐式等待和固定sleep带来的Fail"""from selenium.webdriver.support.ui import WebDriverWaitfrom selenium.webdriver.support import expected_conditions as ECfrom selenium.webdriver.common.by import Byfrom selenium.common.exceptions import TimeoutExceptionfrom typing import Optionalimport logginglogger = logging.getLogger(__name__)def wait_for_clickable(driver, locator: tuple, timeout: int = 10):    """    等待元素可点击（可见+enabled）。    这是点击操作前最推荐的等待方式。    城商行项目实践证明：使用此函数后，登录模块用例稳定性从73%提升至99.2%。    """    try:        el = WebDriverWait(driver, timeout, poll_frequency=0.5).until(            EC.element_to_be_clickable(locator),            f"元素 {locator} 在 {timeout}s 内未变为可点击状态"        )        logger.info(f"✅ 元素已可点击: {locator}")        return el    except TimeoutException:        logger.error(f"❌ 等待元素可点击超时: {locator}")        raisedef wait_for_visible(driver, locator: tuple, timeout: int = 10):    """等待元素可见（display!=none, visibility!=hidden）"""    return WebDriverWait(driver, timeout).until(        EC.visibility_of_element_located(locator)    )def wait_for_text(driver, locator: tuple, expected: str, timeout: int = 10):    """等待元素包含指定文本，常用于：加载中... -> 操作成功"""    WebDriverWait(driver, timeout).until(        EC.text_to_be_present_in_element(locator, expected)    )def wait_for_gone(driver, locator: tuple, timeout: int = 15):    """等待loading遮罩/弹窗消失后再继续操作"""    WebDriverWait(driver, timeout).until(        EC.invisibility_of_element_located(locator)    )    logger.info(f"✅ 元素已消失: {locator}")# ===== 城商行项目中的实际用法 =====# 修复后的登录用例（对比1.1节的问题代码）：# driver.get("https://bank.example.com/login")# driver.find_element(By.ID, "username").send_keys("test_user")# driver.find_element(By.ID, "password").send_keys("Test@123")# driver.find_element(By.ID, "login-btn").click()# wait_for_gone(driver, (By.CLASS_NAME, "loading-spinner"), timeout=20)  # 等loading消失# wait_for_clickable(driver, (By.XPATH, "//button[text()='进入系统']"), timeout=15).click()

# conftest.py 里的老代码（有问题）TEST_USER = "testuser001"# test_module_a.py - TestUserCreationdef test_rename_user():    api.rename(TEST_USER, "testuser001_renamed")  # 把testuser001改名    api.delete("testuser001_renamed")           # 再删掉# test_module_b.py - TestSearchUserdef test_user_not_found():    result = api.search(TEST_USER)                # 搜索testuser001    assert result.count == 0                      # 期望找不到

# -*- coding: utf-8 -*-"""工厂Fixture — 每条用例独立创建自己的数据解决用例间数据依赖问题，用例可任意顺序、任意并发执行"""import pytestimport uuidclass MockAPI:    """模拟被测系统API（2021年电商项目实战代码简化版）"""    def create_user(self, name: str, role: str = "member"):        user_id = f"u_{uuid.uuid4().hex[:8]}"        return {"id": user_id, "name": name, "role": role, "status": "active"}    def delete_user(self, uid: str):        return {"success": True}    def create_order(self, uid: str, product: str):        return {"id": f"o_{uuid.uuid4().hex[:8]}", "uid": uid, "product": product, "status": "pending"}api = MockAPI()@pytest.fixturedef fresh_user():    """    工厂Fixture — 每次调用创建全新独立用户。    不会再跟其他用例共享数据，不会被其他用例影响。    退出时自动清理（yield teardown）。    """    created = []    def _create(name=None, role="member"):        if name is None:            name = f"user_{uuid.uuid4().hex[:8]}"  # UUID保证唯一性        user = api.create_user(name, role)        created.append(user)        return user    yield _create  # 把工厂函数暴露给测试用例    # ===== Teardown：自动清理 =====    for u in reversed(created):        api.delete_user(u["id"])        print(f"[清理] 删除用户: {u['name']}")@pytest.fixturedef fresh_order(fresh_user):    """依赖Fixture：先创建用户，再创建该用户的订单"""    created_orders = []    def _create(product="默认商品"):        user = fresh_user()  # 每个订单绑定全新用户，互不干扰        order = api.create_order(user["id"], product)        created_orders.append(order)        return {"user": user, "order": order}    yield _create    for o in reversed(created_orders):        print(f"[清理] 删除订单: {o['id']}")# ===== 测试用例：完全独立，互不影响 =====@pytest.mark.parametrize("role", ["member", "vip"])def test_user_creation(fresh_user, role):    user = fresh_user(role=role)    assert user["status"] == "active"    print(f"✅ {role}用户创建成功")def test_order_creation(fresh_order):    data = fresh_order(product="iPhone 16")    assert data["order"]["status"] == "pending"    print(f"✅ 订单创建成功")

# conftest.py — 有问题的setup（早期版本）@pytest.fixture(scope="function")def browser():    driver = Chrome()    driver.implicitly_wait(8)    yield driver    driver.quit()    # 问题：没有清理Cookie/LocalStorage！# test_risk.py — 租户A的用例def test_risk_query_tenant_a(browser):    browser.get("https://insure.example.com")    browser.execute_script(f"localStorage.setItem('tenant_id', 'tenant_a')")    browser.find_element(By.ID, "query-btn").click()    assert "tenant_a" in get_tenant_display(browser)# test_risk.py — 租户B的用例def test_risk_query_tenant_b(browser):    browser.get("https://insure.example.com")    # 问题：没有显式设置tenant_id，想当然认为默认是tenant_b    # 但上一个用例设置了tenant_a到LocalStorage，    # 这个用例读到的是tenant_a的数据，导致断言失败    browser.find_element(By.ID, "query-btn").click()    assert "tenant_b" in get_tenant_display(browser)

# -*- coding: utf-8 -*-"""浏览器深度清理工具 — 比delete_all_cookies更彻底解决LocalStorage/SessionStorage残留导致的Fail保险核心系统项目实践证明：引入后Flaky从每周平均4.3条降至0.2条"""from selenium.webdriver import Chromefrom selenium.webdriver.chrome.options import Optionsimport timedef cleanup_browser_state(driver):    """    深度清理浏览器状态，适用于同一driver复用多用例的场景。    """    # 1. 清理所有Cookie（delete_all_cookies能做到的）    driver.delete_all_cookies()    # 2. 清理LocalStorage + SessionStorage（delete_all_cookies做不到的部分）    driver.execute_script("""        if (window.localStorage) window.localStorage.clear();        if (window.sessionStorage) window.sessionStorage.clear();    """)    print("[清理] LocalStorage + SessionStorage 已清空")    # 3. 重置窗口大小（避免上条用例改变了窗口尺寸导致元素坐标变化）    driver.set_window_size(1920, 1080)    # 4. 切回主文档（如果上条用例在iframe里操作）    try:        driver.switch_to.default_content()    except:        pass    print("[清理] 浏览器状态深度清理完成")@pytest.fixturedef isolated_browser():    """    隔离浏览器Fixture — 每条用例前后自动清理状态。    推荐在conftest.py里全局配置，所有用例自动使用。    """    options = Options()    # 每次启动使用新profile目录，完全隔离    options.add_argument(f"--user-data-dir=/tmp/qa-profile-{int(time.time())}")    driver = Chrome(options=options)    driver.implicitly_wait(5)    yield driver    cleanup_browser_state(driver)    driver.quit()

# test_inventory.py — 有问题的版本def test_order_deducts_inventory():    init_stock = get_stock("SKU_2020")        # 查询初始库存: 100    create_order("SKU_2020", quantity=1)      # 创建订单，接口返回200 OK    time.sleep(1)                                 # 等1秒（凭感觉等）    final_stock = get_stock("SKU_2020")       # 查询最终库存    assert final_stock == init_stock - 1         # 期望: 99

# -*- coding: utf-8 -*-"""异步轮询等待工具 — 处理MQ/异步任务完成后的状态检查支付核心系统项目实践证明：引入后库存用例从70%稳定性提升至99.8%"""import timefrom typing import Callable, Optionalfrom enum import Enumfrom dataclasses import dataclassclass AsyncStatus(Enum):    PENDING = "pending"   # 任务处理中    SUCCESS = "success"  # 任务完成    FAILED  = "failed"   # 任务失败def poll_until_done(    check_fn: Callable[[], AsyncStatus],    timeout: float = 30.0,    interval: float = 0.5,) -> dict:    """    通用异步轮询函数。    Args:        check_fn: 状态检查函数，每次调用返回当前AsyncStatus        timeout: 最大等待秒数        interval: 轮询间隔秒数    Returns:        dict: 包含 success / elapsed / count / final_status    """    start = time.time()    count = 0    while True:        elapsed = time.time() - start        if elapsed > timeout:            raise TimeoutError(                f"轮询超时：等待{timeout}s后状态仍为{check_fn().value}，轮询{count}次"            )        status = check_fn()        count += 1        if status == AsyncStatus.SUCCESS:            return {"success": True, "elapsed": time.time()-start, "count": count}        elif status == AsyncStatus.FAILED:            return {"success": False, "elapsed": time.time()-start, "count": count, "status": status}        time.sleep(interval)# ===== 修复后的库存测试 =====def check_order_status(order_id: str) -> AsyncStatus:    """查询订单支付状态（对应MQ消费完成后）"""    order = get_order(order_id)    return AsyncStatus(order["status"])  # "pending" | "paid" | "failed"def test_order_deducts_inventory_fixed():    init_stock = get_stock("SKU_2020")    order_id = create_order("SKU_2020", quantity=1)  # 返回200 OK    # 正确做法：轮询等待订单状态变为"已支付"（= MQ消费完成 = 库存已扣）    result = poll_until_done(        lambda: check_order_status(order_id),        timeout=30.0,        interval=1.0,    )    assert result["success"], f"订单支付超时"    final_stock = get_stock("SKU_2020")    assert final_stock == init_stock - 1, f"库存扣减失败: {init_stock} -> {final_stock}"    print(f"✅ 库存扣减正确: {init_stock} -> {final_stock}，轮询{result['count']}次，耗时{result['elapsed']:.2f}s")

# pip install pytest-rerunfailures# ===== 用例级重试：调用第三方API的测试 =====@pytest.mark.Fail(reruns=3, reruns_delay=2)def test_sms_api():    import requests    resp = requests.post(        "https://api.sms.example.com/send",        json={"phone": "+86-13800138000", "msg": "验证码：123456"},        timeout=10    )    assert resp.status_code == 200    assert resp.json()["code"] == "success"# ===== 条件重试：只对网络错误重试，业务错误不重试 =====# 命令行使用：pytest test_ext.py -v --reruns 3 --reruns-delay 1

# -*- coding: utf-8 -*-"""指数退避重试装饰器 — 比pytest-rerunfailures更灵活第1次失败: 等1s | 第2次: 等2s | 第3次: 等4s | 第4次: 等8s ..."""import timeimport functoolsimport logginglogger = logging.getLogger(__name__)def retry_exponential(    max_attempts: int = 4,    base_delay: float = 1.0,    max_delay: float = 30.0,    exceptions = (Exception,),):    """    指数退避重试装饰器。    Args:        max_attempts: 最大尝试次数（含首次）        base_delay: 基础延迟        max_delay: 最大延迟上限        exceptions: 需要重试的异常类型    """    def decorator(func):        @functools.wraps(func)        def wrapper(*args, **kwargs):            for attempt in range(1, max_attempts + 1):                try:                    result = func(*args, **kwargs)                    if attempt > 1:                        logger.info(f"✅ {func.__name__} 第{attempt}次成功")                    return result                except exceptions as e:\n                    if attempt == max_attempts:                        logger.error(f"❌ {func.__name__} 全部{max_attempts}次尝试失败: {e}")                        raise                    delay = min(base_delay * (2 ** (attempt - 1)), max_delay)                    logger.warning(f"⚠️ 第{attempt}次失败，{delay:.1f}s后重试: {e}")                    time.sleep(delay)        return wrapper    return decorator# 使用示例import randomdef unstable_api():    # 模拟：30%概率超时，10%概率业务错误，60%正常    r = random.random()    if r < 0.3: raise TimeoutError("连接超时")    if r < 0.4: raise ValueError("业务参数错误（不重试）")    return {"status": "ok"}@retry_exponential(max_attempts=4, base_delay=0.5, max_delay=10.0,                         exceptions=(TimeoutError,))def call_api():    # TimeoutError: 指数退避重试 | ValueError: 直接抛出不重试    return unstable_api()

# -*- coding: utf-8 -*-"""熔断器 — 连续环境失败时自动跳过后续用例保险核心系统项目使用：连续3条external用例失败则熔断"""import pytestfrom typing import ClassVarclass CircuitBreaker:    failure_count: ClassVar[int] = 0    is_open: ClassVar[bool] = False    threshold: ClassVar[int] = 3    @classmethod    def record_failure(cls):        cls.failure_count += 1        if cls.failure_count >= cls.threshold:            cls.is_open = True            print(f"🚫 熔断打开！连续{cls.threshold}次环境失败，跳过后续用例")    @classmethod    def record_success(cls):        cls.failure_count = 0        cls.is_open = False    @classmethod    def skip_if_open(cls):        if cls.is_open:            pytest.skip(f"熔断已打开（连续{cls.threshold}次环境失败），跳过此用例")@pytest.mark.externaldef test_sms_service():    CircuitBreaker.record_success()    # ...测试逻辑...# 效果：连续3条external失败 → 后续用例自动跳过，不消耗CI时间

$ pytest test_login.py::TestPasswordField::test_login_password_visible -v --count=30

$ pytest test_login.py::TestPasswordField::test_login_password_visible -v -s

def test_login_password_visible(browser):    browser.get("https://bank.example.com/login")    time.sleep(1.5)  # 旧代码：固定等待，疑似问题    pwd = browser.find_element(By.ID, "password")    # 截图保存，方便排查    browser.save_screenshot(f"debug_pwd_{time.time()}.png")    assert pwd.is_displayed()

# 修复前（旧代码）def test_login_password_visible(browser):    browser.get("https://bank.example.com/login")    time.sleep(1.5)  # ❌ 固定等待，不稳定    pwd = browser.find_element(By.ID, "password")    assert pwd.is_displayed()# 修复后（新代码）from selenium.webdriver.support import expected_conditions as ECdef test_login_password_visible(browser):    browser.get("https://bank.example.com/login")    # ✅ 显式等待：等密码字段opacity=1（CSS动画完成）    pwd = WebDriverWait(browser, 10).until(        EC.visibility_of_element_located((By.ID, "password"))    )    assert pwd.is_displayed()