目錄
Python接口測試課程(第一天)-Python基礎
Python接口測試課程(第二天)-接口測試快速實踐
Python接口測試課程(第三天)-接口安全驗證,參數(shù)化及斷言
Python接口測試課程(第四天)-接口測試框架實現(xiàn)
更多學習資料請加添加作者微信：lockingfree獲取
第四天: Python接口測試框架
什么是框架
目前主流接口測試方案
框架類型
- 錄制回放
- 數(shù)據(jù)驅(qū)動
- 行為驅(qū)動
框架的分層與規(guī)劃
框架分層
- 表示層: (用戶界面)
- 業(yè)務邏輯層: (讀取數(shù)據(jù),配置并組裝發(fā)送請求)+執(zhí)行控制層(pytest)
- 數(shù)據(jù)層: (配置讀取/數(shù)據(jù)讀取/數(shù)據(jù)庫連接/其他(log/email)
框架規(guī)劃
- case: 測試用例目錄
- user: (用戶模塊)
- case.py: 用例公共方法
- data: 數(shù)據(jù)文件目錄
- test_user_data.xlsx: 測試用例數(shù)據(jù)文件
- conf: 配置文件目錄
- report: pytest生成的報告保存路徑
- log: log保存路徑,按天生成log
- common: 公共方法目錄
- config.py: 配置文件讀取
- data.py: 數(shù)據(jù)文件讀取
- db.py: 數(shù)據(jù)庫連接
- log.py: 日志配置
- send_email.py: 發(fā)送郵件配置
框架實現(xiàn)
conf/default.conf: 表示層-項目配置文件
[runtime]
log_level=debug
report_dir=report
log_dir=log
timeout=10
[server]
test = http://127.0.0.1:5000
stage = http://127.0.0.1:6000
prod = http://127.0.0.1:7000
[db_test]
host = localhost
port = 3307
db = api
user = root
passwd =
[db_stage]
[db_prod]
[email]
server = smtp.sina.com
user = test_results@sina.com
pwd = ******
subject = Api Test Ressult
receiver = superhin@126.com
data/test_user_data.xlsx: 表示層-用例數(shù)據(jù)文件
| TestCase |
Url |
Method |
DataType |
Data |
Code |
Msg |
| test_reg_normal |
/api/user/reg/ |
POST |
JSON |
{"name": "{NAME}", "passwd": "123456"} |
100000 |
成功 |
| TestCase |
Url |
Method |
DataType |
Data |
ResponseText |
| test_login_normal |
/api/user/login/ |
POST |
FORM |
{"name": "張三", "passwd": "123456"} |
登錄成功 |
| checkUser |
select * from user where name={NAME} |
| checkUserPasswd |
select * from user where name={NAME} and passwd={PASSWD} |
common/config.py:數(shù)據(jù)層-config文件讀取
"""
1. 從配置文件中獲取各個段信息
2. 返回一個項目的絕對路徑
"""
import os
import configparser
# 相對導入包的問題
pro_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
class Config(object):
def __init__(self, filename="default.conf"):
self.cf = configparser.ConfigParser()
self.cf.read(os.path.join(pro_path,"conf",filename))
def get_runtime(self, option):
return self.cf.get("runtime", option)
def get_server(self, option):
return self.cf.get("server", option)
def get_db_test(self, option):
return self.cf.get("db_test", option)
def get_email(self, option):
return self.cf.get("email",option)
if __name__ == "__main__":
c = Config()
print(c.get_runtime("log_level"))
print(c.get_server("test"))
data.py: 數(shù)據(jù)層-數(shù)據(jù)文件讀取
"""
1. 從Excel中讀取接口的數(shù)據(jù)
2. 讀取Sql命令
"""
import xlrd
import sys
import os
sys.path.append("..")
from common.config import pro_path
class Data(object):
def __init__(self, filename):
data_file_path = os.path.join(pro_path,"data",filename)
self.wb = xlrd.open_workbook("../data/test_user_data.xlsx")
def get_case(self,sheet_name, case_name):
sh = self.wb.sheet_by_name(sheet_name)
for i in range(1, sh.nrows):
if sh.cell(i,0).value == case_name:
return sh.row_values(i)
print("用例名未找到")
return None
def get_sql(self, sql_name):
sh = self.wb.sheet_by_name("SQL")
for i in range(sh.nrows):
if sh.cell(i,0).value == sql_name:
return sh.cell(i,1).value
print("sql未找到")
return None
if __name__ == "__main__":
d = Data("test_user_data.xlsx")
print(d.get_case("reg","test_reg_normal"))
print(d.get_sql("checkUser"))
db.py: 數(shù)據(jù)層-數(shù)據(jù)庫連接
"""
1. 從配置文件中讀取數(shù)據(jù)庫配置
2. 連接數(shù)據(jù)庫
3. 執(zhí)行sql并返回所有結果
"""
import sys
import pymysql
sys.path.append("..")
from common.config import Config
class DB(object):
def __init__(self):
c = Config()
self.conn = pymysql.connect(host=c.get_db_test("host"),
port=int(c.get_db_test("port")),
db=c.get_db_test("db"),
user=c.get_db_test("user"),
passwd=c.get_db_test("passwd"),
charset="utf8")
self.cur = self.conn.cursor()
def do_sql(self, sql):
self.cur.execute(sql)
return self.cur.fetchall()
def __del__(self):
self.cur.close()
self.conn.close()
if __name__ == "__main__":
db = DB()
print(db.do_sql("select * from user"))
log.py: 數(shù)據(jù)層-log配置
"""
1. 配置log輸出格式 time - loglevel - file - func - line - msg
2. 支持輸出到log文件及屏幕
3. 支持返回一個logger,讓其他模塊調(diào)用
"""
import sys
sys.path.append("..")
from common.config import Config, pro_path
import time
import logging
import os
class Log():
@classmethod
def config_log(cls):
cf = Config()
log_dir = os.path.join(pro_path, cf.get_runtime("log_dir"))
today = time.strftime("%Y%m%d", time.localtime(time.time()))
log_file = os.path.join(log_dir, today+".log")
# 獲取一個標準的logger, 配置loglevel
cls.logger = logging.getLogger()
cls.logger.setLevel(eval("logging." + cf.get_runtime("log_level").upper()))
# 建立不同handler
fh = logging.FileHandler(log_file, mode="a",encoding=‘utf-8’)
ch = logging.StreamHandler()
# 定義輸出格式
ft = logging.Formatter("%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s")
fh.setFormatter(ft)
ch.setFormatter(ft)
# 把定制handler 添加到我們logger
cls.logger.addHandler(fh)
cls.logger.addHandler(ch)
@classmethod
def get_logger(cls):
cls.config_log()
return cls.logger
if __name__ == "__main__":
l= Log.get_logger()
l.info("abc")
l.debug("hello, debug")
send_email.py: 數(shù)據(jù)層-郵件服務器連接
"""
1. 從配置文件中讀取stmp配置
2. 從report文件夾下打開report.html,發(fā)送郵件
"""
import smtplib
from email.mime.text import MIMEText
import os
import sys
sys.path.append("..")
from common.config import Config, pro_path
from common.log import Log
def send_email(report_name):
cf = Config()
logger = Log.get_logger()
report_file = os.path.join(pro_path, cf.get_runtime("report_dir"),report_name)
with open(report_file, "rb") as f:
body = f.read()
# 格式化email正文
msg = MIMEText(body, "html", "utf-8")
# 配置email頭
msg["Subject"] = cf.get_email("subject")
msg["From"] = cf.get_email("user")
msg["To"] = cf.get_email("receiver")
# 連接smtp服務器,發(fā)送郵件
smtp = smtplib.SMTP()
smtp.connect(cf.get_email("server"))
smtp.login(cf.get_email("user"),cf.get_email("pwd"))
smtp.sendmail(cf.get_email("user"), cf.get_email("receiver"), msg.as_string())
print("郵件發(fā)送成功")
if __name__ == "__main__":
send_email("report.html")
case/case.py: 業(yè)務邏輯層, 為用例執(zhí)行封裝方法
"""
1. 加載數(shù)據(jù)
2. 發(fā)送接口
3. 為用例封裝一些方法
"""
import sys
sys.path.append("..")
from common.log import Log
from common.config import Config
from common.db import DB
from common.data import Data
import json
import requests
class Case(object):
def __init__(self):
self.logger = Log.get_logger()
self.cf = Config()
def load_data(self, data_file):
self.data = Data(data_file)
def set_env(self, env):
self.env = env
def run_case(self, sheet_name, case_name, var={}):
case_data = self.data.get_case(sheet_name, case_name)
url = self.cf.get_server(self.env) + case_data[1]
data = case_data[4].format(**var)
if case_data[3].lower() == "form":
data = json.loads(data)
headers = {}
else:
headers = {"content-type": "application/json"}
if case_data[2].lower() == "get":
resp = requests.get(url=url)
else:
resp = requests.post(url=url, headers=headers, data=data)
return resp.text
def check_response(self):
pass
def check_db(self, sql_name, vars={}):
sql = self.data.get_sql(sql_name).format(**vars)
return self.db.exec_sql(sql)
if __name__ == "__main__":
c = Case()
c.set_env("test")
c.load_data("test_user_data.xlsx")
r = c.run_case("login", "test_login_normal")
print(r)
case/user/test_user.py: 表示層: 測試用例腳本
import sys
import random
import pytest
sys.path.append("../..")
from case.case import Case
case = Case()
def setup_module(module):
case.set_env('dev')
case.load_data('test_user_data.xlsx')
def test_login_normal():
result case.run("login", "test_login_normal")
if __name__ == '__main__':
pytest.main(["-q", "test_user.py"])
run_all.py: 表示層: 執(zhí)行所有用例入口
import os
import time
from util.config import Config
from util.e_mail import send_email
import pytest
def main():
cf = Config()
report_dir = cf.get_report_dir()
now = time.strftime('%Y%m%d%H%M%S', time.localtime(time.time()))
report_name = os.path.join(report_dir, 'report_' + now + '.html')
pytest.main(["-q", "case", "--html=" + report_name])
send_email(report_name)
if __name__ == '__main__':
main()
