百度360必应搜狗淘宝本站头条
当前位置:网站首页 > 技术文章 > 正文

Python3+pytest+allure接口自动化测试框架搭建

myzbx 2025-03-20 16:59 46 浏览

项目结构

安装依赖

pip install pytest requests pytest-html pytest-xdist jsonpath  

核心代码实现

配置文件 (config/config.py)

class Config:  
    # 环境配置  
    BASE_URL = "https://api.example.com"  
    TIMEOUT = 10  
    
    # 测试账号  
    USERNAME = "test_user"  
    PASSWORD = "test_password"  
    
    # 日志配置  
    LOG_LEVEL = "INFO"

API客户端 (api/client.py)

import requests  
import json  
import logging  
from config.config import Config  

class APIClient:  
    def __init__(self):  
        self.base_url = Config.BASE_URL  
        self.session = requests.Session()  
        self.timeout = Config.TIMEOUT  
        
    def get(self, url, params=None, headers=None):  
        return self._request("GET", url, params=params, headers=headers)  
        
    def post(self, url, data=None, json=None, headers=None):  
        return self._request("POST", url, data=data, json=json, headers=headers)  
        
    def put(self, url, data=None, json=None, headers=None):  
        return self._request("PUT", url, data=data, json=json, headers=headers)  
        
    def delete(self, url, headers=None):  
        return self._request("DELETE", url, headers=headers)  
        
    def _request(self, method, url, **kwargs):  
        full_url = self.base_url + url  
        logging.info(f"Request: {method} {full_url}")  
        
        if "headers" not in kwargs or kwargs["headers"] is None:  
            kwargs["headers"] = {"Content-Type": "application/json"}  
            
        if "timeout" not in kwargs:  
            kwargs["timeout"] = self.timeout  
            
        try:  
            response = self.session.request(method, full_url, **kwargs)  
            logging.info(f"Response status: {response.status_code}")  
            return response  
        except Exception as e:  
            logging.error(f"Request error: {e}")  
            raise
            

测试用例 (tests/test_api.py)

import pytest  
import json  
from api.client import APIClient  

class TestAPI:  
    @pytest.fixture(scope="class")  
    def api_client(self):  
        return APIClient()  
    
    def test_get_users(self, api_client):  
        """测试获取用户列表接口"""  
        response = api_client.get("/users")  
        assert response.status_code == 200  
        data = response.json()  
        assert "data" in data  
        assert isinstance(data["data"], list)  
    
    def test_create_user(self, api_client):  
        """测试创建用户接口"""  
        user_data = {  
            "name": "测试用户",  
            "email": "test@example.com",  
            "password": "password123"  
        }  
        response = api_client.post("/users", json=user_data)  
        assert response.status_code == 201  
        data = response.json()  
        assert data["name"] == user_data["name"]  
        assert data["email"] == user_data["email"]  
        
    def test_update_user(self, api_client):  
        """测试更新用户接口"""  
        # 假设我们已经知道用户ID为1  
        user_id = 1  
        update_data = {"name": "更新名称"}  
        response = api_client.put(f"/users/{user_id}", json=update_data)  
        assert response.status_code == 200  
        data = response.json()  
        assert data["name"] == update_data["name"]

pytest配置 (conftest.py)

import pytest  
import logging  
import os  
from datetime import datetime  

# 设置日志  
def setup_logging():  
    log_dir = "logs"  
    if not os.path.exists(log_dir):  
        os.makedirs(log_dir)  
    
    log_file = f"{log_dir}/test_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"  
    logging.basicConfig(  
        level=logging.INFO,  
        format="%(asctime)s - %(levelname)s - %(message)s",  
        handlers=[  
            logging.FileHandler(log_file),  
            logging.StreamHandler()  
        ]  
    )  

def pytest_configure(config):  
    setup_logging()  
    
@pytest.fixture(scope="session", autouse=True)  
def setup_teardown():  
    # 测试前置操作  
    logging.info("测试开始")  
    yield  
    # 测试后置操作  
    logging.info("测试结束")

测试运行脚本 (run.py)

import pytest  
import argparse  
import os  
from datetime import datetime  

def run_tests():  
    parser = argparse.ArgumentParser(description="运行API自动化测试")  
    parser.add_argument("--html", action="store_true", help="生成HTML报告")  
    parser.add_argument("--parallel", action="store_true", help="并行运行测试")  
    args = parser.parse_args()  
    
    # 创建报告目录  
    report_dir = "reports"  
    if not os.path.exists(report_dir):  
        os.makedirs(report_dir)  
    
    # 构建pytest参数  
    pytest_args = ["-v", "tests/"]  
    
    if args.html:  
        report_file = f"{report_dir}/report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.html"  
        pytest_args.extend(["--html", report_file, "--self-contained-html"])  
    
    if args.parallel:  
        pytest_args.extend(["-n", "auto"])  
    
    # 运行测试  
    pytest.main(pytest_args)  

if __name__ == "__main__":  
    run_tests()

执行测试

# 普通执行  
python run.py  

# 生成HTML报告  
python run.py --html  

# 并行执行测试  
python run.py --parallel  

# 同时生成报告和并行执行  
python run.py --html --parallel

添加Allure报告

更新依赖

pip install pytest requests pytest-html pytest-xdist jsonpath allure-pytest

安装Allure命令行工具Windows

scoop install allure

更新测试用例 (tests/test_api.py)

import pytest  
import json  
import allure  
from api.client import APIClient  

@allure.epic("API测试")  
@allure.feature("用户管理接口")  
class TestAPI:  
    @pytest.fixture(scope="class")  
    def api_client(self):  
        return APIClient()  
    
    @allure.story("获取用户列表")  
    @allure.title("测试获取所有用户")  
    @allure.severity(allure.severity_level.NORMAL)  
    def test_get_users(self, api_client):  
        """测试获取用户列表接口"""  
        with allure.step("发送GET请求到/users"):  
            response = api_client.get("/users")  
        
        with allure.step("验证响应状态码"):  
            assert response.status_code == 200  
        
        with allure.step("验证响应数据结构"):  
            data = response.json()  
            assert "data" in data  
            assert isinstance(data["data"], list)  
            
        allure.attach(json.dumps(data, indent=2),   
                     "响应数据",   
                     allure.attachment_type.JSON)  
    
    @allure.story("创建新用户")  
    @allure.title("测试创建用户功能")  
    @allure.severity(allure.severity_level.CRITICAL)  
    def test_create_user(self, api_client):  
        """测试创建用户接口"""  
        user_data = {  
            "name": "测试用户",  
            "email": "test@example.com",  
            "password": "password123"  
        }  
        
        allure.attach(json.dumps(user_data, indent=2),   
                     "请求数据",   
                     allure.attachment_type.JSON)  
        
        with allure.step("发送POST请求到/users"):  
            response = api_client.post("/users", json=user_data)  
        
        with allure.step("验证响应状态码"):  
            assert response.status_code == 201  
        
        with allure.step("验证创建的用户数据"):  
            data = response.json()  
            assert data["name"] == user_data["name"]  
            assert data["email"] == user_data["email"]  
            
        allure.attach(json.dumps(data, indent=2),   
                     "响应数据",   
                     allure.attachment_type.JSON)  
        
    @allure.story("更新用户")  
    @allure.title("测试更新用户信息")  
    @allure.severity(allure.severity_level.HIGH)  
    def test_update_user(self, api_client):  
        """测试更新用户接口"""  
        # 假设我们已经知道用户ID为1  
        user_id = 1  
        update_data = {"name": "更新名称"}  
        
        allure.attach(json.dumps(update_data, indent=2),   
                     "请求数据",   
                     allure.attachment_type.JSON)  
        
        with allure.step(f"发送PUT请求到/users/{user_id}"):  
            response = api_client.put(f"/users/{user_id}", json=update_data)  
        
        with allure.step("验证响应状态码"):  
            assert response.status_code == 200  
        
        with allure.step("验证更新后的用户数据"):  
            data = response.json()  
            assert data["name"] == update_data["name"]  
            
        allure.attach(json.dumps(data, indent=2),   
                     "响应数据",   
                     allure.attachment_type.JSON)

更新API客户端 (api/client.py)

import requests  
import json  
import logging  
import allure  
from config.config import Config  

class APIClient:  
    def __init__(self):  
        self.base_url = Config.BASE_URL  
        self.session = requests.Session()  
        self.timeout = Config.TIMEOUT  
        
    def get(self, url, params=None, headers=None):  
        return self._request("GET", url, params=params, headers=headers)  
        
    def post(self, url, data=None, json=None, headers=None):  
        return self._request("POST", url, data=data, json=json, headers=headers)  
        
    def put(self, url, data=None, json=None, headers=None):  
        return self._request("PUT", url, data=data, json=json, headers=headers)  
        
    def delete(self, url, headers=None):  
        return self._request("DELETE", url, headers=headers)  
        
    def _request(self, method, url, **kwargs):  
        full_url = self.base_url + url  
        logging.info(f"Request: {method} {full_url}")  
        
        # 记录请求信息到Allure  
        if "json" in kwargs and kwargs["json"] is not None:  
            allure.attach(json.dumps(kwargs["json"], indent=2, ensure_ascii=False),   
                         f"{method} {url} - Request Body",   
                         allure.attachment_type.JSON)  
        elif "data" in kwargs and kwargs["data"] is not None:  
            allure.attach(str(kwargs["data"]),   
                         f"{method} {url} - Request Body",   
                         allure.attachment_type.TEXT)  
        
        if "headers" not in kwargs or kwargs["headers"] is None:  
            kwargs["headers"] = {"Content-Type": "application/json"}  
            
        if "timeout" not in kwargs:  
            kwargs["timeout"] = self.timeout  
            
        try:  
            response = self.session.request(method, full_url, **kwargs)  
            logging.info(f"Response status: {response.status_code}")  
            
            # 记录响应信息到Allure  
            try:  
                response_body = response.json()  
                allure.attach(json.dumps(response_body, indent=2, ensure_ascii=False),   
                             f"{method} {url} - Response Body",   
                             allure.attachment_type.JSON)  
            except:  
                allure.attach(response.text,   
                             f"{method} {url} - Response Body",   
                             allure.attachment_type.TEXT)  
                
            return response  
        except Exception as e:  
            logging.error(f"Request error: {e}")  
            allure.attach(str(e),   
                         f"{method} {url} - Exception",   
                         allure.attachment_type.TEXT)  
            raise

更新conftest.py

import pytest  
import logging  
import os  
import allure  
from datetime import datetime  

# 设置日志  
def setup_logging():  
    log_dir = "logs"  
    if not os.path.exists(log_dir):  
        os.makedirs(log_dir)  
    
    log_file = f"{log_dir}/test_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"  
    logging.basicConfig(  
        level=logging.INFO,  
        format="%(asctime)s - %(levelname)s - %(message)s",  
        handlers=[  
            logging.FileHandler(log_file),  
            logging.StreamHandler()  
        ]  
    )  
    return log_file  

def pytest_configure(config):  
    log_file = setup_logging()  
    
    # 添加环境信息到Allure报告  
    allure.environment(  
        python_version=platform.python_version(),  
        pytest_version=pytest.__version__,  
        platform=platform.system(),  
        host=Config.BASE_URL  
    )  

@pytest.fixture(scope="session", autouse=True)  
def setup_teardown():  
    # 测试前置操作  
    logging.info("测试开始")  
    yield  
    # 测试后置操作  
    logging.info("测试结束")  

# 添加钩子函数记录测试结果    
@pytest.hookimpl(tryfirst=True, hookwrapper=True)  
def pytest_runtest_makereport(item, call):  
    outcome = yield  
    report = outcome.get_result()  
    
    if report.when == "call":  
        if report.failed:  
            # 失败时捕获页面截图或其他信息  
            allure.attach(  
                f"测试用例: {item.function.__name__} 失败",  
                "失败信息",  
                allure.attachment_type.TEXT  
            )

更新运行脚本 (run.py)

import pytest  
import argparse  
import os  
import shutil  
import subprocess  
from datetime import datetime  

def run_tests():  
    parser = argparse.ArgumentParser(description="运行API自动化测试")  
    parser.add_argument("--html", action="store_true", help="生成HTML报告")  
    parser.add_argument("--allure", action="store_true", help="生成Allure报告")  
    parser.add_argument("--parallel", action="store_true", help="并行运行测试")  
    parser.add_argument("--clean", action="store_true", help="清理报告目录")  
    args = parser.parse_args()  
    
    # 创建报告目录  
    report_dir = "reports"  
    allure_result_dir = "allure-results"  
    allure_report_dir = f"{report_dir}/allure-report"  
    
    if args.clean:  
        if os.path.exists(report_dir):  
            shutil.rmtree(report_dir)  
        if os.path.exists(allure_result_dir):  
            shutil.rmtree(allure_result_dir)  
    
    if not os.path.exists(report_dir):  
        os.makedirs(report_dir)  
    
    # 构建pytest参数  
    pytest_args = ["-v", "tests/"]  
    
    if args.html:  
        report_file = f"{report_dir}/report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.html"  
        pytest_args.extend(["--html", report_file, "--self-contained-html"])  
    
    if args.allure:  
        pytest_args.extend(["--alluredir", allure_result_dir])  
    
    if args.parallel:  
        pytest_args.extend(["-n", "auto"])  
    
    # 运行测试  
    pytest.main(pytest_args)  
    
    # 生成Allure报告  
    if args.allure:  
        # 创建Allure报告目录  
        if not os.path.exists(allure_report_dir):  
            os.makedirs(allure_report_dir)  
            
        print("正在生成Allure报告...")  
        try:  
            subprocess.run(["allure", "generate", allure_result_dir, "-o", allure_report_dir, "--clean"], check=True)  
            print(f"Allure报告已生成: {os.path.abspath(allure_report_dir)}")  
            
            # 打开Allure报告  
            print("正在打开Allure报告...")  
            subprocess.run(["allure", "open", allure_report_dir], check=True)  
        except subprocess.CalledProcessError as e:  
            print(f"生成Allure报告失败: {e}")  
        except FileNotFoundError:  
            print("未找到allure命令,请确保Allure已正确安装")  

if __name__ == "__main__":  
    run_tests()

添加缺少的导入到conftest.py中

import platform  
from config.config import Config

执行测试并生成Allure报告

# 清理历史报告并生成Allure报告  
python run.py --allure --clean  

# 同时生成HTML和Allure报告  
python run.py --html --allure  

# 并行执行测试并生成Allure报告  
python run.py --allure --parallel

相关推荐

任天堂Switch OLED:一块沉浸屏,点燃全家欢乐的游戏时光

在一个寻常的周末午后,客厅里弥漫着轻松惬意的氛围。电视屏幕暂时休眠,全家人的目光却聚焦在那台轻巧的掌上设备——任天堂SwitchOLED。父母与孩子挤在沙发上,指尖在Joy-Con手柄上跃动,时而因...

Switch是什么地区的版本?怎么分辨Switch普通版和续航版、OLED版

Switch有国行、港版、日版、美版、欧版、韩版,版本之间又有普通版和续航版的OLED区别。只要看掌机的序列号第二三位,第二位代表的机型,A是普通版,K是续航版,T是OLED版。第三位代表的是区域,比...

这款Switch手柄真别致!开箱体验八位堂Lite蓝牙手柄

【引言】Hi大家好,我是歌布林,今天给大家带来一款八位堂Lite蓝牙手柄的开箱报告!相信已经有不少Switch玩家已经入手了这款产品,讲真,这款手柄的颜值真的吸引到我了~双十一旗舰各大快递公...

switch 的性能提升了 3 倍,我只用了这一招

上一篇《if快还是switch快?解密switch背后的秘密》我们测试了if和switch的性能,得出了要尽量使用switch的结论,因为他的效率比if高很多,具体原因点击上文查看。既...

在对《Nintendo Switch 运动》充满期待的同时,我也有些许不安

任天堂「运动」系列的最新作《NintendoSwitch运动》即将于4月29日正式发售。在等待本作发售的日子里,笔者每一天都在掰着指头计算发售日还有几天才到来。在《NintendoSwi...

switch上最值得玩的五款JRPG(switch最值得玩的大作)

不知道为什么即使现在节奏变得这么快,人也变得浮躁,我依旧喜欢玩JPRG,喜欢沉浸入游戏的角色之中体验一场独特的冒险。JRPG的灵魂是什么?那绝对不是回合制战斗或者古旧的系统,而是令人深刻的剧情及在这漫...

那个 Vue 的路由,路由是干什么用的?

在Vue里,路由就像“页面导航的指挥官”,专门负责管理页面(组件)的切换和显示逻辑。简单来说,它能让单页应用(SPA)像多页应用一样实现“不同URL对应不同页面”的效果,但整个过程不会刷新网页。一、路...

S235JRH材50x50x3-5mm欧标方管疲劳强度分析

在工程与建筑领域,方管作为一种重要的结构材料,广泛应用于各种框架和支撑结构中。S235JRH材质的50x50x3-5mm欧标方管在疲劳强度方面的表现尤其受到关注。本文将以“解答常见误区”的形式,分析该...

纤维丛上的联络与曲率关系之二:七个联络

参阅一文看懂纤维丛(看图说话)纤维丛上的联络与曲率关系之一1.仿射联络(AffineConnection)定义:在光滑流形M的切丛TM上,一个仿射联络是一个双线性映射:仿射联络定义了向量...

安全完整性等级(SIL)分析报告编制与认证实践方法

以下是一篇关于安全完整性等级(SIL)分析报告的文章,涵盖SIL定级方法、验证流程、计算模型及工程实践。安全完整性等级(SIL)是量化安全仪表系统(SIS)性能的核心指标,由IEC61508/615...

欧标方管75x75x3mm8mm S355J0H力学性能测试

欧标方管75x75x3mmS355J0H是一种常见的钢管材料,广泛应用于建筑、机械制造和结构工程等领域。下面将对其力学性能进行详细测试和分析。1.材料概述S355J0H是欧洲标准EN10210中规定的...

p光与s光在SHG中的相位匹配中有哪些不同?

【第六期】p光与s光在SHG中的相位匹配中有哪些不同?在二次谐波生成(SHG)中,p偏振光和s偏振光的相位匹配条件存在显著差异。这些差异主要源于它们与纳米结构相互作用的方式以及它们在激发表面等离子体共...

S355JRH欧标方管机械性能及工程应用解析

S355JRH欧标方管是一种高强度钢管,以其优良的机械性能和良好的加工性在工程领域得到广泛应用。本文将从机械性能、场景应用等方面对S355JRH欧标方管进行解析,以帮助读者更好地理解其特性及应用。1....

《三国志·战棋版》PK7黄天蔽汉,四队共存作业来啦!

哈喽大家好啊,今天要给大家分享的是三棋PK7黄天蔽汉平时四队共存队,这只是其中一种选择供大家参考,后面可能还会分享其他的四队共存方案。队伍共存绝不唯一,大家有什么好的共存方案或者想看哪些队伍的共存也欢...

迈阿密三大巨星首秀被0-0逼平,豪阵为何难赢开罗国民

谁能想到,汇聚梅西、苏亚雷斯、布斯克茨这“银河战舰”级别三巨头,迈阿密国际世俱杯首秀居然0-0被埃及劲旅开罗国民逼平?全场90分钟,球迷的心情从“期待冠军”一路跌到“差点开门黑”。补时第96分钟,梅西...