Python自动化测试九章经（2026版）
·
第一章：筑基篇——测试基础与Python测试生态
1.1 软件测试的本质与价值
测试的终极目的：不是发现Bug，而是建立对软件质量的信心。
在敏捷开发与DevOps时代，测试已从"开发后的验证环节"转变为"贯穿全生命周期的质量活动"：
| 测试阶段 | 目标 | 自动化程度 | Python工具链 |
|---|---|---|---|
| 单元测试 | 验证最小代码单元 | 100% | unittest, pytest |
| 集成测试 | 验证模块间交互 | 90% | pytest + requests |
| 接口测试 | 验证API契约 | 95% | requests, httpx, Schemathesis |
| UI测试 | 验证用户交互流程 | 70% | Selenium, Playwright, Cypress |
| 性能测试 | 验证系统承载能力 | 80% | Locust, k6, JMeter |
| 安全测试 | 发现潜在漏洞 | 60% | Bandit, Safety, OWASP ZAP |
1.2 Python测试生态全景
Python拥有业界最成熟的测试生态系统：
核心框架层
├── unittest (标准库，xUnit风格)
├── pytest (第三方，现代测试标准)
├── doctest (文档即测试)
└── nose2 (unittest扩展)
专项测试层
├── Web UI: Selenium, Playwright, Cypress(Python绑定)
├── API: requests, httpx, Flask-Testing, FastAPI-TestClient
├── 性能: Locust, pytest-benchmark
├── 安全: Bandit, Safety, Semgrep
└── 数据: Hypothesis(属性测试), Factory Boy( fixtures )
质量保障层
├── 覆盖率: coverage.py, pytest-cov
├── 静态分析: mypy, pylint, black, ruff
├── 持续集成: tox, nox, GitHub Actions, GitLab CI
└── 报告生成: Allure, pytest-html, junit-xml
第二章：元始篇——unittest标准库精要
2.1 unittest核心架构
unittest是Python标准库中的xUnit风格测试框架，理解它是掌握Python测试的基础。
import unittest
from typing import List
class Calculator:
"""被测系统(SUT)"""
def add(self, a: float, b: float) -> float:
return a + b
def divide(self, a: float, b: float) -> float:
if b == 0:
raise ValueError("除数不能为零")
return a / b
def batch_add(self, numbers: List[float]) -> float:
return sum(numbers)
class TestCalculator(unittest.TestCase):
"""测试用例类"""
# ========== 生命周期钩子 ==========
@classmethod
def setUpClass(cls):
"""类级别：所有测试前执行一次"""
print("初始化测试环境...")
cls.calc = Calculator()
@classmethod
def tearDownClass(cls):
"""类级别：所有测试后执行一次"""
print("清理测试环境...")
def setUp(self):
"""方法级别：每个测试前执行"""
self.local_calc = Calculator()
def tearDown(self):
"""方法级别：每个测试后执行"""
pass
# ========== 基础断言 ==========
def test_add_basic(self):
"""基础加法测试"""
result = self.calc.add(2, 3)
self.assertEqual(result, 5, "2+3应该等于5")
def test_add_float(self):
"""浮点数精度测试"""
result = self.calc.add(0.1, 0.2)
self.assertAlmostEqual(result, 0.3, places=7, msg="浮点数比较需使用近似相等")
# ========== 异常测试 ==========
def test_divide_by_zero(self):
"""除零异常测试"""
with self.assertRaises(ValueError) as context:
self.calc.divide(10, 0)
self.assertIn("除数不能为零", str(context.exception))
# ========== 批量测试与条件跳过 ==========
@unittest.skip("功能尚未实现")
def test_future_feature(self):
pass
@unittest.skipIf(True, "临时跳过")
def test_conditional_skip(self):
pass
@unittest.expectedFailure
def test_known_bug(self):
"""已知缺陷，预期失败"""
self.assertEqual(1, 2)
# ========== 测试套件组织 ==========
def suite():
"""自定义测试套件"""
loader = unittest.TestLoader()
test_suite = unittest.TestSuite()
test_suite.addTests(loader.loadTestsFromTestCase(TestCalculator))
return test_suite
if __name__ == '__main__':
# 基础运行
# unittest.main()
# 详细输出
runner = unittest.TextTestRunner(verbosity=2)
runner.run(suite())
2.2 unittest高级特性
子测试(SubTest)——参数化测试的基础：
class TestBatchOperations(unittest.TestCase):
def test_batch_add_multiple_cases(self):
"""使用subTest实现参数化"""
test_cases = [
([1, 2, 3], 6),
([], 0),
([-1, 1], 0),
([0.1, 0.2], 0.3),
]
for inputs, expected in test_cases:
with self.subTest(inputs=inputs, expected=expected):
calc = Calculator()
result = calc.batch_add(inputs)
self.assertAlmostEqual(result, expected, places=7)
Mock对象——隔离依赖：
from unittest.mock import Mock, patch, MagicMock, call
class TestWithMocking(unittest.TestCase):
def test_external_api_call(self):
"""模拟外部API调用"""
with patch('requests.get') as mock_get:
# 配置Mock行为
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {'data': 'test'}
mock_get.return_value = mock_response
# 执行被测代码
result = fetch_data_from_api('http://api.example.com')
# 验证调用
mock_get.assert_called_once_with('http://api.example.com')
self.assertEqual(result, {'data': 'test'})
def test_mock_side_effects(self):
"""模拟异常和序列返回值"""
mock_func = Mock()
mock_func.side_effect = [1, 2, 3, Exception('错误')]
self.assertEqual(mock_func(), 1)
self.assertEqual(mock_func(), 2)
self.assertEqual(mock_func(), 3)
with self.assertRaises(Exception):
mock_func()
第三章：进阶篇——pytest现代测试框架
3.1 pytest核心优势
pytest已成为Python测试的事实标准，相比unittest的优势：
| 特性 | unittest | pytest |
|---|---|---|
| 断言方式 | 丰富的assert*方法 | 原生assert，更直观 |
| 插件生态 | 有限 | 800+插件，生态丰富 |
| 夹具(Fixture) | setUp/tearDown | 强大的依赖注入系统 |
| 参数化 | subTest较繁琐 | @pytest.mark.parametrize |
| 失败重试 | 不支持 | pytest-rerunfailures |
| 并行执行 | 不支持 | pytest-xdist |
3.2 pytest基础实战
# test_pytest_basics.py
import pytest
from typing import List, Dict
class ShoppingCart:
def __init__(self):
self.items: List[Dict] = []
self.discount_rate: float = 0.0
def add_item(self, name: str, price: float, quantity: int = 1):
self.items.append({
'name': name,
'price': price,
'quantity': quantity
})
def apply_discount(self, rate: float):
"""rate: 0.0-1.0"""
if not 0 <= rate <= 1:
raise ValueError("折扣率必须在0-1之间")
self.discount_rate = rate
def get_total(self) -> float:
subtotal = sum(item['price'] * item['quantity'] for item in self.items)
return subtotal * (1 - self.discount_rate)
# ========== 基础测试函数 ==========
def test_cart_initialization():
"""测试购物车初始化"""
cart = ShoppingCart()
assert cart.get_total() == 0
assert cart.items == []
def test_add_single_item():
"""测试添加单件商品"""
cart = ShoppingCart()
cart.add_item("Apple", 5.0, 2)
assert len(cart.items) == 1
assert cart.get_total() == 10.0
# ========== 参数化测试 ==========
@pytest.mark.parametrize("items,expected_total", [
([("Apple", 5.0, 2)], 10.0),
([("Apple", 5.0, 1), ("Banana", 3.0, 2)], 11.0),
([], 0.0),
([("Gold", 1000.0, 1)], 1000.0),
])
def test_calculate_total(items, expected_total):
"""参数化测试多种场景"""
cart = ShoppingCart()
for name, price, qty in items:
cart.add_item(name, price, qty)
assert cart.get_total() == expected_total
@pytest.mark.parametrize("rate,expected", [
(0.0, 100.0),
(0.1, 90.0),
(0.5, 50.0),
(1.0, 0.0),
])
def test_discount_calculation(rate, expected):
"""测试不同折扣率"""
cart = ShoppingCart()
cart.add_item("Item", 100.0)
cart.apply_discount(rate)
assert cart.get_total() == expected
# ========== 异常测试 ==========
def test_invalid_discount_rate():
"""测试无效折扣率"""
cart = ShoppingCart()
with pytest.raises(ValueError, match="折扣率必须在0-1之间"):
cart.apply_discount(1.5)
with pytest.raises(ValueError):
cart.apply_discount(-0.1)
# ========== 分组与标记 ==========
@pytest.mark.slow
def test_performance_heavy_calculation():
"""慢测试标记"""
import time
time.sleep(1)
assert True
@pytest.mark.smoke
def test_critical_path():
"""冒烟测试标记"""
cart = ShoppingCart()
cart.add_item("Critical Item", 999.0)
assert cart.get_total() > 0
# 命令行运行特定标记: pytest -m "smoke and not slow"
3.3 Fixture：pytest的灵魂
Fixture是pytest最强大的特性，实现了真正的依赖注入：
# conftest.py - 共享fixture配置
import pytest
import tempfile
import os
from typing import Generator
# ========== 基础Fixture ==========
@pytest.fixture
def empty_cart() -> ShoppingCart:
"""提供空购物车实例"""
return ShoppingCart()
@pytest.fixture
def sample_cart() -> ShoppingCart:
"""提供预填充购物车"""
cart = ShoppingCart()
cart.add_item("Apple", 5.0, 2)
cart.add_item("Banana", 3.0, 3)
return cart
# ========== 作用域控制 ==========
@pytest.fixture(scope="module")
def database_connection():
"""模块级别：只创建一次连接"""
print("创建数据库连接...")
conn = create_db_connection()
yield conn
print("关闭数据库连接...")
conn.close()
@pytest.fixture(scope="session")
def test_config():
"""会话级别：整个测试会话共享"""
return {
"api_url": "http://api.test.com",
"timeout": 30,
"retry_count": 3
}
# ========== 自动使用与清理 ==========
@pytest.fixture(autouse=True, scope="function")
def setup_teardown():
"""每个测试自动执行"""
print("测试前准备")
yield
print("测试后清理")
@pytest.fixture
def temp_data_file() -> Generator[str, None, None]:
"""临时文件fixture"""
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.json') as f:
f.write('{"test": "data"}')
temp_path = f.name
yield temp_path
# 清理
os.unlink(temp_path)
# ========== Fixture组合与参数化 ==========
@pytest.fixture(params=["sqlite", "postgresql", "mysql"])
def database_backend(request):
"""参数化fixture"""
return request.param
@pytest.fixture
def data_store(database_backend):
"""依赖其他fixture"""
if database_backend == "sqlite":
return SQLiteStore()
elif database_backend == "postgresql":
return PostgresStore()
else:
return MySQLStore()
# ========== 在测试中使用 ==========
def test_with_fixtures(empty_cart, sample_cart, temp_data_file):
"""使用多个fixture"""
assert empty_cart.get_total() == 0
assert sample_cart.get_total() > 0
assert os.path.exists(temp_data_file)
3.4 高级pytest技巧
自定义插件与钩子：
# conftest.py
import pytest
def pytest_configure(config):
"""配置阶段钩子"""
config.addinivalue_line(
"markers", "integration: 标记为集成测试"
)
def pytest_collection_modifyitems(config, items):
"""修改测试收集"""
# 自动为所有测试添加超时
for item in items:
if "slow" not in item.keywords:
item.add_marker(pytest.mark.timeout(30))
@pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_makereport(item, call):
"""测试执行报告钩子"""
outcome = yield
report = outcome.get_result()
# 测试失败时自动截图（UI测试）
if report.when == "call" and report.failed:
if "selenium" in item.fixturenames:
driver = item.funcargs['selenium']
driver.save_screenshot(f"screenshots/{item.name}.png")
# ========== 自定义命令行选项 ==========
def pytest_addoption(parser):
parser.addoption(
"--run-integration",
action="store_true",
default=False,
help="运行集成测试"
)
@pytest.fixture
def run_integration(request):
return request.config.getoption("--run-integration")
第四章：契约篇——API自动化测试
4.1 REST API测试体系
# test_api.py
import pytest
import requests
from requests.auth import HTTPBasicAuth
import jsonschema
from jsonschema import validate
# ========== 基础API测试 ==========
BASE_URL = "https://api.example.com/v1"
class TestUserAPI:
"""用户API测试套件"""
@pytest.fixture
def auth_headers(self):
"""认证头fixture"""
return {
"Authorization": "Bearer test_token",
"Content-Type": "application/json"
}
def test_get_user_success(self, auth_headers):
"""测试获取用户成功"""
response = requests.get(
f"{BASE_URL}/users/123",
headers=auth_headers
)
assert response.status_code == 200
data = response.json()
# 验证响应结构
assert "id" in data
assert "username" in data
assert "email" in data
assert isinstance(data["id"], int)
def test_create_user(self, auth_headers):
"""测试创建用户"""
payload = {
"username": "testuser",
"email": "test@example.com",
"role": "user"
}
response = requests.post(
f"{BASE_URL}/users",
headers=auth_headers,
json=payload
)
assert response.status_code == 201
assert response.json()["username"] == "testuser"
def test_user_not_found(self, auth_headers):
"""测试用户不存在"""
response = requests.get(
f"{BASE_URL}/users/999999",
headers=auth_headers
)
assert response.status_code == 404
def test_invalid_auth(self):
"""测试无效认证"""
response = requests.get(f"{BASE_URL}/users/123")
assert response.status_code == 401
# ========== JSON Schema验证 ==========
USER_SCHEMA = {
"type": "object",
"required": ["id", "username", "email"],
"properties": {
"id": {"type": "integer"},
"username": {"type": "string", "minLength": 3},
"email": {"type": "string", "format": "email"},
"role": {"type": "string", "enum": ["admin", "user", "guest"]},
"created_at": {"type": "string", "format": "date-time"}
}
}
def test_response_schema_validation(auth_headers):
"""响应结构Schema验证"""
response = requests.get(f"{BASE_URL}/users/123", headers=auth_headers)
validate(instance=response.json(), schema=USER_SCHEMA)
4.2 使用FastAPI/TestClient的现代测试
# 针对FastAPI应用的测试
from fastapi.testclient import TestClient
from main import app # 假设这是FastAPI应用
client = TestClient(app)
class TestFastAPIEndpoints:
"""FastAPI端点测试"""
def test_read_main(self):
"""测试主端点"""
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello World"}
def test_create_item(self):
"""测试创建项目"""
item_data = {
"name": "Foo",
"description": "A very nice Item",
"price": 35.4,
"tax": 3.2
}
response = client.post("/items/", json=item_data)
assert response.status_code == 200
assert response.json()["name"] == "Foo"
def test_read_item_with_mock_db(self, monkeypatch):
"""使用monkeypatch模拟数据库"""
def mock_get_item(item_id):
return {"id": item_id, "name": "Mocked Item"}
monkeypatch.setattr("crud.get_item", mock_get_item)
response = client.get("/items/1")
assert response.status_code == 200
assert response.json()["name"] == "Mocked Item"
4.3 API测试最佳实践
使用VCR.py录制与回放HTTP交互：
import vcr
# 录制真实API响应，后续测试使用录制内容
@vcr.use_cassette('fixtures/cassettes/user_api.yml')
def test_with_recorded_response():
"""使用录制响应测试"""
response = requests.get('https://api.github.com/users/octocat')
assert response.status_code == 200
assert response.json()['login'] == 'octocat'
使用Httpx进行异步API测试：
import httpx
import pytest
@pytest.mark.asyncio
async def test_async_api():
"""异步API测试"""
async with httpx.AsyncClient() as client:
response = await client.get("https://api.example.com/data")
assert response.status_code == 200
第五章：视界篇——UI自动化测试
5.1 Selenium WebDriver实战
# test_ui_selenium.py
import pytest
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager
class TestECommerceSite:
"""电商网站UI测试"""
@pytest.fixture(scope="class")
def driver(self):
"""配置WebDriver"""
chrome_options = Options()
chrome_options.add_argument("--headless") # 无头模式
chrome_options.add_argument("--no-sandbox")
chrome_options.add_argument("--disable-dev-shm-usage")
service = Service(ChromeDriverManager().install())
driver = webdriver.Chrome(service=service, options=chrome_options)
driver.implicitly_wait(10)
driver.maximize_window()
yield driver
driver.quit()
def test_login_flow(self, driver):
"""测试登录流程"""
# 访问登录页
driver.get("https://shop.example.com/login")
# 填写表单
driver.find_element(By.ID, "username").send_keys("testuser")
driver.find_element(By.ID, "password").send_keys("password123")
driver.find_element(By.ID, "login-btn").click()
# 显式等待验证登录成功
welcome_msg = WebDriverWait(driver, 10).until(
EC.presence_of_element_located((By.CLASS_NAME, "welcome-message"))
)
assert "欢迎回来" in welcome_msg.text
def test_add_to_cart(self, driver):
"""测试添加商品到购物车"""
driver.get("https://shop.example.com/products/1")
# 选择规格
driver.find_element(By.CSS_SELECTOR, "[data-size='M']").click()
driver.find_element(By.CSS_SELECTOR, "[data-color='blue']").click()
# 添加到购物车
add_btn = driver.find_element(By.ID, "add-to-cart")
add_btn.click()
# 验证购物车数量更新
cart_count = WebDriverWait(driver, 5).until(
EC.text_to_be_present_in_element(
(By.CLASS_NAME, "cart-count"), "1"
)
)
assert cart_count
def test_search_functionality(self, driver):
"""测试搜索功能"""
driver.get("https://shop.example.com")
search_box = driver.find_element(By.NAME, "q")
search_box.send_keys("Python书籍")
search_box.submit()
# 验证搜索结果
results = driver.find_elements(By.CLASS_NAME, "product-item")
assert len(results) > 0
# 验证结果相关性
for result in results[:3]:
title = result.find_element(By.CLASS_NAME, "product-title").text
assert "Python" in title or "编程" in title
# ========== Page Object模式 ==========
class LoginPage:
"""登录页面对象"""
def __init__(self, driver):
self.driver = driver
self.url = "https://shop.example.com/login"
def open(self):
self.driver.get(self.url)
return self
def enter_username(self, username):
self.driver.find_element(By.ID, "username").send_keys(username)
return self
def enter_password(self, password):
self.driver.find_element(By.ID, "password").send_keys(password)
return self
def click_login(self):
self.driver.find_element(By.ID, "login-btn").click()
return DashboardPage(self.driver)
class DashboardPage:
"""仪表盘页面对象"""
def __init__(self, driver):
self.driver = driver
def get_welcome_message(self):
return self.driver.find_element(By.CLASS_NAME, "welcome-message").text
# 使用Page Object的测试
def test_login_with_pom(driver):
"""使用页面对象模式测试"""
dashboard = LoginPage(driver).open() \
.enter_username("testuser") \
.enter_password("password123") \
.click_login()
assert "欢迎回来" in dashboard.get_welcome_message()
5.2 Playwright：下一代UI测试
# test_ui_playwright.py
import pytest
from playwright.sync_api import Page, expect
class TestWithPlaywright:
"""使用Playwright的现代UI测试"""
def test_login(self, page: Page):
"""测试登录"""
page.goto("https://shop.example.com/login")
# 填写表单
page.fill("[name='username']", "testuser")
page.fill("[name='password']", "password123")
# 点击登录
page.click("button[type='submit']")
# 验证跳转
expect(page).to_have_url("https://shop.example.com/dashboard")
expect(page.locator(".welcome-message")).to_contain_text("欢迎")
def test_api_mocking(self, page: Page):
"""模拟API响应"""
# 拦截并修改API响应
page.route("**/api/products", lambda route: route.fulfill(
status=200,
content_type="application/json",
body='[{"id": 1, "name": "Mocked Product", "price": 99.99}]'
))
page.goto("https://shop.example.com/products")
expect(page.locator(".product-name")).to_contain_text("Mocked Product")
def test_visual_regression(self, page: Page):
"""视觉回归测试"""
page.goto("https://shop.example.com")
# 截图对比
expect(page).to_have_screenshot("homepage.png")
def test_mobile_viewport(self, page: Page):
"""移动端适配测试"""
page.set_viewport_size({"width": 375, "height": 667})
page.goto("https://shop.example.com")
# 验证移动端菜单
expect(page.locator(".mobile-menu")).to_be_visible()
第六章：压力篇——性能与安全测试
6.1 性能测试：Locust实战
# locustfile.py
from locust import HttpUser, task, between, events
import random
import json
class WebsiteUser(HttpUser):
"""模拟真实用户行为"""
wait_time = between(1, 5) # 思考时间1-5秒
def on_start(self):
"""用户启动时登录"""
self.client.post("/api/login", json={
"username": f"user_{self.user_id}",
"password": "test123"
})
self.cart_items = []
@task(10)
def view_products(self):
"""浏览商品（权重10）"""
product_id = random.randint(1, 1000)
with self.client.get(
f"/api/products/{product_id}",
catch_response=True
) as response:
if response.status_code == 200:
data = response.json()
if "price" in data:
response.success()
else:
response.failure("Invalid response format")
elif response.status_code == 404:
response.success() # 商品不存在是正常情况
@task(5)
def search_products(self):
"""搜索商品（权重5）"""
keywords = ["Python", "Java", "手机", "电脑", "书籍"]
keyword = random.choice(keywords)
self.client.get(f"/api/search?q={keyword}", name="/api/search")
@task(3)
def add_to_cart(self):
"""添加购物车（权重3）"""
product_id = random.randint(1, 1000)
response = self.client.post("/api/cart", json={
"product_id": product_id,
"quantity": random.randint(1, 3)
})
if response.status_code == 201:
self.cart_items.append(product_id)
@task(1)
def checkout(self):
"""结账（权重1，较少执行）"""
if len(self.cart_items) > 0:
self.client.post("/api/orders", json={
"items": self.cart_items,
"payment_method": "credit_card"
})
self.cart_items = []
# 自定义事件监听
@events.request.add_listener
def on_request(request_type, name, response_time, response_length,
response, context, exception, **kwargs):
"""记录详细请求信息"""
if response_time > 1000: # 慢请求告警
print(f"Slow request: {name} took {response_time}ms")
@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
"""测试结束生成报告"""
print("性能测试完成，生成报告中...")
运行命令：
# 启动Locust
locust -f locustfile.py --host=https://api.example.com
# 分布式运行
locust -f locustfile.py --master
locust -f locustfile.py --worker --master-host=192.168.1.1
6.2 安全测试基础
# test_security.py
import pytest
import requests
from bandit import core as bandit_core
from bandit.core import manager as bandit_manager
class TestSecurity:
"""基础安全测试"""
def test_sql_injection(self):
"""SQL注入测试"""
malicious_inputs = [
"' OR '1'='1",
"'; DROP TABLE users; --",
"1' UNION SELECT * FROM passwords --"
]
for payload in malicious_inputs:
response = requests.get(
f"https://api.example.com/search?q={payload}"
)
# 验证没有SQL错误信息泄露
assert "SQL" not in response.text
assert "syntax" not in response.text.lower()
def test_xss_protection(self):
"""XSS防护测试"""
xss_payload = "<script>alert('xss')</script>"
response = requests.post(
"https://api.example.com/comments",
json={"content": xss_payload}
)
# 验证脚本被转义
assert "<script>" not in response.text
assert "<script>" in response.text or "script" not in response.text
def test_sensitive_data_exposure(self):
"""敏感信息泄露测试"""
response = requests.get("https://api.example.com/users/1")
data = response.json()
# 验证不返回敏感字段
assert "password" not in data
assert "ssn" not in data
assert "credit_card" not in data
def test_rate_limiting(self):
"""速率限制测试"""
# 快速发送请求
responses = []
for _ in range(150): # 超过限制
r = requests.get("https://api.example.com/data")
responses.append(r.status_code)
# 验证触发限流
assert 429 in responses # Too Many Requests
# 使用Bandit进行静态安全扫描
def test_bandit_scan():
"""Bandit安全扫描"""
mgr = bandit_manager.BanditManager(
bandit_core.CONFIG,
agg_type='file'
)
mgr.discover_files(['./src'], True)
mgr.run_tests()
issues = mgr.get_issue_list()
high_severity = [i for i in issues if i.severity == 'HIGH']
assert len(high_severity) == 0, f"发现高危安全问题: {high_severity}"
第七章：数据篇——测试数据管理与 fixtures
7.1 测试数据工厂模式
# factories.py
import factory
from factory import Faker
from datetime import datetime
from models import User, Order, Product
class UserFactory(factory.Factory):
"""用户数据工厂"""
class Meta:
model = User
id = factory.Sequence(lambda n: n)
username = factory.LazyAttribute(lambda obj: f"user_{obj.id}")
email = factory.LazyAttribute(lambda obj: f"{obj.username}@example.com")
created_at = factory.LazyFunction(datetime.now)
is_active = True
@factory.post_generation
def set_password(obj, create, extracted, **kwargs):
"""生成后处理：设置密码"""
obj.password_hash = hash_password(extracted or "defaultpass")
class ProductFactory(factory.Factory):
"""商品数据工厂"""
class Meta:
model = Product
name = Faker('catch_phrase')
description = Faker('text', max_nb_chars=200)
price = Faker('pydecimal', left_digits=3, right_digits=2, positive=True)
stock = Faker('random_int', min=0, max=1000)
category = Faker('random_element', elements=['Electronics', 'Books', 'Clothing'])
class OrderFactory(factory.Factory):
"""订单数据工厂"""
class Meta:
model = Order
user = factory.SubFactory(UserFactory)
created_at = factory.LazyFunction(datetime.now)
status = Faker('random_element', elements=['pending', 'paid', 'shipped'])
@factory.post_generation
def add_items(obj, create, extracted, **kwargs):
"""添加订单项"""
if create and extracted:
for _ in range(extracted):
OrderItemFactory(order=obj)
# 使用示例
@pytest.fixture
def sample_user():
return UserFactory()
@pytest.fixture
def premium_user():
return UserFactory(
username="premium_user",
email="premium@example.com"
)
@pytest.fixture
def bulk_products():
"""批量生成商品"""
return ProductFactory.create_batch(50)
7.2 数据库 fixtures 与事务管理
# conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from models import Base
# 内存数据库用于测试
TEST_DATABASE_URL = "sqlite:///:memory:"
@pytest.fixture(scope="session")
def engine():
"""创建数据库引擎"""
engine = create_engine(TEST_DATABASE_URL)
Base.metadata.create_all(engine)
yield engine
Base.metadata.drop_all(engine)
@pytest.fixture
def db_session(engine) -> Session:
"""每个测试的独立数据库会话"""
connection = engine.connect()
transaction = connection.begin()
session = sessionmaker(bind=connection)()
yield session
session.close()
transaction.rollback()
connection.close()
@pytest.fixture
def user_with_orders(db_session):
"""创建带订单的用户"""
user = UserFactory()
db_session.add(user)
db_session.flush()
orders = OrderFactory.create_batch(3, user=user)
db_session.add_all(orders)
db_session.commit()
return user
第八章：流水篇——持续集成与质量门禁
8.1 tox多环境测试
# tox.ini
[tox]
envlist = py38, py39, py310, py311, flake8, mypy
[testenv]
deps =
pytest
pytest-cov
pytest-xdist
-rrequirements.txt
commands =
pytest {posargs} --cov=src --cov-report=xml --cov-report=term
[testenv:flake8]
deps = flake8
commands = flake8 src tests
[testenv:mypy]
deps =
mypy
types-requests
commands = mypy src
[testenv:lint]
deps =
black
isort
ruff
commands =
black --check src tests
isort --check-only src tests
ruff check src tests
8.2 GitHub Actions 工作流
# .github/workflows/test.yml
name: Python CI
on:
push:
branches: [ main, develop ]
pull_request:
branches: [ main ]
jobs:
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ['3.8', '3.9', '3.10', '3.11']
services:
postgres:
image: postgres:13
env:
POSTGRES_PASSWORD: postgres
POSTGRES_DB: test_db
options: >-
--health-cmd pg_isready
--health-interval 10s
--health-timeout 5s
--health-retries 5
ports:
- 5432:5432
redis:
image: redis:6
ports:
- 6379:6379
steps:
- uses: actions/checkout@v3
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v4
with:
python-version: ${{ matrix.python-version }}
- name: Cache pip packages
uses: actions/cache@v3
with:
path: ~/.cache/pip
key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements.txt') }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install -r requirements-test.txt
- name: Lint with ruff
run: |
ruff check src tests
ruff format --check src tests
- name: Type check with mypy
run: mypy src
- name: Security scan with bandit
run: bandit -r src -f json -o bandit-report.json || true
- name: Test with pytest
run: |
pytest \
--cov=src \
--cov-report=xml \
--cov-report=term \
--cov-fail-under=80 \
-n auto \
--dist=loadfile
- name: Upload coverage to Codecov
uses: codecov/codecov-action@v3
with:
file: ./coverage.xml
fail_ci_if_error: true
integration-test:
needs: test
runs-on: ubuntu-latest
if: github.event_name == 'pull_request'
steps:
- uses: actions/checkout@v3
- name: Run integration tests
run: |
docker-compose -f docker-compose.test.yml up --abort-on-container-exit
8.3 质量门禁配置
# setup.cfg
[tool:pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
addopts =
-v
--tb=short
--strict-markers
--disable-warnings
--color=yes
markers =
slow: marks tests as slow (deselect with '-m "not slow"')
integration: marks tests as integration tests
unit: marks tests as unit tests
[coverage:run]
source = src
branch = True
omit =
*/tests/*
*/venv/*
*/migrations/*
[coverage:report]
exclude_lines =
pragma: no cover
def __repr__
raise AssertionError
raise NotImplementedError
if __name__ == .__main__.:
fail_under = 80
precision = 2
[mypy]
python_version = 3.9
warn_return_any = True
warn_unused_configs = True
disallow_untyped_defs = True
disallow_incomplete_defs = True
check_untyped_defs = True
disallow_untyped_decorators = True
no_implicit_optional = True
warn_redundant_casts = True
warn_unused_ignores = True
warn_no_return = True
warn_unreachable = True
strict_equality = True
第九章：化境篇——测试策略与质量文化
9.1 测试金字塔与策略选择
/\
/ \ E2E测试 (少量，关键路径)
/____\ 占比：10%
/ \
/ /\ \ 集成测试 (服务间交互)
/ / \ \ 占比：30%
/___/____\___\
/ \
/ Unit Tests \ 单元测试 (核心，快速反馈)
/__________________\ 占比：60%
关键原则：
1. 单元测试要"快"——执行时间<10ms，不依赖外部资源
2. 集成测试要"稳"——使用测试容器，保证环境一致性
3. E2E测试要"少"——只覆盖核心业务流程，避免脆弱性
9.2 测试驱动开发(TDD)实战
Red-Green-Refactor循环：
# Step 1: Red - 编写失败的测试
def test_calculate_discount():
calculator = PriceCalculator()
result = calculator.calculate(100, "VIP")
assert result == 80 # VIP 8折
# Step 2: Green - 最简实现
class PriceCalculator:
def calculate(self, price, customer_type):
if customer_type == "VIP":
return price * 0.8
return price
# Step 3: Refactor - 优化设计
from enum import Enum
from typing import Dict
class CustomerType(Enum):
REGULAR = 1
VIP = 2
PREMIUM = 3
class PriceCalculator:
DISCOUNT_RATES: Dict[CustomerType, float] = {
CustomerType.REGULAR: 1.0,
CustomerType.VIP: 0.8,
CustomerType.PREMIUM: 0.7,
}
def calculate(self, price: float, customer_type: CustomerType) -> float:
rate = self.DISCOUNT_RATES.get(customer_type, 1.0)
return price * rate
9.3 测试质量评估指标
| 指标 | 目标值 | 监控方式 |
|---|---|---|
| 代码覆盖率 | >80% | pytest-cov |
| 变异测试分数 | >70% | mutmut |
| 测试执行时间 | <5分钟 | CI流水线 |
| ** flaky test 比例** | <1% | 测试报告分析 |
| 缺陷逃逸率 | <5% | 生产缺陷回溯 |
9.4 测试文化建设的实践原则
- 测试即文档：好的测试用例应该清晰表达业务规则
- 测试即设计：难以测试的代码往往意味着设计缺陷
- 快速反馈：本地测试应在30秒内完成，提交前必须全绿
- 测试优先：修复Bug前先写复现测试，新功能必须伴随测试
- 集体所有权：测试代码与生产代码同等重要，共同维护
附录：工具链速查表
| 场景 | 推荐工具 | 安装命令 |
|---|---|---|
| 单元测试 | pytest | pip install pytest |
| 覆盖率 | pytest-cov | pip install pytest-cov |
| Mock | unittest.mock | 内置 |
| 参数化 | pytest | 内置 |
| 异步测试 | pytest-asyncio | pip install pytest-asyncio |
| HTTP测试 | responses / aioresponses | pip install responses |
| UI测试 | Playwright | pip install playwright |
| 性能测试 | Locust | pip install locust |
| 安全扫描 | bandit | pip install bandit |
| 数据工厂 | factory_boy | pip install factory_boy |
| 代码质量 | ruff / black / mypy | pip install ruff black mypy |
| CI/CD | tox / nox | pip install tox |
结语：自动化测试不是目的，而是快速交付高质量软件的手段。在Python生态中，我们拥有业界最成熟的测试工具链，关键在于建立测试优先的思维习惯，将质量内建于开发流程之中。愿每一位Python工程师都能写出既运行正确，又易于测试的优雅代码。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
6
0- 0
已为社区贡献67条内容
所有评论(0)