本文目录导读：
我来为你详细介绍Python自动化测试工具的封装方法和实际案例。
基础测试框架封装
1 通用测试基类
# base_test.py
import unittest
import logging
import json
from datetime import datetime
from typing import Any, Dict, List, Optional
class BaseTest(unittest.TestCase):
"""测试基类"""
@classmethod
def setUpClass(cls):
"""类级别初始化"""
cls.logger = logging.getLogger(cls.__name__)
cls.start_time = datetime.now()
cls.test_results = []
def setUp(self):
"""用例级别初始化"""
self.logger.info(f"开始执行测试: {self._testMethodName}")
self.case_start_time = datetime.now()
def tearDown(self):
"""用例级别清理"""
duration = (datetime.now() - self.case_start_time).total_seconds()
result = self._outcome.result
if result.wasSuccessful():
status = "PASS"
else:
status = "FAIL" if result.failures else "ERROR"
self.test_results.append({
"case": self._testMethodName,
"status": status,
"duration": duration
})
self.logger.info(f"测试完成: {self._testMethodName} - {status} ({duration}s)")
@classmethod
def tearDownClass(cls):
"""类级别清理"""
total_duration = (datetime.now() - cls.start_time).total_seconds()
total = len(cls.test_results)
passed = sum(1 for r in cls.test_results if r['status'] == 'PASS')
cls.logger.info(f"""
========== 测试报告 ==========
总计: {total} | 通过: {passed} | 失败: {total - passed}
耗时: {total_duration:.2f}s
==============================
""")
API测试封装
1 HTTP请求封装
# api_client.py
import requests
import json
from typing import Dict, Optional, Any
from requests.exceptions import RequestException
import time
class APIClient:
"""API测试客户端封装"""
def __init__(self, base_url: str, timeout: int = 30):
self.base_url = base_url.rstrip('/')
self.timeout = timeout
self.session = requests.Session()
self.session.headers.update({
'Content-Type': 'application/json',
'Accept': 'application/json'
})
self.response_history = []
def set_headers(self, headers: Dict[str, str]):
"""设置请求头"""
self.session.headers.update(headers)
def set_auth_token(self, token: str):
"""设置认证令牌"""
self.session.headers.update({'Authorization': f'Bearer {token}'})
def request(self, method: str, endpoint: str, **kwargs) -> requests.Response:
"""通用请求方法"""
url = f"{self.base_url}/{endpoint.lstrip('/')}"
kwargs.setdefault('timeout', self.timeout)
try:
response = self.session.request(method, url, **kwargs)
self.response_history.append({
'method': method,
'url': url,
'status_code': response.status_code,
'timestamp': time.time()
})
return response
except RequestException as e:
raise APIClientError(f"请求失败: {str(e)}")
def get(self, endpoint: str, params: Optional[Dict] = None) -> requests.Response:
"""GET请求"""
return self.request('GET', endpoint, params=params)
def post(self, endpoint: str, data: Any = None, json_data: Any = None) -> requests.Response:
"""POST请求"""
return self.request('POST', endpoint, data=data, json=json_data)
def put(self, endpoint: str, data: Any = None, json_data: Any = None) -> requests.Response:
"""PUT请求"""
return self.request('PUT', endpoint, data=data, json=json_data)
def delete(self, endpoint: str) -> requests.Response:
"""DELETE请求"""
return self.request('DELETE', endpoint)
class APIClientError(Exception):
"""API客户端异常"""
pass
2 API测试用例封装
# api_test.py
from base_test import BaseTest
from api_client import APIClient
import json
class APITestCase(BaseTest):
"""API测试用例基类"""
@classmethod
def setUpClass(cls):
super().setUpClass()
# 从配置文件读取配置
cls.config = cls.load_config()
cls.client = APIClient(cls.config['base_url'])
# 如果需要进行认证
if 'auth_token' in cls.config:
cls.client.set_auth_token(cls.config['auth_token'])
@classmethod
def load_config(cls) -> dict:
"""加载配置文件"""
config_path = getattr(cls, 'config_path', 'config.json')
with open(config_path, 'r') as f:
return json.load(f)
def validate_response(self, response, expected_status: int = 200):
"""验证响应"""
self.assertEqual(response.status_code, expected_status,
f"期望状态码 {expected_status}, 实际 {response.status_code}")
return response.json()
def validate_json_schema(self, data: dict, schema: dict):
"""验证JSON Schema"""
# 这里可以使用 jsonschema 库
from jsonschema import validate
validate(instance=data, schema=schema)
UI自动化测试封装
1 WebDriver封装
# web_driver_manager.py
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.common.exceptions import TimeoutException, NoSuchElementException
from typing import Tuple, Optional, List
import logging
class WebDriverManager:
"""WebDriver管理器"""
def __init__(self, browser: str = 'chrome', headless: bool = False):
self.logger = logging.getLogger(self.__class__.__name__)
self.driver = self._init_driver(browser, headless)
self.wait = WebDriverWait(self.driver, 10)
def _init_driver(self, browser: str, headless: bool) -> webdriver.Remote:
"""初始化浏览器驱动"""
if browser.lower() == 'chrome':
options = webdriver.ChromeOptions()
if headless:
options.add_argument('--headless')
options.add_argument('--no-sandbox')
options.add_argument('--disable-dev-shm-usage')
return webdriver.Chrome(options=options)
elif browser.lower() == 'firefox':
options = webdriver.FirefoxOptions()
if headless:
options.add_argument('--headless')
return webdriver.Firefox(options=options)
else:
raise ValueError(f"不支持的浏览器: {browser}")
def navigate(self, url: str):
"""导航到URL"""
self.logger.info(f"导航到: {url}")
self.driver.get(url)
def find_element(self, by: By, value: str, timeout: int = 10) -> webdriver.Remote:
"""查找元素"""
try:
element = WebDriverWait(self.driver, timeout).until(
EC.presence_of_element_located((by, value))
)
return element
except TimeoutException:
self.logger.error(f"元素未找到: ({by}, {value})")
raise
def click_element(self, by: By, value: str):
"""点击元素"""
element = self.find_element(by, value)
element.click()
self.logger.info(f"点击元素: ({by}, {value})")
def input_text(self, by: By, value: str, text: str):
"""输入文本"""
element = self.find_element(by, value)
element.clear()
element.send_keys(text)
self.logger.info(f"输入文本到 ({by}, {value}): {text}")
def get_text(self, by: By, value: str) -> str:
"""获取文本"""
element = self.find_element(by, value)
return element.text
def take_screenshot(self, filepath: str):
"""截图"""
self.driver.save_screenshot(filepath)
self.logger.info(f"截图保存到: {filepath}")
def close(self):
"""关闭浏览器"""
if self.driver:
self.driver.quit()
2 Page Object模式封装
# page_objects/base_page.py
from web_driver_manager import WebDriverManager
from selenium.webdriver.common.by import By
from typing import Tuple
class BasePage:
"""页面对象基类"""
def __init__(self, driver_manager: WebDriverManager):
self.driver = driver_manager.driver
self.wait = driver_manager.wait
self.dm = driver_manager
def find_element(self, locator: Tuple[By, str]):
"""查找元素"""
return self.dm.find_element(*locator)
def click(self, locator: Tuple[By, str]):
"""点击操作"""
self.dm.click_element(*locator)
def input(self, locator: Tuple[By, str], text: str):
"""输入操作"""
self.dm.input_text(*locator, text)
def get_title(self) -> str:
"""获取页面标题"""
return self.driver.title
def get_current_url(self) -> str:
"""获取当前URL"""
return self.driver.current_url
# page_objects/login_page.py
class LoginPage(BasePage):
"""登录页面"""
# 定位器
USERNAME_INPUT = (By.ID, "username")
PASSWORD_INPUT = (By.ID, "password")
LOGIN_BUTTON = (By.ID, "login-btn")
ERROR_MESSAGE = (By.CLASS_NAME, "error-msg")
def login(self, username: str, password: str):
"""执行登录操作"""
self.input(self.USERNAME_INPUT, username)
self.input(self.PASSWORD_INPUT, password)
self.click(self.LOGIN_BUTTON)
def get_error_message(self) -> str:
"""获取错误信息"""
return self.find_element(self.ERROR_MESSAGE).text
数据驱动测试封装
1 数据提供器
# data_provider.py
import pandas as pd
import json
import yaml
from typing import List, Dict, Any
from pathlib import Path
class DataProvider:
"""测试数据提供器"""
@staticmethod
def from_csv(filepath: str) -> List[Dict[str, Any]]:
"""从CSV文件读取测试数据"""
df = pd.read_csv(filepath)
return df.to_dict('records')
@staticmethod
def from_json(filepath: str) -> List[Dict[str, Any]]:
"""从JSON文件读取测试数据"""
with open(filepath, 'r', encoding='utf-8') as f:
return json.load(f)
@staticmethod
def from_yaml(filepath: str) -> List[Dict[str, Any]]:
"""从YAML文件读取测试数据"""
with open(filepath, 'r', encoding='utf-8') as f:
return yaml.safe_load(f)
@staticmethod
def from_excel(filepath: str, sheet_name: str = 'Sheet1') -> List[Dict[str, Any]]:
"""从Excel文件读取测试数据"""
df = pd.read_excel(filepath, sheet_name=sheet_name)
return df.to_dict('records')
2 参数化测试
# parameterized_test.py
from base_test import BaseTest
from data_provider import DataProvider
from ddt import ddt, data, unpack
import json
@ddt
class ParameterizedTest(BaseTest):
"""参数化测试示例"""
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.data_provider = DataProvider()
@data(*DataProvider.from_csv('test_data.csv'))
@unpack
def test_login(self, username, password, expected):
"""数据驱动的登录测试"""
# 这里执行实际的登录测试
result = self.login_test(username, password)
self.assertEqual(result, expected)
def login_test(self, username, password):
"""模拟登录测试"""
# 实际测试逻辑
if username == "admin" and password == "password123":
return "success"
return "failed"
测试报告生成
1 HTML报告生成器
# report_generator.py
import json
import time
from datetime import datetime
from typing import List, Dict, Any
from pathlib import Path
import jinja2
class TestReportGenerator:
"""测试报告生成器"""
def __init__(self, output_dir: str = "reports"):
self.output_dir = Path(output_dir)
self.output_dir.mkdir(exist_ok=True)
def generate_html_report(self, test_results: List[Dict[str, Any]],
template_file: str = "report_template.html"):
"""生成HTML测试报告"""
# 准备报告数据
total_tests = len(test_results)
passed_tests = sum(1 for r in test_results if r['status'] == 'PASS')
failed_tests = total_tests - passed_tests
pass_rate = (passed_tests / total_tests * 100) if total_tests > 0 else 0
report_data = {
"title": "自动化测试报告",
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"total": total_tests,
"passed": passed_tests,
"failed": failed_tests,
"pass_rate": f"{pass_rate:.2f}%",
"duration": sum(r.get('duration', 0) for r in test_results),
"test_cases": test_results
}
# 渲染模板
template_loader = jinja2.FileSystemLoader(searchpath="./templates")
template_env = jinja2.Environment(loader=template_loader)
template = template_env.get_template(template_file)
html_content = template.render(report_data)
# 保存报告
report_path = self.output_dir / f"test_report_{int(time.time())}.html"
with open(report_path, 'w', encoding='utf-8') as f:
f.write(html_content)
return report_path
完整测试用例示例
# test_suite.py
import unittest
import json
from api_test import APITestCase
from api_client import APIClient
from data_provider import DataProvider
from report_generator import TestReportGenerator
class TestUserAPI(APITestCase):
"""用户API测试"""
config_path = "config/api_config.json"
def test_create_user(self):
"""测试创建用户"""
user_data = {
"username": "testuser1",
"email": "test@example.com",
"password": "Test123!"
}
response = self.client.post("/api/users", json=user_data)
data = self.validate_response(response, 201)
self.assertIn("id", data)
self.assertEqual(data["username"], user_data["username"])
def test_get_user(self):
"""测试获取用户"""
user_id = 1
response = self.client.get(f"/api/users/{user_id}")
data = self.validate_response(response, 200)
self.assertEqual(data["id"], user_id)
def test_get_user_not_found(self):
"""测试获取不存在的用户"""
response = self.client.get("/api/users/99999")
self.assertEqual(response.status_code, 404)
def test_update_user(self):
"""测试更新用户"""
user_id = 1
update_data = {"email": "updated@example.com"}
response = self.client.put(f"/api/users/{user_id}", json=update_data)
data = self.validate_response(response, 200)
self.assertEqual(data["email"], update_data["email"])
def test_delete_user(self):
"""测试删除用户"""
user_id = 1
response = self.client.delete(f"/api/users/{user_id}")
self.validate_response(response, 204)
if __name__ == "__main__":
# 配置日志
import logging
logging.basicConfig(level=logging.INFO)
# 运行测试
suite = unittest.TestLoader().loadTestsFromTestCase(TestUserAPI)
runner = unittest.TextTestRunner(verbosity=2)
result = runner.run(suite)
# 生成报告
generator = TestReportGenerator()
test_results = [
{"case": test._testMethodName, "status": "PASS" if result.wasSuccessful() else "FAIL"}
for test in suite._tests[0]._tests # 简化处理
]
report_path = generator.generate_html_report(test_results)
print(f"测试报告已生成: {report_path}")
配置文件示例
config/api_config.json
{
"base_url": "https://api.example.com",
"auth_token": "your_auth_token_here",
"timeout": 30,
"headers": {
"X-API-Key": "your_api_key"
}
}
test_data.csv
username,password,expected admin,password123,success user1,wrongpass,failed testuser,test123,success
目录结构
自动化测试项目/
├── config/
│ ├── api_config.json
│ └── test_config.yaml
├── data/
│ ├── test_data.csv
│ └── test_data.json
├── page_objects/
│ ├── __init__.py
│ ├── base_page.py
│ └── login_page.py
├── reports/
│ └── test_report_template.html
├── templates/
│ └── report_template.html
├── tests/
│ ├── __init__.py
│ ├── test_api.py
│ ├── test_ui.py
│ └── test_data_driven.py
├── utils/
│ ├── __init__.py
│ ├── api_client.py
│ ├── base_test.py
│ ├── data_provider.py
│ ├── report_generator.py
│ └── web_driver_manager.py
├── requirements.txt
└── run_tests.py这样封装的自动化测试框架具有以下特点：
- 模块化设计：各组件职责清晰，易于维护
- 可扩展性：易于添加新的测试场景和功能
- 数据驱动：支持多种数据源
- 报告生成：自动生成测试报告
- 代码复用：通过基类和工具类减少重复代码
- 易于集成：可以与CI/CD工具集成
你可以根据具体项目需求调整和扩展这些封装。
