Pytest实战API测试框架

https://www.jianshu.com/p/40a0b396465c?utm_campaign=hugo&utm_medium=reader_share&utm_content=note&utm_source=weixin-timeline&from=timeline&isappinstalled=0

功能规划

1. 数据库断言 pymysql -> 封装
2. 环境清理 数据库操做 -> Fixtures
3. 并发执行 pytest-xdist 多进程并行
4. 复合断言 pytest-check
5. 用例重跑 pytest-rerunfailures
6. 环境切换 pytest-base-url
7. 数据分离 pyyaml
8. 配置分离 pytest.ini
9. 报告生成 pytest-html, allure-pytest
10. 用例等级 pytest-level
11. 限制用例超时时间 pytest-timeout
12. 发送报告邮件 经过自定Fixture及Hooks实现

安装相应的包

pip安装时能够经过-i https://pypi.doubanio.com/simple/,指定使用豆瓣的源, 下载稍微快一点

pip install requests pymysql pyyaml pytest pyetst-xdist pytest-check pytest-rerunfailures pytest-base-url pytest-html pytest-level pytest-timeout -i https://pypi.doubanio.com/simple/ 

导出依赖到requirements.txt中

pip freeze > requirments.txt 

结构规划

分层结构

分层设计模式: 每一层为上层提供服务

用例层(测试用例)
  | Fixtures辅助层(全局的数据、数据库操做对象和业务流等) | utils实用方法层(数据库操做, 数据文件操做，发送邮件方法等等) 

静态目录

• data: 存放数据
• reports: 存放报告

目录结构

longteng17/
  - data/
    - data.yaml: 数据文件
  - reports/: 报告目录
  - test_cases/: 用例目录
    - pytest.ini:  pytest配置
    - api_test/:  接口用例目录
      - conftest.py:  集中管理Fixtures方法
    - web_test/:  web用例目录
    - app_test/:  app用例目录
  - utils/: 辅助方法
    - data.py: 封装读取数据文件方法
    - db.py: 封装数据库操做方法
    - api.py: 封装api请求方法
    - notify.py: 封装发送邮件等通知方法
  - conftest.py: 用来放置通用的Fixtures和Hooks方法
  - pytest.ini: Pytest运行配置

规划conftest.py的位置，要确保项目跟目录被导入到环境变量路径(sys.path)中去。
conftest.py及用例的导入机制为：

1. 若是在包（同级有init.py）内，则导入最上层包（最外一个包含init.py）的父目录。
2. 若是所在目录没有init.py，直接导入conftest.py父目录。

数据文件的选择

• 无结构
  • txt: 分行, 无结构的文本数据
• 表格型
  • csv: 表格型, 适合大量同一类型的数据
  • excel: 表格型, 构造数据方便, 文件较大,解析较慢
• 树形
  • json: 能够存储多层数据, 格式严格,不支持备注
  • yaml: 兼容json, 灵活,能够存储多层数据
  • xml: 能够存储多层, 文件格式教繁琐
• 配置型
  • .ini/.properties/.conf: 只能存储1-2层数据, 适合配置文件

因为用例数据经常须要多层级的数据结构，这里选择yaml文件做为本项目的数据文件，示例格式以下：

test_case1:  a: 1  b: 2 

数据第一层以用例名标识某条用例所使用的数据，这里约定要和用例名称彻底一致，方便后面使用Fixture方法自动向用例分配数据。

标记规划

标记: mark, 也称做标签, 用来跨目录分类用例方便灵活的选择执行。

• 按类型: api, web, app
• 标记有bug: bug
• 标记异常流程: negative

也能够根据本身的需求，按模块、按是否有破坏性等来标记用例。

utils实用方法层

数据文件操做: data.py

首先须要安装pyyaml, 安装方法：pip install pyyaml
读取yaml文件数据的方法为：

1. 打开文件 with open(..) as f:
2. 加载数据 data=yaml.safe_load(f)

yaml.safe_load()和yaml.load()的区别:

因为yaml文件也支持任意的Python对象
从文件中直接加载注入Python是极不安全的, safe_load()会屏蔽Python对象类型,只解析加载字典/列表/字符串/数字等级别类型数据

示例以下:

import yaml def load_yaml_data(file_path): with open(file_path, encoding='utf-8') as f: data = yaml.safe_load(f) print("加载yaml文件: {file_path} 数据为: {data}") return data 

为了示例简单，这里没有对文件不存在、文件格式非yaml等异常作处理。异常处理统一放到Fixture层进行。

假如项目要支持多种数据文件, 可使用类来处理。

数据库操做: db.py

这里使用pymysql, 安装方法pip install pymysql

敏感数据处理

数据库配置分离

数据库密码等敏感数据，直接放在代码或配置文件中，会有暴露风险，用户敏感数据咱们能够放到环境变量中，而后从环境变量中读取出来。

注意：部署项目时，应记得在服务器上配置相应的环境变量，才能运行。

Windows在环境变量中添加变量MYSQL_PWD,值为相应用户的数据库密码，也能够将数据库地址，用户等信息也配置到环境变量中。
Linux/Mac用户能够经过在_/.bashrc或/.bash_profile或/etc/profile中添加

export MYSQL_PWD=数据库密码 

而后source相应的文件使之生效,如source ~/.bashrc。

Python中使用os.getenv('MYSQL_PWD')即可以拿到相应环境变量的值。

注意：若是使用PyCharm，设置完环境变量后，要重启PyCharm才能读取到新的环境变量值。

咱们使用字典来存储整个数据库的配置，而后经过字典拆包传递给数据库链接方法。

import os import pymysql DB_CONF = { 'host': '数据库地址', 'port': 3306, 'user': 'test', 'password': os.getenv('MYSQL_PWD'), 'db': 'longtengserver', 'charset': 'utf8' } conn = pymysql.connect(**DB_CONF) 

封装数据库操做方法

数据常见的操做方法有查询，执行修改语句和关闭链接等。对应一种对象的多个方法，咱们使用类来封装。
同时为避免查询语句和执行语句的串扰，咱们在创建链接时使用autocommit=True来确保每条语句执行后都当即提交，完整代码以下。

import os import pymysql DB_CONF = { 'host': '数据库地址', 'port': 3306, 'user': 'test', 'password': os.getenv('MYSQL_PWD'), 'db': 'longtengserver', 'charset': 'utf8' } class DB(object): def __init__(self, db_conf=DB_CONF) self.conn = pymysql.connect(**db_conf, autocommit=True) self.cur = self.conn.cursor(pymysql.cursors.DictCursor) def query(self, sql): self.cur.execute(sql) data = self.cur.fetchall() print(f'查询sql: {sql} 查询结果: {data}') return data def change_db(self, sql): result = self.cur.execute(sql) print(f'执行sql: {sql} 影响行数: {result}') def close(self): print('关闭数据库链接') self.cur.close() self.conn.close() 

其中若是查询中包含中文，要根据数据库指定响应的charset，这里的charset值为utf8不能写成utf-8。
self.conn.cursor(pymysql.cursors.DictCursor)这里使用了字典格式的游标，返回的查询结果会包含响应的表字段名，结果更清晰。

因为全部sql语句都是单条自动提交，不支持事务，所以在change_db时，不须要再做事务异常回滚的操做，对于数据库操做异常，统一在Fixture层简单处理。

封装经常使用数据库操做

# db.py ... class FuelCardDB(DB): def del_card(self, card_number): print(f'删除加油卡: {card_number}') sql = f'DELETE FROM cardinfo WHERE cardNumber="{card_number}"' self.change_db(sql) def check_card(self, card_number): print(f'查询加油卡: {card_number}') sql = f'SELECT id FROM cardinfo WHERE cardNumber="{card_number}"' res = self.query(sql) return True if res else False def add_card(self, card_number): print(f'添加加油卡: {card_number}') sql = f'INSERT INTO cardinfo (cardNumber) VALUES ({card_number})' self.change_db(sql) 

发送邮件通知: notify.py

使用Python发送邮件

发送邮件通常要经过SMTP协议发送。首先要在你的邮箱设置中开启SMTP服务，清楚SMTP服务器地址、端口号已是否必须使用安全加密传输SSL等。
使用Python发送邮件分3步：

1. 组装邮件内容MIMEText
2. 组装邮件头: From、To及Subject
3. 登陆SMTP服务器发送邮件

• 组装邮件内容MIMEText

from email.mime.text import MIMEText import smtplib body = 'Hi, all\n附件中是测试报告, 若有问题请指出' body2 = '<h2>测试报告</h2><p>如下为测试报告内容<p>' # msg = MIMEText(content, 'plain', 'utf-8') msg = MIMEText(content2, 'html', 'utf-8') 

使用MIMEText组装Email消息数据对象，正文支持纯文本plain和html两种格式。

• 组装邮件头: From、To及Subject

...
msg['From'] = 'zhichao.han@qq.com' msg['To'] = 'superhin@126.com' msg['Subject'] = '接口测试报告' 

msg['From']中也能够声明收件人名称，格式为：

msg['From'] = '<韩志超> zhichao.han@qq.com' 

msg['To']中也能够写多个收件人，写到一个字符串中使用英文逗号隔开：

msg['To'] = 'superhin@126.com,ivan-me@163.com' 

注意邮件头的From、To只是一种声明，并不必定是实际的发件人和收件人，好比From写A邮箱，实际发送时，使用B邮箱的SMTP发送，便会造成代发邮件（B表明A发送）。

• 登陆SMTP服务器发送邮件

...
smtp = smtplib.SMTP('邮箱SMTP地址') # smtp = smtplib.SMTP_SSL('邮箱SMTP地址') smtp.login('发件人邮箱', '密码') smtp.sendmail('发件人邮箱', '收件人邮箱', msg.as_string()) 

这里登陆SMTP和SMTP_SSL要看邮箱服务商支持哪一种，链接时也能够指定端口号，如：

smtp = smtplib.SMTP_SSL('邮箱SMTP地址', 465) 

登陆时的密码根据邮箱的支持能够是受权码或登陆密码（通常如QQ邮箱采用受权码，不支持使用登陆密码登陆SMTP）。
sendmail发送邮件时，使用的发件人邮箱和收件人邮箱是实践的发件人和收件人，能够和邮件头中的不一致。可是发件人邮箱必须和登陆SMTP的邮箱一致。
sendmail每次只能给一个收件人发送邮件，当有多个收件人是，可使用屡次sendmail方法，示例以下：

receivers = ['superhin@163.com', 'zhichao.han@qq.com'] for person in receivers: smtp.sendmail('发件人邮箱', person, msg.as_string()) 

msg.as_string()是将msg消息对象序列化为字符串后发送。

发送带附件的邮件

因为邮件正文会过滤掉大部分的样式和JavaScript，所以直接将html报告读取出来，放到邮件正文中每每没有任何格式。这时，咱们能够经过附件来发送测试报告。

邮件附件通常采用二进制流格式（application/octet-stream），正文则采用文本格式。要混合两种格式咱们须要使用MIMEMultipart这种混合的MIME格式，通常步骤为：

1. 创建一个MIMEMultipart消息对象
2. 添加MIMEText格式的正文
3. 添加MIMEText格式的附件（打开附件，按Base64编码转为MIMEText格式）
4. 添加邮件头信息
5. 发送邮件

示例代码以下：

from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart import smtplib # 1. 创建一个MIMEMultipart消息对象 msg = MIMEMultipart() # 2. 添加邮件正文 body = MIMEText('hi, all\n附件中是测试报告,请查收', 'plain', 'utf-8') msg.attach(body) # 3. 添加附件 att = MIMEText(open('report.html', 'rb').read(), 'base64', 'utf-8') att['Content-Type'] = 'application/octet-stream' att["Content-Disposition"] = 'attachment; filename=report.html' msg.attach(att1) # 4. 添加邮件头信息 ... # 5. 发送邮件 ... 

使用消息对象msg的attach方法来添加MIMEText格式的邮件正文和附件。
构造附件MIMEText对象时，要使用rb模式打开文件，使用base64格式编码，同时要声明附件的内容类型Content-Type以及显示排列Content-Dispositon，这里的attachment; filename=report.html，attachment表明附件图标，filename表明显示的文件名，这里表示图标在左，文件名在右，显示为report.html。

添加邮件头信息和发送邮件同发送普通邮件一致。

发送邮件方法封装

一样，咱们能够将敏感信息邮箱密码配置到环境变量中去，这里变量名设置为SMTP_PWD。

import os from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart import smtplib SMTP_HOST = '邮箱SMTP地址' SMTP_USER = '发件人邮箱' SMTP_PWD = os.getenv('SMTP_PWD') def send_email(self, body, subject, receivers, file_path): msg = MIMEMultipart() msg.attach(MIMEText(body, 'html', 'utf-8')) att1 = MIMEText(open(file_path, 'rb').read(), 'base64', 'utf-8') att1['Content-Type'] = 'application/octet-stream' att1["Content-Disposition"] = f'attachment; filename={file_name}' msg.attach(att1) msg['From'] = SMTP_USER msg['To'] = ','.join(receivers) msg['Subject'] = subject smtp = smtplib.SMTP_SSL(SMTP_HOST) smtp.login(SMTP_USER, SMTP_PWD) for person in receivers: print(f'发送邮件给: {person}') smtp.sendmail(SMTP_USER, person, msg.as_string()) print('邮件发送成功') 

一样，为了示例简单，这里并无对SMTP链接、登陆、发送邮件作异常处理，读者能够进行相应的补充。

请求方法封装：api.py

requests自己已经提供了很好的方法，特别是通用的请求方法requests.request()。这里的封装只是简单加了base_url组装、默认的超时时间和打印信息。

import requests TIMEOUT = 30 class Api(object): def __init__(self, base_url=None): self.session = requests.session() self.base_url = base_url def request(self, method, url, **kwargs): url = self.base_url + url if self.base_url else url kwargs['timeout'] = kwargs.get('timeout', TIMEOUT) res = self.session.request(method, url, **kwargs) print(f"发送请求: {method} {url} {kwargs} 响应数据: {res.text}") return res def get(self, url, **kwargs): return self.request('get', url, **kwargs) def post(self, url, **kwargs): return self.request('post', url, **kwargs) 

这里，Api实例化时若是传递了base_url参数，则全部的url都会拼接上base_url。
kwargs['timeout'] = kwargs.get('timeout', TIMEOUT)，设置默认的超时时间设置为30s。

Fixtures方法层

import pytest from utils.data import Data from utils.db import FuelCardDB from utils.api import Api @pytest.fixture(scope='session') def data(request): basedir = request.config.rootdir try: data_file_path = os.path.join(basedir, 'data', 'api_data.yaml') data = Data().load_yaml(data_file_path) except Exception as ex: pytest.skip(str(ex)) else: return data @pytest.fixture(scope='session') def db(): try: db = FuelCardDB() except Exception as ex: pytest.skip(str(ex)) else: yield db db.close() @pytest.fixture(scope='session') def api(base_url): api = Api(base_url) return api 

这里对，utils实用方法层的异常进行简单的skip处理，即当数据链接或数据文件有问题时，全部引用该Fixture的用例都会自动跳过。

在api这个Fixtures中咱们引入了base_url，它来自于插件pytest-base-url，能够在运行时经过命令行选项--base-url或pytest.ini中的配置项base_url来指定。

[pytest] ... base_url=http://....:8080 

按用例名分发用例

Fixture方法经过用例参数，注入到用例中使用。Fixture方法中能够拿到用例所在的模块，模块变量，用例方法对象等数据，这些数据都封装在Fixture方法的上下文参数request中。
原有的data这个Fixture方法为用例返回了数据文件中的全部数据，可是通常用例只须要当前用例的数据便可。咱们在数据文件中第一层使用和用例方法名同名的项来区分各个用例的数据。如：

# api_data.yaml test_add_fuel_card_normal:  data_source_id: bHRz  cardNumber: hzc_00001 ... 

下面的示例演示了根据用例方法名分配数据的Fixture方法：

# conftest.py ... @pytest.fixture def case_data(request, data): case_name = request.function.__name__ return data.get(case_name) 

request是用例请求Fixture方法的上下文参数，里面包含了config对象、各类Pytest运行的上下文信息，能够经过Python的自省方法print(request.__dict__)查看request对象中全部的属性。

• request.function为调用Fixture的函数方法对象，若是是用例直接调用的Fixture，这里即是用例函数对象，经过函数对象的name属性获取到函数名。
• 经过request.module拿到用例所在模块，进而根据模块中某些属性做相应动态配置。
• 经过request.config能够拿到pytest运行时的运行参数、配置参数值等信息。

这样，用例中引入的case_data参数就只是该用例的数据。

用例层

一条完整的用例应包含如下步骤：

1. 环境检查或数据准备
2. 业务操做
3. 不止一条断言语句（包括数据库断言）
4. 环境清理

另一般用例还应加上指定的标记。

import pytest @pytest.mark.level(1) @pytest.mark.api def test_add_fuel_card_normal(api, db, case_data): """正常添加加油卡""" url = '/gasStation/process' data_source_id, card_number = case_data.get('data_source_id'), case_data.get('card_number') # 环境检查 if db.check_card(card_number): pytest.skip(f'卡号: {card_number} 已存在') json_data = {"dataSourceId": data_source_id, "methodId": "00A", "CardInfo": {"cardNumber": card_number}} res_dict = api.post(url, json=json_data).json() # 响应断言 assert 200 == res_dict.get("code")) assert "添加卡成功" == res_dict.get("msg") assert res_dict.get('success') is True # 数据库断言 assert db.check_card(card_number) is True # 环境清理 db.del_card(card_number) 

使用复合断言：pytest-check

使用assert断言时，当某一条断言失败后，该条用例即视为失败，后面的断言不会再进行判断。有时咱们须要每一次能够检查全部的检查点，输出全部断言失败项。此时咱们可使用pytest-check插件进行复合断言。
安装方法pip install pytest-check。
所谓复合断言即，当某条断言失败后仍继续检查下面的断言，最后汇总全部失败项。
pytest-check使用方法

import pytest_check as check ... check.equal(200, es_dict.get("code")) check.equal("添加卡成功",res_dict.get("msg")) check.is_true(res_dict.get('success')) check.is_true(db.check_card(card_number)) 

除此外经常使用的还有:

• check.is_false()：断言值为False
• check.is_none(): 断言值为None
• check.is_not_none()：断言值不为None

标记用例跳过和预期失败

若是某些用例暂时环境不知足没法运行能够标记为skip, 也可使用skipif()判断条件跳过。 对于已知Bug，还没有完成的功能也能够标记为xfail（预期失败）。
使用方法以下：

import os import pytest @pytest.mark.skip(reason="暂时没法运行该用例") def test_a(): pass @pytest.mark.skipif(os.getenv("MYSQL_PWD") is None, reason="缺失环境变量MYSQL_PWD配置") def test_b(): pass @pytest.mark.xfail(reason='还没有解决的已知Bug') def test_c(): pass 

test_b首先对环境变量作了检查，若是没有配置MYSQL_PWD这个环境变量，则会跳过该用例。
test_c为指望失败，这时若是用例正常经过则视为异常的xpass状态，失败则为视为正常的xfail状态，在--strict严格模式下，xfail视为用例经过，xpass视为用例失败。
这里标记运行时分别使用-r/-x/-X显示skip、xfail、xpass的缘由说明：

pytest -rsxX

这里的-s能够在命令行上显示用例中print的一些信息。

另外，也能够在Fixture方法或用例中，使用pytest.skip("跳过缘由"), pytest.xfail("指望失败缘由")来根据条件表用例跳过和指望失败。

标记skip和xfail属于一种临时隔离策略，等问题修复后，应及时去掉该标记。

运行控制

切换环境

运行时经过传入--base-url来切换环境：

pytest --base-url=http://服务地址:端口号

失败用例重跑

默认Pytest支持-lf参数来重跑上次失败的用例。但若是咱们想要本次用例失败后自动重跑的话，可使用pytest-rerunfailures插件。
安装方法pip install pytest-rerunfailures。
运行时使用

pytest --reruns 3 --reruns-delay 1

来指定失败用例延迟1s后自动重跑，最多重跑3次。

对应已知的不稳定用例，咱们能够经过flasky标记,来使之失败时自动重跑，示例以下。

import pytest @pytest.mark.flaky(reruns=3, reruns_delay=1) def test_example(): import random assert random.choice([True, False]) 

按用例等级运行

使用pytest-level能够对用例标记等级，安装方法： pip install pyest-level
使用方法：

@pytest.mark.level(1) def test_basic_math(): assert 1 + 1 == 2 @pytest.mark.level(2) def test_intermediate_math(): assert 10 / 2 == 5 @pytest.mark.level(3) def test_complicated_math(): assert 10 ** 3 == 1000 

运行时经过--level来选择运行的等级。

pytest --level 2

以上只会运行level1和level2的用例（数字越大，优先级约低）

限制用例执行时间

使用插件pytest-timeout能够限制用例的最大运行时间。
安装方法：pip install pytest-timeout
使用方式为

pytest --timeout=30

或配置到pytest.ini中

...
timeout=30 

用例并行

使用pytest-xdist能够开启多个进程运行用例。
安装方法：pip install pytest-xdist
使用方式

pytest -n 4

将全部用例分配到4个进程运行。

完整的项目配置文件

pytest.ini

[pytest] miniversion = 5.0.0 addopts = --strict --html=report_{}.html --self-contained-html base_url = http://115.28.108.130:8080 testpaths = test_cases/ markers = api: api test case web: web test case app: app test case negative: abnormal test case email_subject = Test Report email_receivers = superhin@126.com,hanzhichao@secco.com email_body = Hi,all\n, Please check the attachment for the Test Report. log_cli = true log_cli_level = info log_cli_format = %(asctime)s %(levelname)s %(message)s log_cli_date_format = %Y-%m-%d %H:%M:%S timeout = 10 timeout_func_only = true 

做者：韩志超 连接：https://www.jianshu.com/p/40a0b396465c 來源：简书 简书著做权归做者全部，任何形式的转载都请联系做者得到受权并注明出处。