接口自动化之 数据库操作

如果是对于查询接口，可以验证response是否符合预期就行；
如果是对于增、删、改接口，需要验证db

import pymssql
import json


class Operationdb:
    def __init__(self):
        self.comn = pymssql.connect(server="172.16.20.61", user="sa", password="sa", database="SHHDSN",as_dict=True)  # 获取连接
        self.cur = self.comn.cursor()  # 获取光标


    def search(self,sql):
        self.cur.execute(sql)
        result = self.cur.fetchall()

        #关闭游标
        self.cur.close()
        #关闭连接
        self.comn.close()

return json.dumps(result) if __name__ == '__main__': operadb = Operationdb() result = operadb.search("SELECT * FROM [SHHDSN].[dbo].[Dsn_admin]") print(result) print(type(result))

也可以将数据库配置数据放到config中读取

case.config

[DB]
db_config={"server":"172.16.20.61","user":"sa","password":"sa","database":"SHHDSN","as_dict":True}

读取配置文件 read_config.py

import configparser
from util.project_path import *

class ReadConfig:
    def read_config(self,config_path,section=None,option = None):
        config = configparser.ConfigParser()
        config.read(config_path,encoding="utf-8")
        return config[section][option]


if __name__ == '__main__':
    print(ReadConfig().read_config(config_path,"DB","db_config"))

operation_db.py

import pymssql
import json
from util.read_config import ReadConfig
from util.project_path import *

class Operationdb:
    def __init__(self):
        # self.comn = pymssql.connect(server="172.16.20.61", user="sa", password="sa", database="SHHDSN",as_dict=True)  # 获取连接
        # self.cur = self.comn.cursor()  # 获取光标
        #读取数据库的配置文件
        db_config = eval(ReadConfig().read_config(config_path,"DB","db_config"))
        #获取连接
        self.comn = pymssql.connect(**db_config)
        #获取游标
        self.cur = self.comn.cursor()
    
    def search(self,sql,state = "all"):   #查询只有一个结果用fetchone，返回的是一个元祖，多个结果用fetchall，返回的是嵌套元祖的 列表
        
        #执行查询语句
        self.cur.execute(sql)
        if state == 1:
            result = self.cur.fetchone()
        else:
            result = self.cur.fetchall()
        
        #关闭游标
        self.cur.close()
        #关闭连接
        self.comn.close()
        return json.dumps(result)
    
if __name__ == '__main__':
    operadb = Operationdb()
    result = operadb.search("SELECT * FROM [SHHDSN].[dbo].[Dsn_admin]")
    print(result)
    print(type(result))