python实现智能测试数据工厂

[DataFactory]     │    ├── [UserGenerator] → 生成用户数据（含风控标签）    ├── [OrderGenerator] → 生成订单（关联用户等级）    ├── [PaymentGenerator] → 生成支付（关联订单风险）    │    └── [UniquenessManager] → 全局唯一性控制         └── [RedisStorage] (可选) → 分布式去重‍

# utils/uniqueness.pyimport threadingimport hashlibfrom typing import Setclass UniquenessManager:    _instance = None    _lock = threading.Lock()    _used_values: Set[str] = set()    def __new__(cls):        if cls._instance is None:            with cls._lock:                if cls._instance is None:                    cls._instance = super().__new__(cls)        return cls._instance    def ensure_unique(self, value: str, max_retries: int = 5) -> str:        """确保值全局唯一，自动加后缀"""        base_value = str(value)        if base_value not in self._used_values:            self._used_values.add(base_value)            return base_value        for i in range(1, max_retries + 1):            candidate = f"{base_value}_{i}"            if candidate not in self._used_values:                self._used_values.add(candidate)                return candidate        raise RuntimeError(f"无法生成唯一值: {base_value}")    def reset(self):        """测试间清理（用于 fixture）"""        with self._lock:            self._used_values.clear()

# generators/base.pyfrom abc import ABC, abstractmethodfrom typing import Dict, Anyfrom faker import Fakerfrom utils.uniqueness import UniquenessManagerclass BaseGenerator(ABC):    def __init__(self, locale: str = "zh_CN"):        self.faker = Faker(locale)        self.uniqueness = UniquenessManager()    @abstractmethod    def generate(self, scenario: str = "normal", **kwargs) -> Dict[str, Any]:        """        生成数据        :param scenario: normal | boundary | abnormal        :param kwargs: 业务参数（如 user_level="VIP"）        """        pass‍

# generators/user.pyfrom generators.base import BaseGeneratorimport randomclass UserGenerator(BaseGenerator):    USER_LEVELS = ["NORMAL", "VIP", "BLACKLIST"]    RISK_PROFILES = {        "NORMAL": {"risk_score": 10, "max_order": 5000},        "VIP": {"risk_score": 30, "max_order": 50000},        "BLACKLIST": {"risk_score": 90, "max_order": 0}    }    def generate(self, scenario: str = "normal", **kwargs) -> dict:        # 1. 确定用户等级        level = kwargs.get("level")        if not level:            level = self._determine_level_by_scenario(scenario)        # 2. 生成基础信息        username_base = self.faker.user_name()        email_base = self.faker.free_email()        # 3. 保证唯一性        username = self.uniqueness.ensure_unique(f"auto_{level.lower()}_{username_base}")        email = self.uniqueness.ensure_unique(f"{username}@{email_base.split('@')[1]}")        # 4. 生成身份证（中国）        id_card = self.faker.ssn()        # 5. 添加业务属性        profile = self.RISK_PROFILES[level]        return {            "user_id": None,  # 待注册后填充            "username": username,            "email": email,            "phone": self.faker.phone_number(),            "id_card": id_card,            "level": level,            "risk_score": profile["risk_score"],            "max_order_amount": profile["max_order"]        }    def _determine_level_by_scenario(self, scenario: str) -> str:        if scenario == "abnormal":            return random.choice(["BLACKLIST", "VIP"])  # 异常场景包含高风险用户        elif scenario == "boundary":            return "VIP"        else:            return "NORMAL"

# generators/order.pyfrom generators.base import BaseGeneratorfrom typing import Dictclass OrderGenerator(BaseGenerator):    PRODUCTS = ["LAPTOP", "PHONE", "BOOK"]    def generate(self, scenario: str = "normal", user: Dict = None, **kwargs) -> dict:        if not user:            raise ValueError("订单必须关联用户")        # 1. 确定订单金额        amount = self._calculate_amount(scenario, user, kwargs.get("amount"))        # 2. 生成唯一订单号        order_no_base = f"ORD{self.faker.random_int(100000, 999999)}"        order_no = self.uniqueness.ensure_unique(order_no_base)        return {            "order_no": order_no,            "user_id": user["user_id"],            "amount": amount,            "product": random.choice(self.PRODUCTS),            "risk_level": "HIGH" if user["risk_score"] > 50 else "LOW"        }    def _calculate_amount(self, scenario: str, user: Dict, override: float = None) -> float:        if override is not None:            return override        base = user["max_order_amount"] or 1000        if scenario == "abnormal":            return base * random.uniform(1.0, 2.0)  # 超限交易        elif scenario == "boundary":            return base  # 边界值        else:            return base * random.uniform(0.1, 0.8)  # 正常范围‍

# factory/data_factory.pyfrom generators.user import UserGeneratorfrom generators.order import OrderGeneratorfrom typing import Dict, Anyclass DataFactory:    _generators = {        "user": UserGenerator(),        "order": OrderGenerator(),    }    @classmethod    def create(cls, data_type: str, scenario: str = "normal", **kwargs) -> Dict[str, Any]:        """        智能生成测试数据        示例:          user = DataFactory.create("user", scenario="abnormal", level="VIP")          order = DataFactory.create("order", user=user, scenario="boundary")        """        if data_type not in cls._generators:            raise ValueError(f"不支持的数据类型: {data_type}")        return cls._generators[data_type].generate(scenario=scenario, **kwargs)    @classmethod    def create_scenario(cls, scenario_name: str) -> Dict[str, Any]:        """        一键生成完整业务场景（预留 AI 扩展点）        """        scenarios = {            "high_risk_payment": {                "user": {"scenario": "abnormal", "level": "VIP"},                "order": {"scenario": "abnormal"}            },            "normal_checkout": {                "user": {"scenario": "normal"},                "order": {"scenario": "normal"}            }        }        config = scenarios.get(scenario_name)        if not config:            raise ValueError(f"未知场景: {scenario_name}")        result = {}        # 先生成用户        result["user"] = cls.create("user", **config["user"])        # 再生成订单（依赖用户）        result["order"] = cls.create("order", user=result["user"], **config["order"])        return result‍

# test_payment.pydef test_high_risk_payment_blocked():    # 一行代码生成完整场景    scenario = DataFactory.create_scenario("high_risk_payment")    # 注册用户（假设返回 user_id）    user_resp = register_user(scenario["user"])    scenario["user"]["user_id"] = user_resp["user_id"]    # 创建大额订单    order_resp = create_order(scenario["order"])    # 尝试支付 → 应被风控拦截    pay_resp = pay_order(order_resp["order_id"])    assert pay_resp["code"] == "RISK_BLOCKED"

def test_order_amount_boundary():    user = DataFactory.create("user", level="VIP")    # 生成刚好等于最大限额的订单    order = DataFactory.create("order", user=user, scenario="boundary")    assert order["amount"] == user["max_order_amount"]‍

# 伪代码：通过 LLM 解析指令def create_from_nlp(prompt: str):    # 调用 LLM: "生成一个黑名单用户的超限订单"    # → 返回 {"data_type": "scenario", "name": "high_risk_payment"}    config = llm_parse(prompt)    return DataFactory.create_scenario(config["name"])

# 自动生成 Markdown 文档def generate_data_dictionary():    docs = []    for name, gen in DataFactory._generators.items():        docs.append(f"## {name}\n")        docs.append(f"- 支持场景: normal, boundary, abnormal\n")        # 可通过解析 generate 方法的 docstring 获取字段说明    return "\n".join(docs)‍

# test_uniqueness.pyimport threadingdef test_concurrent_uniqueness():    results = set()    lock = threading.Lock()    def worker():        for _ in range(10):            data = DataFactory.create("user")            with lock:                assert data["username"] not in results                results.add(data["username"])    threads = [threading.Thread(target=worker) for _ inrange(5)]    for t in threads: t.start()    for t in threads: t.join()    assert len(results) == 50  # 5线程 * 10次 = 50唯一值‍