接口自动化框架架构设计
大约 9 分钟
接口自动化框架架构设计
前言：从"能用"到"好用"的架构进化
还记得我第一次搭建自动化测试框架时，就是把所有代码都写在一个文件里，测试数据、业务逻辑、断言验证全部混在一起。当时觉得"能跑就行"，结果随着项目的发展，这个框架变成了一个"巨无霸"，维护起来简直是噩梦。
后来经过多个项目的历练，我逐渐明白了一个道理：好的架构不是一开始就设计出来的，而是在不断的重构和优化中演进出来的。今天，我就来分享一下如何设计一个既灵活又稳定的接口自动化测试框架。
框架设计原则
1. SOLID原则在测试框架中的应用
单一职责原则（SRP）：每个类只负责一个功能
# 错误示例：一个类承担太多职责
class TestHelper:
def send_request(self):
pass
def validate_response(self):
pass
def generate_report(self):
pass
def manage_test_data(self):
pass
# 正确示例：职责分离
class HTTPClient:
def send_request(self):
pass
class ResponseValidator:
def validate_response(self):
pass
class ReportGenerator:
def generate_report(self):
pass
class TestDataManager:
def manage_test_data(self):
pass开闭原则（OCP）：对扩展开放，对修改关闭
from abc import ABC, abstractmethod
class BaseAssertion(ABC):
@abstractmethod
def assert_response(self, response, expected):
pass
class JSONAssertion(BaseAssertion):
def assert_response(self, response, expected):
# JSON响应断言逻辑
pass
class XMLAssertion(BaseAssertion):
def assert_response(self, response, expected):
# XML响应断言逻辑
pass
# 新增断言类型时，不需要修改现有代码
class HTMLAssertion(BaseAssertion):
def assert_response(self, response, expected):
# HTML响应断言逻辑
pass2. 分层架构设计
测试框架分层架构
├── 测试用例层 (Test Cases Layer)
│ ├── 业务测试用例
│ └── 数据驱动测试
├── 业务逻辑层 (Business Logic Layer)
│ ├── 业务操作封装
│ └── 工作流程定义
├── 服务接口层 (Service Interface Layer)
│ ├── API客户端
│ └── 接口封装
├── 基础设施层 (Infrastructure Layer)
│ ├── HTTP客户端
│ ├── 数据库连接
│ ├── 配置管理
│ └── 日志系统
└── 工具支撑层 (Utility Layer)
├── 断言工具
├── 数据生成器
├── 报告生成器
└── 通用工具核心组件设计
1. HTTP客户端层
from abc import ABC, abstractmethod
import requests
from typing import Dict, Any, Optional
class HTTPClientInterface(ABC):
"""HTTP客户端接口"""
@abstractmethod
def get(self, url: str, **kwargs) -> 'Response':
pass
@abstractmethod
def post(self, url: str, **kwargs) -> 'Response':
pass
@abstractmethod
def put(self, url: str, **kwargs) -> 'Response':
pass
@abstractmethod
def delete(self, url: str, **kwargs) -> 'Response':
pass
class RequestsHTTPClient(HTTPClientInterface):
"""基于requests的HTTP客户端实现"""
def __init__(self, base_url: str = "", timeout: int = 30):
self.base_url = base_url.rstrip('/')
self.timeout = timeout
self.session = requests.Session()
self._setup_session()
def _setup_session(self):
"""配置session"""
self.session.headers.update({
'User-Agent': 'AutoTestFramework/1.0',
'Accept': 'application/json'
})
def _build_url(self, url: str) -> str:
"""构建完整URL"""
if url.startswith('http'):
return url
return f"{self.base_url}/{url.lstrip('/')}"
def get(self, url: str, **kwargs) -> requests.Response:
kwargs.setdefault('timeout', self.timeout)
return self.session.get(self._build_url(url), **kwargs)
def post(self, url: str, **kwargs) -> requests.Response:
kwargs.setdefault('timeout', self.timeout)
return self.session.post(self._build_url(url), **kwargs)
def put(self, url: str, **kwargs) -> requests.Response:
kwargs.setdefault('timeout', self.timeout)
return self.session.put(self._build_url(url), **kwargs)
def delete(self, url: str, **kwargs) -> requests.Response:
kwargs.setdefault('timeout', self.timeout)
return self.session.delete(self._build_url(url), **kwargs)
class HTTPClientFactory:
"""HTTP客户端工厂"""
@staticmethod
def create_client(client_type: str = "requests", **kwargs) -> HTTPClientInterface:
if client_type == "requests":
return RequestsHTTPClient(**kwargs)
# 可以扩展其他HTTP客户端实现
# elif client_type == "httpx":
# return HttpxHTTPClient(**kwargs)
else:
raise ValueError(f"不支持的客户端类型: {client_type}")2. 配置管理层
import os
import yaml
from dataclasses import dataclass
from typing import Dict, Any
from pathlib import Path
@dataclass
class DatabaseConfig:
host: str
port: int
username: str
password: str
database: str
@dataclass
class APIConfig:
base_url: str
timeout: int
retry_count: int
verify_ssl: bool
@dataclass
class TestConfig:
environment: str
api: APIConfig
database: DatabaseConfig
debug: bool
log_level: str
class ConfigManager:
"""配置管理器"""
def __init__(self, config_dir: str = "config"):
self.config_dir = Path(config_dir)
self._config_cache = {}
def load_config(self, env: str = None) -> TestConfig:
"""加载配置"""
env = env or os.getenv('TEST_ENV', 'dev')
if env in self._config_cache:
return self._config_cache[env]
config_file = self.config_dir / f"{env}.yaml"
if not config_file.exists():
raise FileNotFoundError(f"配置文件不存在: {config_file}")
with open(config_file, 'r', encoding='utf-8') as f:
config_data = yaml.safe_load(f)
# 解析配置
api_config = APIConfig(**config_data['api'])
db_config = DatabaseConfig(**config_data['database'])
test_config = TestConfig(
environment=env,
api=api_config,
database=db_config,
debug=config_data.get('debug', False),
log_level=config_data.get('log_level', 'INFO')
)
self._config_cache[env] = test_config
return test_config
def get_config(self) -> TestConfig:
"""获取当前环境配置"""
return self.load_config()
# 全局配置实例
config_manager = ConfigManager()3. 日志系统
import logging
import sys
from pathlib import Path
from datetime import datetime
class LoggerManager:
"""日志管理器"""
def __init__(self):
self._loggers = {}
def get_logger(self, name: str = "test_framework") -> logging.Logger:
"""获取日志器"""
if name in self._loggers:
return self._loggers[name]
logger = logging.getLogger(name)
logger.setLevel(logging.DEBUG)
# 避免重复添加handler
if not logger.handlers:
self._setup_handlers(logger)
self._loggers[name] = logger
return logger
def _setup_handlers(self, logger: logging.Logger):
"""设置日志处理器"""
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# 控制台处理器
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(logging.INFO)
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
# 文件处理器
log_dir = Path("logs")
log_dir.mkdir(exist_ok=True)
file_handler = logging.FileHandler(
log_dir / f"test_{datetime.now().strftime('%Y%m%d')}.log",
encoding='utf-8'
)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
# 全局日志管理器
logger_manager = LoggerManager()4. 断言增强层
from typing import Any, Dict, List, Union
import jsonpath
import re
class AssertionError(Exception):
"""自定义断言异常"""
pass
class ResponseAssertion:
"""响应断言类"""
def __init__(self, response):
self.response = response
self.logger = logger_manager.get_logger("assertion")
def assert_status_code(self, expected_code: int) -> 'ResponseAssertion':
"""断言状态码"""
actual_code = self.response.status_code
if actual_code != expected_code:
self.logger.error(f"状态码断言失败: 期望{expected_code}, 实际{actual_code}")
raise AssertionError(f"期望状态码{expected_code}, 实际{actual_code}")
self.logger.info(f"状态码断言成功: {actual_code}")
return self
def assert_json_path(self, json_path: str, expected_value: Any) -> 'ResponseAssertion':
"""断言JSON路径值"""
try:
json_data = self.response.json()
except ValueError:
raise AssertionError("响应不是有效的JSON格式")
actual_values = jsonpath.jsonpath(json_data, json_path)
if not actual_values:
raise AssertionError(f"JSON路径 {json_path} 未找到")
actual_value = actual_values[0]
if actual_value != expected_value:
self.logger.error(f"JSON路径断言失败: 路径{json_path}, 期望{expected_value}, 实际{actual_value}")
raise AssertionError(f"路径 {json_path} 期望值 {expected_value}, 实际值 {actual_value}")
self.logger.info(f"JSON路径断言成功: {json_path} = {actual_value}")
return self
def assert_response_time(self, max_seconds: float) -> 'ResponseAssertion':
"""断言响应时间"""
actual_time = self.response.elapsed.total_seconds()
if actual_time > max_seconds:
self.logger.error(f"响应时间断言失败: 期望<{max_seconds}s, 实际{actual_time}s")
raise AssertionError(f"响应时间 {actual_time}s 超过限制 {max_seconds}s")
self.logger.info(f"响应时间断言成功: {actual_time}s")
return self
def assert_contains(self, text: str) -> 'ResponseAssertion':
"""断言响应包含指定文本"""
response_text = self.response.text
if text not in response_text:
self.logger.error(f"包含文本断言失败: 响应中未找到 '{text}'")
raise AssertionError(f"响应中未找到文本: {text}")
self.logger.info(f"包含文本断言成功: 找到 '{text}'")
return self
def assert_regex_match(self, pattern: str) -> 'ResponseAssertion':
"""断言响应匹配正则表达式"""
response_text = self.response.text
if not re.search(pattern, response_text):
self.logger.error(f"正则匹配断言失败: 模式 '{pattern}' 未匹配")
raise AssertionError(f"响应未匹配正则表达式: {pattern}")
self.logger.info(f"正则匹配断言成功: 模式 '{pattern}'")
return self
def assert_response(response) -> ResponseAssertion:
"""创建响应断言对象"""
return ResponseAssertion(response)5. 业务服务层
from typing import Dict, Any, Optional
class BaseService:
"""基础服务类"""
def __init__(self, http_client: HTTPClientInterface):
self.http_client = http_client
self.logger = logger_manager.get_logger(self.__class__.__name__)
def _log_request(self, method: str, url: str, **kwargs):
"""记录请求日志"""
self.logger.info(f"发送请求: {method.upper()} {url}")
if 'json' in kwargs:
self.logger.debug(f"请求体: {kwargs['json']}")
def _log_response(self, response):
"""记录响应日志"""
self.logger.info(f"收到响应: {response.status_code}")
self.logger.debug(f"响应体: {response.text[:500]}...")
class UserService(BaseService):
"""用户服务"""
def create_user(self, user_data: Dict[str, Any]) -> Dict[str, Any]:
"""创建用户"""
self._log_request("POST", "/api/users", json=user_data)
response = self.http_client.post("/api/users", json=user_data)
self._log_response(response)
assert_response(response).assert_status_code(201)
return response.json()
def get_user(self, user_id: int) -> Dict[str, Any]:
"""获取用户信息"""
self._log_request("GET", f"/api/users/{user_id}")
response = self.http_client.get(f"/api/users/{user_id}")
self._log_response(response)
assert_response(response).assert_status_code(200)
return response.json()
def update_user(self, user_id: int, user_data: Dict[str, Any]) -> Dict[str, Any]:
"""更新用户信息"""
self._log_request("PUT", f"/api/users/{user_id}", json=user_data)
response = self.http_client.put(f"/api/users/{user_id}", json=user_data)
self._log_response(response)
assert_response(response).assert_status_code(200)
return response.json()
def delete_user(self, user_id: int) -> bool:
"""删除用户"""
self._log_request("DELETE", f"/api/users/{user_id}")
response = self.http_client.delete(f"/api/users/{user_id}")
self._log_response(response)
assert_response(response).assert_status_code(204)
return True
def login(self, username: str, password: str) -> Dict[str, Any]:
"""用户登录"""
login_data = {"username": username, "password": password}
self._log_request("POST", "/api/auth/login", json=login_data)
response = self.http_client.post("/api/auth/login", json=login_data)
self._log_response(response)
assert_response(response).assert_status_code(200)
return response.json()
class ProductService(BaseService):
"""商品服务"""
def create_product(self, product_data: Dict[str, Any]) -> Dict[str, Any]:
"""创建商品"""
self._log_request("POST", "/api/products", json=product_data)
response = self.http_client.post("/api/products", json=product_data)
self._log_response(response)
assert_response(response).assert_status_code(201)
return response.json()
def search_products(self, keyword: str, page: int = 1, size: int = 10) -> Dict[str, Any]:
"""搜索商品"""
params = {"keyword": keyword, "page": page, "size": size}
self._log_request("GET", "/api/products/search", params=params)
response = self.http_client.get("/api/products/search", params=params)
self._log_response(response)
assert_response(response).assert_status_code(200)
return response.json()框架集成与使用
1. 服务工厂
class ServiceFactory:
"""服务工厂"""
def __init__(self, config: TestConfig):
self.config = config
self.http_client = HTTPClientFactory.create_client(
base_url=config.api.base_url,
timeout=config.api.timeout
)
def create_user_service(self) -> UserService:
"""创建用户服务"""
return UserService(self.http_client)
def create_product_service(self) -> ProductService:
"""创建商品服务"""
return ProductService(self.http_client)
def authenticate(self, username: str, password: str):
"""认证用户"""
user_service = self.create_user_service()
auth_result = user_service.login(username, password)
# 设置认证头
token = auth_result.get('token')
if token:
self.http_client.session.headers.update({
'Authorization': f'Bearer {token}'
})
return auth_result
# 全局服务工厂
service_factory = None
def get_service_factory() -> ServiceFactory:
"""获取服务工厂实例"""
global service_factory
if service_factory is None:
config = config_manager.get_config()
service_factory = ServiceFactory(config)
return service_factory2. 测试基类
import pytest
class BaseTestCase:
"""测试基类"""
@pytest.fixture(scope="class", autouse=True)
def setup_class(self):
"""类级别设置"""
self.config = config_manager.get_config()
self.service_factory = get_service_factory()
self.logger = logger_manager.get_logger(self.__class__.__name__)
@pytest.fixture(autouse=True)
def setup_method(self):
"""方法级别设置"""
self.logger.info(f"开始执行测试: {self._pytest_current_test}")
yield
self.logger.info(f"测试执行完成: {self._pytest_current_test}")
def get_user_service(self) -> UserService:
"""获取用户服务"""
return self.service_factory.create_user_service()
def get_product_service(self) -> ProductService:
"""获取商品服务"""
return self.service_factory.create_product_service()
class AuthenticatedTestCase(BaseTestCase):
"""需要认证的测试基类"""
@pytest.fixture(scope="class", autouse=True)
def authenticate(self):
"""自动认证"""
self.service_factory.authenticate("admin", "password")3. 实际测试用例
import pytest
import allure
@allure.epic("用户管理")
@allure.feature("用户CRUD操作")
class TestUserCRUD(AuthenticatedTestCase):
"""用户CRUD操作测试"""
@allure.story("创建用户")
@allure.severity(allure.severity_level.CRITICAL)
def test_create_user(self):
"""测试创建用户"""
user_service = self.get_user_service()
user_data = {
"username": "testuser",
"email": "testuser@example.com",
"name": "测试用户"
}
with allure.step("创建用户"):
result = user_service.create_user(user_data)
with allure.step("验证创建结果"):
assert result["username"] == user_data["username"]
assert result["email"] == user_data["email"]
assert "id" in result
@allure.story("查询用户")
@allure.severity(allure.severity_level.NORMAL)
def test_get_user(self):
"""测试查询用户"""
user_service = self.get_user_service()
# 先创建一个用户
user_data = {"username": "queryuser", "email": "query@example.com"}
created_user = user_service.create_user(user_data)
user_id = created_user["id"]
with allure.step("查询用户"):
result = user_service.get_user(user_id)
with allure.step("验证查询结果"):
assert result["id"] == user_id
assert result["username"] == user_data["username"]
@allure.epic("商品管理")
@allure.feature("商品搜索")
class TestProductSearch(BaseTestCase):
"""商品搜索测试"""
@allure.story("关键词搜索")
@allure.severity(allure.severity_level.NORMAL)
def test_search_products(self):
"""测试商品搜索"""
product_service = self.get_product_service()
with allure.step("搜索商品"):
result = product_service.search_products("手机")
with allure.step("验证搜索结果"):
assert "products" in result
assert "total" in result
assert isinstance(result["products"], list)总结
一个好的接口自动化测试框架应该具备以下特点：
架构清晰：分层设计，职责明确 易于扩展：遵循开闭原则，支持插件化 配置灵活：支持多环境，配置外部化 日志完善：详细的日志记录，便于问题定位 断言丰富：提供多种断言方式，支持链式调用 服务封装：业务逻辑封装，提高代码复用性
通过合理的架构设计，我们可以构建出既稳定又灵活的测试框架，为团队的自动化测试提供强有力的支撑。
下一篇文章，我们将探讨如何将这个框架集成到CI/CD流水线中，实现持续测试，敬请期待！