接口测试---PyMysql

PyMysql数据库操作代码

• 安装 : pip install PyMySQL

数据库应用场景

• 校验测试数据 :

  • http请求发送后,明确会修改表中的数据,但响应结果中没体现

    • 如删除员工(is_delete字段)

• 构造测试数据 :

  • 测试数据使用一次就失效,不能重复使用 : 添加员工(手机号码字段)

  • 测试数据在展开测试前无法确定是否存在 : 查询,修改,删除(使用的 员工ID)

操作步骤!!!!

1. 导包 import pymysql

2. 创建链接 : con=pymysql.connect()

3. 创建游标 : cursor=conn.cursor()

4. 执行SQL语句 : conn.execute(“sql语句”)

   • 查询语句selete : 不修改数据库,返回结果集

     • 从结果集中提取想要的数据,cursor.fetch*

   • 增删改insert,update,delete : 没有结果返回,会修改数据库

     • 执行成功 : 提交事务,conn.commit()

     • 执行失败 : 回滚事务,conn.rollback()

5. 关闭游标 : cursor.close()

6. 关闭链接 : conn.close()

事务

• 概念 : 将数据库中连续的一条或多条SQL语句,看作一个整体,作为一个事务存在,要么都成功,要么都失败(eg : 跨行转账)

• 操作方法 :

  • 提交事务 : commit

  • 回滚事务 : rollback

PyMySQL连接数据库

1. 建立连接方法

conn = pymysql.connect(host="",port=0,user="",password="",database="",charset="")
host : 数据库主机ip地址
port : int类型,数据库使用端口号
user : 数据库用户名
password : 数据库用户密码
database : 数据库名
charset : 字符集,这里写utf8
conn : 成功建立的数据库对象

• 实例1 : 查询sql版本

# 导包
import pymysql
# 建立连接
conn=pymysql.connect(host="localhost",port=3306,user="root",password="123456",database="test",charset="utf8")
# 建立游标
my_cursor=conn.cursor()
# 游标执行命令获取版本信息
my_cursor.execute("select version()")
# 执行抓取数据
result=my_cursor.fetchone()
# 打印输出版本信息----fetchone是元组类型数据,取下标[0]输出数据
print("result= :",result)
print("result= :",result[0])
# 关闭游标
my_cursor.close()
# 关闭链接
conn.close()

• 回顾部分sql语法

# 查询语法:
select * from 表名 where 过滤条件;
select 字段名1,字段名2....表名 where 过滤条件;# 添加语法
insert into 表明 字段名=字段值,字段名2=字段值;# 更新语法
update 表名 set 字段名=字段值 where 过滤条件;# 删除语法
delete from 表名 where 过滤条件;

2. 游标

1. 游标负责提取结果集中的数据

2. 新创建的游标指向结果集0条记录

3. 游标提取数据的方式:每次提取游标指向位置的下一条记录

4. 每成功提取一条记录,游标会自动向下游动

3. 常用方法--提取数据

• fetchone() : 从结果集中提取一条记录

• fetchmany(size) : 从结果集中,提取size条记录

• fetchall() : 提取全部数据

• 游标属性rownumber : 设置游标位置,常用于归0,cursor.rownumber=0

---

• 实例2 : 提取第一条/全部/三四条数据

# 导包
import pymysql
# 创建链接
conn=pymysql.connect(host="localhost",port=3306,user="root",password="123456",database="test",charset="utf8")
# 创建游标
my_cursor=conn.cursor()
# 执行
my_cursor.execute("select * from emp;")
print("输出第一条数据 : ",my_cursor.fetchone())
# 游标归0提取全部
my_cursor.rownumber=0
print("输出全部数据 : ",my_cursor.fetchall())
# 游标置于2,提取3,4
my_cursor.rownumber=2
print("输出3,4条数据 : ",my_cursor.fetchmany(2))
# 关闭游标
my_cursor.close()
# 关闭链接
conn.close()

4. 异常捕获

conn和my_cursor定义在try语句块内,finally中访问不到,两变量定义为全局再使用

# 语法:
try:尝试执行的代码
except Exception as err:出现异常后,执行的代码
finally:无论有没有异常,都会执行的代码
---------------------------------
!!!给cursor和conn创建全局变量,方便全局调用
conn=None
cursor=None

# 导包
import pymysql
# 定义全局变量
conn=None
my_cursor=None
try:
# 创建链接conn=pymysql.connect(host="localhost",port=3306,user="root",password="123456",database="test",charset="utf8")# 创建游标my_cursor=conn.cursor()# 执行my_cursor.execute("select * from emp;")print("输出第一条数据 : ",my_cursor.fetchone())# 游标归0提取全部my_cursor.rownumber=0res=my_cursor.fetchall()print("输出全部数据 : ",res)# 游标置于2,提取3,4my_cursor.rownumber=2print("输出3,4条数据 : ",my_cursor.fetchmany(2))except Exception as err:print("提取3,4条",str(err))finally:# 关闭游标my_cursor.close()# 关闭链接conn.close()

5. 对数据库数据实现增删改操作

• 实例3 : 新增一条数据(insert into salgrade values(6,10001,20001))

import pymysql
my_cursor=None
conn=None
# 不出错的写入try
try:# 创建链接conn=pymysql.connect(host="localhost",port=3306,user="root",password="123456",database="test",charset="utf8")# 创建游标my_cursor=conn.cursor()# 执行my_cursor.execute("insert into salgrade values(6,10001,20001)")# 提交数据conn.commit()except Exception as err:print("插入数据异常 : ",str(err))# 报错回滚事务conn.rollback()finally:# 关闭游标my_cursor.close()# 关闭链接conn.close()

• 将grade=6的最低值提高100

# 导包
import pymysql
my_cursor=None
conn=None# 不出错的写入try
try:# 创建链接---这里的自动提交默认是关闭的,改为True会自动提交数据.conn=pymysql.connect(host="localhost",port=3306,user="root",password="123456",database="test",charset="utf8",autocommit=True)# 创建游标my_cursor=conn.cursor()# 执行my_cursor.execute("update salgrade set `losal`=`losal`+ 100 where grade=6;")# 提交数据
except Exception as err:print("更新sql数据失败 : ",str(err))# 报错回滚事务conn.rollback()
finally:# 关闭游标my_cursor.close()# 关闭链接conn.close()

• 删除grade=6的数据,如果需要删除的数据没有,不会报错!!!!

# 执行
my_cursor.execute("delete from salgrade where grade=6;")
# 方法用conn调
print("Affects rows:",conn.affected_rows())

数据库工具类封装

实现类方法

• 目的 : 减少操作,方便使用

• 封装查询与增删改操作

import pymysql# 定义 数据库的工具类
class DBTools(object):# 创建连接游标的方法--创建一个类方法!@classmethoddef create_conn(cls):conn = pymysql.connect(host="localhost", port=3306, user="root", password="123456",database="test", charset="utf8")# 调用完返回return conn# 查一条记录@classmethoddef query_one(cls,sql):res = Nonemy_cursor=Noneconnect1=Nonetry:# 创建连接# 类名.方法名()得到一个返回的conn连接,将conn的数据存入connect1中connect1=DBTools.create_conn()# 创建游标my_cursor=connect1.cursor()# 执行sql语句my_cursor.execute(sql)# 提取一条数据记录res=my_cursor.fetchone()except Exception as err:print("抛出异常 : ",str(err))finally:# 关闭游标my_cursor.close()# 关闭连接connect1.close()# 返回给调用return res# 添加插入删除update,insert,delete@classmethoddef db_uid(cls, sql) :connect2=Nonemy_cursor2=Nonetry:# 创建连接connect2=DBTools.create_conn()# 创建游标my_cursor2=connect2.cursor()# 执行增删改sql语句my_cursor2.execute(sql)# 提交commitconnect2.commit()except Exception as err:print("抛出异常:",str(err))connect2.rollback()finally:# 关闭游标my_cursor2.close()# 关闭连接connect2.close()if __name__ == "__main__":# 接收res结果# result=DBTools.query_one("select * from emp;")# 打印查询语句的结果# print(result)# result3 = DBTools.db_uid("insert into salgrade values(6,10001,20001)")result2 = DBTools.db_uid("update salgrade set `losal`=`losal`+ 100 where grade=6;")

# 对用户不可见可将方法设为私有
@classmethod
def __create_conn(cls):

删除员工接口数据库操作(解决校验与构造测试数据的方法)

• 思路 :

  1. 构造测试数据 :

     • 删除员工id测试前,通过insert语句插入一个员工id到数据库中

     • 直接用这个id,进行删除员工接口

  2. 校验测试数据 :

     • 发送删除员工接口请求,请求成功后,通过select语句查询is_delete字段的值,如果为1,说明删除成功

     • 使用断言实现

删除员工接口

• 基本实现 :

def test_delete_emp_ok(self):resp=requests.delete(url='http://ihrm-java.itheima.net/api/sys/user/77889900',headers={'Authorization':'61adb484-cdf4-4761-8276-17874caaaf9c'})# 打印响应结果
print(resp.json())

• 构造数据

    def setup(self):# 使用直接封装的 数据库工具,执行insert语句,插入一条带有id记录到数据库中ins_sql="insert into bs_user(id,mobile,username) values('77889900','13943970710','abc001');"DBTools.db_uid(ins_sql)def teardown(self):del_sql="delete from bs_user where id = 77889900;"DBTools.db_uid(del_sql)

• 校验is_delete字段

 # 断言77889900所对应的is_delete记录为1res=DBTools.query_one("select is_delete from bs_user where id=77889900;")# 数据为元组类型,要取下标assert 1 == res[0]

• 完整代码实现

import pymysql
import requestsfrom pymysql07_db_tools import DBToolsclass TestIhrmDelete(object):def setup(self):# 使用直接封装的 数据库工具,执行insert语句,插入一条带有id记录到数据库中ins_sql="insert into bs_user(id,mobile,username) values('77889900','13943970710','abc001');"DBTools.db_uid(ins_sql)def teardown(self):del_sql="delete from bs_user where id = 77889900;"DBTools.db_uid(del_sql)def test_delete_emp_ok(self):resp=requests.delete(url='http://ihrm-java.itheima.net/api/sys/user/77889900',headers={'Authorization':'61adb484-cdf4-4761-8276-17874caaaf9c'})# 打印响应结果print(resp.json())assert 200 == resp.status_codeassert True == resp.json().get("success")assert 10000 == resp.json().get("code")assert "操作成功" == resp.json().get("message")# 断言77889900所对应的is_delete记录为1res=DBTools.query_one("select is_delete from bs_user where id=77889900;")assert 1 == res[0]