自动化测试中对数据恢复的思考与实际业务改造实践

[company_data]
x = 1  # page
y = 1000  # page_size
next_y = 242  # index

[product_data]
x = 1
y = 1000
next_y = 146

import configparser, os


def get_next_data(ini_name):
    """
    ini_name:配置文件中，配置的名字，写在[]里面的那串
    方法返回列表，按顺序分别是  x，y，next_y
    对应的意思是  当前要请求的页码，当前请求的条数，当前从第几条获取
    """
    # 取文件路径
    path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    # 合成配置文件的路径
    ini_path = path + '/config/page_config.ini'
    cf = configparser.ConfigParser()
    cf.read(ini_path)
    # 取上一次请求的页码
    x = int(cf.get(ini_name, 'x'))
    y = int(cf.get(ini_name, 'y'))
    next_y = int(cf.get(ini_name, 'next_y'))

    if next_y + 1 < y:
        # 如果+1还小于当前页码数据，那么下次还在当前页数据
        cf.set(ini_name, 'next_y', str(next_y + 1))
    else:
        # 如果+1等于当前页码数据，那么下次就翻页，x也要+1
        cf.set(ini_name, 'x', str(x + 1))
        cf.set(ini_name, 'next_y', str(0))
    cf.write(open(ini_path, 'w'))
    return [x, y, next_y]

import api_test.config.page_config as page_config

@pytest.mark.rs_smoke
@allure.story("企业管理")
def test_company_manager(self, rs_resource, rs_admin_login, rs_get_admin_user_info, use_db):
    """测试企业增改查接口"""
    user_id = rs_admin_login
    cpy_id = rs_get_admin_user_info
    # 调用方法获取上次读取的位置
    config_data = page_config.get_next_data("company_data")
    x = config_data[0]
    y = config_data[1]
    next_y = config_data[2]
    get_company_list = rs_resource.get_company_list(cpy_id, user_id, x, y)
    company_list = get_company_list["d"]
    company_name = company_list[next_y]["a"]
    company_simple = company_list[next_y]["a"]
    company_num = company_list[next_y]["d"]
    company_manager = self.fake.name()
    company_phone = self.fake.phone_number()
    company_pwd = 123456
    company_type = 3
    sort_id = str(random.randint(1, 100))

    try:
        with allure.step("调用添加企业接口"):
            add_company = rs_resource.add_company(cpy_id, user_id, company_name, company_manager, company_phone,
                                                  company_pwd, company_simple, company_num, company_type)
            # 当错误码返回254，说明该企业信息已存在
            # 此时再调用get_next_data方法，next_y +1
            while add_company["a"] == 254:
                config_data = page_config.get_next_data("company_data")
                x = config_data[0]
                y = config_data[1]
                next_y = config_data[2]
                get_company_list = rs_resource.get_company_list(cpy_id, user_id, x, y)
                company_list = get_company_list["d"]
                company_name = company_list[next_y]["a"]
                company_simple = company_list[next_y]["a"]
                company_num = company_list[next_y]["d"]
                add_company = rs_resource.add_company(cpy_id, user_id, company_name, company_manager, company_phone,
                                                      company_pwd, company_simple, company_num, company_type)
        assert add_company["a"] == 200
        self.company_user_id = add_company["d"]
        select_db = use_db.execute_sql(
            f"SELECT * FROM t_r_company_base WHERE user_id = {self.company_user_id}")  # 查询数据库是否存在新增的数据
        assert company_name in str(select_db)

insert into t_r_user_info ( user_id, user_name, phone, `status`, product_id, create_time ) values ( 473036, 浙江华甸, 13912300000, 1, 2022, 1667443614864 )

select user_id, account_name, pass_word, `status`, product_id, create_time, update_time, ext1, ext2, ext3 from t_r_account where account_name = 13912300000 and product_id = 2022 and status = 1 limit 1

select id, company_id, post_code, post_name, remark, `status`, creator, create_time, ext1, ext2, ext3,classify from t_r_post where post_code = 16 and company_id = 0 order by status DESC limit 

insert into t_r_user_post ( user_id, post_id, post_code, post_name, creator, create_time ) values ( 473036, 4, 16, 客商调度, 429381, 1667443614864 )

select id, company_id, post_code, post_name, remark, `status`, creator, create_time, ext1, ext2, ext3,classify from t_r_post where post_code = 17 and company_id = 0 order by status DESC limit 

insert into t_r_user_post ( user_id, post_id, post_code, post_name, creator, create_time ) values ( 473036, 7, 17, 客商管理员, 429381, 1667443614864 )

insert into t_r_user_company(user_id, cpy_id, type, create_time, creator_id, updater_id, update_time, ext1, ext2, ext3,company_id) values (473036, 10306, 2, 1667443614864, 429381, null, null, null, null, null,10306

insert into t_r_company_extend (company_id, ext_name, ext_code, ext_value, create_time) values (10306, 简称,cpySimpleName, 浙江华甸防雷科技, 1667443614864) , (10306, 企业编码,cpyCode, xchdfl, 1667443614864

insert into t_r_location ( company_id, `name`, addr, addr_detail, addr_point, `status`, contact_company, contact_name, contact_phone, user_id, create_time ) values ( 10306, 浙江华甸防雷科技, 山西省-长治市-襄垣县, 山西省长治市襄垣县古韩镇山西机电职业技术学院(东湖校区), 113.065637,36.542198, 1, 浙江华甸防雷科技股份有限公司(String), 浙江华甸, 13912300000, 473036, 1667443614864 )

select l.id, l.company_id, l.`name`, l.addr, l.addr_detail, l.addr_point, l.`status`, l.contact_company, l.contact_name, l.contact_phone, l.user_id, l.create_time from t_r_location l left join t_r_company_base b on l.company_id = b.id where b.ext1 = 1 and b.status = 1

insert into t_r_transport_line ( company_id, `name`, load_location_id, unload_location_id, create_time, creator_user_id ) values ( 10306, 简称2-浙江华甸防雷科技, 10000, 10306, 1667443614864, 429381 

insert into t_r_transport_line_point(line_id, location, sort, type, status, remark, create_time, creator_id, updater_id, update_time, ext1, ext2, ext3) values (10610, 113.065637,36.542198, 1, 1, 1, 山西省长治市襄垣县古韩镇山西机电职业技术学院(东湖校区), 1667443614864, 429381, null, null, null, null, null) , (10610, 118.92252,40.410167, 0, 1, 1, 河北省秦皇岛市青龙满族自治县青龙镇小楸子沟门, 1667443614864, 429381, null, null, null, null, null) , (10611, 113.065637,36.542198, 0, 1, 1, 山西省长治市襄垣县古韩镇山西机电职业技术学院(东湖校区), 1667443614864, 429381, null, null, null, null, null) , (10611, 118.92252,40.410167, 1, 1, 1, 河北省秦皇岛市青龙满族自治县青龙镇小楸子沟门, 1667443614864, 429381, null, null, null, null, null

insert into t_r_orgination ( code, `name`, parent_id, `type`, company_id, product_id, create_time ) values ( 10306, 浙江华甸防雷科技股份有限公司, 0, 1, 10306, 2022, 1667443614864 )

insert into t_r_orgination ( `name`, parent_id, `type`, company_id, product_id, create_time ) values ( 运输中心, 13442, 2, 10306, 2022, 1667443614898 ) 

insert into t_r_orgination ( `name`, parent_id, `type`, company_id, product_id, create_time ) values ( 司机, 13443, 2, 10306, 2022, 1667443614898 ) 

insert into t_r_orgination ( `name`, parent_id, `type`, company_id, product_id, create_time ) values ( 押运员, 13443, 2, 10306, 2022, 1667443614898 ) 

insert into t_r_orgination ( code, `name`, parent_id, `type`, company_id, product_id, create_time ) values ( 473036, 浙江华甸, 13442, 3, 10306, 2022, 1667443614864 )

select id, code, `name`, parent_id, `type`, company_id, product_id, create_time, update_time, ext1, ext2, ext3,person_count,im_group_id from t_r_orgination where id = 13442

update t_r_orgination SET code = 10306, `name` = 浙江华甸防雷科技股份有限公司, parent_id = 0, `type` = 1, company_id = 10306, product_id = 2022, create_time = 1667443614864, person_count = 1, im_group_id = 0 where id = 13442

select id, code, `name`, parent_id, `type`, company_id, product_id, create_time, update_time, ext1, ext2, ext3,person_count,im_group_id from t_r_orgination where id = 0

insert into t_r_operator_log ( order_id, code, `type`, remark, company_id, user_id, create_time ) values ( 473036, PR001, 1, 【企业注册】手机号：13912300000 用户Id：473036, 10000, 429381, 1667443614908 )

Preparing: select id, code, `name`, parent_id, `type`, company_id, product_id, create_time, update_time, ext1, ext2, ext3,person_count,im_group_id from t_r_orgination WHERE company_id = 10306 and code = 10306 and `type` = 1 order by create_time desc,id desc limit 1

update t_r_orgination SET code = 10306, `name` = 浙江华甸防雷科技股份有限公司, parent_id = 0, `type` = 1, company_id = 10306, product_id = 2022, create_time = 1667443614864, update_time = 1667443614915, person_count = 1, im_group_id = 111150 where id = 13442

@staticmethod
def delete_company_data(use_db, phone_number):
    """
    删除新增企业产生的相关数据
    :param use_db: 数据库实例
    :param phone_number: 企业联系人手机号
    :return:
    """
    # 查询企业ID
    select_add_company_id = f"SELECT id FROM t_r_company_base WHERE `contact_phone`={phone_number};"
    # 查询用户ID
    select_add_user_id = f"SELECT user_id FROM t_r_company_base WHERE `contact_phone`={phone_number};"
    add_company_id = use_db.execute_sql(select_add_company_id)[0]
    add_user_id = use_db.execute_sql(select_add_user_id)[0]
    # 查询线路ID
    select_line_id = f"SELECT id FROM t_r_transport_line WHERE `company_id`={add_company_id};"
    line_ids = use_db.execute_sql(is_fetchall=True, sql=select_line_id)
    line_id_1 = line_ids[0][0]
    line_id_2 = line_ids[1][0]
    # 根据company_id删除资质告警资料配置
    delete_sql_1 = f"DELETE FROM t_r_qualification_alert_config WHERE company_id={add_company_id};"
    # 根据company_id删除组织表相关数据
    delete_sql_2 = f"DELETE FROM t_r_orgination WHERE company_id={add_company_id};"
    # 根据company_id删除操作日志
    delete_sql_3 = f"DELETE FROM t_r_operator_log WHERE company_id={add_company_id};"
    # 根据company_id删除操作资料统计
    delete_sql_4 = f"DELETE FROM t_r_res_sum_total WHERE company_id={add_company_id};"
    # 删除途经点信息
    delete_sql_5 = f"DELETE FROM t_r_transport_line_point WHERE line_id={line_id_1} or line_id={line_id_2};"
    # 删除运输线路表数据
    delete_sql_6 = f"DELETE FROM t_r_transport_line WHERE company_id={add_company_id};"
    # 删除坐标表数据
    delete_sql_7 = f"DELETE FROM t_r_location WHERE company_id={add_company_id};"
    # 删除企业扩展表数据
    delete_sql_8 = f"DELETE FROM t_r_company_extend WHERE company_id={add_company_id};"
    # 删除人员客商关系表数据
    delete_sql_9 = f"DELETE FROM t_r_user_company WHERE company_id={add_company_id};"
    # 删除人员岗位表数据
    delete_sql_10 = f"DELETE FROM t_r_user_post WHERE user_id={add_user_id};"
    # 删除账户表数据
    delete_sql_11 = f"DELETE FROM t_r_account WHERE user_id={add_user_id};"
    # 删除用户信息表数据
    delete_sql_12 = f"DELETE FROM t_r_user_info WHERE user_id={add_user_id};"
    # 删除企业物料绑定关系数据
    delete_sql_13 = f"DELETE FROM t_r_msds_company WHERE code={add_company_id};"
    # 删除企业基础表数据
    delete_sql_14 = f"DELETE FROM t_r_company_base WHERE id={add_company_id};"
    # 执行各个SQL
    use_db.execute_sql(delete_sql_1)
    use_db.execute_sql(delete_sql_2)
    use_db.execute_sql(delete_sql_3)
    use_db.execute_sql(delete_sql_4)
    use_db.execute_sql(delete_sql_5)
    use_db.execute_sql(delete_sql_6)
    use_db.execute_sql(delete_sql_7)
    use_db.execute_sql(delete_sql_8)
    use_db.execute_sql(delete_sql_9)
    use_db.execute_sql(delete_sql_10)
    use_db.execute_sql(delete_sql_11)
    use_db.execute_sql(delete_sql_12)
    use_db.execute_sql(delete_sql_13)
    use_db.execute_sql(delete_sql_14)

def test_1111(self, use_db, rs_resource):
    rs_resource.delete_company_data(use_db=use_db, phone_number=13213213132)

@pytest.mark.rs_smoke
@allure.story("企业管理")
def test_06_company_manager(self, rs_resource, rs_admin_login, rs_get_admin_user_info, use_db):
    """测试企业增改查接口"""
    user_id = rs_admin_login
    cpy_id = rs_get_admin_user_info
    # 此处去除读取ini配置文件逻辑
    # 随便传入一个页码，就算每次都从这一页开始，后面也会删除数据，不会导致数据重复
    page = 120
    get_company_list = rs_resource.get_company_list(cpy_id, user_id, page, 10)
    company_list = get_company_list["d"]
    company_name = company_list[0]["a"]
    company_simple = company_name[0:5]
    company_num = company_list[0]["d"]
    company_manager = self.fake.name()
    company_phone = self.fake.phone_number()
    company_pwd = 123456
    company_type = 3
    sort_id = str(random.randint(1, 100))

    try:
        logger.info(f"新增企业'{company_name}'信息...")
        with allure.step("调用添加企业接口"):
            add_company = rs_resource.add_company(cpy_id, user_id, company_name, company_manager, company_phone,
                                                  company_pwd, company_simple, company_num, company_type)
            while add_company["a"] == 254:
                # 以防万一，还是加入了页码自增+1逻辑，防止这一页的数据被手动用过
                page = page + 1
                get_company_list = rs_resource.get_company_list(cpy_id, user_id, page, 10)
                company_list = get_company_list["d"]
                company_name = company_list[0]["a"]
                company_simple = company_name[0:5]
                company_num = company_list[0]["d"]
                add_company = rs_resource.add_company(cpy_id, user_id, company_name, company_manager, company_phone,
                                                      company_pwd, company_simple, company_num, company_type)
        assert add_company["a"] == 200
        self.company_user_id = add_company["d"]
        select_db = use_db.execute_sql(
            f"SELECT * FROM t_r_company_base WHERE user_id = {self.company_user_id}")  # 查询数据库是否存在新增的数据
        assert company_name in str(select_db)
        logger.info(f"企业'{company_name}'新增信息成功")

        logger.info(f"修改企业'{company_name}'信息...")
        with allure.step("调用修改企业信息接口"):
            select_db = use_db.execute_sql(
                f"SELECT id FROM t_r_company_base WHERE user_id = {self.company_user_id}")  # 查询新增的数据的id
            self.company_id = int(select_db[0])
            modify_company = rs_resource.modify_company(cpy_id, user_id, self.company_id, company_name,
                                                        company_simple, company_manager)
        assert modify_company["a"] == 200
        logger.info(f"修改企业'{company_name}'信息成功")

        logger.info(f"修改企业'{company_name}'账号...")
        with allure.step("调用修改企业账号接口"):
            company_new_phone = self.fake.phone_number()
            modify_company_phone = rs_resource.modify_company_phone(cpy_id, user_id, self.company_user_id,
                                                                    self.company_id, company_new_phone)
        assert modify_company_phone["a"] == 200
        logger.info(f"修改企业'{company_name}'账号成功")

        logger.info(f"修改企业'{company_name}'密码...")
        with allure.step("调用修改企业密码接口"):
            modify_company_pwd = rs_resource.modify_company_pwd(cpy_id, user_id, self.company_user_id,
                                                                self.company_id, company_new_phone,
                                                                company_pwd='654321')
        assert modify_company_pwd["a"] == 200
        logger.info(f"修改企业'{company_name}'密码成功")

        logger.info(f"修改企业'{company_name}'排序...")
        with allure.step("调用修改企业排序接口"):
            modify_company_sort = rs_resource.modify_company_sort(cpy_id, user_id, self.company_id, sort_id)
        assert modify_company_sort["a"] == 200
        logger.info(f"修改企业'{company_name}'排序成功")

        logger.info(f"查询企业'{company_name}'信息...")
        with allure.step("调用查询企业信息接口"):
            query_company = rs_resource.query_company(cpy_id, user_id, company_num, company_type)
        assert query_company["a"] == 200
        logger.info(f"查询企业'{company_name}'信息成功")

        logger.info(f"给企业'{company_name}'绑定物料...")
        with allure.step("调用企业绑定物料接口"):
            get_exist_product_list = rs_resource.get_exist_product_list(cpy_id, user_id, x=1, y=1000)
            exist_product_list = get_exist_product_list["d"]
            exist_productID_list = []
            for i in exist_product_list:
                aa = i["aa"]
                exist_productID_list.append(aa)
            exist_product_id = random.choice(exist_productID_list)
            add_company_product = rs_resource.add_company_product(cpy_id, user_id, self.company_id, company_type,
                                                                  exist_product_id)
        assert add_company_product["a"] == 200
        select_db = use_db.execute_sql(
            f"SELECT * FROM t_r_msds_company WHERE code = {self.company_id}")  # 查询数据库是否存在新增的数据
        assert str(exist_product_id) in str(select_db)
        logger.info(f"企业'{self.company_id}'和物料的绑定关系成功")

        logger.info("删除新增企业产生的相关数据")
        rs_resource.delete_company_data(use_db=use_db, phone_number=company_new_phone)

    except AssertionError as e:
        logger.info(f"企业'{company_name}'增改查失败")
        raise e