kg_person/bin/utils/checkInput.py

#!/usr/bin/env python 
# -*- coding:utf-8 -*-
"""
检测外部输入参数是否匹配
Author: songjian at <songjian12@baidu.com>
Created: 2021/1/12
"""
import sys
import os
import json
import logging

work_path = os.path.dirname(os.path.abspath(__file__))
sdk_path = os.path.dirname(os.path.dirname(work_path))

reload(sys)
sys.setdefaultencoding('utf-8')


def replace_para(case_data, prop, value, replace_key):
    """
    用外部输入参数替换变量
    :param case_data:
    :param prop: 变量
    :param value: 替换值
    :return:
    """
    try:
        replace_list = case_data
        parameter = ""
        if not isinstance(replace_list[replace_key], dict) and not isinstance(replace_list[replace_key], list):
            parameter = replace_list[replace_key]
        else:
            parameter = json.dumps(replace_list[replace_key], ensure_ascii=False)
        if isinstance(value[prop], str) or isinstance(value[prop], unicode) \
                or replace_key == 'check' or replace_key == 'skip':
            parameter = parameter.replace('%s' % prop, str(value[prop]))
        elif isinstance(value[prop], int):
            if '\\"%s\\"' % prop in parameter:
                parameter = parameter.replace('\\"%s\\"' % prop, str(value[prop]))
            else:
                parameter = parameter.replace('"%s"' % prop, str(value[prop]))
        else:
            parameter = parameter.replace('"%s"' % prop, json.dumps(value[prop], ensure_ascii=False))

        if not isinstance(replace_list[replace_key], dict) and not isinstance(replace_list[replace_key], list):
            replace_list[replace_key] = parameter
        else:
            replace_list[replace_key] = json.loads(parameter)
        return replace_list

    except Exception as e:
        logging.error(e)
        return case_data


def check_input(case_dict):
    """
    检测输入参数
    :param case_dict:
    :return:
    """
    # 若不依赖外部信息
    if 'input' not in case_dict.keys():
        return case_dict
    
    # 若依赖的外部信息，通过其他方式获取参数
    input_para = case_dict['input']
    if input_para['type'] != 1:
        return case_dict

    conf = {}
    para = []
    case_data = []
    # 处理文件内部依赖
    try:
        conf = case_dict['input']['parameter_content']
        for case in case_dict['test_case']:
            if 'precondition' in case.keys() and 'input.' in case['precondition']['case_name']:
                case_name_list = case['precondition']['case_name'].split(';')
                back_prop_list = case['precondition']['back_prop'].split(';')
                if 'replace_key' not in case['precondition'].keys() \
                        or not case['precondition']['replace_key']:
                    replace_key = ['parameter']
                else:
                    replace_key = case['precondition']['replace_key'].split(';')
                if len(case_name_list) != len(back_prop_list) or len(case_name_list) != len(replace_key):
                    logging.error('precondition parameter length not equal')
                    return case_dict
                for round in range(len(case_name_list)):
                    case_name = case_name_list[round].split('||')
                    back_prop = back_prop_list[round].split('||')
                    for i in range(len(case_name)):
                        if 'input.' in case_name[i]:
                            prop_list = back_prop[i].split('|')
                            for prop in prop_list:
                                # 缺少参数
                                if prop not in conf.keys() and prop not in para:
                                    para.append(prop)
                                # 替换
                                elif prop in conf.keys():
                                    case = replace_para(case, prop, conf, replace_key[round])
            case_data.append(case)
        # 无法完成替换，需要返回错误
        if len(para) != 0:
            logging.error('缺少必要输入参数,%s: %s' % (input_para['info'], ','.join(para)))
            sys.exit(-1)
        else:
            case_dict['test_case'] = case_data
            return case_dict

    except Exception as e:
        logging.error(e)
        sys.exit(-1)