import xlrd
import xlwt
from xlutils import copy
import requests
import json
import logging
from db.SqlHandler import MySqlDb
from log.logHandler import LogHandler
def login():
url = 'http://124.220.179.221:8081/cms/manage/loginJump.do'
data = {
"userAccount": "admin",
"loginPwd": "123456"
}
response = requests.request(method='post', url=url, data=data)
return response.cookies
Log = LogHandler().log_fun()
class ExcelSet:
# 专门读写excel
def __init__(self, login):
self.excel_path = '../day16/test2.xls' # 文件名以及路径，如果路径或者文件名有中文给前面加一个r拜师原生字符。
self.work_book = xlrd.open_workbook(self.excel_path, formatting_info=True)
self.work_sheet = self.work_book.sheet_by_name('示例') # 通过名称获取sheet
self.cookies = login()
# 拷贝
self.new_work_book = copy.copy(self.work_book)
# 打开拷贝的excel 的sheet
self.new_work_sheet = self.new_work_book.get_sheet(0)
def read_excel_data(self):
for row in range(1, self.work_sheet.nrows):
case_id = self.work_sheet.cell_value(row, 0) # 行 列
desc = self.work_sheet.cell_value(row, 1) #
url = self.work_sheet.cell_value(row, 2) #
method = self.work_sheet.cell_value(row, 3) #
header = self.work_sheet.cell_value(row, 4) #
data = self.work_sheet.cell_value(row, 5) #
setup_sql = self.work_sheet.cell_value(row, 6) # 前置sql
teardown_sql = self.work_sheet.cell_value(row, 7) # 后置sql
except_result = self.work_sheet.cell_value(row, 8) # 预期结果
response_result = self.work_sheet.cell_value(row, 9) # 实际结果
except_sql = self.work_sheet.cell_value(row, 10) # 预期数据库返回的结果
if not url or not data:
logging.error(f'用例{case_id},{desc},url或body为空')
continue
my_test_case = MyTestCase(case_id, desc, url, method, header, data, except_result, response_result,
self.cookies, setup_sql)
ret = my_test_case.test_01_cms()
if ret.get('msg') == eval(except_result).get('msg'):
# 第一层校验
logging.info(f'用例{case_id},{desc},第一层校验通过')
# 第二层校验
if teardown_sql and except_sql:
ret_sql = my_test_case.check_sql(teardown_sql)
if ret_sql:
assert ret_sql[0][0] == except_sql
logging.info(f'用例{case_id},{desc},测试通过!')
self.write_excel_data(row, '测试成功')
else:
logging.error(f'用例{case_id},{desc},第一层校验失败!!!')
self.write_excel_data(row, '测试失败')
def write_excel_data(self, row, msg):
"""
写入数据
:return:
"""
# self.new_work_sheet.write('行下标','列下标','内容')
# 写入数据 到新的excel里面
self.new_work_sheet.write(row, 9, msg)
# 保存excel
self.new_work_book.save('result.xls')
class MyTestCase:
my_sql = MySqlDb()
def __init__(self, case_id, desc, url, method, header, data, except_result, response_result, cookies, setup_sql,
):
self.base_url = 'http://124.220.179.221:8081'
self.case_id = case_id
self.desc = desc
self.url = url
self.method = method
self.header = header
# self.data = json.loads(data)
self.data = eval(data)
self.except_result = except_result
self.response_result = response_result
self.cookies = cookies
self.setup_sql = setup_sql
def test_01_cms(self):
self.check_sql(self.setup_sql)
response = requests.request(method=self.method, url= self.base_url+self.url, data=self.data, cookies=self.cookies)
return response.json()
def check_sql(self, sql):
"""
处理sql
:return:
"""
if sql.lower().startswith('select'):
select_sql = self.my_sql.select_data(sql)
return select_sql
elif sql.lower().startswith('delete'):
del_sql = self.my_sql.del_data(sql)
return del_sql
elif sql.lower().startswith('insert'):
insert_sql = self.my_sql.insert_data(sql)
return insert_sql
elif sql.lower().startswith('update'):
update_sql = self.my_sql.update_data(sql)
return update_sql
else:
logging.error(f'用例{self.case_id},{self.desc}接口,sql语句错误!')
if __name__ == '__main__':
e = ExcelSet(login)
e.read_excel_data()
