Python dowhen想你想做

pip install dowhen
import dowhen
print(dowhen.__version__)  # 版本

when(目标函数, "触发点描述")

when(func, "point").do(...).bp().goto(...)

from dowhen import when, do

def  process_data(data):
    # 复杂数据处理逻辑
    result = [x * 2for x in data if x > 0]
    returnsum(result)

# 注入监控点
monitor = when(process_data, "result = [...]").do(
    'print(f"输入: {data}, 中间结果: {result}")'
)

# 测试用例
assert process_data([1, 2, 3]) == 12
monitor.remove()

from dowhen import when, do

def  safe_divide(a, b):
    if b == 0:
        returnfloat('inf')
    return a / b

# 强制触发除零错误
with when(safe_divide, "if b == 0").do("b = 1"):  # 防止除零
    assert safe_divide(10, 0) == 10.0

# 测试错误处理
error_handler = when(safe_divide, "return a / b").do(
    'raise ValueError("模拟计算错误")'
)
try:
    safe_divide(10, 2)
except ValueError as e:
    print(f"成功捕获模拟错误: {e}")
error_handler.remove()

from dowhen import when, goto
import time

def expensive_operation(n):
    # 模拟耗时操作
    time.sleep(1)
    result = n * 2
    return result

# 测试跳过耗时操作
with when(expensive_operation, "time.sleep").goto("return result"):
    start = time.time()
    assert expensive_operation(5) == 10
    print(f"执行时间: {time.time() - start:.4f}秒")  # ≈0秒

from dowhen import when, bp

def  complex_calculation(a, b):
    x = a + b
    y = x * 2
    z = y ** 2
    return z

# 设置条件断点
debugger = when(complex_calculation, "y = x * 2").bp().do(
    'print(f"a={a}, b={b}, x={x}, y={y}")'
)

# 触发调试
complex_calculation(3, 4)  # 进入pdb调试器
debugger.disable()  # 禁用调试器

from dowhen import when, do, goto

def data_pipeline(data):
    data = [x for x in data if x > 0]  # 过滤
    total = sum(data)  # 聚合
    return total / len(data)  # 计算均值

# 多阶段测试
test_handler = (
    when(data_pipeline, "data = [...]")
    .do('print(f"过滤后数据: {data}")')
    .when(data_pipeline, "total = sum")
    .do('print(f"求和结果: {total}")')
    .when(data_pipeline, "return total")
    .goto("data = [...]")  # 循环回起点
)

# 执行测试
result = data_pipeline([1, -2, 3, -4, 5])
print(f"最终结果: {result}")
test_handler.remove()

from dowhen import when, do

def api_client(request):
    # 模拟API调用
    return {"status": "success", "data": request * 2}

# 创建可切换的测试点
test_point = when(api_client, "return").do(
    'print(f"API调用: 请求={request}, 响应={_}");'
    'if "error" in request: _["status"] = "error"'
)

# 正常模式
print("正常响应:", api_client("test")) 

# 测试错误处理
test_point.enable()
print("错误模拟:", api_client({"error": True}))
test_point.disable()

# 清晰命名便于管理
perf_test = when(func, "slow_op").goto(...)
error_test = when(func, "validate").do(...)

# pytest示例
import pytest

@pytest.fixture
def mock_database():
    with when(db, "query").do('return ["mock", "data"]'):
        yield

def test_app(mock_database):
    assert app.run() == ["mock", "data"]