Python接口测试实战3(上)- Python操作数据库

Posted caoxiaowei

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Python接口测试实战3(上)- Python操作数据库相关的知识,希望对你有一定的参考价值。

前言#

在功能、接口测试中常常需要通过数据库的操作,来准备数据、检测环境及核对功能、接口的数据库操作是否正确。
在自动化测试中,就需要我们用代码连接数据库自动完成数据准备、环境检查及数据库断言的功能。
使用Python操作MySQL数据库这里我们需要用到三方库PyMySQl

安装方法:pip install pymysql

数据库操作#

  1. 建立数据库连接 conn = pymysql.connect()
  2. 从连接建立操作游标 cur = conn.cursor()
  3. 使用游标执行sql(读/写) cur.execute(sql)
  4. 获取结果(读)/ 提交更改(写) cur.fetchall() / conn.commit()
  5. 关闭游标及连接 cur.close();conn.close()

Copy

import pymysql

 

# 1. 建立连接

conn = pymysql.connect(host=‘1‘,

                    port=,

                    user=‘‘,

                    passwd=‘‘,   # password也可以

                    db=‘api_test‘,

                    charset=‘utf8‘)   # 如果查询有中文需要指定数据库编码

                    

# 2. 从连接建立游标(有了游标才能操作数据库)

cur = conn.cursor()

 

# 3. 查询数据库(读)

cur.execute("select * from user where name=‘张三‘")

 

# 4. 获取查询结果

result = cur.fetchall()

print(result)

 

# 3. 更改数据库(写)

cur.execute("delete from user where name=‘李四‘")

 

# 4. 提交更改

conn.commit()  # 注意是用的conn不是cur

 

# 5. 关闭游标及连接

cur.close()

conn.close()

什么是游标? 游标类似文件句柄,可以逐条的访问数据库执行结果集。pymysql中只能通过游标来执行sql和获取结果

查询操作
使用cur.execute(), 执行数据库查询后无返回的是影响的行数,而非查询结果。我们要使用cur.fetchone()/cur.fetchmany()/cur.fetchall()来获取查询结果

  • cur.fetchone(): 获取一条数据(同时获取的数据会从结果集删除),返回元祖(‘张三‘,‘123456‘)
  • cur.fetchmany(3): 获取多条数据,返回嵌套元祖((‘张三‘,‘123456‘),(‘李四‘,‘123456‘),("王五","123456"))
  • cur.fetchall(): 获取所有数据,返回嵌套元祖,((‘张三‘,‘123456‘),)(只有一条数据时)

注意: 获取完数据后,数据会从数据集中删除,再次获取获取不到,如:

Copy

cur.execute(select * from user where name=‘张三‘)

print(cur.fetchone()) # 结果: (‘张三‘,‘123456‘)

print(cur.fetchone()) # 结果:None

print(cur.fetchall()) # 结果:()

所以我们需要重复使用查询结果时,需要将查询结果赋给某个变量

Copy

cur.execute(select * from user where name=‘张三‘)

result = cur.fetchall()

print(result)  # 结果: (‘张三‘,‘123456‘)

print(result)  # 结果: (‘张三‘,‘123456‘)

修改操作
执行修改数据库的操作后不立即生效,使用连接conn.commit()提交后才生效,支持事物及回滚

Copy

try:

    cur.execute("insert into user (name,password) values (‘张三‘, ‘123456‘)")

    cur.execute("insert into user (name, passwd) values (‘李四‘), ‘123456‘") # 此处sql出错

    conn.commit()  # 使用连接提交所有更改

except Exception as e:

    conn.rollback()  # 回滚所有更改(注意用的是conn)

    print(str(e))

 

封装数据库操作#

由于经常要使用到数据库操作,建议将所有数据库操作封装成公用的数据库模块

  1. 新建db.py, 代码如下:

Copy

import pymysql

 

 

# 获取连接方法

import pymysql

 

 

# 获取连接方法

def get_db_conn():

    conn = pymysql.connect(host=‘‘,

                           port=,

                           user=‘‘,

                           passwd=‘‘,  

                           db=‘api_test‘,

                           charset=‘utf8‘)  # 如果查询有中文,需要指定测试集编码

 

    return conn

 

 

# 封装数据库查询操作

def query_db(sql):

    conn = get_db_conn()  # 获取连接

    cur = conn.cursor()  # 建立游标

    cur.execute(sql)  # 执行sql

    result = cur.fetchall()  # 获取所有查询结果

    cur.close()  # 关闭游标

    conn.close()  # 关闭连接

    return result  # 返回结果

 

 

# 封装更改数据库操作

def change_db(sql):

    conn = get_db_conn()  # 获取连接

    cur = conn.cursor()  # 建立游标

    try:

        cur.execute(sql)  # 执行sql

        conn.commit()  # 提交更改

    except Exception as e:

        conn.rollback()  # 回滚

    finally:

        cur.close()  # 关闭游标

        conn.close()  # 关闭连接

 

 

# 封装常用数据库操作

def check_user(name):

    # 注意sql中‘‘号嵌套的问题

    sql = "select * from user where name = ‘{}‘".format(name)

    result = query_db(sql)

    return True if result else False

 

 

def add_user(name, password):

    sql = "insert into user (name, passwd) values (‘{}‘,‘{}‘)".format(name, password)

    change_db(sql)

 

 

def del_user(name):

    sql = "delete from user where name=‘{}‘".format(name)

    change_db(sql)

相比用例中直接使用sql操作数据库,封装常用的数据库操作会更安全

调用方法(其他模块)

Copy

from db import *

 

if check_user("张三"):

    del_user("张三")

**补充:**另一种封装方法

由于上面这种封装方法,每一次查询都会建立一次数据库连接,效率较低,也可以采用下面面向对象的封装方法

db2.py

Copy

import pymysql

 

 

class DB:

    def __init__(self):

        self.conn = pymysql.connect(host=‘‘,

                    port=,

                    user=‘t‘,

                    passwd=‘‘,   # passwd 不是 password

                    db=‘api_test‘)

        self.cur = self.conn.cursor()

        

    def __del__(self): # 析构函数,实例删除时触发

        self.cur.close()

        self.conn.close()

        

    def query(self, sql):

        self.cur.execute(sql)

        return self.cur.fetchall()

    

    def exec(self, sql):

        try:

            self.cur.execute(sql)

            self.conn.commit()

        except Exception as e:

            self.conn.rollback()

            print(str(e))

    

    def check_user(self,name):

        result = self.query("select * from user where name=‘{}‘".format(name))

        return True if result else False

        

    def del_user(self, name)

        self.exec("delete from user where name=‘{}‘".format(name))

        

使用方法

Copy

from db2 import DB:

 

db = DB()  # 实例化一个数据库操作对象

if db.check_user("张三"):

    db.del_user("张三")

后言#

    • 数据库连接信息建议写到配置文件中,从配置文件中读取
    • sql语句建议先在手工测试一下没有语法问题再进行封装
    • 通过封装各种sql可以完成各种业务操作
    • 更改数据库有风险,操作需谨慎!!!

以上是关于Python接口测试实战3(上)- Python操作数据库的主要内容,如果未能解决你的问题,请参考以下文章

Python接口测试实战3(下)- unittest测试框架

Python接口测试实战5(上) - Git及Jenkins持续集成

Python接口测试实战5(下) - RESTfulWeb Service及Mock Server

Python接口测试实战3(上)- Python操作数据库

Python接口测试实战3(上)- Python操作数据库

Python接口测试实战4(下) - 框架完善:用例基类,用例标签,重新运行上次失败用例