如何实现Python接口测试数据库封装

发布时间:2020-07-28 11:42:54 作者:小猪
来源:亿速云 阅读:165

这篇文章主要为大家展示了如何实现Python接口测试数据库封装,内容简而易懂,希望大家可以学习一下,学习完之后肯定会有收获的,下面让小编带大家一起来看看吧。

引言

  做接口测试的时候,避免不了操作数据库。因为数据校验需要,测试数据初始化需要、一些参数化场景需要等。

  数据库操作框架设计

  这里主要操作mysql数据库,整体思路:

如何实现Python接口测试数据库封装

  封装实现

  具体代码实现:

import pymysql
import json
 
 
class OperateMysql(object):
  def __init__(self):
    # 数据库初始化连接
    self.connect_interface_testing = pymysql.connect(
      "localhost",
      "root",
      "123456",
      "test",
      charset='utf8mb4',
      cursorclass=pymysql.cursors.DictCursor
    )
 
    # 创建游标操作数据库
    self.cursor_interface_testing = self.connect_interface_testing.cursor()
 
  def select_first_data(self, sql):
    """
    查询第一条数据
    """
    try:
      # 执行 sql 语句
      self.cursor_interface_testing.execute(sql)
    except Exception as e:
      print("执行sql异常:%s"%e)
    else:
      # 获取查询到的第一条数据
      first_data = self.cursor_interface_testing.fetchone()
      # print(first_data)
      # 将返回结果转换成 str 数据格式,禁用acsii编码
      first_data = json.dumps(first_data,ensure_ascii=False)
      # self.connect_interface_testing.close()
      return first_data
 
  def select_all_data(self,sql):
    """
    查询结果集
    """
    try:
      self.cursor_interface_testing.execute(sql)
    except Exception as e:
      print("执行sql异常:%s"%e)
    else:
      first_data = self.cursor_interface_testing.fetchall()
      first_data = json.dumps(first_data,ensure_ascii=False)
      # self.connect_interface_testing.close()
      return first_data
 
  def del_data(self,sql):
    """
    删除数据
    """
    res = {}
    try:
      # 执行SQL语句
      result = self.cursor_interface_testing.execute(sql)
      # print(result)
      if result != 0:
        # 提交修改
        self.connect_interface_testing.commit()
        res = {'删除成功'}
      else:
        res = {'没有要删除的数据'}
    except:
      # 发生错误时回滚
      self.connect_interface_testing.rollback()
      res = {'删除失败'}
    return res
 
  def update_data(self,sql):
    """
    修改数据
    """
    try:
      self.cursor_interface_testing.execute(sql)
      self.connect_interface_testing.commit()
      res = {'更新成功'}
    except Exception as e:
      self.connect_interface_testing.rollback()
      res = {'更新删除'}
    return res
 
  def insert_data(self,sql,data):
    """
    新增数据
    """
 
    try:
      self.cursor_interface_testing.execute(sql,data)
      self.connect_interface_testing.commit()
      res = {data,'新增成功'}
    except Exception as e:
      res = {'新增失败',e}
    return res
  def conn_close(self):
    # 关闭数据库
    self.cursor_interface_testing.close()
 
 
if __name__ == "__main__":
  # ()类的实例化
  om = OperateMysql()
 
  # 新增
  data = [{'id': 1, 'name': '测试', 'age': 15}, {'id': 2, 'name': '老王', 'age': 10}, {'id': 3, 'name': '李四', 'age': 20}]
  for i in data:
    i_data = (i['id'],i['name'],i['age'])
    insert_res = om.insert_data(
      """
       INSERT INTO test_student (id,name,age) VALUES (%s,%s,%s)
      """,i_data
    )
    print(insert_res)
 
  # 查询
  one_data = om.select_first_data(
    """
      SELECT * FROM test_student;
    """
  )
  all_data = om.select_all_data(
    """
    SELECT * FROM test_student;
    """
  )
  print(one_data)
  # all_data字符串类型的list转list
  print("查询总数据:%s",len(json.loads(all_data)),"分别是:%s",all_data)
 
  # 修改
  update_data = om.update_data(
    """
    UPDATE test_student SET name = '王五' WHERE id = 1;
    """
  )
  print(update_data)
 
  # 删除
  del_data = om.del_data(
    """
    DELETE FROM test_student WHERE id in (1,2,3);
    """
  )
  print(del_data)
 
  # 关闭游标
  om.conn_close()

运行结果:

如何实现Python接口测试数据库封装

为了方便演示,先注释删除数据的sql,再执行程序:

如何实现Python接口测试数据库封装

以上就是关于如何实现Python接口测试数据库封装的内容,如果你们有学习到知识或者技能,可以把它分享出去让更多的人看到。

推荐阅读:
  1. python单个接口测试记录
  2. 如何实现Python发送邮件封装

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

python 数据库 如何实现

上一篇:更改jupyter保存地址的方法

下一篇:将gitee远程仓库抓取(克隆)到本地仓库

相关阅读

您好,登录后才能下订单哦!

密码登录
登录注册
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》