Python 之测试环境db自动同步

2022-07-04 20:40:20 浏览数 (1)

这是无量测试之道的第194篇原创
分享主题

多套测试环境,如何做基线的数据库级别的同步更新?

应用场景

工作中测试环境有多套时,为保证基础环境配置的一致性,就需要所有测试环境的数据库结构保持一致。

例如:A需求在 beta1 环境进行测试,且A需求提测单中有新增表的 sql,B需求在 beta2 环境进行测试,由于A需求比B需求先发布上线,此时在B需求测试过程中发布时需要将主干的代码合并到当前需求分支(集成测试的需要,可以提前检测出已上线的需求是否对当前在测的需求有影响),代码合并后对应的相关配置也得跟上,否则程序运行时会报错,所以就需要在 beta2 环境更新 beta1 环境A需求新增表的sql。

因为每一次的发布上线都会做数据库级别的同步更新,如果只是两、三个测试环境,使用人工来手动更新也是可以的,如果测试环境多且数据库更新的内容量大,依然使用人工手动更新,效率就会十分低下,同时也会造成一些人为操作的错误。这时自动化同步更新数据库就显得犹为重要了。在效率和正确率上都是完胜手工更新的。

代码实现
代码语言:javascript复制
#coding:utf-8
import pymysql


dbDict = {"test1":"l-test1.beta.ep.tx1.test.io","test2":"l-test2.beta.ep.tx1.test.io",
          "test3":"l-test3.beta.ep.tx1.test.io","test4":"l-test4.beta.ep.tx1.test.io",
          "test5":"l-test5.beta.ep.tx1.test.io","test6":"l-test6.beta.ep.tx1.test.io"}

#这是定义了一个连接db的类,初始化方法里建立连接,并定义了sql 的执行与查询的两个方法

class DBUtils():
    def __init__(self,test):
        print(dbDict.get(test))
        self.conn = pymysql.connect(dbDict.get(test), "root", "123456")
        self.cursor = self.conn.cursor()

    def dbExcute(self,sql):
        print ("execute sql")
        self.cursor.execute(sql)
        print(sql)
        self.dbClose()

    def dbSelect(self,sql):
        print ("------------------------------------")
        print(sql)
        resultList = []
        self.cursor.execute(sql)
        result = self.cursor.fetchall()
        columns = self.cursor.description
        for val in result:
            tempDict = {}
            for cloNum in range(len(columns)):
                tempDict[str(columns[cloNum][0])] = val[cloNum]
            resultList.append(tempDict)
        print("---------------------打印查询结果----------------------")
        print(resultList)
        self.dbClose()
        return resultList

    def dbClose(self):
        self.conn.commit()
        self.cursor.close()
        self.conn.close(


def main(flag,sql):
    dbname = "test"
    envlist=[1, 2, 3, 4, 5, 6]

    for i in envlist:
        dbname  = str(i)
        print("*" * 20   "正在执行的环境是:", dbname   "*" * 20)
        test= DBUtils(dbname)
        if(flag=="exe"):
            test.dbExcute(sql)
        else:
            test.dbSelect(sql)

        dbname = "test"

if __name__ == "__main__":
    sql="ALTER TABLE change_record change operatorEmail email varchar(100) NOT NULL DEFAULT '' COMMENT 'email'"
    main("exe",sql)
总结

由代码实现部分可以看出,有了这个自动同步的自动化脚本,在数据库更新时,只需要传入更新的 sql 语句就可一键自动同步多套测试环境的数据库信息了,十分高效。

end

0 人点赞