13.pytest的接口自动化封装之接口关联

作者

Pytest 接口自动化封装之接口关联详解

1. 什么是接口关联

1.1 接口关联的基本概念

接口关联是指在接口自动化测试中，一个接口的响应结果作为另一个接口的请求参数或前置条件。简单来说，就是接口之间的数据依赖关系。

1.2 形象比喻

想象一下你去银行办理业务：

场景 1：取款流程

  1. 首先需要登录，获取你的身份凭证（token）
  2. 然后使用这个凭证去查询余额
  3. 最后使用这个凭证去取款

在这个流程中：

  • 登录接口返回的 token 是后续接口的前置条件
  • 查询余额和取款接口都需要使用这个 token
  • 这就是典型的接口关联

场景 2：电商购物流程

  1. 登录 → 获取用户 token
  2. 添加商品到购物车 → 需要 token，返回购物车ID
  3. 创建订单 → 需要 token 和购物车ID
  4. 支付订单 → 需要 token 和订单ID

在这个流程中，每个步骤都依赖前一步的结果，形成了接口关联链。

1.3 接口关联的核心要素

接口关联主要涉及以下几个核心要素：

  1. 关联数据：需要传递的数据（如 token、用户ID、订单ID 等）
  2. 数据提取：从接口响应中提取需要的数据
  3. 数据存储：将提取的数据存储起来，供后续使用
  4. 数据传递：将存储的数据传递给后续接口

1.4 接口关联的常见类型

1.4.1 Token 关联

最常见的关联类型，用于身份认证：

# 示例：登录获取 token
# 请求
POST /api/login
{
    "username": "admin",
    "password": "123456"
}

# 响应
{
    "code": 200,
    "data": {
        "token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
        "user_id": 1001
    }
}

# 后续接口需要使用这个 token
GET /api/user/info
Headers: {
    "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
}

1.4.2 ID 关联

一个接口返回的ID，作为另一个接口的参数：

# 示例：创建订单后获取订单ID
# 请求
POST /api/order/create
{
    "product_id": 100,
    "quantity": 2
}

# 响应
{
    "code": 200,
    "data": {
        "order_id": "ORD20240101001",
        "total_price": 199.00
    }
}

# 后续接口使用订单ID
GET /api/order/detail?order_id=ORD20240101001

1.4.3 数据状态关联

一个接口改变了数据状态，影响另一个接口的结果：

# 示例：先创建用户，再查询用户
# 步骤1：创建用户
POST /api/user/create
{
    "name": "张三",
    "email": "zhangsan@example.com"
}
# 返回：user_id = 1001

# 步骤2：查询刚创建的用户
GET /api/user/1001
# 需要确保用户已创建，才能查询到

2. 为什么需要接口关联

2.1 没有接口关联的问题

让我们看一个实际的例子，说明为什么需要接口关联：

场景：测试用户信息查询接口

没有接口关联的情况：

import requests

def test_get_user_info():
    """测试获取用户信息"""
    # 问题1：硬编码 token，容易过期
    token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.xxx"

    # 问题2：如果 token 过期，测试就会失败
    headers = {
        "Authorization": f"Bearer {token}"
    }

    response = requests.get(
        "http://api.example.com/user/info",
        headers=headers
    )

    assert response.status_code == 200
    assert response.json()["code"] == 200

存在的问题：

  1. Token 过期问题：硬编码的 token 会过期，需要频繁更新
  2. 测试不稳定：每次 token 过期，测试就会失败
  3. 无法自动化：需要手动获取 token，无法完全自动化
  4. 维护成本高：多个测试用例都需要手动维护 token

2.2 使用接口关联的优势

使用接口关联后：

import requests

def test_get_user_info():
    """测试获取用户信息"""
    # 优势1：自动获取最新的 token
    login_response = requests.post(
        "http://api.example.com/login",
        json={"username": "admin", "password": "123456"}
    )
    token = login_response.json()["data"]["token"]

    # 优势2：使用动态获取的 token
    headers = {
        "Authorization": f"Bearer {token}"
    }

    response = requests.get(
        "http://api.example.com/user/info",
        headers=headers
    )

    assert response.status_code == 200
    assert response.json()["code"] == 200

优势：

  1. 自动化程度高：完全自动化，无需手动干预
  2. 测试稳定：每次都获取最新的 token，不会过期
  3. 维护成本低：token 获取逻辑集中管理
  4. 真实场景：模拟真实的业务流程

2.3 接口关联的实际应用场景

场景 1：完整的业务流程测试

# 测试完整的购物流程
def test_shopping_flow():
    # 1. 登录获取 token
    token = login()

    # 2. 添加商品到购物车（需要 token）
    cart_id = add_to_cart(token, product_id=100)

    # 3. 创建订单（需要 token 和 cart_id）
    order_id = create_order(token, cart_id)

    # 4. 支付订单（需要 token 和 order_id）
    payment_result = pay_order(token, order_id)

    # 5. 查询订单状态（需要 token 和 order_id）
    order_status = get_order_status(token, order_id)

    assert payment_result["status"] == "success"
    assert order_status["status"] == "paid"

场景 2：数据依赖测试

# 测试用户相关的所有接口
def test_user_operations():
    # 1. 创建用户，获取 user_id
    user_id = create_user(name="张三", email="zhangsan@example.com")

    # 2. 查询用户（需要 user_id）
    user_info = get_user(user_id)
    assert user_info["name"] == "张三"

    # 3. 更新用户（需要 user_id）
    update_user(user_id, name="李四")

    # 4. 再次查询验证（需要 user_id）
    user_info = get_user(user_id)
    assert user_info["name"] == "李四"

    # 5. 删除用户（需要 user_id）
    delete_user(user_id)

3. 接口关联的实现方式

3.1 方式一：直接在测试用例中实现

这是最简单的方式，直接在测试用例中调用关联接口：

import requests
import pytest

class TestUserAPI:
    """用户接口测试"""

    BASE_URL = "http://api.example.com"

    def test_get_user_info(self):
        """测试获取用户信息"""
        # 步骤1：登录获取 token
        login_url = f"{self.BASE_URL}/login"
        login_data = {
            "username": "admin",
            "password": "123456"
        }
        login_response = requests.post(login_url, json=login_data)
        login_result = login_response.json()

        # 提取 token
        token = login_result["data"]["token"]

        # 步骤2：使用 token 获取用户信息
        headers = {
            "Authorization": f"Bearer {token}"
        }
        user_info_url = f"{self.BASE_URL}/user/info"
        response = requests.get(user_info_url, headers=headers)

        # 断言
        assert response.status_code == 200
        result = response.json()
        assert result["code"] == 200
        assert "data" in result

优点：

  • 简单直接，容易理解
  • 不需要额外的配置

缺点：

  • 代码重复：每个测试用例都要写登录逻辑
  • 维护困难：如果登录接口改变，要修改所有测试用例
  • 效率低：每个测试用例都要重新登录

3.2 方式二：使用函数封装

将关联逻辑封装成函数，在测试用例中调用：

import requests
import pytest

class TestUserAPI:
    """用户接口测试"""

    BASE_URL = "http://api.example.com"

    def login(self):
        """登录并返回 token"""
        login_url = f"{self.BASE_URL}/login"
        login_data = {
            "username": "admin",
            "password": "123456"
        }
        response = requests.post(login_url, json=login_data)
        result = response.json()
        return result["data"]["token"]

    def test_get_user_info(self):
        """测试获取用户信息"""
        # 调用封装的登录函数
        token = self.login()

        # 使用 token 获取用户信息
        headers = {
            "Authorization": f"Bearer {token}"
        }
        user_info_url = f"{self.BASE_URL}/user/info"
        response = requests.get(user_info_url, headers=headers)

        # 断言
        assert response.status_code == 200
        result = response.json()
        assert result["code"] == 200

优点：

  • 代码复用：登录逻辑只写一次
  • 易于维护：修改登录逻辑只需改一个地方

缺点：

  • 每个测试用例仍要调用登录函数
  • 无法在测试用例之间共享 token（每个测试都重新登录）

3.3 方式三：使用 Fixture（推荐）

使用 pytest 的 fixture 机制，这是最推荐的方式：

import requests
import pytest

class TestUserAPI:
    """用户接口测试"""

    BASE_URL = "http://api.example.com"

    @pytest.fixture
    def token(self):
        """登录并返回 token 的 fixture"""
        login_url = f"{self.BASE_URL}/login"
        login_data = {
            "username": "admin",
            "password": "123456"
        }
        response = requests.post(login_url, json=login_data)
        result = response.json()
        token = result["data"]["token"]
        return token

    def test_get_user_info(self, token):
        """测试获取用户信息"""
        # 直接使用 fixture 提供的 token
        headers = {
            "Authorization": f"Bearer {token}"
        }
        user_info_url = f"{self.BASE_URL}/user/info"
        response = requests.get(user_info_url, headers=headers)

        # 断言
        assert response.status_code == 200
        result = response.json()
        assert result["code"] == 200

    def test_update_user_info(self, token):
        """测试更新用户信息"""
        # 同样使用 fixture 提供的 token
        headers = {
            "Authorization": f"Bearer {token}"
        }
        update_url = f"{self.BASE_URL}/user/info"
        update_data = {
            "nickname": "新昵称"
        }
        response = requests.put(update_url, json=update_data, headers=headers)

        # 断言
        assert response.status_code == 200

优点：

  • 代码复用：登录逻辑只写一次
  • 自动注入：pytest 自动将 token 注入到测试用例
  • 易于维护：修改登录逻辑只需改 fixture
  • 支持作用域：可以控制 token 的生命周期

缺点：

  • 需要理解 pytest fixture 的概念（但这是值得的）

4. 使用 Fixture 实现接口关联（详细讲解）

4.1 Fixture 基础回顾

在深入学习接口关联之前，我们先快速回顾一下 fixture 的基础知识：

Fixture 的作用域（Scope）：

  1. function（默认）：每个测试函数执行一次
  2. class：每个测试类执行一次
  3. module：每个测试模块执行一次
  4. package：每个测试包执行一次
  5. session：整个测试会话执行一次

4.2 简单的 Token 关联示例

让我们从一个最简单的例子开始：

import requests
import pytest

# 基础配置
BASE_URL = "http://api.example.com"

@pytest.fixture
def login_token():
    """登录并返回 token"""
    print("n=== 开始登录 ===")
    url = f"{BASE_URL}/login"
    data = {
        "username": "admin",
        "password": "123456"
    }
    response = requests.post(url, json=data)
    result = response.json()

    # 提取 token
    token = result["data"]["token"]
    print(f"=== 登录成功，获取 token: {token[:20]}... ===")

    return token

def test_get_user_info(login_token):
    """测试获取用户信息"""
    print(f"n使用 token: {login_token[:20]}...")

    headers = {
        "Authorization": f"Bearer {login_token}"
    }
    url = f"{BASE_URL}/user/info"
    response = requests.get(url, headers=headers)

    assert response.status_code == 200
    result = response.json()
    assert result["code"] == 200
    print("✓ 获取用户信息成功")

def test_update_user_info(login_token):
    """测试更新用户信息"""
    print(f"n使用 token: {login_token[:20]}...")

    headers = {
        "Authorization": f"Bearer {login_token}"
    }
    url = f"{BASE_URL}/user/info"
    data = {
        "nickname": "新昵称"
    }
    response = requests.put(url, json=data, headers=headers)

    assert response.status_code == 200
    result = response.json()
    assert result["code"] == 200
    print("✓ 更新用户信息成功")

运行结果：

$ pytest test_user_api.py -v -s

test_user_api.py::test_get_user_info 
=== 开始登录 ===
=== 登录成功，获取 token: eyJhbGciOiJIUzI1NiIs... ===

使用 token: eyJhbGciOiJIUzI1NiIs...
✓ 获取用户信息成功
PASSED

test_user_api.py::test_update_user_info 
=== 开始登录 ===
=== 登录成功，获取 token: eyJhbGciOiJIUzI1NiIs... ===

使用 token: eyJhbGciOiJIUzI1NiIs...
✓ 更新用户信息成功
PASSED

注意：每个测试用例都会执行一次登录，因为默认的 fixture 作用域是 function。

4.3 使用 Session 作用域共享 Token

如果 token 在测试会话期间不会过期，我们可以使用 session 作用域，让所有测试用例共享同一个 token：

import requests
import pytest

BASE_URL = "http://api.example.com"

@pytest.fixture(scope="session")
def login_token():
    """登录并返回 token（整个测试会话只执行一次）"""
    print("n=== [SESSION] 开始登录 ===")
    url = f"{BASE_URL}/login"
    data = {
        "username": "admin",
        "password": "123456"
    }
    response = requests.post(url, json=data)
    result = response.json()

    token = result["data"]["token"]
    print(f"=== [SESSION] 登录成功，获取 token: {token[:20]}... ===")

    return token

def test_get_user_info(login_token):
    """测试获取用户信息"""
    print(f"n[test_get_user_info] 使用 token: {login_token[:20]}...")

    headers = {
        "Authorization": f"Bearer {login_token}"
    }
    url = f"{BASE_URL}/user/info"
    response = requests.get(url, headers=headers)

    assert response.status_code == 200
    result = response.json()
    assert result["code"] == 200
    print("✓ 获取用户信息成功")

def test_update_user_info(login_token):
    """测试更新用户信息"""
    print(f"n[test_update_user_info] 使用 token: {login_token[:20]}...")

    headers = {
        "Authorization": f"Bearer {login_token}"
    }
    url = f"{BASE_URL}/user/info"
    data = {
        "nickname": "新昵称"
    }
    response = requests.put(url, json=data, headers=headers)

    assert response.status_code == 200
    result = response.json()
    assert result["code"] == 200
    print("✓ 更新用户信息成功")

运行结果：

$ pytest test_user_api.py -v -s

=== [SESSION] 开始登录 ===
=== [SESSION] 登录成功，获取 token: eyJhbGciOiJIUzI1NiIs... ===

test_user_api.py::test_get_user_info 
[test_get_user_info] 使用 token: eyJhbGciOiJIUzI1NiIs...
✓ 获取用户信息成功
PASSED

test_user_api.py::test_update_user_info 
[test_update_user_info] 使用 token: eyJhbGciOiJIUzI1NiIs...
✓ 更新用户信息成功
PASSED

注意：现在登录只执行了一次，所有测试用例共享同一个 token，大大提高了测试效率。

4.4 提取多个关联数据

有时候，一个接口可能返回多个需要关联的数据：

import requests
import pytest

BASE_URL = "http://api.example.com"

@pytest.fixture(scope="session")
def login_info():
    """登录并返回所有相关信息"""
    print("n=== [SESSION] 开始登录 ===")
    url = f"{BASE_URL}/login"
    data = {
        "username": "admin",
        "password": "123456"
    }
    response = requests.post(url, json=data)
    result = response.json()

    # 提取多个关联数据
    login_data = {
        "token": result["data"]["token"],
        "user_id": result["data"]["user_id"],
        "username": result["data"]["username"],
        "expires_in": result["data"]["expires_in"]
    }

    print(f"=== [SESSION] 登录成功 ===")
    print(f"  Token: {login_data['token'][:20]}...")
    print(f"  User ID: {login_data['user_id']}")
    print(f"  Username: {login_data['username']}")

    return login_data

def test_get_user_info(login_info):
    """测试获取用户信息"""
    # 使用 token
    headers = {
        "Authorization": f"Bearer {login_info['token']}"
    }
    url = f"{BASE_URL}/user/info"
    response = requests.get(url, headers=headers)

    assert response.status_code == 200
    result = response.json()
    assert result["code"] == 200
    assert result["data"]["user_id"] == login_info["user_id"]

def test_get_user_orders(login_info):
    """测试获取用户订单"""
    # 使用 token 和 user_id
    headers = {
        "Authorization": f"Bearer {login_info['token']}"
    }
    url = f"{BASE_URL}/user/{login_info['user_id']}/orders"
    response = requests.get(url, headers=headers)

    assert response.status_code == 200
    result = response.json()
    assert result["code"] == 200

4.5 链式接口关联

有时候，接口关联是链式的，一个接口依赖另一个接口的结果：

import requests
import pytest

BASE_URL = "http://api.example.com"

@pytest.fixture(scope="session")
def login_token():
    """登录获取 token"""
    url = f"{BASE_URL}/login"
    data = {
        "username": "admin",
        "password": "123456"
    }
    response = requests.post(url, json=data)
    result = response.json()
    return result["data"]["token"]

@pytest.fixture(scope="session")
def cart_id(login_token):
    """添加商品到购物车，返回购物车ID（依赖 login_token）"""
    headers = {
        "Authorization": f"Bearer {login_token}"
    }
    url = f"{BASE_URL}/cart/add"
    data = {
        "product_id": 100,
        "quantity": 2
    }
    response = requests.post(url, json=data, headers=headers)
    result = response.json()
    return result["data"]["cart_id"]

@pytest.fixture(scope="session")
def order_id(login_token, cart_id):
    """创建订单，返回订单ID（依赖 login_token 和 cart_id）"""
    headers = {
        "Authorization": f"Bearer {login_token}"
    }
    url = f"{BASE_URL}/order/create"
    data = {
        "cart_id": cart_id
    }
    response = requests.post(url, json=data, headers=headers)
    result = response.json()
    return result["data"]["order_id"]

def test_pay_order(login_token, order_id):
    """测试支付订单（依赖 login_token 和 order_id）"""
    headers = {
        "Authorization": f"Bearer {login_token}"
    }
    url = f"{BASE_URL}/order/pay"
    data = {
        "order_id": order_id
    }
    response = requests.post(url, json=data, headers=headers)

    assert response.status_code == 200
    result = response.json()
    assert result["code"] == 200
    assert result["data"]["status"] == "paid"

说明：

  • cart_id fixture 依赖 login_token fixture
  • order_id fixture 依赖 login_token 和 cart_id fixture
  • pytest 会自动处理这些依赖关系，按正确的顺序执行

5. 使用 conftest.py 共享关联数据

5.1 什么是 conftest.py

conftest.py 是 pytest 的一个特殊文件，用于存放共享的 fixture。放在 conftest.py 中的 fixture 可以被同一目录及子目录下的所有测试文件使用。

5.2 为什么使用 conftest.py

场景：项目中有多个测试文件，都需要使用登录 token

不使用 conftest.py 的情况：

# test_user_api.py
import requests
import pytest

BASE_URL = "http://api.example.com"

@pytest.fixture(scope="session")
def login_token():
    """登录获取 token"""
    url = f"{BASE_URL}/login"
    data = {"username": "admin", "password": "123456"}
    response = requests.post(url, json=data)
    return response.json()["data"]["token"]

def test_get_user_info(login_token):
    # 测试代码...
    pass

# test_order_api.py
import requests
import pytest

BASE_URL = "http://api.example.com"

@pytest.fixture(scope="session")
def login_token():  # 重复定义！
    """登录获取 token"""
    url = f"{BASE_URL}/login"
    data = {"username": "admin", "password": "123456"}
    response = requests.post(url, json=data)
    return response.json()["data"]["token"]

def test_create_order(login_token):
    # 测试代码...
    pass

问题：

  • 代码重复：每个测试文件都要定义 login_token fixture
  • 维护困难：如果登录逻辑改变，要修改多个文件
  • 容易出错：可能在不同文件中定义不一致

使用 conftest.py 后：

# conftest.py（项目根目录）
import requests
import pytest

BASE_URL = "http://api.example.com"

@pytest.fixture(scope="session")
def login_token():
    """登录获取 token（共享 fixture）"""
    print("n=== [SESSION] 开始登录 ===")
    url = f"{BASE_URL}/login"
    data = {
        "username": "admin",
        "password": "123456"
    }
    response = requests.post(url, json=data)
    result = response.json()
    token = result["data"]["token"]
    print(f"=== [SESSION] 登录成功，token: {token[:20]}... ===")
    return token

# test_user_api.py
import requests
import pytest

BASE_URL = "http://api.example.com"

def test_get_user_info(login_token):  # 直接使用 conftest.py 中的 fixture
    """测试获取用户信息"""
    headers = {
        "Authorization": f"Bearer {login_token}"
    }
    url = f"{BASE_URL}/user/info"
    response = requests.get(url, headers=headers)

    assert response.status_code == 200
    result = response.json()
    assert result["code"] == 200

# test_order_api.py
import requests
import pytest

BASE_URL = "http://api.example.com"

def test_create_order(login_token):  # 直接使用 conftest.py 中的 fixture
    """测试创建订单"""
    headers = {
        "Authorization": f"Bearer {login_token}"
    }
    url = f"{BASE_URL}/order/create"
    data = {
        "product_id": 100,
        "quantity": 2
    }
    response = requests.post(url, json=data, headers=headers)

    assert response.status_code == 200
    result = response.json()
    assert result["code"] == 200

优势：

  • 代码复用：登录逻辑只定义一次
  • 易于维护：修改登录逻辑只需改 conftest.py
  • 自动共享：所有测试文件自动可以使用

5.3 conftest.py 的目录结构

conftest.py 的作用域遵循 pytest 的发现规则：

project/
├── conftest.py              # 所有测试文件都可以使用
├── test_user_api.py
├── test_order_api.py
├── api/
│   ├── conftest.py          # api/ 目录下的测试文件可以使用
│   ├── test_login.py
│   └── test_logout.py
└── order/
    ├── conftest.py          # order/ 目录下的测试文件可以使用
    └── test_order.py

规则：

  • 子目录的 conftest.py 可以覆盖父目录的同名 fixture
  • 子目录的测试文件可以使用父目录的 fixture
  • 父目录的测试文件不能使用子目录的 fixture

5.4 完整的 conftest.py 示例

让我们创建一个完整的 conftest.py，包含常用的接口关联 fixture：

# conftest.py
import requests
import pytest
from typing import Dict, Any

# 基础配置
BASE_URL = "http://api.example.com"

# ==================== 登录相关 Fixture ====================

@pytest.fixture(scope="session")
def login_token():
    """登录并返回 token（整个测试会话只执行一次）"""
    print("n" + "="*50)
    print("  [SESSION] 开始登录")
    print("="*50)

    url = f"{BASE_URL}/login"
    data = {
        "username": "admin",
        "password": "123456"
    }

    try:
        response = requests.post(url, json=data, timeout=10)
        response.raise_for_status()  # 如果状态码不是 200，抛出异常

        result = response.json()
        if result["code"] != 200:
            raise Exception(f"登录失败: {result.get('message', '未知错误')}")

        token = result["data"]["token"]
        print(f"  [SESSION] 登录成功")
        print(f"  Token: {token[:30]}...")
        print("="*50 + "n")

        return token
    except Exception as e:
        print(f"  [SESSION] 登录失败: {str(e)}")
        print("="*50 + "n")
        raise

@pytest.fixture(scope="session")
def login_info():
    """登录并返回所有相关信息"""
    print("n" + "="*50)
    print("  [SESSION] 开始登录（获取完整信息）")
    print("="*50)

    url = f"{BASE_URL}/login"
    data = {
        "username": "admin",
        "password": "123456"
    }

    response = requests.post(url, json=data, timeout=10)
    response.raise_for_status()

    result = response.json()
    if result["code"] != 200:
        raise Exception(f"登录失败: {result.get('message', '未知错误')}")

    login_data = {
        "token": result["data"]["token"],
        "user_id": result["data"]["user_id"],
        "username": result["data"]["username"],
        "email": result["data"].get("email", ""),
        "expires_in": result["data"].get("expires_in", 3600)
    }

    print(f"  [SESSION] 登录成功")
    print(f"  User ID: {login_data['user_id']}")
    print(f"  Username: {login_data['username']}")
    print(f"  Token: {login_data['token'][:30]}...")
    print("="*50 + "n")

    return login_data

# ==================== 请求头 Fixture ====================

@pytest.fixture(scope="session")
def auth_headers(login_token):
    """返回带认证信息的请求头"""
    return {
        "Authorization": f"Bearer {login_token}",
        "Content-Type": "application/json"
    }

# ==================== 订单相关 Fixture ====================

@pytest.fixture(scope="function")
def order_id(login_token):
    """创建订单并返回订单ID（每个测试函数执行一次）"""
    headers = {
        "Authorization": f"Bearer {login_token}",
        "Content-Type": "application/json"
    }
    url = f"{BASE_URL}/order/create"
    data = {
        "product_id": 100,
        "quantity": 1
    }

    response = requests.post(url, json=data, headers=headers, timeout=10)
    response.raise_for_status()

    result = response.json()
    if result["code"] != 200:
        raise Exception(f"创建订单失败: {result.get('message', '未知错误')}")

    order_id = result["data"]["order_id"]
    print(f"  [FUNCTION] 创建订单成功，订单ID: {order_id}")

    return order_id

# ==================== 用户相关 Fixture ====================

@pytest.fixture(scope="function")
def test_user(login_token):
    """创建测试用户并返回用户信息（每个测试函数执行一次）"""
    headers = {
        "Authorization": f"Bearer {login_token}",
        "Content-Type": "application/json"
    }
    url = f"{BASE_URL}/user/create"
    data = {
        "name": "测试用户",
        "email": "test@example.com"
    }

    response = requests.post(url, json=data, headers=headers, timeout=10)
    response.raise_for_status()

    result = response.json()
    if result["code"] != 200:
        raise Exception(f"创建用户失败: {result.get('message', '未知错误')}")

    user_info = result["data"]
    print(f"  [FUNCTION] 创建测试用户成功，用户ID: {user_info['user_id']}")

    return user_info

# ==================== 清理 Fixture ====================

@pytest.fixture(scope="function", autouse=True)
def cleanup_test_data():
    """自动清理测试数据（每个测试函数执行后自动执行）"""
    yield  # 测试执行前

    # 测试执行后的清理逻辑
    # 例如：删除测试创建的订单、用户等
    print("  [CLEANUP] 清理测试数据...")

使用示例：

# test_example.py
import requests
import pytest

BASE_URL = "http://api.example.com"

def test_get_user_info(auth_headers):
    """测试获取用户信息（使用 auth_headers fixture）"""
    url = f"{BASE_URL}/user/info"
    response = requests.get(url, headers=auth_headers)

    assert response.status_code == 200
    result = response.json()
    assert result["code"] == 200

def test_create_and_query_order(order_id, auth_headers):
    """测试创建订单并查询（使用 order_id 和 auth_headers fixture）"""
    # order_id 已经通过 fixture 创建好了
    url = f"{BASE_URL}/order/{order_id}"
    response = requests.get(url, headers=auth_headers)

    assert response.status_code == 200
    result = response.json()
    assert result["code"] == 200
    assert result["data"]["order_id"] == order_id

6. 接口关联的高级技巧

6.1 动态提取关联数据

有时候，我们需要从接口响应中动态提取数据，而不是固定路径：

import requests
import pytest
import jsonpath

BASE_URL = "http://api.example.com"

@pytest.fixture(scope="session")
def login_token():
    """登录并返回 token"""
    url = f"{BASE_URL}/login"
    data = {
        "username": "admin",
        "password": "123456"
    }
    response = requests.post(url, json=data)
    result = response.json()

    # 方式1：使用 jsonpath 提取（需要安装 jsonpath 库）
    # token = jsonpath.jsonpath(result, "$.data.token")[0]

    # 方式2：使用字典的 get 方法（更安全）
    token = result.get("data", {}).get("token")

    if not token:
        raise ValueError("无法从响应中提取 token")

    return token

@pytest.fixture(scope="function")
def extract_order_id():
    """从创建订单的响应中提取订单ID"""
    def _extract(response_data: dict) -> str:
        """内部函数，用于提取订单ID"""
        # 尝试多种可能的路径
        order_id = (
            response_data.get("data", {}).get("order_id") or
            response_data.get("order_id") or
            response_data.get("data", {}).get("id")
        )

        if not order_id:
            raise ValueError("无法从响应中提取订单ID")

        return order_id

    return _extract

def test_create_order(login_token, extract_order_id):
    """测试创建订单并提取订单ID"""
    headers = {
        "Authorization": f"Bearer {login_token}"
    }
    url = f"{BASE_URL}/order/create"
    data = {
        "product_id": 100,
        "quantity": 2
    }

    response = requests.post(url, json=data, headers=headers)
    result = response.json()

    # 使用 extract_order_id fixture 提取订单ID
    order_id = extract_order_id(result)

    # 使用订单ID查询订单
    query_url = f"{BASE_URL}/order/{order_id}"
    query_response = requests.get(query_url, headers=headers)

    assert query_response.status_code == 200

6.2 Token 自动刷新

如果 token 会过期，我们需要实现自动刷新机制：

import requests
import pytest
import time
from typing import Optional

BASE_URL = "http://api.example.com"

class TokenManager:
    """Token 管理器，负责 token 的获取和刷新"""

    def __init__(self):
        self.token: Optional[str] = None
        self.expires_at: Optional[float] = None
        self.refresh_token: Optional[str] = None

    def login(self) -> str:
        """登录获取 token"""
        url = f"{BASE_URL}/login"
        data = {
            "username": "admin",
            "password": "123456"
        }
        response = requests.post(url, json=data)
        result = response.json()

        self.token = result["data"]["token"]
        expires_in = result["data"].get("expires_in", 3600)
        self.expires_at = time.time() + expires_in
        self.refresh_token = result["data"].get("refresh_token")

        return self.token

    def refresh(self) -> str:
        """刷新 token"""
        if not self.refresh_token:
            # 如果没有 refresh_token，重新登录
            return self.login()

        url = f"{BASE_URL}/token/refresh"
        data = {
            "refresh_token": self.refresh_token
        }
        response = requests.post(url, json=data)
        result = response.json()

        self.token = result["data"]["token"]
        expires_in = result["data"].get("expires_in", 3600)
        self.expires_at = time.time() + expires_in

        return self.token

    def get_token(self) -> str:
        """获取有效的 token（如果过期则自动刷新）"""
        # 如果 token 不存在或即将过期（提前5分钟刷新）
        if not self.token or (self.expires_at and time.time() > self.expires_at - 300):
            return self.refresh()

        return self.token

# 创建全局的 TokenManager 实例
_token_manager = TokenManager()

@pytest.fixture(scope="session")
def login_token():
    """登录并返回 token（支持自动刷新）"""
    return _token_manager.get_token()

@pytest.fixture(scope="function")
def auto_refresh_token():
    """每次使用前自动刷新 token"""
    return _token_manager.get_token()

6.3 参数化接口关联

有时候，我们需要为不同的用户创建不同的关联数据：

import requests
import pytest

BASE_URL = "http://api.example.com"

# 测试用户数据
TEST_USERS = [
    {"username": "admin", "password": "123456", "role": "admin"},
    {"username": "user1", "password": "123456", "role": "user"},
    {"username": "user2", "password": "123456", "role": "user"},
]

@pytest.fixture(scope="session", params=TEST_USERS)
def user_token(request):
    """为每个测试用户登录并返回 token"""
    user_info = request.param
    print(f"n=== 登录用户: {user_info['username']} ===")

    url = f"{BASE_URL}/login"
    response = requests.post(url, json={
        "username": user_info["username"],
        "password": user_info["password"]
    })
    result = response.json()

    token = result["data"]["token"]
    print(f"=== 用户 {user_info['username']} 登录成功 ===")

    return {
        "token": token,
        "username": user_info["username"],
        "role": user_info["role"]
    }

def test_get_user_info(user_token):
    """测试获取用户信息（会为每个用户执行一次）"""
    headers = {
        "Authorization": f"Bearer {user_token['token']}"
    }
    url = f"{BASE_URL}/user/info"
    response = requests.get(url, headers=headers)

    assert response.status_code == 200
    result = response.json()
    assert result["code"] == 200
    print(f"✓ 用户 {user_token['username']} 获取信息成功")

运行结果：

$ pytest test_user_api.py -v -s

test_user_api.py::test_get_user_info[user_token0] 
=== 登录用户: admin ===
=== 用户 admin 登录成功 ===
✓ 用户 admin 获取信息成功
PASSED

test_user_api.py::test_get_user_info[user_token1] 
=== 登录用户: user1 ===
=== 用户 user1 登录成功 ===
✓ 用户 user1 获取信息成功
PASSED

test_user_api.py::test_get_user_info[user_token2] 
=== 登录用户: user2 ===
=== 用户 user2 登录成功 ===
✓ 用户 user2 获取信息成功
PASSED

6.4 条件关联

有时候，某些关联数据只有在特定条件下才需要：

import requests
import pytest

BASE_URL = "http://api.example.com"

@pytest.fixture(scope="session")
def login_token():
    """登录获取 token"""
    url = f"{BASE_URL}/login"
    data = {"username": "admin", "password": "123456"}
    response = requests.post(url, json=data)
    return response.json()["data"]["token"]

@pytest.fixture(scope="function")
def order_id(login_token, request):
    """创建订单（可选，通过 pytest.mark 控制）"""
    # 检查测试用例是否标记了 skip_order
    if request.node.get_closest_marker("skip_order"):
        pytest.skip("跳过订单创建")

    headers = {"Authorization": f"Bearer {login_token}"}
    url = f"{BASE_URL}/order/create"
    data = {"product_id": 100, "quantity": 1}

    response = requests.post(url, json=data, headers=headers)
    result = response.json()
    return result["data"]["order_id"]

@pytest.mark.skip_order
def test_without_order(login_token):
    """不需要订单的测试"""
    headers = {"Authorization": f"Bearer {login_token}"}
    url = f"{BASE_URL}/user/info"
    response = requests.get(url, headers=headers)
    assert response.status_code == 200

def test_with_order(login_token, order_id):
    """需要订单的测试"""
    headers = {"Authorization": f"Bearer {login_token}"}
    url = f"{BASE_URL}/order/{order_id}"
    response = requests.get(url, headers=headers)
    assert response.status_code == 200

7. 实际案例演示

7.1 案例一：电商系统完整流程

让我们实现一个完整的电商购物流程测试：

# conftest.py
import requests
import pytest
import time

BASE_URL = "http://api.example.com"

@pytest.fixture(scope="session")
def login_token():
    """登录获取 token"""
    url = f"{BASE_URL}/login"
    data = {
        "username": "test_user",
        "password": "123456"
    }
    response = requests.post(url, json=data, timeout=10)
    response.raise_for_status()
    result = response.json()
    return result["data"]["token"]

@pytest.fixture(scope="session")
def auth_headers(login_token):
    """认证请求头"""
    return {
        "Authorization": f"Bearer {login_token}",
        "Content-Type": "application/json"
    }

@pytest.fixture(scope="function")
def cart_id(auth_headers):
    """添加商品到购物车"""
    url = f"{BASE_URL}/cart/add"
    data = {
        "product_id": 100,
        "quantity": 2
    }
    response = requests.post(url, json=data, headers=auth_headers, timeout=10)
    response.raise_for_status()
    result = response.json()
    cart_id = result["data"]["cart_id"]

    yield cart_id

    # 清理：删除购物车
    try:
        delete_url = f"{BASE_URL}/cart/{cart_id}"
        requests.delete(delete_url, headers=auth_headers, timeout=10)
    except:
        pass

@pytest.fixture(scope="function")
def order_id(auth_headers, cart_id):
    """创建订单"""
    url = f"{BASE_URL}/order/create"
    data = {
        "cart_id": cart_id
    }
    response = requests.post(url, json=data, headers=auth_headers, timeout=10)
    response.raise_for_status()
    result = response.json()
    order_id = result["data"]["order_id"]

    yield order_id

    # 清理：取消订单（如果未支付）
    try:
        cancel_url = f"{BASE_URL}/order/{order_id}/cancel"
        requests.post(cancel_url, headers=auth_headers, timeout=10)
    except:
        pass

# test_shopping_flow.py
import requests
import pytest

BASE_URL = "http://api.example.com"

def test_complete_shopping_flow(auth_headers, cart_id, order_id):
    """测试完整的购物流程"""

    # 步骤1：验证购物车
    cart_url = f"{BASE_URL}/cart/{cart_id}"
    cart_response = requests.get(cart_url, headers=auth_headers)
    assert cart_response.status_code == 200
    cart_data = cart_response.json()
    assert cart_data["code"] == 200
    assert len(cart_data["data"]["items"]) > 0
    print(f"✓ 购物车验证成功，商品数量: {len(cart_data['data']['items'])}")

    # 步骤2：验证订单创建
    order_url = f"{BASE_URL}/order/{order_id}"
    order_response = requests.get(order_url, headers=auth_headers)
    assert order_response.status_code == 200
    order_data = order_response.json()
    assert order_data["code"] == 200
    assert order_data["data"]["order_id"] == order_id
    print(f"✓ 订单创建成功，订单ID: {order_id}")

    # 步骤3：支付订单
    pay_url = f"{BASE_URL}/order/{order_id}/pay"
    pay_data = {
        "payment_method": "alipay",
        "amount": order_data["data"]["total_amount"]
    }
    pay_response = requests.post(pay_url, json=pay_data, headers=auth_headers)
    assert pay_response.status_code == 200
    pay_result = pay_response.json()
    assert pay_result["code"] == 200
    assert pay_result["data"]["status"] == "paid"
    print(f"✓ 订单支付成功")

    # 步骤4：验证订单状态
    status_response = requests.get(order_url, headers=auth_headers)
    status_data = status_response.json()
    assert status_data["data"]["status"] == "paid"
    print(f"✓ 订单状态验证成功，状态: {status_data['data']['status']}")

7.2 案例二：用户管理系统

# conftest.py
import requests
import pytest

BASE_URL = "http://api.example.com"

@pytest.fixture(scope="session")
def admin_token():
    """管理员登录"""
    url = f"{BASE_URL}/login"
    data = {"username": "admin", "password": "admin123"}
    response = requests.post(url, json=data)
    return response.json()["data"]["token"]

@pytest.fixture(scope="function")
def test_user(admin_token):
    """创建测试用户"""
    headers = {
        "Authorization": f"Bearer {admin_token}",
        "Content-Type": "application/json"
    }
    url = f"{BASE_URL}/user/create"
    data = {
        "name": "测试用户",
        "email": f"test_{int(time.time())}@example.com",
        "password": "123456"
    }
    response = requests.post(url, json=data, headers=headers)
    result = response.json()
    user_id = result["data"]["user_id"]

    yield {
        "user_id": user_id,
        "name": data["name"],
        "email": data["email"]
    }

    # 清理：删除测试用户
    try:
        delete_url = f"{BASE_URL}/user/{user_id}"
        requests.delete(delete_url, headers=headers)
    except:
        pass

# test_user_management.py
import requests
import pytest

BASE_URL = "http://api.example.com"

def test_user_crud_operations(admin_token, test_user):
    """测试用户的增删改查操作"""
    headers = {
        "Authorization": f"Bearer {admin_token}",
        "Content-Type": "application/json"
    }
    user_id = test_user["user_id"]

    # 1. 查询用户
    get_url = f"{BASE_URL}/user/{user_id}"
    get_response = requests.get(get_url, headers=headers)
    assert get_response.status_code == 200
    user_data = get_response.json()
    assert user_data["data"]["user_id"] == user_id
    assert user_data["data"]["name"] == test_user["name"]
    print(f"✓ 查询用户成功: {user_data['data']['name']}")

    # 2. 更新用户
    update_url = f"{BASE_URL}/user/{user_id}"
    update_data = {
        "name": "更新后的名称",
        "phone": "13800138000"
    }
    update_response = requests.put(update_url, json=update_data, headers=headers)
    assert update_response.status_code == 200
    update_result = update_response.json()
    assert update_result["code"] == 200
    print(f"✓ 更新用户成功")

    # 3. 验证更新
    verify_response = requests.get(get_url, headers=headers)
    verify_data = verify_response.json()
    assert verify_data["data"]["name"] == "更新后的名称"
    assert verify_data["data"]["phone"] == "13800138000"
    print(f"✓ 验证更新成功")

    # 4. 删除用户（由 fixture 的清理逻辑处理）
    # 这里可以手动删除验证
    delete_url = f"{BASE_URL}/user/{user_id}"
    delete_response = requests.delete(delete_url, headers=headers)
    assert delete_response.status_code == 200
    print(f"✓ 删除用户成功")

7.3 案例三：多环境接口关联

# conftest.py
import requests
import pytest
import os

# 从环境变量获取基础URL，支持多环境
BASE_URL = os.getenv("API_BASE_URL", "http://api.example.com")

@pytest.fixture(scope="session")
def login_token():
    """登录获取 token（支持多环境）"""
    url = f"{BASE_URL}/login"

    # 根据环境选择不同的测试账号
    env = os.getenv("TEST_ENV", "dev")
    if env == "prod":
        username = "prod_test_user"
        password = "prod_password"
    elif env == "staging":
        username = "staging_test_user"
        password = "staging_password"
    else:
        username = "dev_test_user"
        password = "dev_password"

    data = {
        "username": username,
        "password": password
    }

    print(f"n=== 环境: {env} ===")
    print(f"=== 登录URL: {url} ===")
    print(f"=== 用户名: {username} ===")

    response = requests.post(url, json=data, timeout=10)
    response.raise_for_status()
    result = response.json()

    if result["code"] != 200:
        raise Exception(f"登录失败: {result.get('message')}")

    token = result["data"]["token"]
    print(f"=== 登录成功 ===")

    return token

# 使用方式
# 开发环境: pytest test_api.py
# 测试环境: TEST_ENV=staging pytest test_api.py
# 生产环境: TEST_ENV=prod API_BASE_URL=https://api.prod.com pytest test_api.py

8. 接口关联的最佳实践

8.1 Fixture 作用域的选择

原则：根据数据的生命周期选择合适的作用域

数据类型 推荐作用域 原因
Token（长期有效） session 整个测试会话只需要登录一次
Token（短期有效） function 每次测试都需要新的 token
订单ID function 每个测试应该使用独立的订单
用户ID function 避免测试之间的数据污染
配置信息 session 配置信息不会改变

示例：

# ✅ 正确：Token 使用 session 作用域
@pytest.fixture(scope="session")
def login_token():
    # Token 在测试期间不会过期
    return get_token()

# ✅ 正确：订单ID 使用 function 作用域
@pytest.fixture(scope="function")
def order_id(login_token):
    # 每个测试使用独立的订单，避免相互影响
    return create_order(login_token)

# ❌ 错误：订单ID 使用 session 作用域
@pytest.fixture(scope="session")
def order_id(login_token):
    # 所有测试共享同一个订单，可能导致测试失败
    return create_order(login_token)

8.2 错误处理

原则：接口关联中的错误应该被明确处理，避免隐藏问题

# ✅ 正确：明确的错误处理
@pytest.fixture(scope="session")
def login_token():
    try:
        response = requests.post(url, json=data, timeout=10)
        response.raise_for_status()
        result = response.json()

        if result["code"] != 200:
            raise Exception(f"登录失败: {result.get('message')}")

        return result["data"]["token"]
    except requests.exceptions.Timeout:
        pytest.fail("登录请求超时")
    except requests.exceptions.RequestException as e:
        pytest.fail(f"登录请求失败: {str(e)}")
    except KeyError as e:
        pytest.fail(f"响应数据格式错误: {str(e)}")

# ❌ 错误：没有错误处理
@pytest.fixture(scope="session")
def login_token():
    response = requests.post(url, json=data)
    return response.json()["data"]["token"]  # 可能抛出异常

8.3 数据清理

原则：测试创建的数据应该在测试结束后清理

# ✅ 正确：使用 yield 进行清理
@pytest.fixture(scope="function")
def test_user(admin_token):
    """创建测试用户"""
    user_id = create_user(admin_token, {...})

    yield {"user_id": user_id, ...}

    # 清理：删除测试用户
    try:
        delete_user(admin_token, user_id)
    except Exception as e:
        print(f"清理失败: {str(e)}")

# ✅ 正确：使用 finalizer 进行清理
@pytest.fixture(scope="function")
def test_user(admin_token, request):
    """创建测试用户"""
    user_id = create_user(admin_token, {...})

    def cleanup():
        try:
            delete_user(admin_token, user_id)
        except Exception as e:
            print(f"清理失败: {str(e)}")

    request.addfinalizer(cleanup)
    return {"user_id": user_id, ...}

8.4 代码组织

原则：将接口关联相关的代码组织在 conftest.py 中

project/
├── conftest.py              # 共享的 fixture
├── tests/
│   ├── conftest.py          # 测试目录特定的 fixture
│   ├── test_user_api.py
│   ├── test_order_api.py
│   └── api/
│       ├── conftest.py      # API 测试特定的 fixture
│       └── test_login.py
└── utils/
    └── api_client.py         # API 客户端封装

8.5 命名规范

原则：使用清晰的命名，让 fixture 的用途一目了然

# ✅ 正确：清晰的命名
@pytest.fixture(scope="session")
def admin_login_token():
    """管理员登录 token"""
    pass

@pytest.fixture(scope="function")
def test_order_id():
    """测试订单ID"""
    pass

# ❌ 错误：模糊的命名
@pytest.fixture(scope="session")
def token():  # 不知道是什么 token
    pass

@pytest.fixture(scope="function")
def data():  # 不知道是什么数据
    pass

8.6 文档注释

原则：为每个 fixture 添加清晰的文档注释

# ✅ 正确：有详细的文档注释
@pytest.fixture(scope="session")
def login_token():
    """
    登录并返回 token

    作用域: session（整个测试会话只执行一次）
    依赖: 无
    返回: str - 登录 token

    示例:
        def test_api(login_token):
            headers = {"Authorization": f"Bearer {login_token}"}
            ...
    """
    pass

# ❌ 错误：没有文档注释
@pytest.fixture(scope="session")
def login_token():
    pass

9. 常见问题与解决方案

9.1 问题一：Token 过期导致测试失败

问题描述：

  • 使用 session 作用域的 token，但 token 在测试过程中过期
  • 导致后续测试失败

解决方案：

# 方案1：使用 function 作用域（如果 token 容易过期）
@pytest.fixture(scope="function")
def login_token():
    """每次测试都重新登录"""
    return get_fresh_token()

# 方案2：实现 token 自动刷新（推荐）
class TokenManager:
    def __init__(self):
        self.token = None
        self.expires_at = None

    def get_token(self):
        if not self.token or time.time() > self.expires_at - 300:
            self.refresh_token()
        return self.token

_token_manager = TokenManager()

@pytest.fixture(scope="session")
def login_token():
    return _token_manager.get_token()

9.2 问题二：测试之间的数据污染

问题描述：

  • 测试A创建了订单，测试B使用了同一个订单
  • 导致测试结果不可预期

解决方案：

# ✅ 正确：每个测试使用独立的数据
@pytest.fixture(scope="function")  # 使用 function 作用域
def order_id(login_token):
    """每个测试创建独立的订单"""
    return create_order(login_token)

# ❌ 错误：所有测试共享数据
@pytest.fixture(scope="session")  # 使用 session 作用域
def order_id(login_token):
    """所有测试共享同一个订单"""
    return create_order(login_token)

9.3 问题三：Fixture 依赖顺序问题

问题描述：

  • Fixture A 依赖 Fixture B，但执行顺序不正确
  • 导致测试失败

解决方案：

# ✅ 正确：明确声明依赖关系
@pytest.fixture(scope="session")
def login_token():
    return get_token()

@pytest.fixture(scope="function")
def order_id(login_token):  # 明确声明依赖 login_token
    return create_order(login_token)

# pytest 会自动处理依赖顺序
def test_order(order_id):  # 会自动先执行 login_token，再执行 order_id
    pass

# ❌ 错误：没有声明依赖
@pytest.fixture(scope="session")
def login_token():
    return get_token()

@pytest.fixture(scope="function")
def order_id():  # 没有声明依赖，但实际需要 login_token
    return create_order(login_token)  # 这里会报错，因为 login_token 未定义

9.4 问题四：接口响应格式变化

问题描述：

  • 接口响应格式改变，导致数据提取失败
  • 测试用例大量失败

解决方案：

# ✅ 正确：使用安全的提取方式
def extract_token(response_data: dict) -> str:
    """安全地提取 token"""
    # 尝试多种可能的路径
    token = (
        response_data.get("data", {}).get("token") or
        response_data.get("token") or
        response_data.get("result", {}).get("token")
    )

    if not token:
        raise ValueError(
            f"无法从响应中提取 token。响应数据: {response_data}"
        )

    return token

# ✅ 正确：添加响应验证
@pytest.fixture(scope="session")
def login_token():
    response = requests.post(url, json=data)
    response.raise_for_status()
    result = response.json()

    # 验证响应格式
    assert "data" in result, f"响应格式错误: {result}"
    assert "token" in result["data"], f"响应格式错误: {result}"

    return result["data"]["token"]

9.5 问题五：并发测试时的数据冲突

问题描述：

  • 使用 pytest-xdist 并行执行测试时
  • 多个进程同时使用同一个 token 或数据，导致冲突

解决方案：

# ✅ 正确：为每个进程创建独立的数据
@pytest.fixture(scope="session")
def login_token():
    """每个进程独立登录"""
    # session 作用域在并行测试中，每个进程是独立的
    return get_token()

@pytest.fixture(scope="function")
def unique_user_id(login_token):
    """使用时间戳确保唯一性"""
    import time
    user_id = f"test_user_{int(time.time() * 1000)}"
    create_user(login_token, user_id)
    return user_id

# ✅ 正确：使用进程ID确保唯一性
import os
@pytest.fixture(scope="function")
def unique_order_id(login_token):
    """使用进程ID和时间戳确保唯一性"""
    import time
    process_id = os.getpid()
    timestamp = int(time.time() * 1000)
    order_id = f"ORD_{process_id}_{timestamp}"
    return order_id

9.6 问题六：接口关联失败时的调试

问题描述：

  • 接口关联失败，但不知道是哪个环节出了问题
  • 难以定位问题

解决方案：

# ✅ 正确：添加详细的日志
import logging

logger = logging.getLogger(__name__)

@pytest.fixture(scope="session")
def login_token():
    """登录获取 token（带详细日志）"""
    logger.info("开始登录...")
    logger.debug(f"登录URL: {url}")
    logger.debug(f"登录数据: {data}")

    try:
        response = requests.post(url, json=data, timeout=10)
        logger.debug(f"响应状态码: {response.status_code}")
        logger.debug(f"响应内容: {response.text}")

        response.raise_for_status()
        result = response.json()

        if result["code"] != 200:
            logger.error(f"登录失败: {result.get('message')}")
            raise Exception(f"登录失败: {result.get('message')}")

        token = result["data"]["token"]
        logger.info(f"登录成功，token: {token[:20]}...")

        return token
    except Exception as e:
        logger.exception(f"登录异常: {str(e)}")
        raise

# ✅ 正确：使用 pytest 的打印功能
@pytest.fixture(scope="session")
def login_token():
    """登录获取 token（带打印信息）"""
    print("n" + "="*50)
    print("  [FIXTURE] 开始登录")
    print("="*50)

    response = requests.post(url, json=data)
    print(f"  响应状态码: {response.status_code}")
    print(f"  响应内容: {response.text[:200]}...")

    result = response.json()
    token = result["data"]["token"]

    print(f"  登录成功")
    print(f"  Token: {token[:30]}...")
    print("="*50 + "n")

    return token

发表评论