本文目录导读：
我来为你详细介绍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工具集成
你可以根据具体项目需求调整和扩展这些封装。
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
