kg_person/bin/utils/readYaml.py

#!/usr/bin/env python 
# -*- coding:utf-8 -*-
"""
读yaml文件
Author: songjian at <songjian12@baidu.com>
Created: 2020/4/21
"""

import sys
import yaml
import json
import logging
import os

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 join_path(yamlPath, paraPath):
    """
    相对路径替换成绝对路径
    :param yamlPath:
    :param paraPath:
    :return:
    """
    join_path = os.path.join(os.path.dirname(yamlPath), paraPath)
    if not os.path.isfile(paraPath) and os.path.isfile(join_path):
        return join_path
    return None


def read_yaml(path):
    """
    读yaml文件
    """
    conf = {}
    with open(sdk_path + '/conf/conf.yaml', 'r') as fc:
        conf = yaml.load(fc.read(), Loader=yaml.FullLoader)
    with open(path, 'r') as f:
        data = yaml.load(f.read(), Loader=yaml.FullLoader)
        logging.info('[%s]test_info:%s' % (os.path.basename(path), data['test_info']))
        logging.info('[%s]yaml_data:%s' % (os.path.basename(path), json.dumps(data, ensure_ascii=False)))
        if 'input' in data.keys():
            if 'type' not in data['input'].keys() or not data['input']['type']:
                data['input']['type'] = 0
            elif data['input']['type'] == 1 and 'parameter_path' in data['input'].keys():
                abs_path = join_path(path, str(data['input']['parameter_path']))
                if abs_path:
                    data['input']['parameter_path'] = abs_path
        for case in data['test_case']:
            try:
                if 'parameter_path' in case.keys():
                    abs_path = join_path(path, str(case['parameter_path']))
                    if abs_path:
                        case['parameter_path'] = abs_path
                if 'check' in case.keys() and 'assert_words' in case['check'].keys():
                    abs_path = join_path(path, str(case['check']['assert_words']))
                    if abs_path:
                        case['check']['assert_words'] = abs_path
                if 'parameter_type' in case.keys() and case['parameter_type'] == 'form-data':
                    lines = []
                    with open(case['parameter_path'], 'r') as fr:
                        lines = fr.readlines()
                    with open(case['parameter_path'], 'w') as fw:
                        fw.write('')
                    with open(case['parameter_path'], 'a') as fa:
                        path_para = case['parameter_path']
                        for line in lines:
                            line = line.decode('utf-8').encode('utf-8').strip()
                            if os.path.isfile(line):
                                abs_path = line
                            elif os.path.isabs(line):
                                abs_path = join_path(path_para, os.path.basename(line))
                            else:
                                abs_path = join_path(path_para, line)
                            if abs_path:
                                fa.write(abs_path)
                                fa.write('\n')
                if 'host' in case.keys():
                    case['host'] = conf['hosts'][case['host']]['hostname']
                else:
                    case['host'] = conf['hosts']['host_1']['hostname']
                if 'invalid' not in case.keys():
                    case['invalid'] = False
                if 'custom' not in case.keys():
                    case['custom'] = False
                if 'request_type' not in case.keys():
                    case['request_type'] = 'post'
                if 'headers' not in case.keys():
                    case['headers'] = None
                if 'parameter_type' not in case.keys():
                    case['parameter_type'] = 'raw'
                if 'para_file' not in case.keys():
                    case['para_file'] = False
                if 'parameter' not in case.keys():
                    case['parameter'] = None
                if 'parameter_path' not in case.keys():
                    case['para_file'] = None
                if 'file_key' not in case.keys():
                    case['file_key'] = 'file'
                if 'level' not in case.keys():
                    case['level'] = []
                elif not isinstance(case['level'], list):
                    case['level'] = []
                if 'cookie' in case.keys():
                    case['cookie'] = conf['users'][case['cookie']]['cookie']
                else:
                    case['cookie'] = conf['users']['user_1']['cookie']
                if 'waittime' not in case.keys():
                    case['waittime'] = 1
                elif not case['waittime'] or type(case['waittime']) != int:
                    case['waittime'] = 0
                if case['cookie']:
                    if case['headers']:
                        case['headers']['cookie'] = case['cookie']
                    else:
                        case['headers'] = {'cookie': case['cookie']}

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


if __name__ == "__main__":
    path = sdk_path + '/test_data/standard/test_homepage/4_homepage.yaml'
    print read_yaml(path)