Pytest接口自动化测试实战演练

news2024/11/26 10:50:48

结合单元测试框架pytest+数据驱动模型+allure

目录

api: 存储测试接口

conftest.py :设置前置操作

             目前前置操作:1、获取token并传入headers,2、获取命令行参数给到环境变量,指定运行环境

commmon:存储封装的公共方法

         connect_mysql.py:连接数据库
         http_requests.py: 封装自己的请求方法
         logger.py: 封装输出日志文件
         read_yaml.py:读取yaml文件测试用例数据
         read_save_data.py:读取保存的数据文件

case: 存放所有的测试用例
      
data:存放测试需要的数据
      save_data: 存放接口返回数据、接口下载文件
      test_data: 存放测试用例依赖数据
      upload_data: 存放上传接口文件

logs: 存放输出的日志文件

report: 存放测试输出报告

getpathinfo.py :封装项目测试路径

pytest.int :配置文件

requirement.txt: 本地python包(pip install -r requirements.txt 安装项目中所有python包)

run_main.py: 项目运行文件

结构设计

1.每一个接口用例组合在一个测试类里面生成一个py文件

2.将每个用例调用的接口封装在一个测试类里面生成一个py文件

3.将测试数据存放在yml文件中通过parametrize进行参数化,实现数据驱动

4.通过allure生成测试报告

代码展示

api/api_service.py #需要测试的一类接口

'''
Code description:服务相关接口
Create time: 2020/12/3
Developer: 叶修
'''
import os
from common.http_requests import HttpRequests


class Api_Auth_Service(object):

    def __init__(self):
        self.headers = HttpRequests().headers

    def api_home_service_list(self):
        # 首页服务列表
        # url = "http://192.168.2.199:9092/v1/auth/auth_service/findAuthService"
        url = os.environ["host"] + "/v1/auth/auth_service/findAuthService"  # 读取conftest.py文件地址进行拼接
        response = HttpRequests().get(url, headers=self.headers, verify=False)
        # print(response.json())
        return response

    def get_service_id(self):
        #获取银行卡三要素认证服务id
        url = "http://192.168.2.199:9092/v1/auth/auth_service/findAuthService"
        #url = os.environ["host"] + "/v1/auth/auth_service/findAuthService"  # 读取conftest.py文件地址进行拼接
        response = HttpRequests().get(url,headers=self.headers)
        #print(response.json()['data'][0]['service_list'][0]['id'])
        service_id = response.json()['data'][0]['service_list'][1]['id']
        return service_id

    def api_service_info(self,serviceId='0b6cf45bec757afa7ee7209d30012ce1',developerId=''):
        #服务详情
        body = {
            "serviceId" :serviceId,
            "developerId":developerId
        }
        url = "http://192.168.2.199:9092/v1/auth/auth_service/findServiceDetail"
        #url = os.environ["host"] + "/v1/auth/auth_service/findServiceDetail"#读取conftest.py文件地址进行拼接
        response = HttpRequests().get(url,headers=self.headers,params = body,verify=False)
        #print(response.json())
        return response

    def api_add_or_update_service(self,api_param_req,api_param_res,description,error_code,icon,id,interface_remarks,
                              name,product_info,request_method,sample_code,sort,type,url):
        #服务添加或者更新
        body={
                "api_param_req": api_param_req,
                "api_param_res": api_param_res,
                "description": description,
                "error_code": error_code,
                "icon": icon,
                "id": id,
                "interface_remarks": interface_remarks,
                "name": name,
                "product_info": product_info,
                "request_method": request_method,
                "sample_code": sample_code,
                "sort": sort,
                "type": type,
                "url": url,
        }
        #url = "http://192.168.2.199:9092/v1/auth/auth_service/insertOrUpdateService"
        url = os.environ["host"] + "/v1/auth/auth_service/insertOrUpdateService"  # 读取conftest.py文件地址进行拼接
        response = HttpRequests().post(url,json=body,headers=self.headers,verify=False)
        return response

    def api_add_service_price(self,id,max_number,money,service_id,small_number):
        #服务价格添加
        body = {
            "id": id,
            "max_number": max_number,
            "money": money,
            "service_id": service_id,
            "small_number": small_number
        }
        # url = "http://192.168.2.199:9092/v1/auth/auth_service/insertServicePrice"
        url = os.environ["host"] + "/v1/auth/auth_service/insertServicePrice"  # 读取conftest.py文件地址进行拼接
        response = HttpRequests().post(url, json=body, headers=self.headers, verify=False)
        return response

    def api_apply_service(self,developer_id,service_id):
        #申请服务
        body ={
            "developer_id": developer_id,
            "service_id": service_id
        }
        # url = "http://192.168.2.199:9092/v1/auth/auth_service/applyService"
        url = os.environ["host"] + "/v1/auth/auth_service/applyService"  # 读取conftest.py文件地址进行拼接
        response = HttpRequests().post(url, json=body, headers=self.headers, verify=False)
        return response


if __name__ == '__main__':
    #Auth_Service().api_home_service_list()
    Api_Auth_Service().get_service_id()
    #Auth_Service().api_service_info()

 api/get_token.py#获取登录token

'''
Code description:获取token
Create time:2020-12-03
Developer:叶修
'''
import os

import urllib3
from common.http_requests import HttpRequests


class Get_Token(object):

    def get_token(self,account='****',password='****'):
        #url = "http://192.168.2.199:9092/v1/auth/developer/accountLogin"
        url = os.environ["host"]+"/v1/auth/developer/accountLogin"
        body = {
            "account": account,
            "password": password,
        }
        urllib3.disable_warnings()
        r = HttpRequests().post(url, json=body,verify=False)
        #print(r.json())
        token = r.json()['data']['token']
        params = {
            "access_token": token
        }
        HttpRequests().params.update(params)#更新token到session
        return token

if __name__ == '__main__':
    print(Get_Token().get_token())

case/test_service_info.py #上面接口某一测试用例

'''
Code description: 服务详情
Create time: 2020/12/3
Developer: 叶修
'''
import sys
import allure
import pytest
from common.logger import Log
from common.read_yaml import ReadYaml
from api.api_auth_service.api_auth_service import Api_Auth_Service

testdata = ReadYaml("auth_service.yml").get_yaml_data()  # 读取数据


@allure.feature('服务详情')
class Test_Service_Info(object):
    log = Log()

    @pytest.mark.process
    @pytest.mark.parametrize('serviceId,developerId,expect', testdata['service_info'],
                             ids=['服务详情'])
    def test_service_info(self,serviceId,developerId,expect):
        self.log.info('%s{%s}' % ((sys._getframe().f_code.co_name,'------服务详情接口-----')))
        with allure.step('获取服务id'):
            serviceId = Api_Auth_Service().get_service_id()
        with allure.step('服务详情'):
            msg = Api_Auth_Service().api_service_info(serviceId,developerId)
        self.log.info('%s:%s' % ((sys._getframe().f_code.co_name, '获取请求结果:%s' % msg.json())))
        # 断言
        assert msg.json()["result_message"] == expect['result_message']
        assert msg.json()['result_code'] == expect['result_code']
        assert 'url' in msg.json()['data']

 conftest.py

'''
Code description:配置信息
Create time: 2020/12/3
Developer: 叶修
'''
import os
import pytest
from api.get_token import Get_Token
from common.http_requests import HttpRequests


@pytest.fixture(scope="session")
def get_token():
    '''前置操作获取token并传入headers'''
    Get_Token().get_token()
    if not HttpRequests().params.get("access_token", ""):#没有get到token,跳出用例
        pytest.skip("未获取token跳过用例")
    yield HttpRequests().req
    HttpRequests().req.close()

def pytest_addoption(parser):
    parser.addoption(
        "--cmdhost", action="store", default="http://192.168.1.54:32099",
        help="my option: type1 or type2"
    )
@pytest.fixture(scope="session",autouse=True)
def host(request):
    '''获取命令行参数'''
    #获取命令行参数给到环境变量
    #pytest --cmdhost 运行指定环境
    os.environ["host"] = request.config.getoption("--cmdhost")
    print("当前用例运行测试环境:%s" % os.environ["host"])

 common/connect_mysql.py

'''
Code description: 配置连接数据库
Create time: 2020/12/3
Developer: 叶修
'''
import pymysql

dbinfo = {
    "host":"******",
    "user":"root",
    "password":"******",
    "port":31855
}
class DbConnect():
    def __init__(self,db_conf,database=""):
        self.db_conf = db_conf
        #打开数据库
        self.db = pymysql.connect(database = database,
                                  cursorclass = pymysql.cursors.DictCursor,
                                  **db_conf)
        #使用cursor()方式获取操作游标
        self.cursor = self.db.cursor()

    def select(self,sql):
        #sql查询
        self.cursor.execute(sql)#执行sql
        results = self.cursor.fetchall()
        return results

    def execute(self,sql):
        #sql 删除 提示 修改
        try:
            self.cursor.execute(sql)#执行sql
            self.db.commit()#提交修改
        except:
            #发生错误时回滚
            self.db.rollback()

    def close(self):
        self.db.close()#关闭连接

def select_sql(select_sql):
    '''查询数据库'''
    db = DbConnect(dbinfo,database='auth_platform')
    result = db.select(select_sql)
    db.close()
    return result

def execute_sql(sql):
    '''执行SQL'''
    db = DbConnect(dbinfo,database='auth_platform')
    db.execute(sql)
    db.close()

if __name__ == '__main__':
    sql = "SELECT * FROM auth_platform.auth_service where name='四要素认证'"
    sel = select_sql(sql)[0]['name']
    print(sel)

 common/http_requests.py

'''
Code description: 封装自己的请求类型
Create time: 2020/12/3
Developer: 叶修
'''

import requests


# 定义一个HttpRequests的类
class HttpRequests(object):

    req = requests.session()#定义session会话
    # 定义公共请求头
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.125 Safari/537.36',
        'cookie':''
    }
    params = {
        'access_token':''
    }
    # 封装自己的get请求,获取资源
    def get(self, url='', params='', data='', headers=None, cookies=None,stream=None,verify=None):
        response = self.req.get(url,params=params,data=data,headers=headers,cookies=cookies,stream=stream,verify=verify)
        return response

    # 封装自己的post方法,创建资源
    def post(self, url='', params='',data='', json='', headers=None, cookies=None,stream=None,verify=None):
        response = self.req.post(url,params=params,data=data,json=json,headers=headers,cookies=cookies,stream=stream,verify=verify)
        return response

    # 封装自己的put方法,更新资源
    def put(self, url='', params='', data='', headers=None, cookies=None,verify=None):
        response = self.req.put(url, params=params, data=data, headers=headers, cookies=cookies,verify=verify)
        return response

    # 封装自己的delete方法,删除资源
    def delete(self, url='', params='', data='', headers=None, cookies=None,verify=None):
        response = self.req.delete(url, params=params, data=data, headers=headers, cookies=cookies,verify=verify)
        return response

 common/logger.py

'''
Code description: 封装输出日志文件
Create time: 2020/12/3
Developer: 叶修
'''
import logging,time
import os
import getpathinfo

path = getpathinfo.get_path()#获取本地路径
log_path = os.path.join(path,'logs')# log_path是存放日志的路径
# 如果不存在这个logs文件夹,就自动创建一个
if not os.path.exists(log_path):os.mkdir(log_path)

class Log():
    def __init__(self):
        #文件的命名
        self.logname = os.path.join(log_path,'%s.log'%time.strftime('%Y_%m_%d'))
        self.logger = logging.getLogger()
        self.logger.setLevel(logging.DEBUG)
        #日志输出格式
        self.formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

    def __console(self,level,message):
        #创建一个fileHander,用于写入本地
        fh = logging.FileHandler(self.logname,'a',encoding='utf-8')
        fh.setLevel(logging.DEBUG)
        fh.setFormatter(self.formatter)
        self.logger.addHandler(fh)

        #创建一个StreamHandler,用于输入到控制台
        ch = logging.StreamHandler()
        ch.setLevel(logging.DEBUG)
        ch.setFormatter(self.formatter)
        self.logger.addHandler(ch)

        if level == 'info':
            self.logger.info(message)
        elif level == 'debug':
            self.logger.debug(message)
        elif level == 'warning':
            self.logger.warning(message)
        elif level == 'error':
            self.logger.error(message)
        #避免日志重复
        self.logger.removeHandler(fh)
        self.logger.removeHandler(ch)
        #关闭打开文件
        fh.close()

    def debug(self,message):
        self.__console('debug',message)

    def info(self,message):
        self.__console('info',message)

    def warning(self,message):
        self.__console('warning',message)

    def error(self,message):
        self.__console('error',message)

if __name__ == '__main__':
    log = Log()
    log.info('测试')
    log.debug('测试')
    log.warning('测试')
    log.error('测试')

 common/read_save_data.py

'''
Code description: 读取保存数据
Create time: 2020/12/8
Developer: 叶修
'''
import os
import yaml
import getpathinfo
class Read_Save_Date():
    def __init__(self):
        path = getpathinfo.get_path()#获取本地路径
        self.head_img_path = os.path.join(path,'data','save_data')+"/"+'head_img_path.txt'# head_img_path文件地址
        self.order_id_path = os.path.join(path, 'data', 'save_data') + "/" + 'order_id.txt'  # order_id.txt文件地址

    def get_head_img_path(self):
        # 获取head_img_path
        with open(self.head_img_path, "r", encoding="utf-8")as f:
            return f.read()

    def get_order_id(self):
        # 获取order_id
        with open(self.order_id_path, "r", encoding="utf-8")as f:
            return f.read()

if __name__ == '__main__':
    print(Read_Save_Date().get_head_img_path())
    print(Read_Save_Date().get_order_id())

 common/read_yaml.py

'''
Code description: 读取yml文件测试数据
Create time: 2020/12/3
Developer: 叶修
'''
import os
import yaml
import getpathinfo
class ReadYaml():
    def __init__(self,filename):
        path = getpathinfo.get_path()#获取本地路径
        self.filepath = os.path.join(path,'data','test_data')+"/"+filename#拼接定位到data文件夹

    def get_yaml_data(self):
        with open(self.filepath, "r", encoding="utf-8")as f:
            # 调用load方法加载文件流
            return yaml.load(f,Loader=yaml.FullLoader)

if __name__ == '__main__':
    data = ReadYaml("auth_service.yml").get_yaml_data()['add_or_update_service']
    print(data)

data/

 data/test_data/auth_service.yml

home_service_list:
  - [{'result_code': '0', 'result_message': '处理成功'}]
service_info:
  - ['','',{'result_code': '0', 'result_message': '处理成功'}]
add_or_update_service:
  - [['1','1','1','1','1','123456','1','测试','1','1','1','1','1','1'],{'result_code': '0', 'result_message': '处理成功'}]
add_service_price:
  - ['123456789','10','0','','0',{'result_code': '0', 'result_message': '处理成功'}]
apply_service:
  - ['','',{'result_code': '0', 'result_message': '处理成功'}]

 logs/

 report/

 getpathinfo.py

'''
Code description:配置文件路径
Create time: 2020/12/3
Developer: 叶修
'''
import os

def get_path():
    # 获取当前路径
    curpath = os.path.dirname(os.path.realpath(__file__))
    return curpath

if __name__ == '__main__':# 执行该文件,测试下是否OK
    print('测试路径是否OK,路径为:', get_path())

pytest.ini

#pytest.ini
[pytest]
markers = process
addopts = -p no:warnings
#addopts = -v --reruns 1 --html=./report/report.html --self-contained-html
#addopts = -v --reruns 1 --alluredir ./report/allure_raw
#addopts = -v -s -p no:warnings --reruns  1 --pytest_report ./report/Pytest_Report.html

requirements.txt

allure-pytest==2.8.18
allure-python-commons==2.8.18
BeautifulReport==0.1.3
beautifulsoup4==4.9.3
ddt==1.4.1
Faker==4.18.0
Flask==1.1.1
httpie==1.0.3
httplib2==0.9.2
HttpRunner==1.5.8
py==1.9.0
PyMySQL==0.10.1
pytest==6.1.1
pytest-base-url==1.4.2
pytest-cov==2.10.1
pytest-forked==1.3.0
pytest-html==2.1.1
pytest-instafail==0.4.2
pytest-metadata==1.10.0
pytest-mock==3.3.1
pywin32==228
PyYAML==5.3.1
requests==2.22.0
requests-oauthlib==1.3.0
requests-toolbelt==0.9.1

run_main.py

'''
Code description: 运行主流程测试用例
Create time: 2020/11/5
Developer: 叶修
'''
import os
import pytest
if __name__ == '__main__':
    pytest.main(['-m','process', '-s','--alluredir', 'report/tmp'])#-m运行mark标记文件
    os.system('allure generate report/tmp -o report/html --clean') # /report/tmp 为存放报告的源文件目录

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/460793.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

C. Magic Ship(二分 + 前缀和)

Problem - C - Codeforces 你是一艘船的船长。最初你站在一个点(x1,y1)上(很明显,海上的所有位置都可以用笛卡尔平面描述),你想要前往一个点(x2,y2)。 你知道天气预报——长度为n的字符串s,仅由…

实战详解Docker快速搭建部署ELK

一.安装前须知 以下步骤在 VMware 中的 centos 7 中操作,ip 地址为:192.168.161.128; 注意安装的时候最好统一版本,否则后面会出现许多问题,进官网搜索对应镜像,查看 Tags 标签下的版本,目前我…

记一次死锁问题

最近在做一个需求,碰到了死锁的问题,记录下解决问题的过程 背景 这个需求要改动一个接口,我这边称为A接口,原先的逻辑是A接口内部会调用c方法,c方法是一个dubbo方法, 现在需要再A接口里添加调用B方法&…

springcloud之Feign、ribbon设置超时时间和重试机制的总结

目录标题 超时时间ribbon和Feignribbon和Feign默认超时时间关于ribbon和Feign超时时间配置说明 关于hystrix默认超时时间与配置说明 ribbon的重试机制重试的次数hystrix超时时间举个例子 超时时间 feign/ribbon对应的是请求的时间 hystrix对应的是断路器的时间 一般情况下 都是…

【Linux】2. 常见指令

1. 操作系统的定义 在真正了解Linux操作系统之前,我们需要初步明确什么是操作系统 Linux下的基本指令 指令的作用是什么,其实就是相当于Windows的基本操作,Linux操作系统是以命令行显示的,而Windows则是以图形化界面的方式展现…

图像融合方向:《Deep Image Blending》论文理解

《Deep Image Blending》论文理解 论文:《Deep Image Blending》WACV 2020 链接:Deep Image Blending 本文目录 《Deep Image Blending》论文理解论文创新点具体实现思路文中使用的基准方法文章内容解析使用模型整体架构两阶段算法详情第1阶段第2阶段 损…

一款高效、可靠的自动化测试平台,提升效率、降低测试成本

一、开源项目简介 扬帆测试平台是一款高效、可靠的自动化测试平台,旨在帮助团队提升测试效率、降低测试成本。该平台包括用例管理、定时任务、执行记录等功能模块,支持多种类型的测试用例,目前支持API(http和grpc协议)、性能,并且…

轻量级服务器nginx:负载均衡

负载均衡就是让每个设备,以同样的概率,处理用户对于服务器的任务请求,默认采用的负载调度策略就是轮流询问,Nginx作为反向代理服务器安装在服务端,Nginx的功能就是把请求转发给后面的应用服务器. 这里写目录标题 一 负…

差分优化算法——DE

🍎道阻且长,行则将至。🍓 目录 一、DE1.步骤2.特点 二、DE Optimiza1.函数最小值问题2.差分进化算法求解2.Java 实现与结果绘图 一、DE 差分进化算法是一种基于群体智能的优化算法,由Storn和Price于1995年提出,最早用…

vue3+ts开发微信小程序----使用FontAwesome图标的步骤(只适用于面性图标)

vue3ts开发微信小程序----使用FontAwesome图标的步骤(只适用于面性图标) 因为用的vue3ts 但是对于npm引进FontAwesome来说,好像对于小程序不支持,所以用了引入文件的方式。 下载FontAwesome字体,我用的是5版本的。[官方…

从初识RabbitMQ到安装了解

一、同步和异步通讯 微服务间通讯有同步和异步两种方式: 同步通讯:就像打电话,需要实时响应。 异步通讯:就像发邮件,不需要马上回复。 两种方式各有优劣,打电话可以立即得到响应,但是你却不…

记 SAM React Demo 转 Vue 遇到的一些问题

前言 SegmentAnything Model (SAM) 是 Meta 开源的分割万物 AI 模型,因笔者主要使用 Vue,因此对官方 Demo 进行了工程迁移工作,这里记录了迁移过程遇到的一些问题。 没有使用多线程加速 官方的 React Demo 项目使用ort-wasm-simd-threaded…

态路小课堂丨浅谈IDC数据中心综合布线互联

在数据中心发展过程中,服务器、光模块等设备更新周期短,迭代快。但是,对于IDC数据中心综合布线系统来说,其使用周期一般都比较长。因此,为了能够匹配IDC数据中心的发展,关于综合布线的设计尤为重要。 目前&…

【vue】elementUI中el-table拖动滚动条的时候固定某一列

文章目录 前言:需求 前言: 写项目的时候有一个需求,就是当el-table的滚动条滑动的时候,可见区域如果没有该列则固定到左侧,由elementUi官方网站可知el-table固定一列的代码如下: fixed为主要的标识 头部固定设置的是t…

leetcode 2336. Smallest Number in Infinite Set(有限集中的最小数字)

实现一个SmallestInfiniteSet类(后面用set简称),满足以下功能: 初始化时,set含有所有的正整数。 pop函数返回set中最小的整数。 add函数在set中添加一个整数。 思路: set具有以下的功能: 可以取出最小的数字。 可以查…

FPGA入门系列13--异步串口通信

文章简介 本系列文章主要针对FPGA初学者编写,包括FPGA的模块书写、基础语法、状态机、RAM、UART、SPI、VGA、以及功能验证等。将每一个知识点作为一个章节进行讲解,旨在更快速的提升初学者在FPGA开发方面的能力,每一个章节中都有针对性的代码…

Baklib如何帮助企业设计并维护FAQ页面?

作为现代企业的一部分,客户支持服务是为客户提供解决方案、回答问题和解决技术难题的关键部分。而其中最重要的一个基本工具是FAQ页面(Frequently Asked Questions),它可以有效地减轻客户支持压力和提高工作效率。然而&#xff0c…

Linux——进度条与git的使用

目录 main.c Makefile process.c \r process.h 进度条的演示动图 git的使用 在gitee上创建一个仓库 复制HTTPS到xshell git clone .gitignore ​编辑 git add git commit -m 这里写日志信息 git push sudo yum -y install git git log git status git mv / …

vue2之echarts的封装 折线图,饼图,大图

目录 vue2之echarts的封装 折线图,饼图,大图折线图,饼图chartPan.vue使用 chartPan.vue 之饼图效果使用 chartPan.vue 之折线图效果展开大图大图组件 maxChart.vue大图效果 vue2之echarts的封装 折线图,饼图,大图 折线…

视觉回归测试—UI自动化的最后1分钟

视觉回归测试是一种验证应用GUI是否正确地展示给用户的操作。测试目标是找出应用在可视化上存在的软件缺陷,例如,字体、布局和渲染问题。这使得所发现的软件缺陷可在被最终用户看到前得到修正。此外,视觉测试可用于验证页面的内容&#xff0c…