四、构建与依赖管理
4.1 依赖管理策略
# pyproject.toml - 完整的依赖管理配置
[project]
name = "myapp"
version = "1.0.0"
description = "My Application"
authors = [{name = "Team", email = "team@example.com"}]
license = {text = "MIT"}
readme = "README.md"
requires-python = ">=3.11,<3.12"
# 运行时依赖
dependencies = [
"fastapi>=0.100.0,<0.101.0",
"uvicorn[standard]>=0.23.0,<0.24.0",
"sqlalchemy>=2.0.0,<2.1.0",
"alembic>=1.11.0,<1.12.0",
"redis>=4.6.0,<5.0.0",
"celery>=5.3.0,<5.4.0",
"pydantic>=2.0.0,<3.0.0",
"pydantic-settings>=2.0.0,<3.0.0",
"python-jose[cryptography]>=3.3.0,<4.0.0",
"passlib[bcrypt]>=1.7.4,<2.0.0",
"httpx>=0.24.0,<0.25.0",
"python-multipart>=0.0.6",
]
# 开发依赖
[project.optional-dependencies]
dev = [
"pytest>=7.4.0,<8.0.0",
"pytest-cov>=4.1.0,<5.0.0",
"pytest-asyncio>=0.21.0,<0.22.0",
"black>=23.0.0,<24.0.0",
"ruff>=0.0.280",
"mypy>=1.4.0,<2.0.0",
"isort>=5.12.0,<6.0.0",
"pre-commit>=3.3.0,<4.0.0",
"ipython>=8.14.0,<9.0.0",
"watchfiles>=0.19.0",
]
# 测试依赖
test = [
"pytest>=7.4.0",
"pytest-cov>=4.1.0",
"pytest-asyncio>=0.21.0",
"pytest-mock>=3.11.0",
"factory-boy>=3.2.0",
"faker>=19.0.0",
]
# 生产依赖（排除开发依赖）
prod = [
"gunicorn>=21.0.0",
"prometheus-client>=0.17.0",
]
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"
[tool.poetry.group.dev.dependencies]
# 额外的开发依赖
pdbpp = "^0.10.3"
pytest-xdist = "^3.3.0"
4.2 依赖安全扫描
# .github/workflows/dependency-scan.yml
name: Dependency Security Scan
on:
schedule:
- cron: '0 0 * * *' # 每天运行
push:
branches: [main]
paths:
- 'pyproject.toml'
- 'poetry.lock'
jobs:
scan:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Setup Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
pip install poetry
poetry export -f requirements.txt --output requirements.txt
- name: Run safety check
uses: pyupio/safety-action@v1
with:
requirements-file: requirements.txt
safety-args: '--full-report'
- name: Run bandit security linter
run: |
pip install bandit
bandit -r src/ -f json -o bandit-report.json
- name: Upload security report
uses: actions/upload-artifact@v3
with:
name: security-reports
path: |
bandit-report.json
- name: Create security issue
if: failure()
uses: actions/github-script@v6
with:
script: |
github.rest.issues.create({
owner: context.repo.owner,
repo: context.repo.repo,
title: '⚠️ Security vulnerabilities detected in dependencies',
body: 'Please check the security scan report for details.',
labels: ['security', 'auto-generated']
})
4.3 构建优化
# scripts/build.py - 优化的构建脚本
import subprocess
import sys
import argparse
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor
import time
class BuildOptimizer:
"""构建优化器 - 并行构建、缓存利用"""
def __init__(self):
self.project_root = Path(__file__).parent.parent
self.dist_dir = self.project_root / "dist"
self.cache_dir = self.project_root / ".build-cache"
def parallel_lint(self, paths: list):
"""并行执行代码检查"""
with ThreadPoolExecutor(max_workers=4) as executor:
futures = []
# 并行运行black检查
futures.append(executor.submit(
subprocess.run,
["black", "--check"] + paths,
capture_output=True
))
# 并行运行ruff检查
futures.append(executor.submit(
subprocess.run,
["ruff", "check"] + paths,
capture_output=True
))
# 并行运行mypy
futures.append(executor.submit(
subprocess.run,
["mypy"] + paths,
capture_output=True
))
results = [f.result() for f in futures]
# 检查结果
all_passed = all(r.returncode == 0 for r in results)
return all_passed
def incremental_build(self, changed_files: list):
"""增量构建 - 只构建变更的部分"""
# 检测变更的模块
changed_modules = set()
for file in changed_files:
if file.startswith("src/"):
module = file.split("/")[1]
changed_modules.add(module)
# 只构建变更的模块
for module in changed_modules:
self.build_module(module)
def build_module(self, module_name: str):
"""构建单个模块"""
print(f"Building module: {module_name}")
# 构建逻辑...
def build(self, clean: bool = False, parallel: bool = True):
"""主构建函数"""
start_time = time.time()
if clean:
print("Cleaning previous build...")
subprocess.run(["rm", "-rf", str(self.dist_dir)])
# 1. 安装依赖（使用缓存）
print("Installing dependencies...")
subprocess.run(["poetry", "install", "--no-dev", "--no-interaction"])
# 2. 运行类型检查
print("Running type checks...")
subprocess.run(["mypy", "src/"])
# 3. 运行测试
print("Running tests...")
subprocess.run(["pytest", "--no-cov"])
# 4. 构建Docker镜像
print("Building Docker image...")
image_tag = f"myapp:{self.get_version()}"
subprocess.run([
"docker", "build",
"-t", image_tag,
"--cache-from", "myapp:latest",
"."
])
elapsed = time.time() - start_time
print(f"Build completed in {elapsed:.2f}s")
def get_version(self) -> str:
"""从pyproject.toml获取版本"""
import tomli
with open(self.project_root / "pyproject.toml", "rb") as f:
data = tomli.load(f)
return data["project"]["version"]
# 使用示例
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--clean", action="store_true", help="Clean before build")
parser.add_argument("--no-parallel", action="store_true", help="Disable parallel build")
args = parser.parse_args()
builder = BuildOptimizer()
builder.build(clean=args.clean, parallel=not args.no_parallel)
五、自动化测试体系
5.1 测试分层策略
# 测试金字塔
"""
/\
/ \
/ \
/ e2e \
/--------\
/ \
/ 集成测试 \
/-------------\
/ \
/ 单元测试 \
/---------------------\
测试比例建议:
- 单元测试: 70% (快速、稳定、细粒度)
- 集成测试: 20% (验证模块协作)
- E2E测试: 10% (验证关键用户流程)
"""
# tests/unit/test_order_calculator.py - 单元测试示例
import pytest
from decimal import Decimal
from src.domain.order import Order, OrderItem, Coupon
from src.domain.product import Product, ProductType
class TestOrderCalculator:
"""订单计算器单元测试"""
def test_calculate_subtotal_with_multiple_items(self):
"""测试：多个商品的小计计算"""
order = Order()
product1 = Product(id="p1", name="商品1", price=Decimal("10.00"), product_type=ProductType.PHYSICAL)
product2 = Product(id="p2", name="商品2", price=Decimal("25.50"), product_type=ProductType.PHYSICAL)
order.add_item(product1, 2)
order.add_item(product2, 1)
assert order.subtotal == Decimal("45.50") # 10*2 + 25.5
def test_apply_percentage_coupon(self):
"""测试：应用百分比优惠券"""
order = Order()
product = Product(id="p1", name="商品", price=Decimal("100.00"), product_type=ProductType.PHYSICAL)
order.add_item(product, 1)
coupon = Coupon(code="SAVE10", type="percentage", value=Decimal("10"))
order.apply_coupon(coupon)
assert order.total == Decimal("90.00")
def test_apply_fixed_coupon(self):
"""测试：应用固定金额优惠券"""
order = Order()
product = Product(id="p1", name="商品", price=Decimal("100.00"), product_type=ProductType.PHYSICAL)
order.add_item(product, 1)
coupon = Coupon(code="SAVE20", type="fixed", value=Decimal("20"))
order.apply_coupon(coupon)
assert order.total == Decimal("80.00")
def test_coupon_cannot_make_total_negative(self):
"""测试：优惠券不能使总价为负"""
order = Order()
product = Product(id="p1", name="商品", price=Decimal("10.00"), product_type=ProductType.PHYSICAL)
order.add_item(product, 1)
coupon = Coupon(code="SAVE100", type="fixed", value=Decimal("100"))
order.apply_coupon(coupon)
assert order.total == Decimal("0.00")
def test_cannot_add_item_to_paid_order(self):
"""测试：已支付的订单不能添加商品"""
order = Order()
product = Product(id="p1", name="商品", price=Decimal("10.00"), product_type=ProductType.PHYSICAL)
order.add_item(product, 1)
order.mark_as_paid()
with pytest.raises(OrderAlreadyProcessedError):
order.add_item(product, 1)
@pytest.mark.parametrize("quantity,expected", [
(1, Decimal("10.00")),
(5, Decimal("50.00")),
(0, None), # 0应该抛出异常
(-1, None), # 负数应该抛出异常
])
def test_add_item_with_various_quantities(self, quantity, expected):
"""测试：不同数量的商品添加（参数化测试）"""
if quantity <= 0:
with pytest.raises(InvalidQuantityError):
order = Order()
product = Product(id="p1", name="商品", price=Decimal("10.00"), product_type=ProductType.PHYSICAL)
order.add_item(product, quantity)
else:
order = Order()
product = Product(id="p1", name="商品", price=Decimal("10.00"), product_type=ProductType.PHYSICAL)
order.add_item(product, quantity)
assert order.subtotal == expected
# tests/integration/test_order_repository.py - 集成测试示例
import pytest
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from src.infrastructure.repositories.order_repository import OrderRepository
from src.domain.order import Order, OrderItem
from src.domain.product import Product
@pytest.mark.integration
class TestOrderRepository:
"""订单仓储集成测试（使用真实数据库）"""
@pytest.fixture
async def db_session(self):
"""创建测试数据库session"""
engine = create_async_engine("postgresql+asyncpg://test:test@localhost:5432/test_db")
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async with AsyncSession(engine) as session:
yield session
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
@pytest.fixture
def repository(self, db_session):
return OrderRepository(db_session)
async def test_save_and_find_order(self, repository, db_session):
"""测试：保存并查找订单"""
# 创建订单
order = Order(user_id="user_123")
product = Product(id="p1", name="商品", price=Decimal("10.00"))
order.add_item(product, 2)
# 保存
await repository.save(order)
await db_session.commit()
# 查找
found = await repository.find_by_id(order.id)
assert found is not None
assert found.id == order.id
assert found.user_id == "user_123"
assert len(found.items) == 1
assert found.items[0].quantity == 2
async def test_update_order_status(self, repository, db_session):
"""测试：更新订单状态"""
# 创建并保存订单
order = Order(user_id="user_123")
await repository.save(order)
await db_session.commit()
# 更新状态
order.mark_as_paid()
await repository.save(order)
await db_session.commit()
# 验证
found = await repository.find_by_id(order.id)
assert found.status == OrderStatus.PAID
assert found.paid_at is not None
# tests/e2e/test_user_journey.py - E2E测试示例
import pytest
from httpx import AsyncClient
from src.main import app
@pytest.mark.e2e
class TestUserJourney:
"""端到端测试 - 完整用户旅程"""
@pytest.fixture
async def client(self):
async with AsyncClient(app=app, base_url="http://test") as client:
yield client
async def test_complete_order_flow(self, client):
"""测试：完整的下单流程"""
# 1. 用户注册
register_response = await client.post("/api/auth/register", json={
"email": "test@example.com",
"password": "SecurePass123!",
"name": "Test User"
})
assert register_response.status_code == 201
user_id = register_response.json()["user_id"]
# 2. 用户登录
login_response = await client.post("/api/auth/login", json={
"email": "test@example.com",
"password": "SecurePass123!"
})
assert login_response.status_code == 200
token = login_response.json()["access_token"]
headers = {"Authorization": f"Bearer {token}"}
# 3. 浏览商品
products_response = await client.get("/api/products", headers=headers)
assert products_response.status_code == 200
products = products_response.json()
assert len(products) > 0
# 4. 添加商品到购物车
cart_response = await client.post("/api/cart/items", headers=headers, json={
"product_id": products[0]["id"],
"quantity": 2
})
assert cart_response.status_code == 200
# 5. 查看购物车
cart_response = await client.get("/api/cart", headers=headers)
assert cart_response.status_code == 200
cart = cart_response.json()
assert cart["total"] > 0
# 6. 创建订单
order_response = await client.post("/api/orders", headers=headers, json={
"shipping_address": {
"street": "123 Main St",
"city": "Beijing",
"zipcode": "100000"
}
})
assert order_response.status_code == 201
order = order_response.json()
order_id = order["order_id"]
# 7. 支付订单
payment_response = await client.post(f"/api/orders/{order_id}/pay", headers=headers, json={
"payment_method": "credit_card",
"card_number": "4242424242424242",
"expiry": "12/25",
"cvv": "123"
})
assert payment_response.status_code == 200
# 8. 查看订单状态
order_response = await client.get(f"/api/orders/{order_id}", headers=headers)
assert order_response.status_code == 200
assert order_response.json()["status"] == "paid"
# 9. 查看订单列表
orders_response = await client.get("/api/orders", headers=headers)
assert orders_response.status_code == 200
assert len(orders_response.json()) >= 1
5.2 测试数据工厂
# tests/factories.py - 测试数据工厂
import factory
from factory import Faker, SubFactory
from datetime import datetime, timedelta
from decimal import Decimal
import random
class ProductFactory(factory.Factory):
"""商品工厂"""
class Meta:
model = Product
id = factory.Sequence(lambda n: f"product_{n}")
name = Faker("word")
price = factory.LazyFunction(lambda: Decimal(str(random.uniform(1, 1000))).quantize(Decimal("0.01")))
product_type = factory.Iterator([ProductType.PHYSICAL, ProductType.DIGITAL])
stock = factory.Faker("random_int", min=0, max=1000)
class UserFactory(factory.Factory):
"""用户工厂"""
class Meta:
model = User
id = factory.Sequence(lambda n: f"user_{n}")
email = Faker("email")
name = Faker("name")
level = factory.Iterator(["normal", "vip", "svip"])
created_at = Faker("date_time_this_year")
class OrderItemFactory(factory.Factory):
"""订单项工厂"""
class Meta:
model = OrderItem
product = SubFactory(ProductFactory)
quantity = factory.Faker("random_int", min=1, max=10)
class OrderFactory(factory.Factory):
"""订单工厂 - 创建测试订单"""
class Meta:
model = Order
id = factory.Sequence(lambda n: f"order_{n}")
user_id = factory.Sequence(lambda n: f"user_{n}")
items = factory.List([SubFactory(OrderItemFactory) for _ in range(3)])
status = OrderStatus.PENDING
created_at = Faker("date_time_this_month")
@classmethod
def _create(cls, model_class, *args, **kwargs):
"""自定义创建逻辑"""
order = model_class(*args, **kwargs)
# 计算总价
order._total_amount = sum(item.subtotal for item in order.items)
return order
class OrderWithStatusFactory(OrderFactory):
"""带状态的订单工厂"""
status = OrderStatus.PAID
paid_at = factory.LazyFunction(lambda: datetime.now() - timedelta(hours=1))
# 使用示例
def test_with_factories():
# 创建单个订单
order = OrderFactory()
# 创建10个订单
orders = OrderFactory.create_batch(10)
# 创建已支付的订单
paid_order = OrderWithStatusFactory()
# 创建特定用户的订单
user_orders = OrderFactory.create_batch(5, user_id="specific_user")
5.3 测试覆盖率报告
# .coveragerc - 覆盖率配置
[run]
source = src
omit =
src/migrations/*
src/tests/*
*/__pycache__/*
*/test_*
[report]
exclude_lines =
pragma: no cover
def __repr__
if self.debug:
if __name__ == .__main__.:
raise NotImplementedError
pass
fail_under = 80
show_missing = True
skip_covered = False
[html]
directory = htmlcov
title = "MyApp Coverage Report"
