一个框架，让ERP测试从30天缩到3天，效率提升900%_千锋教育

class ERPDataFactory:
    def create_purchase_order(self, supplier_id, items):
        # 创建采购订单
        po_id = self.db.execute("INSERT INTO po_header VALUES (...) RETURNING id")
        for item in items:
            self.db.execute(f"INSERT INTO po_item VALUES ({po_id}, {item})")
        # 自动生成库存预留
        self.db.execute(f"UPDATE inventory SET reserved = reserved + {total_qty}")
        return po_id

class BusinessFlow:
    def execute_purchase_flow(self):
        # 创建采购订单
        po_id = self.create_po()
        assert self.get_status(po_id) == '草稿'
        
        # 提交审批
        self.approve(po_id)
        assert self.get_status(po_id) == '已审批'
        
        # 收货
        self.receive_goods(po_id)
        assert self.inventory_check() == '库存增加'
        
        # 财务记账
        self.post_invoice(po_id)
        assert self.get_account_balance() == '应付增加'

# 批量生成1000个采购订单数据
data_factory = ERPDataFactory()
for i in range(1000):
    po_id = data_factory.create_purchase_order(
        supplier_id=random.choice(suppliers),
        items=[{'material': 'A001', 'qty': 100}]
    )
    # 记录每个订单的预期状态
    expected_states[po_id] = '草稿'

# 并行审批1000个订单
with ThreadPoolExecutor(max_workers=50) as executor:
    futures = []
    for po_id in po_ids:
        future = executor.submit(self.approve_flow, po_id)
        futures.append(future)
    
    for future in as_completed(futures):
        result = future.result()
        if result['status'] != '已审批':
            error_log.append(f"PO {result['id']} 审批失败")

# 验证库存、财务、供应商数据一致性
def validate_data_consistency():
    for po_id in po_ids:
        # 检查库存
        inventory = db.query(f"SELECT * FROM inventory WHERE po_id={po_id}")
        assert inventory.qty == 100, f"PO {po_id} 库存数量不符"
        
        # 检查财务
        finance = db.query(f"SELECT * FROM ap WHERE po_id={po_id}")
        assert finance.amount > 0, f"PO {po_id} 财务未记账"
        
        # 检查供应商状态
        supplier = db.query(f"SELECT * FROM supplier WHERE po_id={po_id}")
        assert supplier.status == '已付款', f"PO {po_id} 供应商未付款"

# 测试结束后自动清理数据
def cleanup_test_data():
    db.execute("DELETE FROM po_item WHERE created_by = 'test_script'")
    db.execute("DELETE FROM po_header WHERE created_by = 'test_script'")
    db.execute("DELETE FROM inventory_log WHERE source = 'auto_test'")

# 模拟100个用户并发操作
def concurrent_test():
    users = [create_test_user() for _ in range(100)]
    with ThreadPoolExecutor(max_workers=100) as executor:
        futures = [executor.submit(user.create_order) for user in users]
        results = [f.result() for f in as_completed(futures)]
    
    # 检查是否有资源竞争
    assert len(set([r.order_id for r in results])) == 100
    assert all([r.status == '成功' for r in results])