阿发聊测试 · 技术进阶

Python自动化测试入门：从零搭建你的第一个测试框架

从0到1构建可扩展的测试框架，附完整代码示例

你好，我是阿发。

很多刚入门自动化测试的同学都会问："我已经学会了Python和pytest，接下来该怎么做？"答案是：搭建一个属于自己的测试框架。一个好的测试框架能让你的测试工作事半功倍，不仅能提高效率，还能让代码更规范、更易于维护。今天我就带着大家从零开始，一步一步搭建一个完整的Python自动化测试框架。从项目结构设计到配置管理，从API封装到报告生成，全程实战，看完就能上手。

SECTION 01

为什么需要搭建测试框架？

没有框架的痛点

· 代码重复：每个测试用例都要写相同的配置、相同的请求代码

· 维护困难：接口地址变了，要改几十个文件

· 数据混乱：测试数据散落在各个文件中，难以管理

· 报告简陋：默认报告信息不全，排查问题困难

· 扩展性差：想加新功能需要改动大量代码

有框架的优势

· 代码复用：公共逻辑封装成模块，一次编写，多处调用

· 易于维护：配置集中管理，修改一处即可

· 数据分离：测试数据与测试代码分离，便于管理

· 报告丰富：自定义报告，包含日志、截图等信息

· 扩展性强：模块化设计，新增功能只需添加新模块

常见误区：你真的需要框架吗？

· 误区一："我们项目小，不需要框架"。其实项目越小，越需要框架来规范代码，否则很快就会变成一团乱麻。

· 误区二："框架要大而全"。框架应该按需搭建，不要一开始就加入所有可能用到的功能，否则维护成本会很高。

· 误区三："直接用现成的框架就好了"。现成框架虽然强大，但学习成本高，而且不一定完全符合你的业务需求。自己搭建能更好地理解框架的原理，也更容易定制。

· 正确的做法：从简单开始，逐步迭代。先搭建一个能跑通测试的最小框架，然后根据实际需求添加功能。

框架选型：为什么选择Python？

· 生态丰富：Python有大量优秀的测试库，如pytest、requests、selenium、playwright等，几乎能满足所有测试需求

· 语法简洁：Python语法接近自然语言，易于理解和编写，降低测试人员的学习成本，即使是不懂编程的测试人员也能快速上手

· 跨平台：Python可以在Windows、Linux、Mac等多个平台上运行，适合团队协作，无论是开发还是测试都能在自己熟悉的环境中工作

· 社区活跃：Python拥有庞大的开发者社区，遇到问题时能快速找到解决方案，开源项目也非常丰富

· 与AI融合：Python是人工智能和机器学习领域的主流语言，测试框架可以很容易地集成AI功能，如智能测试用例生成、异常检测、测试数据自动生成等，为测试工作带来新的可能性

阿发说实话

搭建测试框架不是一步到位的，而是一个持续迭代的过程。先搭一个最小可用版本，然后根据实际需求不断完善。不要一开始就追求完美，先让框架跑起来再说。

SECTION 02

项目结构设计

推荐的项目结构

```
test_framework/
├── api/ # API接口封装层
│ ├── __init__.py
│ ├── base_api.py # 基础API封装
│ └── user_api.py # 用户相关接口
├── config/ # 配置文件
│ ├── __init__.py
│ └── settings.py # 全局配置
├── data/ # 测试数据
│ ├── __init__.py
│ └── test_data.yaml # 测试数据文件
├── reports/ # 测试报告
│ └── .gitkeep
├── tests/ # 测试用例
│ ├── __init__.py
│ ├── test_user.py # 用户模块测试
│ └── conftest.py # pytest配置
├── utils/ # 工具类
│ ├── __init__.py
│ ├── logger.py # 日志工具
│ └── assertions.py # 断言工具
├── requirements.txt # 依赖文件
└── run.py # 运行入口
```

目录职责说明

· api/：存放API接口的封装代码，每个接口对应一个方法

· config/：存放全局配置，如环境地址、超时时间等

· data/：存放测试数据，支持yaml、json等格式

· reports/：存放测试报告，由pytest自动生成

· tests/：存放测试用例，按模块划分

· utils/：存放工具类，如日志、断言、数据库操作等

· run.py：测试框架的运行入口，提供命令行参数

常见问题：项目结构应该怎么调整？

· 项目初期：可以简化结构，把api、utils、tests放在根目录，不用分文件夹

· 中等项目：按照我们上面的结构划分，每个模块一个文件夹

· 大型项目：可以进一步细分，比如api/下按业务模块分文件夹（user/、order/、product/），tests/下按功能模块分文件夹

核心原则：高内聚，低耦合。相关的代码放在一起，不相关的代码分开。

代码示例：创建项目结构脚本

```python
import os

# 定义项目结构
project_structure = {
"test_framework": {
"api": ["__init__.py", "base_api.py", "user_api.py"],
"config": ["__init__.py", "settings.py"],
"data": ["__init__.py", "test_data.yaml"],
"reports": [".gitkeep"],
"tests": ["__init__.py", "test_user.py", "conftest.py"],
"utils": ["__init__.py", "logger.py", "assertions.py"],
"root": ["requirements.txt", "run.py"]
}
}

def create_project(base_path: str = "."):
"""创建项目结构"""
root = list(project_structure.keys())[0]
project_path = os.path.join(base_path, root)

# 创建根目录
if not os.path.exists(project_path):
os.makedirs(project_path)

# 创建子目录和文件
for folder, files in project_structure[root].items():
if folder == "root":
# 创建根目录下的文件
for file in files:
file_path = os.path.join(project_path, file)
if not os.path.exists(file_path):
with open(file_path, "w", encoding="utf-8") as f:
pass
else:
# 创建子目录
folder_path = os.path.join(project_path, folder)
if not os.path.exists(folder_path):
os.makedirs(folder_path)

# 创建子目录下的文件
for file in files:
file_path = os.path.join(folder_path, file)
if not os.path.exists(file_path):
with open(file_path, "w", encoding="utf-8") as f:
pass

print(f"项目结构创建完成: {project_path}")

# 打印目录结构
for dirpath, dirnames, filenames in os.walk(project_path):
level = dirpath.replace(project_path, "").count(os.sep)
indent = "│ " * level
subindent = "│ " * (level + 1)
print(f"{indent}{os.path.basename(dirpath)}/")
for filename in filenames:
print(f"{subindent}{filename}")

if __name__ == "__main__":
create_project()
```

SECTION 03

配置管理

代码示例：配置文件 settings.py

```python
import os
from dotenv import load_dotenv

# 加载.env文件
load_dotenv()

class Config:
"""基础配置类"""

# 基础URL
BASE_URL = os.getenv("BASE_URL", "https://api.example.com")

# 请求超时时间（秒）
TIMEOUT = int(os.getenv("TIMEOUT", 30))

# 日志配置
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
LOG_FILE = os.path.join(os.path.dirname(__file__), "..", "logs", "test.log")

# 报告配置
REPORT_DIR = os.path.join(os.path.dirname(__file__), "..", "reports")
REPORT_FILE = "test_report.html"

# 数据库配置
DB_HOST = os.getenv("DB_HOST", "localhost")
DB_PORT = int(os.getenv("DB_PORT", 3306))
DB_USER = os.getenv("DB_USER", "root")
DB_PASSWORD = os.getenv("DB_PASSWORD", "")
DB_NAME = os.getenv("DB_NAME", "testdb")

class DevelopmentConfig(Config):
"""开发环境配置"""
BASE_URL = "https://dev.api.example.com"

class TestingConfig(Config):
"""测试环境配置"""
BASE_URL = "https://test.api.example.com"

class ProductionConfig(Config):
"""生产环境配置"""
BASE_URL = "https://api.example.com"

# 根据环境变量选择配置
config_map = {
"dev": DevelopmentConfig,
"test": TestingConfig,
"prod": ProductionConfig
}

def get_config() -> Config:
"""获取当前环境的配置"""
env = os.getenv("ENV", "test")
return config_map.get(env, TestingConfig)
```

配置管理的最佳实践

· 分层配置：基础配置类定义默认值，环境配置类覆盖特定配置

· 环境变量优先：环境变量的值应该覆盖配置文件中的默认值

· 敏感信息保护：敏感信息（如密码、密钥）应该通过环境变量或密钥管理工具管理

· 配置验证：在启动时验证必要的配置项是否存在

· 类型转换：配置值从环境变量读取时要进行类型转换（字符串转整数等）

· 文档说明：为每个配置项添加注释，说明用途和默认值

代码示例：.env文件

```
# 环境选择: dev/test/prod
ENV=test

# API地址
BASE_URL=https://test.api.example.com

# 请求超时时间
TIMEOUT=30

# 日志级别: DEBUG/INFO/WARNING/ERROR
LOG_LEVEL=INFO

# 数据库配置
DB_HOST=localhost
DB_PORT=3306
DB_USER=root
DB_PASSWORD=123456
DB_NAME=testdb
```

阿发说实话

配置文件一定要使用环境变量管理敏感信息，比如数据库密码、API密钥等。不要把敏感信息硬编码在代码里，也不要提交到Git仓库。

SECTION 04

基础API封装

代码示例：base_api.py — 基础API封装

```python
import requests
from requests import Response
from config.settings import get_config

config = get_config()

class BaseAPI:
"""基础API类，封装通用的HTTP请求方法"""

def __init__(self):
self.base_url = config.BASE_URL
self.timeout = config.TIMEOUT
self.session = requests.Session()
self._setup_session()

def _setup_session(self):
"""配置Session，设置通用请求头"""
self.session.headers.update({
"Content-Type": "application/json",
"User-Agent": "TestFramework/1.0.0"
})

def set_token(self, token: str):
"""设置认证token"""
self.session.headers.update({"Authorization": f"Bearer {token}"})

def get(self, url: str, **kwargs) -> Response:
"""发送GET请求"""
full_url = self._build_url(url)
return self.session.get(full_url, timeout=self.timeout, **kwargs)

def post(self, url: str, **kwargs) -> Response:
"""发送POST请求"""
full_url = self._build_url(url)
return self.session.post(full_url, timeout=self.timeout, **kwargs)

def put(self, url: str, **kwargs) -> Response:
"""发送PUT请求"""
full_url = self._build_url(url)
return self.session.put(full_url, timeout=self.timeout, **kwargs)

def delete(self, url: str, **kwargs) -> Response:
"""发送DELETE请求"""
full_url = self._build_url(url)
return self.session.delete(full_url, timeout=self.timeout, **kwargs)

def _build_url(self, url: str) -> str:
"""构建完整URL"""
if url.startswith("http"):
return url
return f"{self.base_url}{url}"

def close(self):
"""关闭Session"""
self.session.close()
```

代码示例：request_helper.py — 请求辅助工具

```python
import json
from typing import Dict, Any, Optional
from requests import Response
from utils.logger import logger
from utils.assertions import Assertions

class RequestHelper:
"""请求辅助工具类，提供通用的请求处理方法"""

@staticmethod
def send_request(api_instance, method: str, url: str, **kwargs) -> Dict[str, Any]:
"""发送请求并返回JSON结果"""
logger.info(f"发送请求: {method.upper()} {url}")

method_map = {
"get": api_instance.get,
"post": api_instance.post,
"put": api_instance.put,
"delete": api_instance.delete
}

request_method = method_map.get(method.lower())
if not request_method:
raise ValueError(f"不支持的HTTP方法: {method}")

if kwargs.get("params"):
logger.info(f"请求参数: {kwargs['params']}")
if kwargs.get("json"):
logger.info(f"请求体: {kwargs['json']}")

response = request_method(url, **kwargs)

logger.info(f"响应状态码: {response.status_code}")
logger.info(f"响应时间: {response.elapsed.total_seconds()}s")

try:
result = response.json()
logger.info(f"响应内容: {json.dumps(result, ensure_ascii=False, indent=2)}")
return result
except json.JSONDecodeError:
logger.warning("响应不是JSON格式")
return {"status_code": response.status_code, "text": response.text}

@staticmethod
def get_response_time(response: Response) -> float:
"""获取响应时间（秒）"""
return response.elapsed.total_seconds()

@staticmethod
def validate_response(response: Dict[str, Any], expected_code: int = 0) -> None:
"""验证响应是否成功"""
code = response.get("code")
Assertions.assert_equal(code, expected_code, f"响应码验证失败: {code}")
```

常见问题：API封装的最佳实践

· 分层封装：基础API类（BaseAPI）处理HTTP请求，业务API类（UserAPI）处理具体业务逻辑，这样的分层设计使得代码结构清晰，易于维护和扩展

· 参数校验：在API方法中对参数进行基本校验，提前发现问题，避免无效请求浪费时间和资源

· 返回值处理：API方法应该返回解析后的JSON数据，而不是原始的Response对象，这样测试用例可以直接使用数据，无需再次解析

· 错误处理：在API层统一处理网络异常、超时等错误，提供友好的错误信息，便于定位和排查问题

· 请求重试：对于网络不稳定的情况，可以在BaseAPI中实现请求重试机制，提高测试的稳定性

· 请求日志：在发送请求前记录请求参数，在收到响应后记录响应内容和响应时间，便于调试和分析

· Session复用：使用requests.Session复用TCP连接，提高性能

代码示例：user_api.py — 用户接口封装

```python
from api.base_api import BaseAPI
from typing import Dict, Any

class UserAPI(BaseAPI):
"""用户模块API封装"""

def register(self, username: str, password: str, email: str = None) -> Dict[str, Any]:
"""用户注册接口"""
data = {
"username": username,
"password": password
}
if email:
data["email"] = email

response = self.post("/api/user/register", json=data)
return response.json()

def login(self, username: str, password: str) -> Dict[str, Any]:
"""用户登录接口"""
data = {
"username": username,
"password": password
}

response = self.post("/api/user/login", json=data)
result = response.json()

# 登录成功后自动设置token
if result.get("code") == 0 and result.get("data", {}).get("token"):
self.set_token(result["data"]["token"])

return result

def get_user_info(self, user_id: int = None) -> Dict[str, Any]:
"""获取用户信息接口"""
url = "/api/user/info"
if user_id:
url = f"/api/user/{user_id}"

response = self.get(url)
return response.json()

def update_user_info(self, **kwargs) -> Dict[str, Any]:
"""更新用户信息接口"""
data = {k: v for k, v in kwargs.items() if v is not None}
response = self.put("/api/user/info", json=data)
return response.json()

def delete_user(self, user_id: int) -> Dict[str, Any]:
"""删除用户接口"""
response = self.delete(f"/api/user/{user_id}")
return response.json()
```

阿发说实话

API封装的核心思想是"业务化"。不要让测试用例直接调用requests，而是把每个接口封装成一个有业务含义的方法。这样测试用例的可读性会大大提高，维护也更方便。

SECTION 05

测试数据管理

代码示例：test_data.yaml — 测试数据文件

```yaml
# 用户模块测试数据
user:
# 注册测试数据
register:
success:
username: "testuser_${random}"
password: "Test@123"
email: "test_${random}@example.com"
fail_empty_username:
username: ""
password: "Test@123"
expected_code: 400
fail_short_password:
username: "testuser"
password: "123"
expected_code: 400
fail_exist_username:
username: "existing_user"
password: "Test@123"
expected_code: 409

# 登录测试数据
login:
success:
username: "testuser"
password: "Test@123"
fail_wrong_password:
username: "testuser"
password: "Wrong@123"
expected_code: 401
fail_not_exist:
username: "nonexistent"
password: "Test@123"
expected_code: 404

# 用户信息测试数据
user_info:
valid_user_id: 1
invalid_user_id: 99999
update_data:
nickname: "阿发"
phone: "13800138000"
email: "afa@example.com"
```

代码示例：data_loader.py — 数据加载工具

```python
import yaml
import os
import random
import string
from typing import Dict, Any

def load_test_data(file_name: str = "test_data.yaml") -> Dict[str, Any]:
"""加载测试数据"""
data_path = os.path.join(
os.path.dirname(__file__),
"..",
"data",
file_name
)

with open(data_path, "r", encoding="utf-8") as f:
data = yaml.safe_load(f)

return data

def generate_random_string(length: int = 8) -> str:
"""生成随机字符串"""
return "".join(random.choices(string.ascii_lowercase + string.digits, k=length))

def replace_placeholders(data: Dict[str, Any]) -> Dict[str, Any]:
"""替换数据中的占位符"""
result = {}

for key, value in data.items():
if isinstance(value, str):
if "${random}" in value:
result[key] = value.replace("${random}", generate_random_string())
else:
result[key] = value
elif isinstance(value, dict):
result[key] = replace_placeholders(value)
else:
result[key] = value

return result

# 全局测试数据对象
test_data = load_test_data()

def get_data(path: str) -> Any:
"""通过路径获取测试数据"""
keys = path.split(".")
data = test_data

for key in keys:
if isinstance(data, dict):
data = data.get(key)
elif isinstance(data, list):
data = data[int(key)]
else:
return None

if data is None:
return None

# 如果是dict，替换占位符
if isinstance(data, dict):
return replace_placeholders(data)

return data
```

常见问题：如何管理大规模测试数据？

· 按模块划分：每个业务模块一个YAML文件，如user_data.yaml、order_data.yaml

· 参数化设计：使用占位符${random}、${timestamp}、${uuid}等动态生成数据

· 数据模板：定义基础数据模板，测试用例根据需要继承和覆盖

· 数据库初始化：使用fixture在测试前初始化测试数据，测试后清理

· 数据版本控制：测试数据文件也要纳入版本控制，便于追溯和回滚

代码示例：数据生成器 data_generator.py

```python
import random
import string
import uuid
from datetime import datetime, timedelta

class DataGenerator:
"""测试数据生成器"""

@staticmethod
def random_string(length: int = 8, prefix: str = "") -> str:
"""生成随机字符串"""
chars = string.ascii_lowercase + string.digits
return prefix + "".join(random.choices(chars, k=length))

@staticmethod
def random_int(min_val: int = 1, max_val: int = 1000) -> int:
"""生成随机整数"""
return random.randint(min_val, max_val)

@staticmethod
def random_email(domain: str = "example.com") -> str:
"""生成随机邮箱"""
username = DataGenerator.random_string(8)
return f"{username}@{domain}"

@staticmethod
def random_phone(country_code: str = "+86") -> str:
"""生成随机手机号"""
phone = "1" + str(random.randint(3, 9))
phone += "".join(random.choices(string.digits, k=9))
return f"{country_code}{phone}"

@staticmethod
def random_uuid() -> str:
"""生成UUID"""
return str(uuid.uuid4())

@staticmethod
def random_date(days_ago: int = 365, days_later: int = 365) -> str:
"""生成随机日期"""
start_date = datetime.now() - timedelta(days=days_ago)
end_date = datetime.now() + timedelta(days=days_later)
random_date = start_date + (end_date - start_date) * random.random()
return random_date.strftime("%Y-%m-%d")

@staticmethod
def random_datetime(days_ago: int = 30, days_later: int = 30) -> str:
"""生成随机日期时间"""
start_date = datetime.now() - timedelta(days=days_ago)
end_date = datetime.now() + timedelta(days=days_later)
random_date = start_date + (end_date - start_date) * random.random()
return random_date.strftime("%Y-%m-%d %H:%M:%S")

@staticmethod
def random_choice(items: list) -> any:
"""从列表中随机选择一个元素"""
return random.choice(items)

@staticmethod
def random_sample(items: list, count: int = 1) -> list:
"""从列表中随机选择多个元素"""
return random.sample(items, count)

# 使用示例
if __name__ == "__main__":
print(DataGenerator.random_string())
print(DataGenerator.random_email())
print(DataGenerator.random_phone())
print(DataGenerator.random_uuid())
print(DataGenerator.random_date())
```

SECTION 06

日志与断言工具

代码示例：logger.py — 日志工具

```python
import logging
import os
from config.settings import get_config

config = get_config()

def setup_logger(name: str = "test_framework") -> logging.Logger:
"""配置日志"""
# 创建日志目录
log_dir = os.path.dirname(config.LOG_FILE)
if not os.path.exists(log_dir):
os.makedirs(log_dir)

# 创建logger
logger = logging.getLogger(name)
logger.setLevel(getattr(logging, config.LOG_LEVEL))

# 避免重复添加handler
if logger.handlers:
return logger

# 创建控制台handler
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)

# 创建文件handler
file_handler = logging.FileHandler(config.LOG_FILE, encoding="utf-8")
file_handler.setLevel(getattr(logging, config.LOG_LEVEL))

# 定义日志格式
formatter = logging.Formatter(
"%(asctime)s - %(name)s - %(levelname)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)

console_handler.setFormatter(formatter)
file_handler.setFormatter(formatter)

# 添加handler
logger.addHandler(console_handler)
logger.addHandler(file_handler)

return logger

# 全局日志对象
logger = setup_logger()
```

日志记录的最佳实践

· 分级记录：DEBUG级别记录详细信息，INFO级别记录关键步骤，WARNING级别记录潜在问题，ERROR级别记录错误信息

· 结构化日志：对于复杂数据，使用JSON格式记录，便于后续分析和过滤

· 上下文信息：在日志中包含请求ID、用户ID、时间戳等上下文信息，便于追踪问题

· 敏感信息脱敏：避免在日志中记录密码、token等敏感信息

· 日志轮转：配置日志文件的大小和数量限制，避免日志文件过大

· 日志分析：使用ELK、Splunk等工具对日志进行集中管理和分析

代码示例：assertions.py — 断言工具

```python
import json
from utils.logger import logger

class Assertions:
"""自定义断言工具类"""

@staticmethod
def assert_equal(actual, expected, msg: str = None):
"""断言相等"""
try:
assert actual == expected, msg or f"Expected {expected}, got {actual}"
logger.info(f"✓ 断言通过: {actual} == {expected}")
except AssertionError as e:
logger.error(f"✗ 断言失败: {e}")
raise

@staticmethod
def assert_not_equal(actual, expected, msg: str = None):
"""断言不相等"""
try:
assert actual != expected, msg or f"Expected not {expected}, got {actual}"
logger.info(f"✓ 断言通过: {actual} != {expected}")
except AssertionError as e:
logger.error(f"✗ 断言失败: {e}")
raise

@staticmethod
def assert_in(item, container, msg: str = None):
"""断言包含"""
try:
assert item in container, msg or f"Expected {item} in {container}"
logger.info(f"✓ 断言通过: {item} in {container}")
except AssertionError as e:
logger.error(f"✗ 断言失败: {e}")
raise

@staticmethod
def assert_status_code(response, expected_code: int):
"""断言HTTP状态码"""
try:
assert response.status_code == expected_code, \n f"Expected status code {expected_code}, got {response.status_code}"
logger.info(f"✓ 状态码断言通过: {response.status_code}")
except AssertionError as e:
logger.error(f"✗ 状态码断言失败: {e}")
logger.error(f"响应内容: {response.text}")
raise

@staticmethod
def assert_api_success(response):
"""断言API请求成功（默认code为0）"""
try:
data = response.json()
code = data.get("code")
assert code == 0, f"Expected code 0, got {code}"
logger.info(f"✓ API请求成功: code={code}")
except (AssertionError, json.JSONDecodeError) as e:
logger.error(f"✗ API请求失败: {e}")
logger.error(f"响应内容: {response.text}")
raise

@staticmethod
def assert_api_failure(response, expected_code: int = None):
"""断言API请求失败"""
try:
data = response.json()
code = data.get("code")

if expected_code:
assert code == expected_code, \n f"Expected code {expected_code}, got {code}"
else:
assert code != 0, f"Expected failure, got code {code}"

logger.info(f"✓ API请求失败断言通过: code={code}")
except (AssertionError, json.JSONDecodeError) as e:
logger.error(f"✗ API请求失败断言失败: {e}")
logger.error(f"响应内容: {response.text}")
raise
```

代码示例：数据库操作工具 db_utils.py

```python
import pymysql
from config.settings import get_config
from utils.logger import logger

config = get_config()

class DBUtils:
"""数据库操作工具类"""

def __init__(self):
self.connection = None
self.cursor = None

def connect(self):
"""建立数据库连接"""
try:
self.connection = pymysql.connect(
host=config.DB_HOST,
port=config.DB_PORT,
user=config.DB_USER,
password=config.DB_PASSWORD,
database=config.DB_NAME,
charset="utf8mb4",
cursorclass=pymysql.cursors.DictCursor
)
self.cursor = self.connection.cursor()
logger.info(f"✓ 数据库连接成功: {config.DB_HOST}:{config.DB_PORT}/{config.DB_NAME}")
except Exception as e:
logger.error(f"✗ 数据库连接失败: {e}")
raise

def execute_query(self, sql: str, params: tuple = None) -> list:
"""执行查询语句"""
try:
if not self.connection:
self.connect()

self.cursor.execute(sql, params)
result = self.cursor.fetchall()
logger.info(f"✓ SQL查询成功: {sql}")
return result
except Exception as e:
logger.error(f"✗ SQL查询失败: {sql}, 错误: {e}")
raise

def execute_update(self, sql: str, params: tuple = None) -> int:
"""执行更新语句"""
try:
if not self.connection:
self.connect()

affected_rows = self.cursor.execute(sql, params)
self.connection.commit()
logger.info(f"✓ SQL更新成功: {sql}, 影响行数: {affected_rows}")
return affected_rows
except Exception as e:
self.connection.rollback()
logger.error(f"✗ SQL更新失败: {sql}, 错误: {e}")
raise

def close(self):
"""关闭数据库连接"""
if self.cursor:
self.cursor.close()
if self.connection:
self.connection.close()
logger.info("✓ 数据库连接已关闭")

def __enter__(self):
"""上下文管理器进入"""
self.connect()
return self

def __exit__(self, exc_type, exc_val, exc_tb):
"""上下文管理器退出"""
self.close()

# 使用示例
if __name__ == "__main__":
with DBUtils() as db:
users = db.execute_query("SELECT * FROM users LIMIT 10")
print(users)
```

SECTION 07

测试用例编写

代码示例：conftest.py — pytest配置

```python
import pytest
from api.user_api import UserAPI
from utils.logger import logger
from config.settings import get_config

config = get_config()

@pytest.fixture(scope="session")
def user_api() -> UserAPI:
"""用户API fixture"""
api = UserAPI()
logger.info("初始化UserAPI")
yield api
api.close()
logger.info("关闭UserAPI")

@pytest.fixture(scope="function")
def login_user(user_api: UserAPI) -> dict:
"""登录用户fixture"""
login_data = {
"username": "testuser",
"password": "Test@123"
}
result = user_api.login(**login_data)
assert result.get("code") == 0, f"登录失败: {result}"
logger.info(f"登录成功: {login_data['username']}")
return result

@pytest.fixture(autouse=True)
def log_test_name(request):
"""自动记录测试用例名称"""
logger.info(f"========== 开始测试: {request.node.name} ==========")
yield
logger.info(f"========== 测试结束: {request.node.name} ==========")
```

代码示例：test_user.py — 用户模块测试用例

```python
import pytest
from api.user_api import UserAPI
from utils.assertions import Assertions
from utils.data_loader import get_data
from utils.logger import logger

class TestUserRegister:
"""用户注册测试用例"""

@pytest.mark.parametrize("data", [get_data("user.register.success")])
def test_register_success(self, user_api: UserAPI, data: dict):
"""测试用户注册成功"""
logger.info(f"测试注册成功: {data['username']}")

result = user_api.register(
username=data["username"],
password=data["password"],
email=data.get("email")
)

Assertions.assert_equal(result.get("code"), 0)
Assertions.assert_in("data", result)
Assertions.assert_in("token", result["data"])

@pytest.mark.parametrize("data", [get_data("user.register.fail_empty_username")])
def test_register_fail_empty_username(self, user_api: UserAPI, data: dict):
"""测试注册失败-用户名为空"""
logger.info("测试注册失败-用户名为空")

result = user_api.register(
username=data["username"],
password=data["password"]
)

Assertions.assert_equal(result.get("code"), data["expected_code"])

@pytest.mark.parametrize("data", [get_data("user.register.fail_short_password")])
def test_register_fail_short_password(self, user_api: UserAPI, data: dict):
"""测试注册失败-密码过短"""
logger.info("测试注册失败-密码过短")

result = user_api.register(
username=data["username"],
password=data["password"]
)

Assertions.assert_equal(result.get("code"), data["expected_code"])

class TestUserLogin:
"""用户登录测试用例"""

@pytest.mark.parametrize("data", [get_data("user.login.success")])
def test_login_success(self, user_api: UserAPI, data: dict):
"""测试登录成功"""
logger.info(f"测试登录成功: {data['username']}")

result = user_api.login(
username=data["username"],
password=data["password"]
)

Assertions.assert_equal(result.get("code"), 0)
Assertions.assert_in("data", result)
Assertions.assert_in("token", result["data"])

@pytest.mark.parametrize("data", [get_data("user.login.fail_wrong_password")])
def test_login_fail_wrong_password(self, user_api: UserAPI, data: dict):
"""测试登录失败-密码错误"""
logger.info("测试登录失败-密码错误")

result = user_api.login(
username=data["username"],
password=data["password"]
)

Assertions.assert_equal(result.get("code"), data["expected_code"])

class TestUserInfo:
"""用户信息测试用例"""

def test_get_user_info(self, user_api: UserAPI, login_user: dict):
"""测试获取用户信息"""
logger.info("测试获取用户信息")

result = user_api.get_user_info()

Assertions.assert_equal(result.get("code"), 0)
Assertions.assert_in("data", result)
Assertions.assert_in("username", result["data"])

def test_update_user_info(self, user_api: UserAPI, login_user: dict):
"""测试更新用户信息"""
logger.info("测试更新用户信息")

update_data = get_data("user.user_info.update_data")
result = user_api.update_user_info(**update_data)

Assertions.assert_equal(result.get("code"), 0)

# 验证更新后的数据
info_result = user_api.get_user_info()
Assertions.assert_equal(info_result["data"]["nickname"], update_data["nickname"])
```

测试用例编写的最佳实践

· 用例命名规范：测试方法名应该清晰描述测试场景，如test_login_success、test_register_fail_empty_username

· 单一职责：每个测试用例只测试一个功能点，不要把多个测试场景放在一个用例中

· 参数化测试：使用@pytest.mark.parametrize实现数据驱动测试，减少重复代码

· 测试隔离：每个测试用例应该是独立的，互不影响，使用fixture管理前置条件

· 断言明确：断言应该明确验证预期结果，避免模糊的断言

· 测试标记：使用@pytest.mark标记测试用例，便于按类型筛选执行

· 文档注释：为测试类和测试方法添加文档字符串，说明测试目的和场景

代码示例：test_order.py — 订单模块测试用例

```python
import pytest
from api.order_api import OrderAPI
from api.user_api import UserAPI
from utils.assertions import Assertions
from utils.data_loader import get_data
from utils.logger import logger

class TestOrder:
"""订单模块测试用例"""

@pytest.fixture(scope="function")
def order_api(self, login_user) -> OrderAPI:
"""订单API fixture，依赖已登录用户"""
return OrderAPI()

@pytest.mark.smoke
def test_create_order(self, order_api: OrderAPI):
"""测试创建订单"""
logger.info("测试创建订单")

order_data = get_data("order.create")
result = order_api.create_order(**order_data)

Assertions.assert_equal(result.get("code"), 0)
Assertions.assert_in("order_id", result.get("data", {}))

@pytest.mark.smoke
def test_get_order_detail(self, order_api: OrderAPI):
"""测试获取订单详情"""
logger.info("测试获取订单详情")

# 先创建订单
order_data = get_data("order.create")
create_result = order_api.create_order(**order_data)
order_id = create_result["data"]["order_id"]

# 获取订单详情
result = order_api.get_order_detail(order_id)

Assertions.assert_equal(result.get("code"), 0)
Assertions.assert_equal(result["data"]["order_id"], order_id)

@pytest.mark.regression
@pytest.mark.parametrize("status", ["pending", "paid", "shipped", "completed"])
def test_get_orders_by_status(self, order_api: OrderAPI, status: str):
"""测试按状态查询订单"""
logger.info(f"测试按状态查询订单: {status}")

result = order_api.get_orders(status=status)

Assertions.assert_equal(result.get("code"), 0)
Assertions.assert_in("list", result.get("data", {}))
```

SECTION 08

测试报告与运行入口

代码示例：run.py — 测试运行入口

```python
import argparse
import os
import sys
import subprocess
from utils.logger import logger
from config.settings import get_config

config = get_config()

def run_tests(args):
"""运行测试"""
# 构建pytest命令
pytest_args = ["pytest"]

# 添加测试目录
if args.test_dir:
pytest_args.append(args.test_dir)
else:
pytest_args.append("tests")

# 添加测试用例过滤
if args.kwargs:
pytest_args.append(f"-k {args.kwargs}")

# 添加标记过滤
if args.mark:
pytest_args.append(f"-m {args.mark}")

# 添加报告参数
report_path = os.path.join(config.REPORT_DIR, config.REPORT_FILE)
pytest_args.extend([
f"--html={report_path}",
"--self-contained-html",
"--tb=short",
"-v",
f"--log-file={config.LOG_FILE}"
])

# 添加并发参数（如果安装了pytest-xdist）
if args.workers:
pytest_args.append(f"-n {args.workers}")

# 打印命令
logger.info(f"执行命令: {' '.join(pytest_args)}")

# 创建报告目录
if not os.path.exists(config.REPORT_DIR):
os.makedirs(config.REPORT_DIR)

# 执行命令
try:
result = subprocess.run(pytest_args, check=True, capture_output=True, text=True)
logger.info(f"测试执行成功")
logger.info(f"报告路径: {report_path}")
return 0
except subprocess.CalledProcessError as e:
logger.error(f"测试执行失败: {e.returncode}")
logger.error(f"错误输出: {e.stderr}")
return e.returncode

def main():
"""主函数"""
parser = argparse.ArgumentParser(description="测试框架运行入口")

parser.add_argument(
"-d", "--test-dir",
help="测试目录路径，默认为tests"
)

parser.add_argument(
"-k", "--kwargs",
help="测试用例过滤关键字，如: 'register and success'"
)

parser.add_argument(
"-m", "--mark",
help="测试标记过滤，如: 'smoke'"
)

parser.add_argument(
"-n", "--workers",
type=int,
help="并发执行的worker数量，需要安装pytest-xdist"
)

parser.add_argument(
"-e", "--env",
choices=["dev", "test", "prod"],
default="test",
help="测试环境，默认为test"
)

args = parser.parse_args()

# 设置环境变量
os.environ["ENV"] = args.env

logger.info(f"测试环境: {args.env}")
logger.info(f"基础URL: {config.BASE_URL}")

# 运行测试
exit_code = run_tests(args)
sys.exit(exit_code)

if __name__ == "__main__":
main()
```

代码示例：requirements.txt — 依赖文件

```
# 核心依赖
pytest>=7.0.0
requests>=2.28.0
PyYAML>=6.0
python-dotenv>=1.0.0

# 测试报告
pytest-html>=4.0.0

# 可选依赖
pytest-xdist>=3.0.0 # 并发执行
allure-pytest>=2.12.0 # Allure报告
mysql-connector-python>=8.0 # MySQL数据库
redis>=4.0.0 # Redis缓存
```

常见问题：如何运行测试？

· 运行所有测试：`python run.py`

· 运行指定模块：`python run.py -d tests/test_user.py`

· 按关键字过滤：`python run.py -k "register"`

· 按标记过滤：`python run.py -m "smoke"`

· 指定环境：`python run.py -e dev`

· 并发执行：`python run.py -n 4`（需要安装pytest-xdist）

阿发说实话

测试报告是测试结果的最终呈现，一定要重视。建议在测试报告中包含：用例执行时间、通过率、失败原因截图、API请求日志等信息。这样出问题时才能快速定位。

SECTION 09

CI/CD集成

代码示例：.github/workflows/test.yml — GitHub Actions配置

```yaml
name: 自动化测试

on:
push:
branches: [main, develop]
pull_request:
branches: [main]

jobs:
test:
runs-on: ubuntu-latest

steps:
- name: 检出代码
uses: actions/checkout@v4

- name: 设置Python环境
uses: actions/setup-python@v5
with:
python-version: "3.11"

- name: 安装依赖
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt

- name: 创建环境变量文件
run: |
echo "ENV=test" > .env
echo "BASE_URL=${{ secrets.BASE_URL }}" >> .env
echo "DB_HOST=${{ secrets.DB_HOST }}" >> .env
echo "DB_USER=${{ secrets.DB_USER }}" >> .env
echo "DB_PASSWORD=${{ secrets.DB_PASSWORD }}" >> .env

- name: 运行测试
run: |
python run.py -e test

- name: 上传测试报告
uses: actions/upload-artifact@v4
with:
name: test-report
path: reports/
if-no-files-found: error

- name: 失败时发送通知
if: failure()
uses: actions/github-script@v7
with:
script: |
github.rest.issues.create({
owner: context.repo.owner,
repo: context.repo.repo,
title: "测试失败: ${{ github.workflow }}",
body: "工作流: ${{ github.workflow }}\n分支: ${{ github.ref }}\n提交: ${{ github.sha }}\n\n请检查测试报告。"
})
```

代码示例：Jenkinsfile — Jenkins CI配置

```groovy
pipeline {
agent any

environment {
ENV = 'test'
BASE_URL = credentials('base-url')
DB_HOST = credentials('db-host')
DB_USER = credentials('db-user')
DB_PASSWORD = credentials('db-password')
}

stages {
stage('检出代码') {
steps {
checkout scm
}
}

stage('设置Python环境') {
steps {
sh 'python --version'
sh 'pip --version'
}
}

stage('安装依赖') {
steps {
sh 'pip install -r requirements.txt'
}
}

stage('运行测试') {
steps {
sh 'python run.py -e test'
}
post {
always {
junit 'reports/**/*.xml'
htmlpublisher htmlDir: 'reports',
keepAll: true,
reportName: '测试报告',
reportTitles: '自动化测试报告'
}
failure {
emailext to: 'test-team@example.com',
subject: "测试失败: ${JOB_NAME} #${BUILD_NUMBER}",
body: "构建失败，请检查报告: ${BUILD_URL}"
}
}
}
}

post {
success {
echo '测试全部通过！'
}
failure {
echo '测试失败，请检查日志！'
}
}
}
```

常见问题：CI/CD集成的最佳实践

· 环境隔离：CI环境应该与生产环境一致，使用相同的依赖版本

· 缓存依赖：使用CI缓存机制缓存Python依赖，加速构建过程

· 并行执行：使用pytest-xdist并行执行测试，减少执行时间

· 超时设置：为测试任务设置合理的超时时间，避免无限等待

· 通知机制：测试失败时及时通知相关人员，确保问题能快速解决

· 报告归档：保留历史测试报告，便于趋势分析和问题追溯

SECTION 10

最佳实践与扩展建议

最佳实践

· 遵循单一职责原则：每个模块只负责一件事，API层只管接口调用，测试用例只管业务逻辑

· 使用fixture管理资源：数据库连接、登录状态等资源通过pytest fixture管理，确保每个测试用例独立

· 测试数据隔离：每个测试用例使用独立的测试数据，避免数据污染

· 日志分级：调试时用DEBUG级别，正式运行时用INFO级别

· 代码审查：测试代码也要进行代码审查，确保质量

· 持续集成：把测试集成到CI/CD流程中，实现自动化

扩展建议

· 数据库操作封装：添加数据库操作模块，支持SQL查询、数据清理等

· UI自动化支持：集成Selenium或Playwright，支持Web UI测试

· Allure报告：集成Allure报告，生成更美观的测试报告

· 定时任务：使用APScheduler实现定时测试

· 消息通知：测试完成后发送邮件、钉钉或企业微信通知

· Mock服务：集成responses或pytest-mock，实现接口Mock

· 性能测试：集成Locust或JMeter，支持接口性能测试

· 安全测试：集成安全扫描工具，检测常见安全漏洞

· 测试覆盖率：使用coverage.py统计测试覆盖率，确保代码质量

常见测试场景与解决方案

场景一：接口需要复杂签名认证

解决方案：在BaseAPI中封装签名算法，每次请求前自动生成签名。可以使用装饰器或继承方式实现，确保所有请求都自动带上正确的签名参数。

场景二：测试数据需要动态生成

解决方案：使用DataGenerator生成动态数据，结合YAML文件中的占位符替换机制。例如在测试数据中使用${random_email}，数据加载器会自动替换为随机邮箱。

场景三：测试环境不稳定导致测试失败

解决方案：使用pytest-rerunfailures插件自动重试失败用例，设置合理的重试次数和间隔时间。同时在测试前检查环境状态，确保测试环境可用。

场景四：需要批量执行不同环境的测试

解决方案：通过命令行参数或配置文件指定测试环境，使用pytest的参数化功能实现多环境并行测试。

场景五：测试报告需要发送给多个相关人员

解决方案：使用Notifier工具类，支持邮件、钉钉、企业微信等多种通知方式。在测试完成后自动发送测试报告和结果摘要。

代码示例：消息通知工具 notify.py

```python
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import requests
from config.settings import get_config
from utils.logger import logger

config = get_config()

class Notifier:
"""消息通知工具类"""

@staticmethod
def send_email(to_email: str, subject: str, content: str, attachment_path: str = None):
"""发送邮件通知"""
try:
msg = MIMEMultipart()
msg["From"] = config.EMAIL_FROM
msg["To"] = to_email
msg["Subject"] = subject

msg.attach(MIMEText(content, "html", "utf-8"))

# 添加附件
if attachment_path:
with open(attachment_path, "rb") as f:
attachment = MIMEText(f.read(), "base64", "utf-8")
attachment["Content-Type"] = "application/octet-stream"
attachment["Content-Disposition"] = f"attachment; filename={attachment_path.split('/')[-1]}"
msg.attach(attachment)

# 发送邮件
with smtplib.SMTP(config.EMAIL_HOST, config.EMAIL_PORT) as server:
server.starttls()
server.login(config.EMAIL_USER, config.EMAIL_PASSWORD)
server.send_message(msg)

logger.info(f"✓ 邮件发送成功: {to_email}")
except Exception as e:
logger.error(f"✗ 邮件发送失败: {e}")

@staticmethod
def send_dingtalk(message: str, webhook_url: str = None):
"""发送钉钉通知"""
try:
url = webhook_url or config.DINGTALK_WEBHOOK
data = {
"msgtype": "text",
"text": {"content": message}
}

response = requests.post(url, json=data)
response.raise_for_status()
logger.info("✓ 钉钉消息发送成功")
except Exception as e:
logger.error(f"✗ 钉钉消息发送失败: {e}")

@staticmethod
def send_wework(message: str, webhook_url: str = None):
"""发送企业微信通知"""
try:
url = webhook_url or config.WEWORK_WEBHOOK
data = {
"msgtype": "text",
"text": {"content": message}
}

response = requests.post(url, json=data)
response.raise_for_status()
logger.info("✓ 企业微信消息发送成功")
except Exception as e:
logger.error(f"✗ 企业微信消息发送失败: {e}")
```

常见问题：如何处理测试环境不稳定？

· 重试机制：使用pytest-rerunfailures插件，失败的用例自动重试

· 超时设置：为每个请求设置合理的超时时间，避免无限等待

· 环境监控：在测试前检查环境是否可用，不可用时跳过或等待

· 数据清理：测试前后清理数据，确保测试环境干净

· 降级策略：核心接口失败时停止测试，非核心接口失败时继续执行

代码示例：pytest.ini — pytest配置文件

```ini
[pytest]
# 测试用例目录
testpaths = tests

# 测试文件匹配模式
python_files = test_*.py
python_classes = Test*
python_functions = test_*

# 报告配置
addopts = -v --tb=short --html=reports/test_report.html --self-contained-html

# 并发执行配置（需要安装pytest-xdist）
# addopts = -v --tb=short -n 4

# 失败重试配置（需要安装pytest-rerunfailures）
# addopts = -v --tb=short --reruns=2 --reruns-delay=1

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

# 自定义标记
markers =
smoke: 冒烟测试用例
regression: 回归测试用例
performance: 性能测试用例
integration: 集成测试用例

# 忽略的目录
ignore = tests/__pycache__
```

测试框架性能优化指南

· 缓存机制：对频繁访问的数据进行缓存，减少重复请求

· 并发执行：使用pytest-xdist并行执行测试，提高执行效率

· 测试分层：将测试分为单元测试、集成测试、端到端测试，按需执行

· 数据预热：在测试前准备好测试数据，避免测试过程中等待

· 资源复用：复用数据库连接、HTTP Session等资源，减少创建开销

· 测试优先级：优先执行核心功能测试，确保关键路径畅通

搭建测试框架的核心不是技术有多复杂，而是在于"规范"和"复用"。一个好的框架能让团队成员用同样的方式写测试，让测试工作更高效、更可靠。

最后

今天我们从零开始搭建了一个完整的Python自动化测试框架，涵盖了项目结构设计、配置管理、API封装、测试数据管理、日志断言、测试用例编写、报告生成和CI集成等所有核心环节。这个框架虽然简单，但已经具备了一个成熟测试框架的基本功能。你可以根据自己的实际需求，逐步扩展更多功能。记住，搭建框架是一个持续迭代的过程，不要追求一步到位，先让框架跑起来，再慢慢完善。

学习搭建测试框架不仅仅是掌握技术，更是培养一种系统化、工程化的思维方式。在这个过程中，你会学会如何设计可维护的代码结构、如何管理配置和数据、如何编写高质量的测试用例，这些技能在任何技术领域都是宝贵的财富。

如果你在搭建过程中遇到问题，或者有任何改进建议，欢迎随时与我交流。记住，技术学习是一个不断探索和实践的过程，保持好奇心，持续学习，你会越来越优秀。希望这篇文章能帮助你迈出自动化测试的第一步，祝你在测试领域取得更大的进步！

关于Python自动化测试、测试框架搭建、AI测试转型等问题，都可以通过公众号菜单栏添加我微信 testafa 私信交流。需要学习资料也可以私信！