数据迁移测试经验总结

一、背景

数据迁移测试，往往涉及到功能又涉及到底层数据，在新旧版本之间都需要做兼容，需要关注的内容比较多，涉及的面比较广，可能带来的问题也比较隐蔽，本文以一个相对复杂的数据迁移测试场景为例，简单总结了一些数据迁移测试过程当中的核心关注点及测试方法。

二、测试分析

在进行测试之前，需要对核心的测试点做重点的梳理，其中主要包括本次迁移涉及到的后端接口、对应的数据库表以及数据迁移的接口

2.1涉及到的后端接口

涉及到了15个接口，需要覆盖的场景较多，对应场景及接口按照大的类别来区分，具体的场景在设计用例时以等价类代表用例覆盖

2.2对应数据库关系及注意点

涉及到了4个库7张表，关系复杂，需要验证的数据内容较多，针对性的列出了对应场景和表表直接关系，以及一些细节关注点

2.3数据迁移接口及数据校验接口

三、测试流程

3.1数据校验

保证数据准确性是本次数据迁移最为核心的关键一步，需要对增量数据及全量数据做全面的校验，由于手工测试费时费力，将可执行的数据对比操作都做了脚本自动化来进行校验

增量数据校验脚本

实例转商品，对应实例表数据变更


import pymysql

def SimpleAddData():

# 连接database

con_custom = pymysql.connect(host="localhost", user="root", password="123456", database="customdb", charset="utf8")

con_product = pymysql.connect(host="localhost", user="root", password="123456", database="dcs_product", charset="utf8")

# 得到一个可以执行SQL语句的光标对象

cur_custom = con_custom.cursor()

cur_product = con_product.cursor()
sql1 = "select * from  parammodelcollectiontag where tagid = 40702;"
sql2 = "select * from tag_folder_migration where  tagid = 40702;"
sql3 = "select * from tagforparammodelassemblycollection where tagid = 40702;"

# 执行SQL语句
try:
    cur_custom.execute(sql2)
    custom_res1 = cur_custom.execute(sql1).fetchall()  # 获取对应实例数据
    custom_res2 = cur_custom.execute(sql2).fetchall()  # 获取实例转商品关联数据
    custom_res3 = cur_custom.execute(sql3).fetchall()  # 获取收藏夹和实例关联表对应数据

    # 确定数据不为空
    while custom_res1 and custom_res2 and custom_res3:
        for row in custom_res3:
            collectionid = row[2]

        sql4 = "select * from parammodelassemblycollection  where collectionid = %s;" %collectionid
        product_res = cur_product.execute(sql4).fetchall()  # 通过collectionid获取对应实例
        while product_res:
            print "关联表插入成功且实例生成成功"
except Exception as e:
    raise e
finally:
    con_custom.close()  # 关闭连接
    con_product.close()

if name == 'main': SimpleAddData()

增量数据校验-商品


实例转商品，对应商品表数据变更
import pymysql
def BrandgoodAddData():
# 连接database
con_custom = pymysql.connect(host="localhost", user="root", password="123456", database="customdb", charset="utf8")
con_fenshua = pymysql.connect(host="localhost", user="root", password="123456", database="fenshua123",charset="utf8")

# 得到一个可以执行SQL语句的光标对象
cur_custom = con_custom.cursor()
cur_fenshua = con_fenshua.cursor()


sql1 = "select * from parammodelcollectiontag where tagid=40702;"

# 执行SQL语句
try:

    custom_res = cur_custom.execute().fetchall()
    for row in custom_res:
        collectionid = row[2]  # 获取对应实例的 collectionid
        sql2 = "select * from brandgood where itemid=%s and itemtype=15;" %collectionid
        fenshua_res = cur_fenshua.execute(sql2)  # 查询对应实例是否转商品
        while fenshua_res:
            print "对应实例转商品成功"
except Exception as e:
    raise e
finally:
    con_custom.close()  # 关闭连接
    con_fenshua.close()

if name == 'main': BrandgoodAddData()

全量数据校验脚本

全量保存


import pymysql
from pymongo import MongoClient
def AllDataCompare():
# 连接database
con_product = pymysql.connect(host="localhost", user="root", password="123456", database="dcs_product", charset="utf8")
con_model = pymysql.connect(host="localhost", user="root", password="123456", database="modelcollection",charset="utf8")

# 得到一个可以执行SQL语句的光标对象
cur_product = con_product.cursor()
cur_model = con_model.cursor()



# 执行SQL语句
try:
    sql1 = "select count(1) from parammodelassemblycollection  where brandgoodid is null and deleted=0 ;"
    product_count = cur_product.execute(sql1) #统计实例表中数据总量

    model_count10 = 0
    model_count54 = 0
    for i in range(9):
        sql2 = "select count(1) from prod_favorite_000%s as pf,prod_favorite_folder_000%s as pff  " \
               "where pf.folderid=pff.folderid and pff.folder_type in (3,4,5,6);"%i # 商品收藏数据统计0~9

        model_count10=cur_model.execute(sql2)+model_count10
    for i in range(9,65):
        sql3 = "select count(1) from prod_favorite_00%s as pf,prod_favorite_folder_00%s as pff  " \
               "where pf.folderid=pff.folderid and pff.folder_type in (3,4,5,6);" % i
        model_count54 = cur_model.execute(sql3) + model_count54  # 商品收藏数据统计10~64

    model_count = model_count10 + model_count54
    while product_count == model_count :
            print "实例与商品总数对应"
except Exception as e:
    raise e
finally:
    con_product.close()  # 关闭连接
    con_model.close()

if name == 'main': AllDataCompare()