数据驱动 - 不同数据源的读取方式(ddt、数据文件、mysql)
阅读原文时间:2023年07月08日阅读:1

1. ddt 装饰器传参

2. ddt 读取数据文件

3. 读取 txt 文件

4. 读取 excel 文件

5. 连接 mysql

1. ddt 装饰器传参

python 的数据驱动模块 ddt

安装:pip install ddt

4 种使用模式:

  1. 引入的装饰器 @ddt
  2. 导入数据的 @data
  3. 拆分数据的 @unpack
  4. 导入外部数据的 @file_data

测试程序

1 from selenium import webdriver
2 import unittest, time
3 import logging, traceback
4 import ddt
5 from selenium.common.exceptions import NoSuchElementException
6
7
8 # 初始化日志对象
9 logging.basicConfig(
10 # 日志级别
11 level = logging.INFO,
12 # 日志格式
13 # 时间、代码所在文件名、代码行号、日志级别名字、日志信息
14 format = '%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
15 # 打印日志的时间
16 datefmt = '%a, %d %b %Y %H:%M:%S',
17 # 日志文件存放的目录(目录必须存在)及日志文件名
18 filename = 'report.log',
19 # 打开日志文件的方式
20 filemode = 'a'
21 )
22
23 @ddt.ddt
24 class TestDemo(unittest.TestCase):
25
26 def setUp(self):
27 self.driver = webdriver.Chrome()
28 self.driver.set_page_load_timeout(10)
29
30 @ddt.data(["hello", "哈喽"],
31 ["hiphop", "嘻哈"],
32 ["morning", "早晨"])
33 @ddt.unpack # 将测试数据解包对应到 testdata 和 expectdata;即该测试方法会执行3次
34 def test_dataDrivenByObj(self, testdata, expectdata):
35 url = "https://www.iciba.com"
36 # 访问百度首页
37 self.driver.get(url)
38 # 设置隐式等待时间为5秒
39 #self.driver.implicitly_wait(5)
40 try:
41 # 找到搜索输入框,并输入测试数据
42 self.driver.find_element_by_xpath("//input[@type='search']").send_keys(testdata)
43 # 找到搜索按钮,并点击
44 self.driver.find_element_by_xpath('//input[@placeholder="请输入您要翻译的单词"]/following-sibling::div').click()
45 time.sleep(5)
46 # 断言期望结果是否出现在页面源代码中
47 self.assertTrue(expectdata in self.driver.page_source)
48 except NoSuchElementException as e:
49 logging.error("查找的页面元素不存在,异常堆栈信息:" + str(traceback.format_exc()))
50 raise NoSuchElementException
51 except AssertionError as e:
52 logging.info("搜索“%s”,期望“%s”,失败" % (testdata, expectdata))
53 raise e
54 except Exception as e:
55 logging.error("未知错误,错误信息:" + str(traceback.format_exc()))
56 raise e
57 else:
58 logging.info("搜索“%s”,期望“%s”通过" % (testdata, expectdata))
59
60 def tearDown(self):
61 self.driver.quit()
62
63
64 if __name__ == '__main__':
65 unittest.main()

当前目录生成的 report.log

Fri, 22 Jan 2021 22:37:42 test.py[line:59] INFO 搜索“hello”,期望“哈喽”通过
Fri, 22 Jan 2021 22:38:07 test.py[line:59] INFO 搜索“glory”,期望“光荣”通过
Fri, 22 Jan 2021 22:38:33 test.py[line:59] INFO 搜索“morning”,期望“早晨”通过

2. ddt 读取数据文件

数据文件:test_data_list.json

[
"邓肯||蒂姆",
"乔丹||迈克尔",
"库里||斯蒂芬",
"杜兰特||凯文",
"詹姆斯||勒布朗"
]

测试报告模板:ReportTemplate.py

1 #encoding=utf-8
2
3 def htmlTemplate(trData):
4 htmlStr = u'''
5 6 7 单元测试报告 8 47 48 49

测试报告


50 51 52 53 54 55 56 57 58 59 ''' # 示例: 60 61 endStr = u''' 62
Search WordsAssert WordsStart TimeWaste Time(s)Status
飞人乔丹21:13:2315s成功
63 64 '''
65 # 拼接完整的测试报告HTML页面代码
66 html = htmlStr + trData + endStr
67 print (html)
68 # 生成.html文件
69 with open(u"testTemplate.html", "w") as fp:
70 fp.write(html)

测试程序

1 from selenium import webdriver
2 import unittest, time
3 import logging, traceback
4 import ddt
5 from ReportTemplate import htmlTemplate
6 from selenium.common.exceptions import NoSuchElementException
7
8
9 # 如果有no json的报错信息,请将json文件存储为utf-8 with Bom
10 # 初始化日志对象
11 logging.basicConfig(
12 # 日志级别
13 level = logging.INFO,
14 # 日志格式
15 # 时间、代码所在文件名、代码行号、日志级别名字、日志信息
16 format = '%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
17 # 打印日志的时间
18 datefmt = '%a, %Y-%m-%d %H:%M:%S',
19 # 日志文件存放的目录(目录必须存在)及日志文件名
20 filename = 'report.log',
21 # 打开日志文件的方式
22 filemode = 'w'
23 )
24
25 @ddt.ddt
26 class TestDemo(unittest.TestCase):
27
28 @classmethod
29 def setUpClass(cls):
30 # 整个测试过程只被调用一次
31 TestDemo.trStr = ""
32
33 def setUp(self):
34 self.driver = webdriver.Chrome()
35 status = None # 用于存放测试结果状态,失败'fail',成功'pass'
36 flag = 0 # 数据驱动测试结果的标志,失败置0,成功置1
37
38 @ddt.file_data("test_data_list.json") # 读取当前目录下的测试数据,每组数据执行该方法一次
39 def test_dataDrivenByFile(self, value):
40 # 决定测试报告中状态单元格中内容的颜色
41 flagDict = {0: 'red', 1: '#00AC4E'}
42
43 url = "https://www.baidu.com"
44 # 访问百度首页
45 self.driver.get(url)
46 # 将浏览器窗口最大化
47 self.driver.maximize_window()
48 print (value)
49 # 将从.json文件中读取出的数据用“||”进行分隔成测试数据
50 # 和期望数据
51 testdata, expectdata = tuple(value.strip().split("||"))
52 # 设置隐式等待时间为10秒
53 self.driver.implicitly_wait(10)
54
55 try:
56 # 获取当前的时间戳,用于后面计算查询耗时用
57 start = time.time()
58 # 获取当前时间的字符串,表示测试开始时间
59 startTime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
60 # 找到搜索输入框,并输入测试数据
61 self.driver.find_element_by_id("kw").send_keys(testdata)
62 # 找到搜索按钮,并点击
63 self.driver.find_element_by_id("su").click()
64 time.sleep(3)
65 # 断言期望结果是否出现在页面源代码中
66 self.assertTrue(expectdata in self.driver.page_source)
67 except NoSuchElementException as e:
68 logging.error(u"查找的页面元素不存在,异常堆栈信息:" + str(traceback.format_exc()))
69 status = 'fail'
70 flag = 0
71 except AssertionError as e:
72 logging.info(u"搜索“%s”,期望“%s”,失败" % (testdata, expectdata))
73 status = 'fail'
74 flag = 0
75 except Exception as e:
76 logging.error(u"未知错误,错误信息:" + str(traceback.format_exc()))
77 status = 'fail'
78 flag = 0
79 else:
80 logging.info(u"搜索“%s”,期望“%s”通过" % (testdata, expectdata))
81 status = 'pass'
82 flag = 1
83 # 计算耗时,从将测试数据输入到输入框中到断言期望结果之间所耗时
84 wasteTime = time.time() - start - 3 # 减去强制等待的3秒
85 # 每一组数据测试结束后,都将其测试结果信息插入表格行
86 # 的HTML代码中,并将这些行HTML代码拼接到变量trStr变量中,
87 # 等所有测试数据都被测试结束后,传入htmlTemplate()函数中
88 # 生成完整测试报告的HTML代码
89 TestDemo.trStr += u'''
90 91 %s 92 %s 93 %s 94 %.2f 95 %s 96
''' % (testdata, expectdata,startTime, wasteTime, flagDict[flag], status)
97
98 def tearDown(self):
99 self.driver.quit()
100
101 @classmethod
102 def tearDownClass(cls):
103 # 写自定义的html测试报告
104 # 整个测试过程只被调用一次
105 htmlTemplate(TestDemo.trStr)
106
107
108 if __name__ == '__main__':
109 unittest.main()

测试日志

Sun, 2021-01-24 20:52:30 data_drivern_by_file.py[line:80] INFO 搜索“邓肯”,期望“蒂姆”通过
Sun, 2021-01-24 20:53:01 data_drivern_by_file.py[line:80] INFO 搜索“乔丹”,期望“迈克尔”通过
Sun, 2021-01-24 20:53:32 data_drivern_by_file.py[line:80] INFO 搜索“库里”,期望“斯蒂芬”通过
Sun, 2021-01-24 20:54:04 data_drivern_by_file.py[line:80] INFO 搜索“杜兰特”,期望“凯文”通过
Sun, 2021-01-24 20:54:35 data_drivern_by_file.py[line:80] INFO 搜索“詹姆斯”,期望“勒布朗”通过

测试报告:

3. 读取 txt 文件

数据文件:data.txt

hiphop||嘻哈
摔跤爸爸||阿米尔
超人||电影

测试程序

1 from selenium import webdriver
2 import time
3 import traceback
4 import os
5
6
7 # 获取当前目录下的数据文件
8 with open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "data.txt")) as fp:
9 data = fp.readlines()
10
11 driver = webdriver.Chrome()
12 # 存放每组数据的测试结果
13 test_result = []
14
15 # 读取每一行测试数据
16 for i in range(len(data)):
17 try:
18 driver.get("http://www.baidu.com")
19 driver.find_element_by_id("kw").send_keys(data[i].split("||")[0].strip())
20 driver.find_element_by_id("su").click()
21 time.sleep(3)
22 assert data[i].split('||')[1].strip() in driver.page_source
23 test_result.append(data[i].strip()+u"||成功\n")
24 print (data[i].split('||')[0].strip()+u"搜索测试执行成功")
25 except AssertionError as e:
26 print (data[i].split('||')[1].strip()+u"测试断言失败")
27 test_result.append(data[i].strip()+u"||断言失败\n")
28 traceback.print_exc()
29 except Exception as e:
30 print (data[i].split('||')[1].strip()+u"测试执行失败")
31 test_result.append(data[i].strip()+u"||异常失败\n")
32 traceback.print_exc()
33
34 # 新建txt文件存放测试结果
35 with open(u"result.txt","w") as fp:
36 fp.writelines(test_result)
37
38 driver.quit()

测试结果:result.txt

hiphop||嘻哈||成功
摔跤爸爸||阿米尔||成功
超人||电影||成功

4. 读取 excel 文件

数据文件:data.xlsx

测试程序

1 from selenium import webdriver
2 import time
3 import datetime
4 from openpyxl import *
5
6
7 wb = load_workbook('data.xlsx')
8 ws = wb.active # 获取第一个sheet
9 print("最大行号:", ws.max_row)
10
11 driver = webdriver.Chrome()
12 test_result = []
13
14 # openpyxl读取excel的行号和列号是从1开始的,所以这里从2(1是标题行)开始迭代遍历测试数据
15 # 且使用切片,必须有结束行的索引号,不能写为[1:]
16 for row in ws[2: ws.max_row]:
17 print(row[1], row[2]) # 该索引方式则从0开始
18 try:
19 driver.get("http://www.baidu.com")
20 driver.find_element_by_id("kw").send_keys(row[1].value)
21 driver.find_element_by_id("su").click()
22 time.sleep(3)
23 assert row[2].value in driver.page_source
24 row[3].value = time.strftime('%Y-%m-%d %H:%M:%S')
25 row[4].value = "成功"
26 except AssertionError as e:
27 row[3].value = time.strftime('%Y-%m-%d %H:%M:%S')
28 row[4].value = "断言失败"
29 except Exception as e:
30 row[3].value = time.strftime('%Y-%m-%d %H:%M:%S')
31 row[4].value = "出现异常失败"
32
33 driver.quit()
34 wb.save(u"data.xlsx") # 注意:该方式将直接覆盖,不是更新

执行结果:data.xlsx

数据文件:data.xlsx

excel 操作的工具类:ExcelUtil.py

1 from openpyxl import load_workbook
2
3
4 class ParseExcel:
5
6 def __init__(self, excelPath, sheetName):
7 # 将要读取的excel加载到内存
8 self.wb = load_workbook(excelPath)
9 # 通过工作表名称获取一个工作表对象
10 self.sheet = self.wb.get_sheet_by_name(sheetName)
11 # 获取工作表中存在数据的区域的最大行号
12 self.maxRowNum = self.sheet.max_row
13
14 def getDatasFromSheet(self):
15 # 用于存放从工作表中读取出来的数据
16 dataList = []
17 # 因为工作表中的第一行是标题行,所以需要去掉
18 for line in self.sheet.rows:
19 # 遍历工作表中数据区域的每一行,
20 # 并将每行中各个单元格的数据取出存于列表tmpList中,
21 # 然后再将存放一行数据的列表添加到最终数据列表dataList中
22 tmpList = []
23 tmpList.append(line[1].value)
24 tmpList.append(line[2].value)
25 dataList.append(tmpList)
26 # 将获取工作表中的所有数据的迭代对象返回
27 return dataList[1:]
28
29
30 if __name__ == '__main__':
31 excelPath = 'data.xlsx'
32 sheetName = "搜索数据表"
33 pe = ParseExcel(excelPath, sheetName)
34 print (pe.getDatasFromSheet())
35 for i in pe.getDatasFromSheet():
36 print (i[0], i[1])

执行结果:

[['邓肯', '蒂姆111'], ['乔丹', '迈克尔'], ['库里', '斯蒂芬']]
邓肯 蒂姆111
乔丹 迈克尔
库里 斯蒂芬

测试程序

1 # encoding=utf-8
2 from selenium import webdriver
3 import unittest, time
4 import logging, traceback
5 import ddt
6 from ExcelUtil import ParseExcel
7 from selenium.common.exceptions import NoSuchElementException
8
9
10 # 初始化日志对象
11 logging.basicConfig(
12 # 日志级别
13 level = logging.INFO,
14 # 日志格式
15 # 时间、代码所在文件名、代码行号、日志级别名字、日志信息
16 format = '%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
17 # 打印日志的时间
18 datefmt = '%a, %Y-%m-%d %H:%M:%S',
19 # 日志文件存放的目录(目录必须存在)及日志文件名
20 filename = 'dataDriveRreport.log',
21 # 打开日志文件的方式
22 filemode = 'w'
23 )
24
25 excelPath = 'data.xlsx'
26 sheetName = "搜索数据表"
27
28 # 创建ParseExcel类的实例对象
29 excel = ParseExcel(excelPath, sheetName)
30
31 @ddt.ddt
32 class TestDemo(unittest.TestCase):
33
34 def setUp(self):
35 self.driver = webdriver.Chrome()
36
37 @ddt.data(*excel.getDatasFromSheet()) # 解包二维列表中的每组子列表数据
38 def test_dataDrivenByFile(self, data):
39 print("*****", data)
40 testData, expectData = tuple(data)
41 url = "http://www.baidu.com"
42 # 访问百度首页
43 self.driver.get(url)
44 # 将浏览器窗口最大化
45 self.driver.maximize_window()
46 print (testData, expectData)
47 # 设置隐式等待时间为10秒
48 self.driver.implicitly_wait(10)
49
50 try:
51 # 获取当前的时间戳,用于后面计算查询耗时用
52 start = time.time()
53 # 获取当前时间的字符串,表示测试开始时间
54 startTime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
55 # 找到搜索输入框,并输入测试数据
56 self.driver.find_element_by_id("kw").send_keys(testData)
57 # 找到搜索按钮,并点击
58 self.driver.find_element_by_id("s").click()
59 time.sleep(3)
60 # 断言期望结果是否出现在页面源代码中
61 self.assertTrue(expectData in self.driver.page_source)
62 end=time.time()
63 print("搜索%s,期望%s" %(testData, expectData))
64 except NoSuchElementException as e:
65 logging.error("查找的页面元素不存在,异常堆栈信息:" + str(traceback.format_exc()))
66 except AssertionError as e:
67 print("断言失败了")
68 logging.info("搜索“%s”,期望“%s”,失败" % (testData, expectData))
69 except Exception as e:
70 logging.error("未知错误,错误信息:" + str(traceback.format_exc()))
71 else:
72 logging.info("搜索“%s”,期望“%s”通过,耗时%s秒" % (testData, expectData, (end-start)/1000))
73
74 def tearDown(self):
75 self.driver.quit()
76
77
78 if __name__ == '__main__':
79 unittest.main()

执行结果:dataDriveRreport.log

Fri, 2021-01-22 23:56:42 data_drivern_by_excel.py[line:67] INFO 搜索“邓肯”,期望“蒂姆111”,失败
Fri, 2021-01-22 23:57:15 data_drivern_by_excel.py[line:71] INFO 搜索“乔丹”,期望“迈克尔”通过,耗时0.004345748424530029秒
Fri, 2021-01-22 23:57:48 data_drivern_by_excel.py[line:71] INFO 搜索“库里”,期望“斯蒂芬”通过,耗时0.004197439908981323秒

5. 连接 mysql

数据表等创建语句:Sql.py

1 # 创建test数据库sql语句
2 create_database = 'CREATE DATABASE IF NOT EXISTS test DEFAULT CHARSET utf8 COLLATE utf8_general_ci;'
3
4 # 创建testdata表
5 drop_table_if_exist_sql="drop table if exists testdata;"
6
7 create_table = """
8 create table testdata(
9 id int not null auto_increment comment '主键',
10 bookname varchar(40) unique not null comment '书名',
11 author varchar(30) not null comment '作者',
12 test_result varchar(30) default null,
13 primary key(id)
14 )engine=innodb character set utf8 comment '测试数据表';
15 """

初始化数据操作:DatabaseInit.py

1 import pymysql
2 from Sql import *
3
4
5 # 本类用于完成初始化数据操作
6 # 创建数据库,创建数据表,向表中插入测试数据
7 class DataBaseInit(object):
8
9 def __init__(self, host, port, dbName, username, password, charset):
10 self.host = host
11 self.port = port
12 self.db = dbName
13 self.user = username
14 self.passwd = password
15 self.charset = charset
16
17 def create(self):
18 try:
19 # 连接mysql数据库
20 conn = pymysql.connect(
21 host = self.host,
22 port = self.port,
23 user = self.user,
24 passwd = self.passwd,
25 charset = self.charset
26 )
27 # 获取数据库游标
28 cur = conn.cursor()
29 # 创建数据库
30 cur.execute(create_database)
31 # 选择创建好的gloryroad数据库
32 conn.select_db("gloryroad")
33 # 创建测试表
34 cur.execute(drop_table_if_exist_sql)
35 cur.execute(create_table)
36 except pymysql.Error as e:
37 raise e
38 else:
39 # 关闭游标
40 cur.close()
41 # 提交操作
42 conn.commit()
43 # 关闭连接
44 conn.close()
45 print (u"创建数据库及表成功")
46
47 def insertDatas(self):
48 try:
49 # 连接mysql数据库中具体某个库
50 conn = pymysql.connect(
51 host = self.host,
52 port = self.port,
53 db = self.db,
54 user = self.user,
55 passwd = self.passwd,
56 charset = self.charset
57 )
58 cur = conn.cursor()
59 # 向测试表中插入测试数据
60 sql = "insert into testdata(bookname, author) values(%s, %s);"
61 res = cur.executemany(sql, [('Selenium WebDriver实战宝典', '吴晓华'),
62 ('HTTP权威指南', '古尔利'),
63 ('探索式软件测试', '惠特克'),
64 ('暗时间', '刘未鹏')])
65 except pymysql.Error as e:
66 raise e
67 else:
68 conn.commit()
69 print (u"初始数据插入成功")
70 # 确认插入数据成功
71 cur.execute("select * from testdata;")
72 for i in cur.fetchall():
73 print (i[1], i[2])
74 cur.close()
75 conn.close()
76
77
78 if __name__ == '__main__':
79 db = DataBaseInit(
80 host="localhost",
81 port=3306,
82 dbName="xxx",
83 username="xxx",
84 password="xxx",
85 charset="utf8"
86 )
87 db.create()
88 db.insertDatas()
89 print ("数据库初始化结束")

mysql 操作工具类:MysqlUtil.py

1 import pymysql
2 from DatabaseInit import DataBaseInit
3
4
5 class MyMySQL(object):
6
7 def __init__(self, host, port, dbName, username, password, charset):
8 # 进行数据库初始化
9 dbInit = DataBaseInit(host, port, dbName, username, password, charset)
10 dbInit.create()
11 dbInit.insertDatas()
12 self.conn = pymysql.connect(
13 host = host,
14 port = port,
15 db = dbName,
16 user = username,
17 passwd = password,
18 charset = charset
19 )
20 self.cur = self.conn.cursor()
21
22 def getDataFromDataBases(self):
23 # 从testdata表中获取需要的测试数据
24 # bookname作为搜索关键词,author作为预期关键词
25 self.cur.execute("select bookname, author from testdata;")
26 # 从查询区域取回所有查询结果
27 datasTuple = self.cur.fetchall()
28 return datasTuple
29
30 def closeDatabase(self):
31 # 数据库后期清理工作
32 self.cur.close()
33 self.conn.commit()
34 self.conn.close()
35
36
37 if __name__ == '__main__':
38 db = MyMySQL(
39 host = "localhost",
40 port = 3306,
41 dbName = "xxx",
42 username = "xxx",
43 password = "xxx",
44 charset = "utf8"
45 )
46 print (db.getDataFromDataBases())

测试程序

1 from selenium import webdriver
2 import time
3 import pymysql
4
5
6 def get_test_data():
7 conn = pymysql.connect(
8 host = "127.0.0.1",
9 port = 3306,
10 user = "xxx",
11 passwd = "xxx" ,
12 db = "xxx",
13 charset = "utf8"
14 )
15 # 使用cursor()方法获取数据库的操作游标
16 cursor = conn.cursor()
17 cursor.execute("select * from testdata;")
18 resSet = cursor.fetchall()
19 print("共%s条数据。" % len(resSet))
20 print(resSet)
21 # 关闭游标
22 cursor.close()
23 # 提交事务
24 conn.commit()
25 # 关闭数据库连接
26 conn.close()
27 return resSet
28
29 def update_test_result(data,result):
30 conn = pymysql.connect(
31 host = "127.0.0.1",
32 port = 3306,
33 user = "root",
34 passwd = "gloryroad" ,
35 db = "gloryroad",
36 charset = "utf8"
37 )
38 # 使用cursor()方法获取数据库的操作游标
39 cursor = conn.cursor()
40 print('update testdata set test_result="'+result+'" where bookname="'+data+'";')
41 update=cursor.execute('update testdata set test_result="'+result+'" where bookname="'+data+'";')
42
43 print( u"修改语句受影响的行数:", update)
44 # 关闭游标
45 cursor.close()
46 # 提交事务
47 conn.commit()
48 # 关闭数据库连接
49 conn.close()
50
51 driver=webdriver.Ie(executable_path="e:\\IEDriverServer")
52 test_result=[]
53
54 for data in get_test_data():
55 print("-------------", data)
56 try:
57 driver.get("http://www.baidu.com")
58 driver.find_element_by_id("kw").send_keys(data[1])
59 driver.find_element_by_id("su").click()
60 time.sleep(3)
61 assert data[2] in driver.page_source
62 update_test_result(data[1], "成功")
63 except AssertionError as e:
64 print( data[2] + "断言失败")
65 update_test_result(data[1], "断言失败")
66 except Exception as e:
67 print(e)
68 print(data[1] + "测试执行出现异常")
69 update_test_result(data[1], "执行出现异常")
70
71 driver.quit()

手机扫一扫

移动阅读更方便

阿里云服务器
腾讯云服务器
七牛云服务器