# 高手做法：工厂函数+Fixture，自动生成+自动清理from faker import Fakerimport pytestfake = Faker('zh_CN')  # 支持中文数据生成# 定义用户数据工厂class UserFactory:    @classmethod    def create(cls, **kwargs):        # 基础数据，确保唯一        base_data = {            "username": fake.user_name() + str(random.randint(1000, 9999)),            "email": fake.unique.email(),            "phone": fake.phone_number()        }        # 支持自定义参数覆盖，灵活适配不同场景        return {**base_data, **kwargs}# Fixture管理数据生命周期，测试后自动清理@pytest.fixturedef clean_user():    user = UserFactory.create()    # 模拟插入数据库    user_id = db.insert(user)    yield user  # 测试用例执行期间使用该用户    db.delete(user_id)  # 测试结束，自动清理数据# 测试用例直接注入使用，无需手动造数据def test_user_order(clean_user):    response = api.create_order(clean_user)    assert response.ok  # 测试结束，数据自动消失

# 数据库连接工厂，支持多环境切换class DBFactory:    @staticmethod    def create_db(kind: str, **cfg):        if kind == "mysql":            return MySQL(**cfg)  # MySQL数据库实例        if kind == "pg":            return Postgres(**cfg)  # Postgres数据库实例        raise ValueError(f"不支持的数据库类型：{kind}")# 测试环境切换，只需修改参数# 测试MySQL环境mysql_db = DBFactory.create_db("mysql", host="127.0.0.1", port=3306)# 测试Postgres环境pg_db = DBFactory.create_db("pg", host="127.0.0.1", port=5432)

# 安装依赖：pip install faker pytest requestsfrom faker import Fakerimport pytestimport requestsimport random# 初始化Faker，生成中文数据fake = Faker('zh_CN')# 1. 定义用户数据工厂（适配用户相关接口）class UserFactory:    @classmethod    def create(cls, role="普通用户", **kwargs):        """        生成用户数据，支持自定义角色和其他参数        :param role: 角色（普通用户/管理员/游客）        :param kwargs: 自定义参数，用于覆盖默认值        :return: 结构化用户数据        """        # 基础数据（确保唯一）        base_user = {            "username": fake.user_name() + str(random.randint(1000, 9999)),            "password": fake.password(length=10, special_chars=True),            "email": fake.unique.email(),            "phone": fake.phone_number(),            "role": role,            "register_time": fake.date_time_this_year().strftime("%Y-%m-%d %H:%M:%S")        }        # 允许自定义参数覆盖（比如指定手机号、邮箱）        return {**base_user, **kwargs}# 2. 定义Fixture，管理用户数据生命周期@pytest.fixture(scope="function")  # 每个用例独立，避免数据污染def unique_user():    """生成唯一用户，测试后自动删除（模拟接口删除）"""    # 生成用户数据    user = UserFactory.create()    # 调用注册接口，创建用户    register_url = "http://xxx.xxx.xxx/api/user/register"    requests.post(register_url, json=user)    yield user  # 测试用例使用该用户    # 测试结束，调用删除接口，清理数据    delete_url = f"http://xxx.xxx.xxx/api/user/delete?username={user['username']}"    requests.post(delete_url)# 3. 测试用例（直接注入用户，无需手动造数据）def test_user_profile(unique_user):    """测试用户资料查询接口"""    url = "http://xxx.xxx.xxx/api/user/profile"    headers = {"Authorization": f"Bearer {unique_user['token']}"}  # 假设返回token    response = requests.get(url, headers=headers)    # 断言：查询到的用户名与工厂生成的一致    assert response.json()["username"] == unique_user["username"]    assert response.status_code == 200

# 安装依赖：pip install selenium webdriver-managerfrom selenium import webdriverfrom webdriver_manager.chrome import ChromeDriverManagerfrom webdriver_manager.firefox import GeckoDriverManagerfrom webdriver_manager.microsoft import EdgeChromiumDriverManager# 浏览器实例工厂class BrowserFactory:    @staticmethod    def create_browser(browser_type="chrome"):        """        生成浏览器实例，支持多浏览器切换        :param browser_type: 浏览器类型（chrome/firefox/edge）        :return: 浏览器实例        """        if browser_type.lower() == "chrome":            # 初始化Chrome浏览器            driver = webdriver.Chrome(ChromeDriverManager().install())        elif browser_type.lower() == "firefox":            # 初始化Firefox浏览器            driver = webdriver.Firefox(GeckoDriverManager().install())        elif browser_type.lower() == "edge":            # 初始化Edge浏览器            driver = webdriver.Edge(EdgeChromiumDriverManager().install())        else:            raise ValueError(f"不支持的浏览器类型：{browser_type}")        # 统一设置浏览器窗口大小        driver.maximize_window()        return driver# 测试用例（灵活切换浏览器）def test_baidu_search():    # 1. 生成Chrome浏览器实例    driver = BrowserFactory.create_browser("chrome")    try:        driver.get("https://www.baidu.com")        driver.find_element(By.ID, "kw").send_keys("Python工厂函数 自动化测试")        driver.find_element(By.ID, "su").click()        # 断言：搜索结果包含关键词        assert "Python工厂函数" in driver.title    finally:        # 关闭浏览器        driver.quit()# 切换到Firefox浏览器，只需修改参数def test_baidu_search_firefox():    driver = BrowserFactory.create_browser("firefox")    try:        driver.get("https://www.baidu.com")        # 后续测试逻辑相同...    finally:        driver.quit()

import pytestimport randomfrom faker import Fakerfake = Faker('zh_CN')# 批量用户数据工厂class BulkUserFactory:    @classmethod    def create_bulk_users(cls, count=10, prefix="perf_"):        """        批量生成用户数据        :param count: 生成数量        :param prefix: 用户名前缀，用于区分性能测试数据        :return: 用户列表        """        users = []        for i in range(count):            user = {                "username": prefix + fake.user_name() + str(i),                "email": fake.unique.email(),                "phone": fake.phone_number(),                "password": fake.password()            }            users.append(user)        return users# Fixture：预生成1000个用户，供所有性能测试用例共享@pytest.fixture(scope="session")  # 全局只创建一次def bulk_test_users():    """性能测试：预生成1000个用户，测试后统一清理"""    print("⏳ 正在生成1000个性能测试用户...")    users = BulkUserFactory.create_bulk_users(count=1000, prefix="perf_")    # 批量插入数据库    db.bulk_insert(users)    yield users  # 所有性能测试用例共享该用户列表    # 测试结束，批量清理数据    print("🧹 正在清理性能测试数据...")    usernames = [user["username"] for user in users]    db.bulk_delete(usernames)# 性能测试用例：测试批量查询接口性能def test_bulk_user_query(bulk_test_users):    """测试批量查询1000个用户的响应时间"""    url = "http://xxx.xxx.xxx/api/user/bulk_query"    params = {"usernames": [user["username"] for user in bulk_test_users[:100]]}    response = requests.get(url, params=params)    # 断言：响应时间小于1秒，接口正常返回    assert response.elapsed.total_seconds() < 1.0    assert response.status_code == 200    assert len(response.json()["data"]) == 100

import pytestimport random# 1. 商品数据工厂class ProductFactory:    @classmethod    def create(cls, **kwargs):        base_product = {            "product_id": f"PROD{random.randint(100000, 999999)}",            "product_name": fake.product_name(),            "price": round(random.uniform(10.0, 1000.0), 2),            "stock": random.randint(10, 1000)        }        return {**base_product, **kwargs}# 2. 订单数据工厂（依赖用户和商品）class OrderFactory:    @classmethod    def create(cls, user, product, **kwargs):        """创建订单，依赖用户和商品数据"""        base_order = {            "order_no": f"ORDER{random.randint(1000000, 9999999)}",            "user_id": user["username"],            "product_id": product["product_id"],            "order_amount": product["price"],            "create_time": fake.date_time_this_hour().strftime("%Y-%m-%d %H:%M:%S")        }        return {**base_order, **kwargs}# 3. Fixture：生成订单依赖链（用户→商品→订单）@pytest.fixture(scope="class")  # 整个测试类共享，避免重复创建def order_chain():    """创建订单依赖链，测试后反向清理"""    # 1. 创建用户    user = UserFactory.create()    db.insert_user(user)    # 2. 创建商品    product = ProductFactory.create()    db.insert_product(product)    # 3. 创建订单    order = OrderFactory.create(user=user, product=product)    db.insert_order(order)    # 返回完整依赖链    yield {"user": user, "product": product, "order": order}    # 测试结束，反向清理（先删订单，再删商品，最后删用户）    db.delete_order(order["order_no"])    db.delete_product(product["product_id"])    db.delete_user(user["username"])# 订单流程测试类class TestOrderFlow:    """测试订单完整流程：创建→支付→查询"""    def test_create_order(self, order_chain):        """测试创建订单接口"""        # 直接使用order_chain中的订单数据        response = requests.post("http://xxx.xxx.xxx/api/order/create", json=order_chain["order"])        assert response.status_code == 200        assert response.json()["order_no"] == order_chain["order"]["order_no"]    def test_pay_order(self, order_chain):        """测试订单支付接口"""        pay_data = {            "order_no": order_chain["order"]["order_no"],            "pay_amount": order_chain["order"]["order_amount"],            "pay_type": "wechat"        }        response = requests.post("http://xxx.xxx.xxx/api/order/pay", json=pay_data)        assert response.json()["status"] == "paid"

# 注册表模式优化多类型工厂_REGISTRY = {}# 注册装饰器def register_db(name: str):    def deco(cls):        _REGISTRY[name] = cls        return cls    return deco# 注册不同数据库类@register_db("mysql")class MySQL:    pass@register_db("pg")class Postgres:    pass# 工厂函数，无需分支判断def make_db(kind: str, **cfg):    try:        return _REGISTRY[kind](**cfg)    except KeyError:        raise ValueError(f"不支持的数据库类型：{kind}，可选类型：{list(_REGISTRY.keys())}")