kg_person/bin/utils/case_lib.py

#!/usr/bin/env python 
# -*- coding:utf-8 -*-
"""
Created: 2021/01/29
"""
import sys
import pytest
import os
import logging
import time
import json

from bin.utils.sendApi import send_request
from bin.utils.checkResult import check_result
from bin.utils.initRelevance import init_relevance
from bin.user import customer

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

def run_case(case_data, request, case_result):
    """
        单个测试用例运行函数
        case_data: 表示为运行的函数
        request: 表示缓存区的内容
        case_result : 表示case的结果
    """
    logging.info('[%s]info:%s' % (case_data['test_name'], case_data['info']))
    logging.info('[%s]request_data:%s' % (case_data['test_name'], json.dumps(case_data, ensure_ascii=False)))
    if 'detail' in case_data.keys():
        logging.info(case_data['detail'])
    #step1 若为无效的用例，则直接跳过，基本废弃
    if case_data['invalid']:
        assert case_data['invalid'] == True
        return
    if 'skip' in case_data.keys() \
            and check_result(case_data['skip'],
                             [(200, case_result[case_data['skip']['case_name']])]) == case_data['skip']['type']:
        assert True
        return
    # step2 有效测试用例的判断，动态替换参数
    cache_data = request.config.cache.get('cache_data', {})
    case_data, code = init_relevance(case_data, case_result, cache_data)
    assert code == True
    result = {}
    # step3: 自定义函数
    if case_data['custom']:
        func = 'customer.%s' % case_data['custom']
        try:
            result = eval(func)
            case_result[case_data['test_name']] = result[0]
            code = result[1]
        except Exception as e:
            logging.error(e)
            code = False
        finally:
            assert code == True
    else:#非用户自定义函数
        result = send_request(case_data)
        case_result[case_data['test_name']] = result[0][1]
        assert check_result(case_data['check'], result) == True
    
    #step4:断言执行成功，获取正确的结果，将其存储cache
    #存储的模式 [output][case_name]
    data = {}
    if 'output' in case_data.keys():
        key = "%s.%s" % (case_data['output'], case_data['test_name'])
        if key not in cache_data.keys():
            cache_data[key] = {}
        if not case_data['custom']:
            data = result[0][1]
        else:
            data = result[0]
        cache_data[key] = data
        request.config.cache.set('cache_data', cache_data)
    time.sleep(case_data['waittime'])