Python + Pytest 接口自动化框架搭建实战

作者：上上签
系列文章：Python 自动化测试系列（第3篇）

一、前言

为什么需要接口自动化测试

在微服务架构盛行的今天，接口测试已经成为质量保障的核心环节。相比 UI 自动化，接口自动化测试具有独特优势：

1. 运行效率高
接口测试无需启动浏览器、渲染页面，执行速度是 UI 测试的 10 倍以上。在 CI/CD 流水线中，快速的反馈循环至关重要。

2. 稳定性强
接口测试不依赖页面元素定位，不会因为 UI 改版而大面积失效。接口定义稳定后，测试用例维护成本极低。

3. 覆盖更全面
接口测试可以覆盖异常场景、边界条件、权限控制等 UI 难以触达的测试点，还能进行性能、安全等专项测试。

4. 开发阶段介入
接口文档确定后即可开始编写测试用例，实现测试左移，提前发现问题。

Pytest 框架简介和优势

Pytest 是 Python 生态中最流行的测试框架，相比 unittest，它具有以下核心优势：

特性Pytestunittest
代码量少（无需继承 TestCase）多（必须继承、定义方法前缀）
断言方式原生 assert 关键字self.assertEqual() 等方法
参数化@pytest.mark.parametrize需要第三方库
插件生态丰富（500+ 插件）相对较少
测试发现自动发现 test_*.py需要手动组织
失败重跑pytest-rerunfailures不支持

与 UI 自动化的对比

维度接口自动化UI 自动化
执行速度秒级分钟级
维护成本低（接口稳定）高（UI 频繁变化）
覆盖范围业务逻辑、数据验证用户体验、交互流程
适用场景回归测试、冒烟测试端到端流程验证
技术门槛中等较高

最佳实践：接口自动化覆盖 70% 的测试场景，UI 自动化覆盖核心业务流程的 30%，形成分层测试体系。

二、环境搭建

Python 环境要求

  • Python 版本：3.8+（推荐 3.10+）
  • 包管理器：pip 或 poetry
  • 虚拟环境：强烈建议使用虚拟环境隔离项目依赖

安装核心依赖

# 创建虚拟环境
python -m venv venv

# 激活虚拟环境
# Windows
venv\Scripts\activate
# Linux/Mac
source venv/bin/activate

# 安装核心依赖
pip install pytest requests allure-pytest pytest-html
pip install pyyaml python-dotenv  # 配置管理
pip install jsonpath faker  # 测试数据工具

requirements.txt

# 测试框架
pytest==7.4.3
pytest-html==4.1.1
allure-pytest==2.13.2
pytest-rerunfailures==13.0

# HTTP 请求
requests==2.31.0
urllib3==2.1.0

# 配置管理
pyyaml==6.0.1
python-dotenv==1.0.0

# 数据处理
jsonpath==0.82
faker==20.1.0

# 日志和报告
loguru==0.7.2

# 工具库
pydantic==2.5.0  # 数据校验

项目结构设计

api_test_framework/
├── config/                    # 配置文件目录
│   ├── config.yaml           # 主配置文件
│   ├── config_dev.yaml       # 开发环境配置
│   ├── config_prod.yaml      # 生产环境配置
│   └── .env                  # 环境变量（敏感信息）
├── api/                       # 接口封装层
│   ├── __init__.py
│   ├── base_api.py           # 基础 API 封装
│   ├── user_api.py           # 用户模块接口
│   └── order_api.py          # 订单模块接口
├── core/                      # 核心模块
│   ├── __init__.py
│   ├── request.py            # Requests 封装
│   ├── logger.py             # 日志模块
│   └── exceptions.py         # 自定义异常
├── testcases/                 # 测试用例
│   ├── conftest.py           # Pytest 配置和 fixtures
│   ├── test_login.py         # 登录测试
│   ├── test_user.py          # 用户模块测试
│   └── test_order.py         # 订单模块测试
├── testdata/                  # 测试数据
│   ├── login_data.yaml       # 登录测试数据
│   └── order_data.yaml       # 订单测试数据
├── utils/                     # 工具函数
│   ├── __init__.py
│   ├── data_loader.py        # 数据加载器
│   └── assertions.py         # 自定义断言
├── reports/                   # 测试报告
│   └── allure-results/       # Allure 原始数据
├── logs/                      # 日志文件
├── pytest.ini                # Pytest 配置
├── requirements.txt          # 依赖清单
└── README.md                 # 项目说明

三、项目架构设计

推荐的目录结构

采用分层架构设计，遵循单一职责原则：

┌─────────────────────────────────────────┐
│          Test Cases Layer               │  测试用例层
│  (test_login.py, test_user.py, ...)     │
└─────────────────┬───────────────────────┘
                  │
┌─────────────────▼───────────────────────┐
│          API Layer                      │  接口封装层
│  (user_api.py, order_api.py, ...)       │
└─────────────────┬───────────────────────┘
                  │
┌─────────────────▼───────────────────────┐
│          Core Layer                     │  核心层
│  (request.py, logger.py, config.py)     │
└─────────────────────────────────────────┘

核心模块划分

模块职责关键文件
core底层封装，HTTP 请求、日志、异常request.py, logger.py
api接口业务封装，组合 HTTP 方法base_api.py, user_api.py
config环境配置管理，多环境切换config.yaml, config.py
testcases测试用例，断言验证test_*.py
testdata测试数据分离，数据驱动*.yaml, *.json
utils通用工具函数data_loader.py, assertions.py

配置管理（config）

config/config.yaml - 主配置文件：

# 环境配置
env: dev

# 测试环境配置
environments:
  dev:
    base_url: "http://dev-api.example.com"
    db_host: "dev-db.example.com"
    db_port: 3306
  test:
    base_url: "http://test-api.example.com"
    db_host: "test-db.example.com"
    db_port: 3306
  prod:
    base_url: "https://api.example.com"
    db_host: "prod-db.example.com"
    db_port: 3306

# 请求配置
request:
  timeout: 10
  retry_times: 3
  retry_delay: 1

# 日志配置
logging:
  level: INFO
  format: "{time:YYYY-MM-DD HH:mm:ss} | {level} | {message}"
  file: logs/test_{time}.log

core/config.py - 配置加载器：

import os
import yaml
from pathlib import Path
from typing import Any, Dict
from dotenv import load_dotenv


class Config:
    """配置管理类，支持多环境切换"""
    
    def __init__(self, env: str = None):
        # 加载环境变量
        load_dotenv()
        
        # 确定环境
        self.env = env or os.getenv("TEST_ENV", "dev")
        
        # 加载配置文件
        self.config_dir = Path(__file__).parent.parent / "config"
        self._load_config()
    
    def _load_config(self):
        """加载配置文件"""
        # 加载主配置
        config_file = self.config_dir / "config.yaml"
        with open(config_file, "r", encoding="utf-8") as f:
            self._config = yaml.safe_load(f)
        
        # 加载环境特定配置
        env_file = self.config_dir / f"config_{self.env}.yaml"
        if env_file.exists():
            with open(env_file, "r", encoding="utf-8") as f:
                env_config = yaml.safe_load(f)
                # 合并配置
                self._merge_config(self._config, env_config)
    
    def _merge_config(self, base: Dict, override: Dict):
        """递归合并配置"""
        for key, value in override.items():
            if key in base and isinstance(base[key], dict) and isinstance(value, dict):
                self._merge_config(base[key], value)
            else:
                base[key] = value
    
    def get(self, key: str, default: Any = None) -> Any:
        """获取配置项（支持点分隔符）"""
        keys = key.split(".")
        value = self._config
        
        for k in keys:
            if isinstance(value, dict) and k in value:
                value = value[k]
            else:
                return default
        
        return value
    
    @property
    def base_url(self) -> str:
        """获取基础 URL"""
        return self.get(f"environments.{self.env}.base_url")
    
    @property
    def timeout(self) -> int:
        """获取请求超时时间"""
        return self.get("request.timeout", 10)


# 全局配置实例
config = Config()

日志模块（logger）

core/logger.py：

import sys
from pathlib import Path
from loguru import logger
from config.config import config


class LogManager:
    """日志管理器"""
    
    _initialized = False
    
    @classmethod
    def setup(cls):
        """初始化日志配置"""
        if cls._initialized:
            return
        
        # 移除默认处理器
        logger.remove()
        
        # 控制台输出
        logger.add(
            sys.stdout,
            level=config.get("logging.level", "INFO"),
            format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
                   "<level>{level: <8}</level> | "
                   "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
                   "<level>{message}</level>",
            colorize=True
        )
        
        # 文件输出
        log_dir = Path(__file__).parent.parent / "logs"
        log_dir.mkdir(exist_ok=True)
        
        logger.add(
            log_dir / "test_{time:YYYY-MM-DD}.log",
            level="DEBUG",
            format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}",
            rotation="00:00",  # 每天轮转
            retention="7 days",  # 保留7天
            compression="zip",  # 压缩旧日志
            encoding="utf-8"
        )
        
        # 错误日志单独记录
        logger.add(
            log_dir / "error_{time:YYYY-MM-DD}.log",
            level="ERROR",
            format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}\n{exception}",
            rotation="00:00",
            retention="30 days",
            encoding="utf-8"
        )
        
        cls._initialized = True
    
    @classmethod
    def get_logger(cls):
        """获取日志实例"""
        cls.setup()
        return logger


# 初始化并导出
log = LogManager.get_logger()

四、Requests 封装

封装 HTTP 请求方法

core/request.py - HTTP 客户端封装：

import time
import requests
from typing import Dict, Any, Optional, Union
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from core.logger import log
from core.exceptions import RequestException, TimeoutException
from config.config import config


class HTTPClient:
    """HTTP 客户端封装类"""
    
    def __init__(self, base_url: str = None):
        self.base_url = base_url or config.base_url
        self.session = self._create_session()
        self.headers = {
            "Content-Type": "application/json",
            "Accept": "application/json"
        }
    
    def _create_session(self) -> requests.Session:
        """创建带有重试机制的 Session"""
        session = requests.Session()
        
        # 配置重试策略
        retry_strategy = Retry(
            total=config.get("request.retry_times", 3),
            backoff_factor=config.get("request.retry_delay", 1),
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["HEAD", "GET", "OPTIONS", "POST", "PUT", "DELETE"]
        )
        
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("http://", adapter)
        session.mount("https://", adapter)
        
        return session
    
    def request(
        self,
        method: str,
        endpoint: str,
        params: Dict = None,
        data: Dict = None,
        json: Dict = None,
        headers: Dict = None,
        files: Dict = None,
        **kwargs
    ) -> "Response":
        """
        统一请求方法
        
        Args:
            method: 请求方法 GET/POST/PUT/DELETE
            endpoint: 接口路径（会拼接 base_url）
            params: URL 参数
            data: 表单数据
            json: JSON 数据
            headers: 请求头（会合并默认 headers）
            files: 上传文件
        
        Returns:
            Response 对象
        """
        url = f"{self.base_url}{endpoint}"
        
        # 合并请求头
        request_headers = {**self.headers, **(headers or {})}
        
        # 记录请求日志
        log.info(f"请求: {method} {url}")
        log.debug(f"请求头: {request_headers}")
        if json:
            log.debug(f"请求体: {json}")
        
        try:
            response = self.session.request(
                method=method.upper(),
                url=url,
                params=params,
                data=data,
                json=json,
                headers=request_headers,
                files=files,
                timeout=config.timeout,
                **kwargs
            )
            
            # 记录响应日志
            log.info(f"响应: {response.status_code}")
            log.debug(f"响应体: {response.text[:500]}")
            
            # 封装响应对象
            return Response(response)
            
        except requests.Timeout:
            log.error(f"请求超时: {method} {url}")
            raise TimeoutException(f"请求超时: {url}")
        except requests.RequestException as e:
            log.error(f"请求异常: {e}")
            raise RequestException(f"请求失败: {e}")
    
    def get(self, endpoint: str, params: Dict = None, **kwargs) -> "Response":
        """GET 请求"""
        return self.request("GET", endpoint, params=params, **kwargs)
    
    def post(
        self,
        endpoint: str,
        data: Dict = None,
        json: Dict = None,
        **kwargs
    ) -> "Response":
        """POST 请求"""
        return self.request("POST", endpoint, data=data, json=json, **kwargs)
    
    def put(
        self,
        endpoint: str,
        data: Dict = None,
        json: Dict = None,
        **kwargs
    ) -> "Response":
        """PUT 请求"""
        return self.request("PUT", endpoint, data=data, json=json, **kwargs)
    
    def delete(self, endpoint: str, **kwargs) -> "Response":
        """DELETE 请求"""
        return self.request("DELETE", endpoint, **kwargs)
    
    def set_auth_token(self, token: str, token_type: str = "Bearer"):
        """设置认证 Token"""
        self.headers["Authorization"] = f"{token_type} {token}"
    
    def close(self):
        """关闭 Session"""
        self.session.close()


class Response:
    """响应封装类"""
    
    def __init__(self, response: requests.Response):
        self._response = response
        self.status_code = response.status_code
        self.headers = response.headers
        self.url = response.url
        self.elapsed = response.elapsed.total_seconds()
        
        # 解析响应体
        try:
            self.json_data = response.json()
        except Exception:
            self.json_data = None
            self.text = response.text
    
    @property
    def data(self) -> Any:
        """获取响应数据"""
        return self.json_data if self.json_data else self.text
    
    def assert_status_code(self, expected: int) -> "Response":
        """断言状态码"""
        assert self.status_code == expected, \
            f"状态码断言失败: 期望 {expected}, 实际 {self.status_code}"
        return self
    
    def assert_response_time(self, max_seconds: float) -> "Response":
        """断言响应时间"""
        assert self.elapsed <= max_seconds, \
            f"响应时间断言失败: 期望 <= {max_seconds}s, 实际 {self.elapsed}s"
        return self
    
    def assert_json_path(self, json_path: str, expected: Any) -> "Response":
        """断言 JSON 路径值"""
        from jsonpath import jsonpath
        actual = jsonpath(self.json_data, json_path)
        assert actual and actual[0] == expected, \
            f"JSON 路径断言失败: {json_path} 期望 {expected}, 实际 {actual}"
        return self
    
    def jsonpath(self, path: str) -> Any:
        """使用 JSONPath 提取数据"""
        from jsonpath import jsonpath
        result = jsonpath(self.json_data, path)
        return result[0] if result else None

自定义异常

core/exceptions.py：

class FrameworkException(Exception):
    """框架基础异常"""
    pass


class RequestException(FrameworkException):
    """请求异常"""
    pass


class TimeoutException(FrameworkException):
    """超时异常"""
    pass


class AssertionException(FrameworkException):
    """断言异常"""
    pass


class ConfigException(FrameworkException):
    """配置异常"""
    pass

Session 管理和鉴权

api/base_api.py - API 基类：

from core.request import HTTPClient
from core.logger import log
from typing import Dict, Any


class BaseAPI:
    """API 基类"""
    
    def __init__(self, client: HTTPClient = None):
        self.client = client or HTTPClient()
    
    def set_auth_token(self, token: str):
        """设置认证 Token"""
        self.client.set_auth_token(token)
    
    def _extract_data(self, response, json_path: str = None) -> Any:
        """提取响应数据"""
        if json_path:
            return response.jsonpath(json_path)
        return response.data


class AuthService:
    """认证服务"""
    
    def __init__(self, client: HTTPClient):
        self.client = client
        self._token = None
    
    def login(self, username: str, password: str) -> Dict[str, Any]:
        """登录获取 Token"""
        response = self.client.post(
            "/auth/login",
            json={"username": username, "password": password}
        )
        
        response.assert_status_code(200)
        
        # 提取 Token
        self._token = response.jsonpath("$.data.token")
        self.client.set_auth_token(self._token)
        
        log.info(f"登录成功: {username}")
        
        return response.data
    
    def logout(self):
        """登出"""
        self.client.post("/auth/logout")
        self._token = None
        log.info("登出成功")
    
    @property
    def token(self) -> str:
        """获取当前 Token"""
        return self._token

五、Pytest 进阶用法

fixture 高级用法

testcases/conftest.py - Pytest 配置文件：

import pytest
from core.request import HTTPClient
from core.logger import log
from api.base_api import AuthService
from config.config import config


@pytest.fixture(scope="session")
def env_config():
    """环境配置 fixture"""
    return config


@pytest.fixture(scope="session")
def base_client():
    """HTTP 客户端 fixture（会话级别）"""
    client = HTTPClient()
    yield client
    client.close()


@pytest.fixture(scope="function")
def auth_client(base_client):
    """已认证的客户端 fixture"""
    # 使用测试账号登录
    auth = AuthService(base_client)
    auth.login(
        username=config.get("test_account.username"),
        password=config.get("test_account.password")
    )
    
    yield base_client
    
    # 清理：登出
    auth.logout()


@pytest.fixture(scope="function")
def clean_user_data():
    """清理测试数据 fixture"""
    # Setup: 准备测试数据
    test_user_id = None
    
    yield test_user_id
    
    # Teardown: 清理数据
    if test_user_id:
        log.info(f"清理测试用户: {test_user_id}")
        # 调用删除接口清理数据


@pytest.fixture(params=[
    {"username": "admin", "role": "admin"},
    {"username": "user", "role": "user"},
    {"username": "guest", "role": "guest"}
])
def user_role(request):
    """参数化 fixture"""
    return request.param


# 自动执行 fixture
@pytest.fixture(autouse=True, scope="function")
def setup_test_env():
    """自动为每个测试设置环境"""
    log.info("=" * 50)
    log.info("测试开始")
    yield
    log.info("测试结束")
    log.info("=" * 50)

参数化测试

import pytest
from core.request import HTTPClient


class TestLoginParameterized:
    """参数化登录测试"""
    
    # 基础参数化
    @pytest.mark.parametrize("username,password,expected_code", [
        ("admin", "admin123", 200),
        ("user", "user123", 200),
        ("invalid", "invalid", 401),
        ("", "", 400),
    ])
    def test_login_scenarios(
        self,
        base_client: HTTPClient,
        username: str,
        password: str,
        expected_code: int
    ):
        """测试不同登录场景"""
        response = base_client.post(
            "/auth/login",
            json={"username": username, "password": password}
        )
        assert response.status_code == expected_code
    
    # 多参数组合
    @pytest.mark.parametrize("page,size", [
        (1, 10),
        (2, 20),
        (5, 50),
    ])
    def test_pagination(
        self,
        auth_client: HTTPClient,
        page: int,
        size: int
    ):
        """测试分页查询"""
        response = auth_client.get(
            "/users",
            params={"page": page, "size": size}
        )
        
        assert response.status_code == 200
        data = response.json_data
        assert data["page"] == page
        assert len(data["items"]) <= size
    
    # 从文件加载参数
    @pytest.mark.parametrize("test_data", [
        {"username": "test1", "email": "test1@example.com"},
        {"username": "test2", "email": "test2@example.com"},
    ])
    def test_create_user_from_data(
        self,
        auth_client: HTTPClient,
        test_data: dict
    ):
        """从数据文件测试创建用户"""
        response = auth_client.post("/users", json=test_data)
        assert response.status_code == 201

钩子函数（conftest.py）

# 在 conftest.py 中添加

import pytest
from pathlib import Path


def pytest_configure(config):
    """Pytest 配置钩子"""
    # 注册自定义标记
    config.addinivalue_line(
        "markers", "smoke: 冒烟测试标记"
    )
    config.addinivalue_line(
        "markers", "regression: 回归测试标记"
    )
    config.addinivalue_line(
        "markers", "slow: 慢速测试标记"
    )


@pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_makereport(item, call):
    """测试报告钩子 - 截图和日志记录"""
    outcome = yield
    report = outcome.get_result()
    
    # 测试失败时记录详细信息
    if report.when == "call" and report.failed:
        # 记录失败信息
        log.error(f"测试失败: {item.name}")
        log.error(f"失败原因: {call.excinfo}")


def pytest_collection_modifyitems(config, items):
    """测试收集修改钩子"""
    # 为没有标记的测试添加默认标记
    for item in items:
        if not list(item.iter_markers()):
            item.add_marker(pytest.mark.regression)


@pytest.fixture(scope="session", autouse=True)
def setup_allure_environment():
    """设置 Allure 环境信息"""
    allure_dir = Path("reports/allure-results")
    allure_dir.mkdir(parents=True, exist_ok=True)
    
    env_file = allure_dir / "environment.properties"
    with open(env_file, "w") as f:
        f.write(f"Base.URL={config.base_url}\n")
        f.write(f"Environment={config.env}\n")
        f.write(f"Python.Version={sys.version}\n")

标记和过滤（markers）

pytest.ini - Pytest 配置文件：

[pytest]
# 测试发现路径
testpaths = testcases

# 测试文件命名规则
python_files = test_*.py
python_classes = Test*
python_functions = test_*

# 命令行参数
addopts = 
    -v
    -s
    --strict-markers
    --alluredir=reports/allure-results
    --clean-alluredir

# 注册标记
markers =
    smoke: 冒烟测试
    regression: 回归测试
    slow: 慢速测试
    p0: 优先级 P0
    p1: 优先级 P1
    p2: 优先级 P2

# 日志配置
log_cli = true
log_cli_level = INFO
log_cli_format = %(asctime)s [%(levelname)s] %(message)s

使用标记：

import pytest


class TestUserAPI:
    """用户接口测试"""
    
    @pytest.mark.smoke
    @pytest.mark.p0
    def test_login_success(self, base_client):
        """登录成功 - 冒烟测试"""
        pass
    
    @pytest.mark.regression
    @pytest.mark.p1
    def test_user_profile(self, auth_client):
        """用户资料 - 回归测试"""
        pass
    
    @pytest.mark.slow
    @pytest.mark.p2
    def test_bulk_import(self, auth_client):
        """批量导入 - 慢速测试"""
        pass

运行特定标记的测试：

# 运行冒烟测试
pytest -m smoke

# 运行 P0 和 P1 优先级测试
pytest -m "p0 or p1"

# 排除慢速测试
pytest -m "not slow"

# 运行特定模块的冒烟测试
pytest testcases/test_user.py -m smoke

六、测试用例实战

登录接口测试示例

api/user_api.py - 用户接口封装：

from api.base_api import BaseAPI
from core.request import HTTPClient
from typing import Dict, Any


class UserAPI(BaseAPI):
    """用户接口封装"""
    
    def login(self, username: str, password: str) -> Dict[str, Any]:
        """登录接口"""
        response = self.client.post(
            "/auth/login",
            json={"username": username, "password": password}
        )
        return response
    
    def get_profile(self) -> Dict[str, Any]:
        """获取用户资料"""
        response = self.client.get("/user/profile")
        return response
    
    def update_profile(self, data: Dict) -> Dict[str, Any]:
        """更新用户资料"""
        response = self.client.put("/user/profile", json=data)
        return response
    
    def get_user_list(self, page: int = 1, size: int = 10) -> Dict[str, Any]:
        """获取用户列表"""
        response = self.client.get(
            "/users",
            params={"page": page, "size": size}
        )
        return response

testcases/test_login.py - 登录测试：

import pytest
from api.user_api import UserAPI
from core.request import HTTPClient
from config.config import config


class TestLogin:
    """登录接口测试"""
    
    @pytest.mark.smoke
    def test_login_success(self, base_client: HTTPClient):
        """测试登录成功"""
        # Arrange
        user_api = UserAPI(base_client)
        username = config.get("test_account.username")
        password = config.get("test_account.password")
        
        # Act
        response = user_api.login(username, password)
        
        # Assert
        response.assert_status_code(200)
        response.assert_response_time(2.0)
        
        # 验证返回数据
        assert response.jsonpath("$.code") == 0
        assert response.jsonpath("$.data.token") is not None
        assert response.jsonpath("$.data.user.username") == username
    
    @pytest.mark.parametrize("invalid_credential,expected_code", [
        ({"username": "invalid", "password": "invalid"}, 401),
        ({"username": "", "password": ""}, 400),
        ({"username": "admin"}, 400),
        ({}, 400),
    ])
    def test_login_invalid_credentials(
        self,
        base_client: HTTPClient,
        invalid_credential: dict,
        expected_code: int
    ):
        """测试登录失败场景"""
        user_api = UserAPI(base_client)
        
        response = user_api.client.post(
            "/auth/login",
            json=invalid_credential
        )
        
        assert response.status_code == expected_code
    
    def test_login_and_access_profile(self, base_client: HTTPClient):
        """测试登录后访问受保护接口"""
        user_api = UserAPI(base_client)
        
        # 先登录
        login_response = user_api.login(
            config.get("test_account.username"),
            config.get("test_account.password")
        )
        
        assert login_response.status_code == 200
        
        # 设置 Token
        token = login_response.jsonpath("$.data.token")
        user_api.set_auth_token(token)
        
        # 访问用户资料
        profile_response = user_api.get_profile()
        
        assert profile_response.status_code == 200
        assert profile_response.jsonpath("$.data.username") is not None

数据驱动测试示例

testdata/login_data.yaml：

login_scenarios:
  - name: "管理员登录"
    username: "admin"
    password: "admin123"
    expected_code: 200
    expected_role: "admin"
  
  - name: "普通用户登录"
    username: "user"
    password: "user123"
    expected_code: 200
    expected_role: "user"
  
  - name: "错误密码"
    username: "admin"
    password: "wrong_password"
    expected_code: 401
    expected_message: "密码错误"
  
  - name: "用户不存在"
    username: "notexist"
    password: "any"
    expected_code: 401
    expected_message: "用户不存在"

utils/data_loader.py - 数据加载器：

import yaml
import json
from pathlib import Path
from typing import Any, List


class DataLoader:
    """测试数据加载器"""
    
    def __init__(self, data_dir: str = "testdata"):
        self.data_dir = Path(__file__).parent.parent / data_dir
    
    def load_yaml(self, filename: str) -> Any:
        """加载 YAML 文件"""
        file_path = self.data_dir / filename
        with open(file_path, "r", encoding="utf-8") as f:
            return yaml.safe_load(f)
    
    def load_json(self, filename: str) -> Any:
        """加载 JSON 文件"""
        file_path = self.data_dir / filename
        with open(file_path, "r", encoding="utf-8") as f:
            return json.load(f)
    
    def get_test_data(self, filename: str, key: str) -> List[dict]:
        """获取测试数据列表"""
        data = self.load_yaml(filename)
        return data.get(key, [])


# 全局实例
data_loader = DataLoader()

testcases/test_login_data_driven.py：

import pytest
from api.user_api import UserAPI
from core.request import HTTPClient
from utils.data_loader import data_loader


class TestLoginDataDriven:
    """数据驱动的登录测试"""
    
    @pytest.mark.parametrize(
        "test_data",
        data_loader.get_test_data("login_data.yaml", "login_scenarios")
    )
    def test_login_scenarios(
        self,
        base_client: HTTPClient,
        test_data: dict
    ):
        """测试多种登录场景"""
        user_api = UserAPI(base_client)
        
        response = user_api.login(
            test_data["username"],
            test_data["password"]
        )
        
        # 断言状态码
        assert response.status_code == test_data["expected_code"]
        
        # 如果登录成功，验证返回数据
        if test_data["expected_code"] == 200:
            assert response.jsonpath("$.data.token") is not None
            assert response.jsonpath("$.data.user.role") == test_data["expected_role"]
        else:
            # 验证错误消息
            assert test_data["expected_message"] in response.jsonpath("$.message")

断言最佳实践

utils/assertions.py - 自定义断言：

from typing import Any, Dict, List
from core.exceptions import AssertionException
from core.logger import log


class AssertUtils:
    """断言工具类"""
    
    @staticmethod
    def assert_equal(actual: Any, expected: Any, message: str = None):
        """相等断言"""
        if actual != expected:
            msg = message or f"断言失败: 期望 {expected}, 实际 {actual}"
            log.error(msg)
            raise AssertionException(msg)
        log.debug(f"断言通过: {actual} == {expected}")
    
    @staticmethod
    def assert_contains(container: Any, item: Any, message: str = None):
        """包含断言"""
        if item not in container:
            msg = message or f"断言失败: {container} 不包含 {item}"
            log.error(msg)
            raise AssertionException(msg)
    
    @staticmethod
    def assert_not_empty(value: Any, message: str = None):
        """非空断言"""
        if not value:
            msg = message or f"断言失败: 值为空"
            log.error(msg)
            raise AssertionException(msg)
    
    @staticmethod
    def assert_in_range(value: Any, min_val: Any, max_val: Any):
        """范围断言"""
        if not (min_val <= value <= max_val):
            raise AssertionException(
                f"断言失败: {value} 不在范围 [{min_val}, {max_val}] 内"
            )
    
    @staticmethod
    def assert_schema(data: Dict, schema: Dict):
        """JSON Schema 断言"""
        from jsonschema import validate, ValidationError
        
        try:
            validate(instance=data, schema=schema)
            log.debug("Schema 验证通过")
        except ValidationError as e:
            log.error(f"Schema 验证失败: {e}")
            raise AssertionException(f"Schema 验证失败: {e.message}")


# 使用示例
"""
from utils.assertions import AssertUtils

# 在测试中使用
AssertUtils.assert_equal(response.status_code, 200)
AssertUtils.assert_not_empty(response.jsonpath("$.data.token"))

# Schema 验证
user_schema = {
    "type": "object",
    "required": ["username", "email"],
    "properties": {
        "username": {"type": "string"},
        "email": {"type": "string", "format": "email"}
    }
}
AssertUtils.assert_schema(response.json_data, user_schema)
"""

七、报告与持续集成

Allure 报告配置

安装和生成报告：

# 安装 Allure 命令行工具
# Windows (使用 Scoop)
scoop install allure

# Mac (使用 Homebrew)
brew install allure

# 运行测试并生成 Allure 数据
pytest --alluredir=reports/allure-results --clean-alluredir

# 生成并打开报告
allure serve reports/allure-results

# 生成静态报告
allure generate reports/allure-results -o reports/allure-report --clean

在测试中使用 Allure 特性：

import allure
import pytest
from api.user_api import UserAPI


@allure.feature("用户管理")
@allure.story("登录功能")
class TestLoginAllure:
    """登录测试 - Allure 增强版"""
    
    @allure.title("登录成功测试")
    @allure.description("验证正确的用户名和密码能够成功登录")
    @allure.severity(allure.severity_level.BLOCKER)
    @allure.tag("P0", "冒烟测试")
    def test_login_success(self, base_client):
        """测试登录成功"""
        with allure.step("准备登录数据"):
            user_api = UserAPI(base_client)
            username = "admin"
            password = "admin123"
        
        with allure.step("执行登录请求"):
            response = user_api.login(username, password)
            
            # 附加请求和响应信息到报告
            allure.attach(
                str(response.json_data),
                name="响应数据",
                attachment_type=allure.attachment_type.JSON
            )
        
        with allure.step("验证登录结果"):
            assert response.status_code == 200
            assert response.jsonpath("$.data.token") is not None
    
    @allure.title("登录失败测试 - {test_data[name]}")
    def test_login_failure(
        self,
        base_client,
        test_data  # 从数据文件加载
    ):
        """测试登录失败场景"""
        with allure.step(f"测试场景: {test_data['name']}"):
            user_api = UserAPI(base_client)
            response = user_api.login(
                test_data["username"],
                test_data["password"]
            )
            
            allure.attach(
                f"用户名: {test_data['username']}\n"
                f"预期状态码: {test_data['expected_code']}",
                name="测试参数",
                attachment_type=allure.attachment_type.TEXT
            )
            
            assert response.status_code == test_data["expected_code"]

与 Jenkins/GitLab CI 集成

Jenkins Pipeline 配置：

// Jenkinsfile
pipeline {
    agent any
    
    environment {
        PYTHON_VERSION = '3.10'
    }
    
    stages {
        stage('Checkout') {
            steps {
                checkout scm
            }
        }
        
        stage('Setup Environment') {
            steps {
                sh '''
                    python -m venv venv
                    . venv/bin/activate
                    pip install -r requirements.txt
                '''
            }
        }
        
        stage('Run Tests') {
            steps {
                sh '''
                    . venv/bin/activate
                    pytest -v \
                           --alluredir=reports/allure-results \
                           --clean-alluredir \
                           -m "not slow"
                '''
            }
        }
        
        stage('Generate Report') {
            steps {
                allure includeProperties: false, 
                       jdk: '', 
                       results: [[path: 'reports/allure-results']]
            }
        }
    }
    
    post {
        always {
            // 清理工作空间
            cleanWs()
        }
        failure {
            // 发送失败通知
            emailext subject: '测试失败通知',
                     body: '测试执行失败，请查看报告',
                     to: 'team@example.com'
        }
    }
}

GitLab CI 配置：

# .gitlab-ci.yml
stages:
  - test
  - report

variables:
  PYTHON_VERSION: "3.10"

test:
  stage: test
  image: python:${PYTHON_VERSION}
  
  before_script:
    - pip install -r requirements.txt
    - apt-get update && apt-get install -y openjdk-11-jre-headless
    - wget -q https://github.com/allure-framework/allure2/releases/download/2.24.0/allure-2.24.0.tgz
    - tar -zxvf allure-2.24.0.tgz -C /opt/
    - ln -s /opt/allure-2.24.0/bin/allure /usr/bin/allure
  
  script:
    - pytest -v --alluredir=reports/allure-results --clean-alluredir
  
  artifacts:
    when: always
    paths:
      - reports/allure-results/
    expire_in: 1 week

allure_report:
  stage: report
  dependencies:
    - test
  script:
    - allure generate reports/allure-results -o reports/allure-report --clean
  artifacts:
    paths:
      - reports/allure-report/
    expire_in: 30 days

八、总结与最佳实践

核心要点回顾

模块关键实践
项目架构分层设计，职责清晰，易于维护和扩展
请求封装统一入口，重试机制，异常处理，日志记录
配置管理多环境支持，敏感信息隔离，配置热加载
测试设计数据驱动，参数化，fixture 复用
断言验证语义化断言，Schema 验证，错误信息友好
报告集成Allure 可视化，CI/CD 自动化，失败通知

最佳实践清单

1. 项目组织

  • ✅ 测试数据与测试代码分离
  • ✅ 接口封装层统一管理 API 调用
  • ✅ 使用 conftest.py 管理共享 fixture

2. 代码质量

  • ✅ 遵循 PEP 8 编码规范
  • ✅ 使用类型注解提高代码可读性
  • ✅ 编写清晰的测试文档字符串

3. 测试稳定性

  • ✅ 添加合理的等待和重试机制
  • ✅ 测试之间相互独立，无依赖
  • ✅ 测试后清理数据，避免污染

4. 执行效率

  • ✅ 使用并行执行（pytest-xdist）
  • ✅ 按优先级标记测试用例
  • ✅ 排除不必要的慢速测试

5. 持续改进

  • ✅ 定期重构和优化测试代码
  • ✅ 监控测试执行时间和稳定性
  • ✅ 收集覆盖率数据指导测试补充

进阶方向

  1. 性能测试集成：使用 locust 或 pytest-benchmark 进行接口性能测试
  2. Mock 服务：使用 responses 或 mock-server 模拟第三方接口
  3. 契约测试：引入 Pact 进行消费者驱动的契约测试
  4. 测试数据工厂：使用 factory_boy 或 model_bakery 构建测试数据
  5. Docker 化：容器化测试环境，实现一键部署和执行

系列预告：下一篇将讲解《接口测试数据驱动与 Mock 实战》，敬请期待！

代码仓库：完整示例代码已上传至 GitHub，关注公众号「测试开发技术」获取仓库地址。

参考资源：

# pytest # 自动化 # 运维 # python # 测试用例 # 测试工具
Logo
乐奇 Rokid 开放社区

更多推荐

cover

面向逻辑思维的程序设计方法——人形机器人最佳训练场景

avatar 乐奇 Rokid 开放社区
cover

支持国产硬件的AR巡检系统哪家好

avatar 乐奇 Rokid 开放社区
cover

AR巡检哪家好

avatar 乐奇 Rokid 开放社区