测试学习记录仅供参考连接MySQL数据库及数据库断言MySQL数据库连接封装MySQL数据库连接操作一般像这种连接之类的是做到配置文件中这里已安装过数据库服务如若没有可自行安装在测试环境测试功能性需求测试完成之后清理测试时产生的测试数据需要到测试库里面去操作实现测试结束后清理测试数据数据库调用完成后释放资源例如每次跑完测试用例之后需要到数据库中把注册用户给删除清理掉下次再跑测试用例时可以重复的使用测试数据若没有清理测试数据的话每次跑测试用例之前需要手动修改相关的测试数据测试场景每次跑完测试用例数据之后连接到数据库把本次的数据删除掉下次再执行时每次就可以使用同一套数据去验证相关的页面功能数据之间不会相互污染目的研究数据库做执行完的后置操作“清理数据”建议可参考借鉴使用 Python 语言 从 0 到 1 搭建完整 Web UI自动化测试学习系列 35--二次封装MySQL数据库连接操作1、修改项目“configs”目录文件下的“config.ini”配置文件内容烦请自行准备可连接数据库表[Host] host http://127.0.0.1:8787 [MySQL] ;host localhost host 192.168.88.148 port 3306 user root password 123456789 database testlocal2、在项目“unit_tools”软件包下新建名称为“db_connector”的Python软件包在“db_connector”软件包下新建名称为“connectMysql.py” 的 Python 文件输入以下内容# 导包 import pymysql from unit_tools.handle_data.configParse import ConfigParser # 实例化 conf ConfigParser() # 创建一个类 class ConnectMysql: 连接MySQL数据库进行增删改查操作 # 初始化构造函数 def __init__(self): # 定义一个字典专门去存储配置文件 self.conf { host: conf.get_mysql_conf(host), port: int(conf.get_mysql_conf(port)), user: conf.get_mysql_conf(user), password: conf.get_mysql_conf(password), database: conf.get_mysql_conf(database) } # 添加异常————像这种连接的最好加上异常处理 try: self.conn pymysql.connect(**self.conf) # 获取操作游标 self.cursor self.conn.cursor() print(f成功连接到数据库数据库ip{self.conf.get(host)}) except Exception as e: print(f连接数据库失败原因{e}) def query(self, sql, fetchallFalse): 查询数据库数据 :param sql: 查询的SQL语句 :param fetchall: 查询全部数据默认为False则查询单条数据 :return: try: # 通过 self. 调用游标cursor--调用执行函数去执行sql self.cursor.execute(sql) # 调用构造函数里面 conn 连接这个数据库对象--去commit()提交操作 self.conn.commit() # 判断这个值fetchall等于True时 if fetchall: # 等于True时--通过self.cursor 游标调用方法fetchall()去查询 全部 res self.cursor.fetchall() else: # 不等于True时--通过self.cursor 游标调用方法fetchone()去查询 单条 res self.cursor.fetchone() # 把结果返回出去res return res except Exception as e: print(f查询数据库内容出现异常{e}) # 调试一下查看 if __name__ __main__: # 直接实例化这个类 con ConnectMysql() sql select * from test where age 40 # 此时 fetchallFalse --默认查询单条数据 res con.query(sql) res_all con.query(sql, fetchallTrue) print(res) print(res_all)3、运行“connectMysql.py”文件测试验证可以看到能够成功连接到数据库4、修改“connectMysql.py” 文件内容添加一个方法在获取游标时设置一下输出数据库中的对应字段信息若是通过这种方式后面需要在别的地方使用时可以通过键key拿到对应的value值# 导包 import pymysql from unit_tools.handle_data.configParse import ConfigParser # 实例化 conf ConfigParser() # 创建一个类 class ConnectMysql: 连接MySQL数据库进行增删改查操作 # 初始化构造函数 def __init__(self): # 定义一个字典专门去存储配置文件 self.conf { host: conf.get_mysql_conf(host), port: int(conf.get_mysql_conf(port)), user: conf.get_mysql_conf(user), password: conf.get_mysql_conf(password), database: conf.get_mysql_conf(database) } # 添加异常————像这种连接的最好加上异常处理 try: self.conn pymysql.connect(**self.conf) # 获取操作游标 # cursorpymysql.cursors.DictCursor --将数据库表字段显示以key-value形式显示 self.cursor self.conn.cursor(cursorpymysql.cursors.DictCursor) print(f成功连接到数据库数据库ip{self.conf.get(host)}) except Exception as e: print(f连接数据库失败原因{e}) def query(self, sql, fetchallFalse): 查询数据库数据 :param sql: 查询的SQL语句 :param fetchall: 查询全部数据默认为False则查询单条数据 :return: try: # 通过 self. 调用游标cursor--调用执行函数去执行sql self.cursor.execute(sql) # 调用构造函数里面 conn 连接这个数据库对象--去commit()提交操作 self.conn.commit() # 判断这个值fetchall等于True时 if fetchall: # 等于True时--通过self.cursor 游标调用方法fetchall()去查询 全部 res self.cursor.fetchall() else: # 不等于True时--通过self.cursor 游标调用方法fetchone()去查询 单条 res self.cursor.fetchone() # 把结果返回出去res return res except Exception as e: print(f查询数据库内容出现异常{e}) # 调试一下查看 if __name__ __main__: # 直接实例化这个类 con ConnectMysql() sql select * from test where age 40 # 此时 fetchallFalse --默认查询单条数据 res con.query(sql) res_all con.query(sql, fetchallTrue) print(res) print(res_all)5、继续修改“connectMysql.py” 文件内容添加删除操作可自行测试验证# 导包 import pymysql from unit_tools.handle_data.configParse import ConfigParser # 实例化 conf ConfigParser() # 创建一个类 class ConnectMysql: 连接MySQL数据库进行增删改查操作 # 初始化构造函数 def __init__(self): # 定义一个字典专门去存储配置文件 self.conf { host: conf.get_mysql_conf(host), port: int(conf.get_mysql_conf(port)), user: conf.get_mysql_conf(user), password: conf.get_mysql_conf(password), database: conf.get_mysql_conf(database) } # 添加异常————像这种连接的最好加上异常处理 try: self.conn pymysql.connect(**self.conf) # 获取操作游标 # cursorpymysql.cursors.DictCursor --将数据库表字段显示以key-value形式显示 self.cursor self.conn.cursor(cursorpymysql.cursors.DictCursor) print(f成功连接到数据库数据库ip{self.conf.get(host)}) except Exception as e: print(f连接数据库失败原因{e}) def query(self, sql, fetchallFalse): 查询数据库数据 :param sql: 查询的SQL语句 :param fetchall: 查询全部数据默认为False则查询单条数据 :return: try: # 通过 self. 调用游标cursor--调用执行函数去执行sql self.cursor.execute(sql) # 调用构造函数里面 conn 连接这个数据库对象--去commit()提交操作 self.conn.commit() # 判断这个值fetchall等于True时 if fetchall: # 等于True时--通过self.cursor 游标调用方法fetchall()去查询 全部 res self.cursor.fetchall() else: # 不等于True时--通过self.cursor 游标调用方法fetchone()去查询 单条 res self.cursor.fetchone() # 把结果返回出去res return res except Exception as e: print(f查询数据库内容出现异常{e}) def delete(self, sql): 删除数据库内容 :param sql: 删除的SQL语句 :return: try: self.cursor.execute(sql) self.conn.commit() print(数据库数据删除成功) except Exception as e: print(f删除数据库数据出现异常{e}) # 调试一下查看 if __name__ __main__: # 直接实例化这个类 con ConnectMysql() sql select * from test where age 40 # 此时 fetchallFalse --默认查询单条数据 res con.query(sql) res_all con.query(sql, fetchallTrue) print(res) print(res_all)6、再次修改“connectMysql.py” 文件内容数据库表操作写完之后得去关闭来释放数据库表资源关闭连接对象# 导包 import pymysql from unit_tools.handle_data.configParse import ConfigParser # 实例化 conf ConfigParser() # 创建一个类 class ConnectMysql: 连接MySQL数据库进行增删改查操作 # 初始化构造函数 def __init__(self): # 定义一个字典专门去存储配置文件 self.conf { host: conf.get_mysql_conf(host), port: int(conf.get_mysql_conf(port)), user: conf.get_mysql_conf(user), password: conf.get_mysql_conf(password), database: conf.get_mysql_conf(database) } # 添加异常————像这种连接的最好加上异常处理 try: self.conn pymysql.connect(**self.conf) # 获取操作游标 # cursorpymysql.cursors.DictCursor --将数据库表字段显示以key-value形式显示 self.cursor self.conn.cursor(cursorpymysql.cursors.DictCursor) print(f成功连接到数据库数据库ip{self.conf.get(host)}) except Exception as e: print(f连接数据库失败原因{e}) # 关闭连接对象 def close(self): # 先判断 对象和游标 存在 if self.conn and self.cursor: # 关闭对象 self.conn.close() # 关闭连接 self.cursor.close() return True def query(self, sql, fetchallFalse): 查询数据库数据 :param sql: 查询的SQL语句 :param fetchall: 查询全部数据默认为False则查询单条数据 :return: try: # 通过 self. 调用游标cursor--调用执行函数去执行sql self.cursor.execute(sql) # 调用构造函数里面 conn 连接这个数据库对象--去commit()提交操作 self.conn.commit() # 判断这个值fetchall等于True时 if fetchall: # 等于True时--通过self.cursor 游标调用方法fetchall()去查询 全部 res self.cursor.fetchall() else: # 不等于True时--通过self.cursor 游标调用方法fetchone()去查询 单条 res self.cursor.fetchone() # 把结果返回出去res return res except Exception as e: print(f查询数据库内容出现异常{e}) # 使用 finally --不管有没有异常都会走这里 finally: # 直接调用close去关闭 self.close() def delete(self, sql): 删除数据库内容 :param sql: 删除的SQL语句 :return: try: self.cursor.execute(sql) self.conn.commit() print(数据库数据删除成功) except Exception as e: print(f删除数据库数据出现异常{e}) finally: self.close() # 调试一下查看 if __name__ __main__: # 直接实例化这个类 con ConnectMysql() sql select * from test where age 40 # 此时 fetchallFalse --默认查询单条数据 res con.query(sql) res_all con.query(sql, fetchallTrue) print(res) print(res_all)7、此时运行“connectMysql.py”文件测试验证可以看到报错信息因为目前还是使用之前的测试调试代码需注意改动8、最后修改“connectMysql.py” 文件内容更改测试调试代码# 导包 import pymysql from unit_tools.handle_data.configParse import ConfigParser # 实例化 conf ConfigParser() # 创建一个类 class ConnectMysql: 连接MySQL数据库进行增删改查操作 # 初始化构造函数 def __init__(self): # 定义一个字典专门去存储配置文件 self.conf { host: conf.get_mysql_conf(host), port: int(conf.get_mysql_conf(port)), user: conf.get_mysql_conf(user), password: conf.get_mysql_conf(password), database: conf.get_mysql_conf(database) } # 添加异常————像这种连接的最好加上异常处理 try: self.conn pymysql.connect(**self.conf) # 获取操作游标 # cursorpymysql.cursors.DictCursor --将数据库表字段显示以key-value形式显示 self.cursor self.conn.cursor(cursorpymysql.cursors.DictCursor) print(f成功连接到数据库数据库ip{self.conf.get(host)}) except Exception as e: print(f连接数据库失败原因{e}) # 关闭连接对象 def close(self): # 先判断 对象和游标 存在 if self.conn and self.cursor: # 关闭对象 self.conn.close() # 关闭连接 self.cursor.close() return True def query(self, sql, fetchallFalse): 查询数据库数据 :param sql: 查询的SQL语句 :param fetchall: 查询全部数据默认为False则查询单条数据 :return: try: # 通过 self. 调用游标cursor--调用执行函数去执行sql self.cursor.execute(sql) # 调用构造函数里面 conn 连接这个数据库对象--去commit()提交操作 self.conn.commit() # 判断这个值fetchall等于True时 if fetchall: # 等于True时--通过self.cursor 游标调用方法fetchall()去查询 全部 res self.cursor.fetchall() else: # 不等于True时--通过self.cursor 游标调用方法fetchone()去查询 单条 res self.cursor.fetchone() # 把结果返回出去res return res except Exception as e: print(f查询数据库内容出现异常{e}) # 使用 finally --不管有没有异常都会走这里 finally: # 直接调用close去关闭 self.close() def delete(self, sql): 删除数据库内容 :param sql: 删除的SQL语句 :return: try: self.cursor.execute(sql) self.conn.commit() print(数据库数据删除成功) except Exception as e: print(f删除数据库数据出现异常{e}) finally: self.close() # 调试一下查看 if __name__ __main__: # 直接实例化这个类 con ConnectMysql() sql select * from test where age 40 # 此时 fetchallFalse --默认查询单条数据 res con.query(sql) print(res)9、运行“connectMysql.py”文件测试验证可以看到能够测试成功数据库断言此处仅作示例烦请根据项目实际情况进行操作10、修改项目“testcases”软件包“userManager”目录文件路径下“adduser.yaml”文件内容添加数据库断言字段信息一般情况下是需要跟添加的“username:testadduser ”相同这里仅供参考- baseInfo: api_name: 用户新增 url: /dar/user/addUser method: post # headers: ${get_headers(data)} testCase: - case_name: 用户新增正常校验 data: username: testadduser password: tset6789890 role_id: 123456789 dates: 2000-12-31 phone: 13800000000 token: ${get_extract_data(token)} validation: - code: 200 - eq: {msg:新增成功} - db: select * from test where testnameAnd Me11、修改项目“unit_tools”软件包下的“assertion_utils.py”文件内容添加数据库断言方式封装。# -*- coding:utf-8 -*- # 引入内置模块--类型检测工具--指定类型 from typing import Callable,Any # 引入自定义异常 # from exception_utils.exception import AssertTypeError from unit_tools.exception_utils.exception import AssertTypeError from unit_tools.db_connector.connectMysql import ConnectMysql import jsonpath import operator # 创建一个断言类 Assertions--里面有多种断言模式 class Assertions: 接口断言模式封装 1、状态码断言 2、包含模式断言 3、相等断言 4、不相等断言 5、数据库断言 # 封装成一个类方法 classmethod--说明这个方法不需要访问实例属性 classmethod # 定义一个方法status_code_assert--跟上两个参数 def status_code_assert(cls, expected_result, status_code): 接口的响应状态码断言 :param expected_result: yaml文件code模式中的预期状态码 :param status_code: 接口实际返回的状态码 :return: # 断言状态码标识0 表示成功其他表示失败 failure_count 0 # 先判断它如果不是一个整数类型--使用isinstance判断数据类型函数来判断expected_result不是整数类型 if not isinstance(expected_result, int): # 直接调用int方法给expected_result预期结果转换成整数类型然后把转化之后的数据再传给它自己 expected_result int(expected_result) # 直接判断 预期结果 等于 接口的实际返回状态码 if expected_result status_code: print(f状态码断言成功接口实际返回状态码 {status_code} {expected_result}) else: print(f状态码断言失败接口实际返回状态码 {status_code} {expected_result}) # 标识加1--加1之后就不等于0了表示失败 failure_count 1 # 最后把失败统计的标识返回出去 return failure_count classmethod # 包含模式断言 def contain_assert(cls, expected_result, response): 字符串包含模式断言预期结果字符串是否包含在接口的实际响应返回信息中 :param expected_result: yaml文件中的预期结果 :param response: 接口的实际返回值 :return: failure_count 0 # 调用字典中的items()分别拿到key值、value值 for assert_key, assert_value in expected_result.items(): # print(fkey值{assert_key} -- value值{assert_value}) # 通过调用jsonpath.jsonpath()方法去接口的实际响应结果response中 根据表达式 $.. 去提取 response_list jsonpath.jsonpath(response, f$..{assert_key}) # print(response_list) # 判断提取到的结果存不存在--并且是不是字符串类型 if response_list and isinstance(response_list[0], str): # 如果是字符串类型的话则重新组成一个字符串--把list数据转成字符串 response_str .join(response_list) # print(response_str) success_message f包含模式断言成功预期结果【{assert_value}】存在实际结果【{response_str}】中 failure_message f包含模式断言失败预期结果【{assert_value}】未存在没有在实际结果【{response_str}】中找到 # 判断assert_value是不是包含在response_str里面 if assert_value in response_str: print(success_message) else: failure_count 1 print(failure_message) return failure_count classmethod def equal_assert(cls, expected_result, response): 相等断言根据yaml里面的validation关键词下面的eq模式数据去跟接口实际响应信息对比 :param expected_result: dictyaml里面的eq值 :param response: dict接口实际响应结果 :return: failure_count 0 if isinstance(response, dict) and isinstance(expected_result, dict): # print(fyaml的预期结果{expected_result}) # print(f接口的实际结果{response}) # 找出实际结果与预期结果共同的key值 common_key list(expected_result.keys() response.keys()) if common_key: common_key common_key[0] # print(common_key) # 根据相同的key值去实际结果中获取并重新生成一个实际结果的字典 new_actual_result {common_key: response[common_key]} # print(new_actual_result) # 调用operator.eq()方法 eq_assert operator.eq(new_actual_result, expected_result) if eq_assert: print(f相等断言成功接口实际结果 {new_actual_result} 预期结果{expected_result}) else: failure_count 1 print(f相等断言失败接口实际结果 {new_actual_result} ! 预期结果{expected_result}) else: failure_count 1 print(相等断言失败请检查yaml文件eq模式的预期结果或接口返回值是否正确) return failure_count classmethod def not_equal_assert(cls, expected_result, response): 不相等断言根据yaml里面的validation关键词下面的ne模式数据去跟接口实际响应信息对比 :param expected_result: dictyaml里面的ne值 :param response: dict接口实际响应结果 :return: failure_count 0 if isinstance(response, dict) and isinstance(expected_result, dict): # print(fyaml的预期结果{expected_result}) # print(f接口的实际结果{response}) # 找出实际结果与预期结果共同的key值 common_key list(expected_result.keys() response.keys()) if common_key: common_key common_key[0] # print(common_key) # 根据相同的key值去实际结果中获取并重新生成一个实际结果的字典 new_actual_result {common_key: response[common_key]} # print(new_actual_result) # 调用operator.ne()方法 eq_assert operator.ne(new_actual_result, expected_result) if eq_assert: print(f不相等断言成功接口实际结果 {new_actual_result} ! 预期结果{expected_result}) else: failure_count 1 print(f不相等断言失败接口实际结果 {new_actual_result} 预期结果{expected_result}) else: failure_count 1 print(不相等断言失败请检查yaml文件ne模式的预期结果或接口返回值是否正确) return failure_count classmethod def database_assert(cls, expected_result, status_codeNone): 数据库断言 :param expected_result: yaml文件db模式中的SQL语句预期结果 :param status_code: 不做任何操作,仅做占位 :return: failure_count 0 conn ConnectMysql() db_value conn.query(expected_result) if db_value is not None: print(数据库断言成功) else: failure_count 1 print(数据库断言失败请检查数据库是否存在该数据) return failure_count classmethod # 定义一个方法assert_result 用于管理多种断言模式 def assert_result(cls, expected_result, response, status_code): 断言主函数通过all_flag标记如all_flag 0 表示测试成功否则为失败 :param expected_result: listyaml文件validation关键词下面的预期结果 :param response: (dict)接口的实际响应信息 :param status_code: 接口的实际响应状态码 :return: all_flag 0 # 通过字典映射的关系去管理方法 assert_methods { code: cls.status_code_assert, contain: cls.contain_assert, eq: cls.equal_assert, ne: cls.not_equal_assert, db: cls.database_assert } try: # 先打印查看一下预期结果是什么类型 # print(f打印结果{expected_result}, 打印类型, type(expected_result)) print(f打印结果{expected_result}, 打印类型, type(expected_result)) # 通过for循环 拿到断言模式 for yq in expected_result: # print(yq) # 先拿到 断言模式assert_mode预期结果值assert_value--通过字典items()函数获取到每一个值 for assert_mode, assert_value in yq.items(): # print(f断言模式{assert_mode}--预期值{assert_value}) # 通过assert_methods.get(assert_mode)它的字典去读取对应的断言方法 # 表示assert_method是一个接收两个参数类型为Any表示可以是任意类型并返回整数的可调用对象 assert_method:Callable[[Any, Any], int] assert_methods.get(assert_mode) # 判断获取到的值存在 if assert_method: # 调用对应的断言方法传递适当的参数 # if assert_mode code: if assert_mode in [code, db]: # 返回一下标识 flag flag assert_method(assert_value, status_code) else: flag assert_method(assert_value, response) all_flag flag else: print(f不支持{assert_mode}该断言模式) # 手动抛出异常 raise AssertTypeError(f不支持{assert_mode}该断言模式) except Exception as exceptions: # print(exceptions) # 抛出异常 raise exceptions # 通过assert关键词去断言 标识all_flag 等于0即测试成功不等于0测试失败 assert all_flag 0, 测试失败 print(测试成功)12、运行 run.py 文件执行“__main__”主函数测试验证能够成功连接到数据表数据库断言验证成功。未完待续。。。