您现在的位置是：首页 >技术教程 >接口自动化-让你了解数据库相关知识网站首页技术教程

接口自动化-让你了解数据库相关知识

自由家 2024-10-07 00:01:03
简介接口自动化-让你了解数据库相关知识

目录

python接口自动化实战

数据库

写一个对数据库操作的类

python接口自动化实战

目标

  • 学习数据库相关，用例增加对数据库校验

  • 利用数据库完成对数据查询

  • 如何校验数据库数据，怎样添加校验让程序自己校验（充值、提现、投资接口对金额的校验）

知识补充

  • 用例之间关联性不要太强，比如每个模块的用例都需要登录状态，在每个模块最开始加上一次登录的用例即可。

  • 关于登录、充值啥的，用例里面的手机号最好不要写死，万一该手机号数据有脏数据，处理起来会很麻烦，所以一般在用例里面用变量，将数据放在指定的sheet里面方便读取与修改

  • 什么接口需要数据库校验？钱的变动、增减

数据库

  • 安装mysql.connector模块 pip2 install --user mysql-connector

  • 基本使用

import mysql.connector
db_config = {
    'host':'47.107.168.87',
    'user':'python',
    'password':'python666',
    'port':'3306',
    'database':'future'
}
#创建一个数据库连接
con = mysql.connector.connect(**db_config)
#建立游标cursor
cursor = con.cursor()
#写sql语句
query_sql = 'select * from member where MobilePhone = 13755120064'
#执行sql语句
cursor.execute(query_sql)
#获取结果，打印结果
# res = cursor.fetchone()  #元组  只对一条数据
res = cursor.fetchall()  #列表   多条数据
print res
#关闭游标
cursor.close()
#关闭连接
con.close()

  • 说明

query_sql = 'select * from member'

  • 因为数据库有很多数据，控制台只输出一条，还有报错，是因为cursor.fetchone()

  1. cursor.fetchone() 返回是元组，只针对一条数据

  2. cursor.fetchall() 返回是列表，针对多行数据，列表嵌套元组

  3. 要返回所有结果，就采用cursor.fetchall()

  • 前面说的初始化手机号的方法，可以通过数据库拿到目前最大的手机号进行+1操作，这样保证初始化的手机号不会和数据库里的号码有冲突

写一个对数据库操作的类

  • 在配置文件中配置数据库相关信息

[DB]
db_config = {
    'host':'47.107.168.87',
    'user':'python',
    'password':'python666',
    'port':3306,
    'database':'future',
    'buffered':True
    }  #这是正确写法

今天遇到一个错误，解决了一整天，心累，原因竟然是“配置文件括号缩进”的问题

这个是错误的写法，遇到的同学注意一下，这是个坑

[DB]
db_config = {
    'host':'47.107.168.87',
    'user':'python',
    'password':'python666',
    'port':3306,
    'database':'future',
    'buffered':True
}   #这个是错误的写法

  • 操作数据库的类

import json
import sys
reload(sys)
sys.setdefaultencoding('utf-8')
import mysql.connector
from project_path import *
from read_config import ReadConfig
class DoMysql:
    def do_mysql(self,query_sql,state='all'):  #query查询语句  state-->all  多天   one   一条
        db_config = eval(ReadConfig().get_config(case_config_path,'DB','db_config')) #利用这个类从配置文件里面读取db info
        #创建一个数据库连接
        con =mysql.connector.connect(**db_config)
        #建立游标cursor
        cursor = con.cursor()
        #执行sql语句
        cursor.execute(query_sql)
        # 获取结果，打印结果
        if state==1:
            res = cursor.fetchone()  # 元组  只对一条数据
        else:
            res = cursor.fetchall()  #列表   多条数据
        #关闭游标
        cursor.close()
        #关闭连接
        con.close()
        return res  #这个返回值要放在关闭游标、关闭连接之后

当第二个接口的参数是第一个接口的执行的结果时，该怎么处理

在请求之前完成loan_id的替换，因为loan_id是上一个接口返回的，所以不能在do_excel里面提前处理，

test_http_request.py文件
import unittest
from http_request import HttpRequest
from ddt import ddt,data
from project_path import *
from do_excel import DoExcel
from get_data import GetData
from do_mysql import DoMysql
test_data = DoExcel().get_data(test_case_path)
from my_log import MyLog
my_logger = MyLog()
@ddt()
class TestHttpRequest(unittest.TestCase):
    def setUp(self):
        pass
    @data(*test_data)
    def test_api1(self,item):
        my_logger.info('开始执行用例{0}'.format(item['用例名称']))
        #在请求之前完成loan_id替换，因为loan_id是上一个接口返回的，所以不能在do_excel里面提前处理，
        if item['data'].find('${loan_id}') != -1:
            if getattr(GetData,'loan_id') == None:
                query_sql = 'select max(Id) from loan where MemberID={0}'.format(getattr(GetData,'loan_member_id'))
                loan_id = DoMysql().do_mysql(query_sql)[0][0]
                item['data'] = item['data'].replace('${loan_id}',str(loan_id))
                setattr(GetData,'loan_id',loan_id)  #利用这个反射做存储结果
            else:
                item['data'] = item['data'].replace('${loan_id}', getattr(GetData,'loan_id'))
        res = HttpRequest().http_request(item['url'],item['data'],item['method'])
        try:
            self.assertEqual(item['except'],res.json()['code'])
            TestResult = 'pass'   #成功
        except AssertionError as e:
            TestResult = 'Failed'   #失败
            raise e
        finally:
            DoExcel().write_back(test_case_path,item['sheet_name'],item['case_id']+1,str(res.json()),TestResult)
            print res.json()
            my_logger.debug('获取的结果是：{0}'.format(res.json()))
    def tearDown(self):
        pass

存储数据的脚本get_data.py文件

from project_path import *
import pandas as pd
from openpyxl import load_workbook
class GetData:
    Cookie = None
    loan_id = None
    # NotRegTel = int(pd.read_excel(test_case_path,sheet_name='init').ix[0,0])  #采用pandas获取excel里面的数据
    NotRegTel = load_workbook(test_case_path)['init'].cell(2,1).value
    normal_tel = load_workbook(test_case_path)['init'].cell(3,1).value
    admin_tel = load_workbook(test_case_path)['init'].cell(4,1).value
    loan_member_id = load_workbook(test_case_path)['init'].cell(5,1).value
    memberID = load_workbook(test_case_path)['init'].cell(6,1).value

对于需要进行数据库校验的接口，在请求前后进行数据库值对比。

如：充值前和充值后都要对数据库数据进行对照，应有的余额是否正确。那么在请求前需要查一次余额，请求后需要查一次余额

  • 解决：配置文件中新增配置,通过配置sheet_name来判断是否需要请求前后查询余额

[CHECKLEAVEAMOUNT]
check_list=["recharge","invest"]

使用

#get_data.py
check_list=eval(ReadConfig.get_config(case_config_path, "CHECKLEAVEAMOUNT", "check_list"))
#test_http_request.py
       
        if item['sheet_name'] in getattr(GetData,'check_list'):
            #请求之前，根据用户id查询数据库余额
            query_sql = 'SELECT a.LeaveAmount FROM member a WHERE a.MobilePhone={0}'.format(eval(item['data'])['MobilePhone'])
            Before_amount = DoMysql.do_mysql(query_sql)[0][0]
            res = HttpRequest().http_request(item['url'],item['data'],item['method'],getattr(GetData,'Cookie'))
            #请求之后，根据用户id再次查询数据库余额
            query_sql = 'SELECT a.LeaveAmount FROM member a WHERE a.MobilePhone={0}'.format(eval(item['data'])['MobilePhone'])
            After_amount = DoMysql.do_mysql(query_sql)[0][0]
            #判断（请求后余额-请求前余额）是否等于充值余额，存储判断结果
            if abs(Before_amount-After_amount) == eval(item['data'])['amount']:
                check_res = '金额正确'
            else:
                check_res = '金额错误'
                #上述方法相当于一个模块所有用例都要去check一次，但实际中并不一定需要。
        else:
            res = HttpRequest().http_request(item['url'],item['data'],item['method'],getattr(GetData,'Cookie'))

  • 另一种解决思路：

  1. 上述方法相当于一个模块所有用例都要去check一次，但实际中并不一定需要。

  2. 另一个解决思路：在excel中新增一个字段check_sql，格式：{sql:SELECT a.LeaveAmount FROM member a WHERE a.MobilePhone=1234},检测如果有query_sql就校验数据库，query_sql为None就不校验数据库

 #test_http_request.py
if item["check_sql"]!=None:
    query_sql=eval(item["check_sql"]["sql"])
    #TODO:请求之前数据库余额    
    res = HttpRequest().http_request(item['url'],item['data'],item['method'],getattr(GetData,'Cookie'))
    #TODO:请求之后数据库余额
    
    #TODO:判断（请求后余额-请求前余额）的绝对值是否等于充值余额，存储判断结果
    if （请求后余额-请求前余额）的绝对值==充值余额:
        check_sql_result="数据库校验通过"
    else:
        check_sql_result="数据库校验不通过"
 #TODO:将结果写回
    DoExcel().write_back_data(test_data_path,item["sheet_name"],item["case_id"]+1,10,str(check_sql_result))
else:
    res = HttpRequest().http_request(item['url'],item['data'],item['method'],getattr(GetData,'Cookie'))
把结果写回的函数修改,记得把result和TestResult字段的代码修改：
#do_excel.py
@staticmethod
def write_back_data(file_name,sheet_name,row,col,result):
    wb=load_workbook(file_name)
    sheet=wb[sheet_name]
    sheet.cell(row,col).value=result
    wb.save(file_name) #保存
#result写回
DoExcel.write_back_data(test_data_path,item["sheet_name"],item["case_id"]+1,8,str(json.dumps(res.json(), encoding="utf-8", ensure_ascii=False)))
#TestResult写回
DoExcel.write_back_data(test_data_path,item["sheet_name"],item["case_id"]+1,9,str(TestResult))

  • 怎么把数据库校验结果写到excel中？

思路：多一个CheckResult字段

  • 检查多个点，需要多个sql语句怎么办?

思路：[{sql1:xxxx},{sql2:xxx},…]

  • 技巧：代码调试时，多使用日志，写完后，可以通过改日志级别控制输出的日志。

  作为一位过来人也是希望大家少走一些弯路，希望能对你带来帮助。（WEB自动化测试、app自动化测试、接口自动化测试、持续集成、自动化测试开发、大厂面试真题、简历模板等等），相信能使你更好的进步！ 

留【自动化测试】即可【自动化测试交流】：574737577（备注ccc）icon-default.png?t=N4P3http://qm.qq.com/cgi-bin/qm/qr?_wv=1027&k=eSv73sPKaVXpq9JeRq1Kboh1D-w19xXu&authKey=Qs1Sduo9tp2wfkOYvRyyFCoblyT7fBhH4PPisEb1G63n1UckS6wQU50e9ar1LAOz&noverify=0&group_code=574737577

 

风语者！平时喜欢研究各种技术，目前在从事后端开发工作，热爱生活、热爱工作。

站长推荐