2026/2/19 2:11:08
网站建设
项目流程
win2008r2做网站服务器,南宁快速优化排名,沂源网站开发,郑州建站系统在线咨询测试学习记录#xff0c;仅供参考#xff01;MySQL数据库连接封装MySQL数据库连接操作#xff1b;一般像这种连接之类的是做到配置文件中#xff1b;在测试环境测试功能性需求#xff0c;测试完成之后清理测试时产生的测试数据#xff0c;需要到测试库里面去操作#xf…测试学习记录仅供参考MySQL数据库连接封装MySQL数据库连接操作一般像这种连接之类的是做到配置文件中在测试环境测试功能性需求测试完成之后清理测试时产生的测试数据需要到测试库里面去操作封装方法怎么去读取数据库里面的数据信息封装增删改查自行封装每一种类型都封装一种方法一、查询1、优化 connectMysql.py 文件添加 查询数据库 方法# 导包 import pymysql from util_tools.handle_data.configParse import ConfigParse from util_tools.logs_util.recordlog import logs # 在最外层先实例化读取ini文件对象 conf ConfigParse() class ConnectMysql: # 初始化构造函数 def __init__(self): # 定义一个字典,里面去写数据库的配置参数值通过配置文件去读取 self.conf { host: conf.get_section_mysql(host), port: int(conf.get_section_mysql(port)), user: conf.get_section_mysql(username), password: conf.get_section_mysql(password), database: conf.get_section_mysql(database) } try: # 返回出去给实例变量 self.conn--拿到数据库连接对象 # self.conn pymysql.connect(hostlocalhost, port3306, userroot, password, databaseecshop) # 关键字传参在python中可以通过 **self.conf 以keyvalue形式--拿到数据库连接对象 self.conn pymysql.connect(**self.conf) # 获取操作游标--通过连接对象 点 .cursor() 去获取连接数据库的游标--然后把它的对象给返回出去 self.cursor self.conn.cursor() logs.info(f成功连接到数据库数据库ip为{self.conf.get(host)}) except Exception as e: logs.error(f连接数据库失败原因为{e}) # 关闭连接对象 def close(self): # if判断一下 连接 和 游标 存在的情况 if self.conn and self.cursor: # 直接调用里面close()方法关闭 self.conn.close() self.cursor.close() return True def query(self, sql, fetchallFalse): 查询数据库数据 :param sql: 查询的SQL语句 :param fetchall: 查询全部数据默认为False则查询单条数据 :return: try: # 通过游标去调用execute()方法去执行要传入的sql语句 self.cursor.execute(sql) # 通过数据库连接对象调用commit()方法去提交sql到数据库中去查询数据 self.conn.commit() if fetchall: # 通过游标获取全部的数据 res self.cursor.fetchall() else: # 若fetchallFalse的话则通过游标获取单条数据 res self.cursor.fetchone() return res except Exception as e: logs.error(f查询数据库内容出现异常原因为{e}) finally: self.close() # 测试调试 if __name__ __main__: # 实例化类 conn ConnectMysql() # sql语句 sql select * from ecs_users where user_nametest01 # 调用查询语言 res conn.query(sql) # 打印结果 print(res)2、封装好查询数据库方法后调试查看INFO - 2021-05-20 21:21:21,210 - connectMysql.py:29 -[connectMysql:__init__] - 成功连接到数据库数据库ip为localhost (2, 11126.com, test01, e10adc3949ba59abbe56e057f20f883e, , , 0, 0000-00-00, Decimal(0.00), Decimal(0.00), 0, 0, 0, 1754991723, 1754991723, 0000-00-00 00:00:00, 0.0.0.0, 1, 0, 0, None, 0, 0, 0, , 22qq.com, 123456, 123, 456, 789, 0, Decimal(0.00), motto, 注册测试) 进程已结束退出代码为 03、默认查单条的情况下不使用条件是返回表中第一条数据# 测试调试 if __name__ __main__: # 实例化类 conn ConnectMysql() # sql语句 sql select * from ecs_users # 调用查询语言 res conn.query(sql) # 打印结果 print(res) INFO - 2021-05-20 21:21:21,210 - connectMysql.py:29 -[connectMysql:__init__] - 成功连接到数据库数据库ip为localhost (1, 123126.com, admin123, 4ade30bb981069c6788f8ac73d0c786b, , , 0, 0000-00-00, Decimal(0.00), Decimal(0.00), 0, 0, 1, 1754373833, 1755672594, 0000-00-00 00:00:00, 0.0.0.0, 73, 0, 0, 9622, 0, 0, 0, , 123126.com, 1, 1, 1, 1, 0, Decimal(0.00), motto, 1) 进程已结束退出代码为 04、fetchallTrue 返回全部数据最外层套了一个元组# 测试调试 if __name__ __main__: # 实例化类 conn ConnectMysql() # sql语句 sql select * from ecs_users # 调用查询语言 # res conn.query(sql) # 返回全部数据 res conn.query(sql, fetchallTrue) # 打印结果 print(res) 运行结果 INFO - 2021-05-20 21:21:21,210 - connectMysql.py:31 -[connectMysql:__init__] - 成功连接到数据库数据库ip为localhost ((1, 123126.com, admin123, 4ade30bb981069c6788f8ac73d0c786b, , , 0, 0000-00-00, Decimal(0.00), Decimal(0.00), 0, 0, 1, 1754373833, 1755672594, 0000-00-00 00:00:00, 0.0.0.0, 73, 0, 0, 9622, 0, 0, 0, , 123126.com, 1, 1, 1, 1, 0, Decimal(0.00), motto, 1), (2, 11126.com, test01, e10adc3949ba59abbe56e057f20f883e, , , 0, 0000-00-00, Decimal(0.00), Decimal(0.00), 0, 0, 0, 1754991723, 1754991723, 0000-00-00 00:00:00, 0.0.0.0, 1, 0, 0, None, 0, 0, 0, , 22qq.com, 123456, 123, 456, 789, 0, Decimal(0.00), motto, 注册测试), (3, 12126.com, test02, e10adc3949ba59abbe56e057f20f883e, , , 0, 0000-00-00, Decimal(0.00), Decimal(0.00), 0, 0, 0, 1755008277, 1755008277, 0000-00-00 00:00:00, 0.0.0.0, 1, 0, 0, None, 0, 0, 0, , 22qq.com, 123456, 123, 456, 789, 0, Decimal(0.00), motto, 注册测试), (4, 0123126.com, test0123, e10adc3949ba59abbe56e057f20f883e, , , 0, 0000-00-00, Decimal(0.00), Decimal(0.00), 0, 0, 0, 1755074515, 1755074516, 0000-00-00 00:00:00, 0.0.0.0, 1, 0, 0, None, 0, 0, 0, , 22qq.com, 123456, 123, 456, 789, 0, Decimal(0.00), motto, 注册测试), (5, 03126.com, test03, e10adc3949ba59abbe56e057f20f883e, , , 0, 0000-00-00, Decimal(0.00), Decimal(0.00), 0, 0, 0, 1755077137, 1755077137, 0000-00-00 00:00:00, 0.0.0.0, 1, 0, 0, None, 0, 0, 0, , 22qq.com, 123456, 123, 456, 789, 0, Decimal(0.00), motto, 注册测试), (6, 04126.com, test04, e10adc3949ba59abbe56e057f20f883e, , , 0, 0000-00-00, Decimal(0.00), Decimal(0.00), 0, 0, 0, 1755079769, 1755079769, 0000-00-00 00:00:00, 0.0.0.0, 1, 0, 0, None, 0, 0, 0, , 22qq.com, 123456, 123, 456, 789, 0, Decimal(0.00), motto, 注册测试), (7, 05126.com, test05, e10adc3949ba59abbe56e057f20f883e, , , 0, 0000-00-00, Decimal(0.00), Decimal(0.00), 0, 0, 0, 1755354356, 1755354356, 0000-00-00 00:00:00, 0.0.0.0, 1, 0, 0, None, 0, 0, 0, , 22qq.com, 123456, 123, 456, 789, 0, Decimal(0.00), motto, 注册测试), (8, 06126.com, test06, e10adc3949ba59abbe56e057f20f883e, , , 0, 0000-00-00, Decimal(0.00), Decimal(0.00), 0, 0, 0, 1755596875, 1755596875, 0000-00-00 00:00:00, 0.0.0.0, 1, 0, 0, None, 0, 0, 0, , 22qq.com, 123456, 123, 456, 789, 0, Decimal(0.00), motto, 注册测试)) 进程已结束退出代码为 05、以 key:value 形式打印输入结果在外层加一个参数 cursorpymysql.cursors.DictCursor将数据库表字段显示以key-value形式展示 # 导包 import pymysql from util_tools.handle_data.configParse import ConfigParse from util_tools.logs_util.recordlog import logs # 在最外层先实例化读取ini文件对象 conf ConfigParse() class ConnectMysql: # 初始化构造函数 def __init__(self): # 定义一个字典,里面去写数据库的配置参数值通过配置文件去读取 self.conf { host: conf.get_section_mysql(host), port: int(conf.get_section_mysql(port)), user: conf.get_section_mysql(username), password: conf.get_section_mysql(password), database: conf.get_section_mysql(database) } try: # 返回出去给实例变量 self.conn--拿到数据库连接对象 # self.conn pymysql.connect(hostlocalhost, port3306, userroot, password, databaseecshop) # 关键字传参在python中可以通过 **self.conf 以keyvalue形式--拿到数据库连接对象 self.conn pymysql.connect(**self.conf) # 获取操作游标--通过连接对象 点 .cursor() 去获取连接数据库的游标--然后把它的对象给返回出去 # cursorpymysql.cursors.DictCursor将数据库表字段显示以key-value形式展示 self.cursor self.conn.cursor(cursorpymysql.cursors.DictCursor) logs.info(f成功连接到数据库数据库ip为{self.conf.get(host)}) except Exception as e: logs.error(f连接数据库失败原因为{e}) # 关闭连接对象 def close(self): # if判断一下 连接 和 游标 存在的情况 if self.conn and self.cursor: # 直接调用里面close()方法关闭 self.conn.close() self.cursor.close() return True def query(self, sql, fetchallFalse): 查询数据库数据 :param sql: 查询的SQL语句 :param fetchall: 查询全部数据默认为False则查询单条数据 :return: try: # 通过游标去调用execute()方法去执行要传入的sql语句 self.cursor.execute(sql) # 通过数据库连接对象调用commit()方法去提交sql到数据库中去查询数据 self.conn.commit() if fetchall: # 通过游标获取全部的数据 res self.cursor.fetchall() else: # 若fetchallFalse的话则通过游标获取单条数据 res self.cursor.fetchone() return res except Exception as e: logs.error(f查询数据库内容出现异常原因为{e}) finally: self.close() # 测试调试 if __name__ __main__: # 实例化类 conn ConnectMysql() # sql语句 sql select * from ecs_users where user_nametest01 # 调用查询语言 res conn.query(sql) # 打印结果 print(res)6、测试查看运行结果最外层返回的是一个字典INFO - 2021-05-20 21:21:21,210 - connectMysql.py:30 -[connectMysql:__init__] - 成功连接到数据库数据库ip为localhost {user_id: 2, email: 11126.com, user_name: test01, password: e10adc3949ba59abbe56e057f20f883e, question: , answer: , sex: 0, birthday: 0000-00-00, user_money: Decimal(0.00), frozen_money: Decimal(0.00), pay_points: 0, rank_points: 0, address_id: 0, reg_time: 1754991723, last_login: 1754991723, last_time: 0000-00-00 00:00:00, last_ip: 0.0.0.0, visit_count: 1, user_rank: 0, is_special: 0, ec_salt: None, salt: 0, parent_id: 0, flag: 0, alias: , msn: 22qq.com, qq: 123456, office_phone: 123, home_phone: 456, mobile_phone: 789, is_validated: 0, credit_line: Decimal(0.00), passwd_question: motto, passwd_answer: 注册测试} 进程已结束退出代码为 0二、删除7、优化 connectMysql.py 文件添加 删除数据 方法# 导包 import pymysql from util_tools.handle_data.configParse import ConfigParse from util_tools.logs_util.recordlog import logs # 在最外层先实例化读取ini文件对象 conf ConfigParse() class ConnectMysql: # 初始化构造函数 def __init__(self): # 定义一个字典,里面去写数据库的配置参数值通过配置文件去读取 self.conf { host: conf.get_section_mysql(host), port: int(conf.get_section_mysql(port)), user: conf.get_section_mysql(username), password: conf.get_section_mysql(password), database: conf.get_section_mysql(database) } try: # 返回出去给实例变量 self.conn--拿到数据库连接对象 # self.conn pymysql.connect(hostlocalhost, port3306, userroot, password, databaseecshop) # 关键字传参在python中可以通过 **self.conf 以keyvalue形式--拿到数据库连接对象 self.conn pymysql.connect(**self.conf) # 获取操作游标--通过连接对象 点 .cursor() 去获取连接数据库的游标--然后把它的对象给返回出去 # cursorpymysql.cursors.DictCursor将数据库表字段显示以key-value形式展示 self.cursor self.conn.cursor(cursorpymysql.cursors.DictCursor) logs.info(f成功连接到数据库数据库ip为{self.conf.get(host)}) except Exception as e: logs.error(f连接数据库失败原因为{e}) # 关闭连接对象 def close(self): # if判断一下 连接 和 游标 存在的情况 if self.conn and self.cursor: # 直接调用里面close()方法关闭 self.conn.close() self.cursor.close() return True def query(self, sql, fetchallFalse): 查询数据库数据 :param sql: 查询的SQL语句 :param fetchall: 查询全部数据默认为False则查询单条数据 :return: try: # 通过游标去调用execute()方法去执行要传入的sql语句 self.cursor.execute(sql) # 通过数据库连接对象调用commit()方法去提交sql到数据库中去查询数据 self.conn.commit() if fetchall: # 通过游标获取全部的数据 res self.cursor.fetchall() else: # 若fetchallFalse的话则通过游标获取单条数据 res self.cursor.fetchone() return res except Exception as e: logs.error(f查询数据库内容出现异常原因为{e}) finally: self.close() def delete(self, sql): 删除数据库内容 :param sql: 删除的SQL语句 :return: try: self.cursor.execute(sql) self.conn.commit() logs.info(数据库数据删除成功) except Exception as e: logs.error(f查询数据库内容出现异常原因为{e}) finally: self.close() # 测试调试 if __name__ __main__: # 实例化类 conn ConnectMysql() # sql语句--删除语句 sql delete from ecs_users where user_nametest06 # 调用查询语言 conn.query(sql)8、查看项目根目录 data 目录文件下 register_success.yaml注册成功数据 文件内容更改删除语句的用户名为“test06”user_nametest06- test06, 06126.com, 123456, 123456, 22qq.com, 123456, 123, 456, 789, 注册测试9、可以在 ecs_users 数据库表中查看到确实有 user_name 为 test06 这样的一条数据自行使用数据库管理工具查看10、运行删除语句成功INFO - 2021-05-20 21:21:21,210 - connectMysql.py:30 -[connectMysql:__init__] - 成功连接到数据库数据库ip为localhost 进程已结束退出代码为 011、刷新 ecs_users 数据库表能够发现 user_name 为 test06 的数据已经不见了确定删除成功三、使用怎么在测试框架中用到以注册页面为例测试用例执行完成之后需要把注册成功的用户名数据给删掉方便下次重跑测试脚本12、在前后置操作中添加“后置”操作应用可以在项目根目录 testcase 软件包下 conftest.py 文件中去编写亦可重新创建一个conftest.py 文件可以根据需求在不同目录下创建多个13、在项目根目录下创建一个 conftest.py 文件# 导包 import pytest from util_tools.connectMysql import ConnectMysql from util_tools.logs_util.recordlog import logs # 添加后置应用--数据清理 pytest.fixture(scopesession, autouseTrue) def data_cleaning(): 测试结束后清理测试数据 # 初始化连接数据库对象 conn ConnectMysql() yield logs.info(正在清理测试数据...) # 删除语句注册成功用户--若有别的可以继续写 sql delete from ecs_users where user_nametest06 # 调用删除语句 conn.delete(sql)14、运行主函数 run.py 文件把之前编写的测试用例登录、注册等全部执行一遍到最后会自动去清理注册的数据测试用例执行的有点慢因为之前测试调试时在页面中添加了很多强制等待会造成测试时间很长当把脚本调试成功后根据实际场景有的页面确实需要等待时间就加上有的页面不需要的话就把强制等待给删掉15、测试调试完成后删除多余的调试程序代码# 导包 import pymysql from util_tools.handle_data.configParse import ConfigParse from util_tools.logs_util.recordlog import logs # 在最外层先实例化读取ini文件对象 conf ConfigParse() class ConnectMysql: # 初始化构造函数 def __init__(self): # 定义一个字典,里面去写数据库的配置参数值通过配置文件去读取 self.conf { host: conf.get_section_mysql(host), port: int(conf.get_section_mysql(port)), user: conf.get_section_mysql(username), password: conf.get_section_mysql(password), database: conf.get_section_mysql(database) } try: # 返回出去给实例变量 self.conn--拿到数据库连接对象 # self.conn pymysql.connect(hostlocalhost, port3306, userroot, password, databaseecshop) # 关键字传参在python中可以通过 **self.conf 以keyvalue形式--拿到数据库连接对象 self.conn pymysql.connect(**self.conf) # 获取操作游标--通过连接对象 点 .cursor() 去获取连接数据库的游标--然后把它的对象给返回出去 # cursorpymysql.cursors.DictCursor将数据库表字段显示以key-value形式展示 self.cursor self.conn.cursor(cursorpymysql.cursors.DictCursor) logs.info(f成功连接到数据库数据库ip为{self.conf.get(host)}) except Exception as e: logs.error(f连接数据库失败原因为{e}) # 关闭连接对象 def close(self): # if判断一下 连接 和 游标 存在的情况 if self.conn and self.cursor: # 直接调用里面close()方法关闭 self.conn.close() self.cursor.close() return True def query(self, sql, fetchallFalse): 查询数据库数据 :param sql: 查询的SQL语句 :param fetchall: 查询全部数据默认为False则查询单条数据 :return: try: # 通过游标去调用execute()方法去执行要传入的sql语句 self.cursor.execute(sql) # 通过数据库连接对象调用commit()方法去提交sql到数据库中去查询数据 self.conn.commit() if fetchall: # 通过游标获取全部的数据 res self.cursor.fetchall() else: # 若fetchallFalse的话则通过游标获取单条数据 res self.cursor.fetchone() return res except Exception as e: logs.error(f查询数据库内容出现异常原因为{e}) finally: self.close() def delete(self, sql): 删除数据库内容 :param sql: 删除的SQL语句 :return: try: self.cursor.execute(sql) self.conn.commit() logs.info(数据库数据删除成功) except Exception as e: logs.error(f查询数据库内容出现异常原因为{e}) finally: self.close()未完待续。。。