Python 接口自动化:10 个装饰器,从懵逼到上手

def say_hello(func):    def wrapper(*args, **kwargs):        print("开始执行")        result = func(*args, **kwargs)        print("执行完毕")        return result    return wrapper@say_hellodef add(a, b):    return a + badd(1, 2)

import timeimport functoolsdef retry(max_times=3, interval=1):    """    max_times: 最多重试几次    interval:  每次重试间隔几秒    """    def decorator(func):        @functools.wraps(func)        def wrapper(*args, **kwargs):            for attempt in range(1, max_times + 1):                try:                    return func(*args, **kwargs)                except Exception as e:                    print(f"第 {attempt} 次执行失败：{e}")                    if attempt < max_times:                        time.sleep(interval)                    else:                        raise  # 重试完了还是失败，就把错误抛出去        return wrapper    return decorator# ---- 使用示例 ----import requests@retry(max_times=3, interval=2)def test_order_api():    resp = requests.post("http://example.com/api/order", json={"item_id": 1})    assert resp.status_code == 200, f"接口返回异常：{resp.status_code}"test_order_api()

import timeimport functoolsdef timeit(func):    @functools.wraps(func)    def wrapper(*args, **kwargs):        start = time.perf_counter()        result = func(*args, **kwargs)        end = time.perf_counter()        print(f"[耗时] {func.__name__} 执行了 {end - start:.3f} 秒")        return result    return wrapper# ---- 使用示例 ----@timeitdef test_login_api():    resp = requests.post("http://example.com/api/login", json={        "username": "test_user",        "password": "123456"    })    assert resp.status_code == 200

import functoolsimport requests# 模拟一个全局 token 缓存_token_cache = {}def get_token():    """统一登录，获取 token"""    if "token" not in _token_cache:        resp = requests.post("http://example.com/api/login", json={            "username": "auto_test",            "password": "test123"        })        _token_cache["token"] = resp.json()["token"]    return _token_cache["token"]def with_token(func):    @functools.wraps(func)    def wrapper(*args, **kwargs):        token = get_token()        # 把 token 通过 kwargs 传给测试函数        kwargs["token"] = token        return func(*args, **kwargs)    return wrapper# ---- 使用示例 ----@with_tokendef test_get_user_info(token):    headers = {"Authorization": f"Bearer {token}"}    resp = requests.get("http://example.com/api/user/info", headers=headers)    assert resp.status_code == 200    assert resp.json()["username"] == "auto_test"test_get_user_info()

import functoolsimport tracebackdef log_on_failure(func):    @functools.wraps(func)    def wrapper(*args, **kwargs):        try:            return func(*args, **kwargs)        except Exception as e:            print(f"\n{'='*50}")            print(f"用例失败：{func.__name__}")            print(f"错误信息：{e}")            print(f"完整堆栈：\n{traceback.format_exc()}")            print(f"{'='*50}\n")            raise  # 依然抛出异常，让测试框架记录失败    return wrapper# ---- 使用示例 ----@log_on_failuredef test_create_user():    resp = requests.post("http://example.com/api/user", json={        "name": "张三",        "email": "invalid-email"  # 故意传个非法邮箱触发失败    })    assert resp.status_code == 201, f"实际响应：{resp.status_code}, body: {resp.text}"test_create_user()

import osimport functoolsdef skip_in_env(*envs):    """    如果当前环境在 envs 列表里，就跳过这个测试    用法: @skip_in_env("prod", "staging")    """    def decorator(func):        @functools.wraps(func)        def wrapper(*args, **kwargs):            current_env = os.getenv("TEST_ENV", "dev")            if current_env in envs:                print(f"[跳过] {func.__name__} 在 {current_env} 环境下不执行")                return None            return func(*args, **kwargs)        return wrapper    return decorator# ---- 使用示例 ----@skip_in_env("prod", "staging")  # 生产和预发环境跳过def test_delete_all_test_data():    resp = requests.delete("http://example.com/api/testdata/clear")    assert resp.status_code == 200# 设置环境变量os.environ["TEST_ENV"] = "prod"test_delete_all_test_data()# 输出: [跳过] test_delete_all_test_data 在 prod 环境下不执行

import functoolsdef validate_schema(expected_keys):    """    检查响应 JSON 里是否包含所有必需字段    expected_keys: 必须存在的字段列表，如 ["id", "name", "email"]    """    def decorator(func):        @functools.wraps(func)        def wrapper(*args, **kwargs):            resp = func(*args, **kwargs)            if resp is None:                return resp            body = resp.json()            missing = [k for k in expected_keys if k not in body]            if missing:                raise AssertionError(f"响应缺少字段：{missing}，实际返回：{list(body.keys())}")            print(f"[校验通过] {func.__name__} 响应结构正常")            return resp        return wrapper    return decorator# ---- 使用示例 ----@validate_schema(expected_keys=["id", "username", "email", "created_at"])def test_get_user_detail():    resp = requests.get("http://example.com/api/user/1")    assert resp.status_code == 200    return resp  # 注意：装饰器需要函数返回 resptest_get_user_detail()

import functoolsimport threadingdef timeout(seconds=10):    """    超过 seconds 秒没执行完，抛出 TimeoutError    """    def decorator(func):        @functools.wraps(func)        def wrapper(*args, **kwargs):            result = [None]            exception = [None]            def target():                try:                    result[0] = func(*args, **kwargs)                except Exception as e:                    exception[0] = e            thread = threading.Thread(target=target)            thread.daemon = True            thread.start()            thread.join(timeout=seconds)            if thread.is_alive():                raise TimeoutError(f"{func.__name__} 执行超时（>{seconds}秒）")            if exception[0]:                raise exception[0]            return result[0]        return wrapper    return decorator# ---- 使用示例 ----@timeout(seconds=5)def test_slow_api():    # 模拟一个需要 8 秒才响应的接口    import time    time.sleep(8)    resp = requests.get("http://example.com/api/report/generate")    assert resp.status_code == 200test_slow_api()# 输出: TimeoutError: test_slow_api 执行超时（>5秒）

import functoolsdef cleanup(cleanup_func):    """    测试执行完后（不管成功还是失败），自动调用 cleanup_func 清理数据    """    def decorator(func):        @functools.wraps(func)        def wrapper(*args, **kwargs):            created_ids = []            try:                result = func(*args, created_ids=created_ids, **kwargs)                return result            finally:                if created_ids:                    cleanup_func(created_ids)        return wrapper    return decoratordef delete_test_users(user_ids):    print(f"清理测试用户：{user_ids}")    for uid in user_ids:        requests.delete(f"http://example.com/api/user/{uid}")# ---- 使用示例 ----@cleanup(cleanup_func=delete_test_users)def test_create_multiple_users(created_ids):    for i in range(3):        resp = requests.post("http://example.com/api/user", json={"name": f"测试用户{i}"})        assert resp.status_code == 201        created_ids.append(resp.json()["id"])  # 把创建的 id 存进去，cleanup 会拿去删test_create_multiple_users()# 用例跑完后自动输出: 清理测试用户：[101, 102, 103]

import functoolsimport jsonfrom unittest.mock import patch# 简单模拟一个日志收集器http_records = []def record_http(func):    @functools.wraps(func)    def wrapper(*args, **kwargs):        original_request = requests.Session.request        def patched_request(self, method, url, **req_kwargs):            resp = original_request(self, method, url, **req_kwargs)            record = {                "test_case": func.__name__,                "method": method.upper(),                "url": url,                "request_body": req_kwargs.get("json") or req_kwargs.get("data"),                "status_code": resp.status_code,                "response_body": resp.text[:500]  # 只记前 500 个字符            }            http_records.append(record)            print(f"[HTTP记录] {method.upper()}{url} -> {resp.status_code}")            return resp        with patch.object(requests.Session, "request", patched_request):            return func(*args, **kwargs)    return wrapper# ---- 使用示例 ----@record_httpdef test_user_workflow():    # 登录    requests.post("http://example.com/api/login", json={"username": "u1", "password": "p1"})    # 查用户信息    requests.get("http://example.com/api/user/info")test_user_workflow()print("\n完整 HTTP 记录：")print(json.dumps(http_records, ensure_ascii=False, indent=2))

import functoolsimport threadingdef concurrent(users=10):    """    同时用 users 个线程并发执行这个测试函数    任何一个线程失败，整体标记为失败    """    def decorator(func):        @functools.wraps(func)        def wrapper(*args, **kwargs):            errors = []            lock = threading.Lock()            def run():                try:                    func(*args, **kwargs)                except Exception as e:                    with lock:                        errors.append(str(e))            threads = [threading.Thread(target=run) for _ in range(users)]            for t in threads:                t.start()            for t in threads:                t.join()            if errors:                raise AssertionError(                    f"{len(errors)}/{users} 个并发请求失败：\n" + "\n".join(errors[:5])                )            print(f"[并发测试] {users} 个并发全部通过")        return wrapper    return decorator# ---- 使用示例 ----@concurrent(users=10)def test_search_api_concurrent():    resp = requests.get("http://example.com/api/search", params={"q": "python"})    assert resp.status_code == 200    assert "results" in resp.json()test_search_api_concurrent()# 输出: [并发测试] 10 个并发全部通过

@timeit@retry(max_times=3, interval=1)@log_on_failuredef test_payment_api():    resp = requests.post("http://example.com/api/pay", json={"order_id": "123"})    assert resp.status_code == 200test_payment_api()