Python全栈自动化测试：从Pytest底层原理到分布式测试平台架构实践

import pytest
import pymysql

@pytest.fixture(scope="session")  # session级别：整个测试会话只创建一次
def db_connection():
    """数据库连接fixture，自动管理生命周期"""
    conn = pymysql.connect(
        host="test_db_host",
        user="tester",
        password="tester_pwd",
        database="test_db"
    )
    yield conn  # 此处返回连接，测试执行到这里暂停
    conn.close()  # 测试全部结束后执行清理

def test_insert_order(db_connection):
    cursor = db_connection.cursor()
    cursor.execute("INSERT INTO orders(order_id) VALUES('T001')")
    db_connection.commit()
    # 无需手动close，fixture的yield后代码自动执行

# 你的代码
assert response.json()["data"]["status"] == "SUCCESS"

# pytest重写后的等效逻辑（伪代码）
__assertion_context = {
    'expression': 'response.json()["data"]["status"] == "SUCCESS"',
    'left': response.json()["data"]["status"],  # 实际值
    'right': 'SUCCESS'
}
if __assertion_context['left'] != __assertion_context['right']:
    raise AssertionError(f"实际值: {__assertion_context['left']}, 期望值: SUCCESS")

import requests
from tenacity import retry, stop_after_attempt, wait_exponential

class APIClient:
    def __init__(self, base_url, auth_token=None):
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({"Authorization": f"Bearer {auth_token}"})

    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
    def post_with_retry(self, endpoint, json_data):
        """带指数退避的重试POST，解决网络波动"""
        response = self.session.post(
            f"{self.base_url}/{endpoint}",
            json=json_data,
            timeout=5
        )
        response.raise_for_status()  # 抛出HTTPError
        return response

from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

class BasePage:
    def __init__(self, driver):
        self.driver = driver
        self.wait = WebDriverWait(driver, 10)

    def find_and_click(self, locator):
        """智能等待+点击，内嵌重试机制"""
        for _ in range(3):
            try:
                element = self.wait.until(EC.element_to_be_clickable(locator))
                element.click()
                return
            except Exception:
                # 处理元素被遮挡/状态变化等场景
                continue
        raise Exception(f"元素 {locator} 无法点击")

class LoginPage(BasePage):
    # 定位器策略：DATA驱动 + 显性等待
    USERNAME_INPUT = ("id", "username")
    PASSWORD_INPUT = ("css", "[data-test='password']")
    SUBMIT_BTN = ("xpath", "//button[text()='登录']")

    def login(self, user, pwd):
        self.find_and_click(self.USERNAME_INPUT).send_keys(user)
        self.find_and_click(self.PASSWORD_INPUT).send_keys(pwd)
        self.find_and_click(self.SUBMIT_BTN)
        # 返回下一个页面对象，实现链式调用
        return DashboardPage(self.driver)

@pytest.fixture
def clean_db(db_connection):
    """事务级Fixture：每个测试用例独立事务，结束后回滚"""
    db_connection.begin()  # 开启事务
    yield db_connection
    db_connection.rollback()  # 测试结束后强制回滚

def test_create_order(clean_db):
    cursor = clean_db.cursor()
    cursor.execute("INSERT INTO orders(order_id) VALUES('T001')")
    clean_db.execute("SELECT * FROM orders WHERE order_id='T001'")
    assert cursor.fetchone() is not None
    # 退出测试后，事务回滚，数据库中无该条记录

pytest -n 8 --dist loadscope --maxfail=5

import redis
import json

r = redis.Redis(host='test-redis', decode_responses=True)

@pytest.hookimpl(tryfirst=True)
def pytest_runtest_makereport(item, call):
    """pytest钩子：在每个测试用例完成后触发"""
    outcome = "passed" if call.excinfo is None else "failed"
    report = {
        "test_id": item.nodeid,
        "outcome": outcome,
        "duration": call.duration,
        "worker": os.getpid()
    }
    r.rpush("test_reports", json.dumps(report))

pipeline {
    agent any
    stages {
        stage('Build & Unit Test') {
            steps {
                sh 'pytest --cov=src --cov-report=xml'
                sh 'coverage xml -o coverage.xml'  // 生成覆盖率报告
            }
        }
        stage('API & UI Full Regression') {
            steps {
                sh 'pytest -n 4 --alluredir=allure-results'
            }
        }
    }
    post {
        always {
            allure includeProperties: false, jdk: '', results: [[path: 'allure-results']]
            junit 'reports/*.xml'
        }
        unstable {
            // 当测试通过率低于95%时，标记为不稳定
            sh 'pytest --passed-percentage=95'
        }
        failure {
            // 发送钉钉告警
            dingtalk(webhook: 'your-webhook', message: '构建失败！')
        }
    }
}

@pytest.fixture(scope="module")
def driver(worker_id):
    # worker_id 由 xdist 提供，形如 'gw0', 'gw1'
    driver = webdriver.Remote(
        command_executor=f'http://selenium-hub:4444/wd/hub',
        desired_capabilities={
            'browserName': 'chrome',
            'selenoid:options': {'name': worker_id}
        }
    )
    yield driver
    driver.quit()