tester.py

import requests
import json
import time
import datetime
import re
from utils import common
import ast
from bson import ObjectId
from threading import Thread

import ssl

ssl._create_default_https_context = ssl._create_unverified_context


def async_test(f):
    def wrapper(*args, **kwargs):
        thr = Thread(target=f, args=args, kwargs=kwargs)
        thr.start()

    return wrapper


# 基础测试类，负责获取测试用例的参数，请求，验证等信息后，进行测试，测试通过则返回True，不通过则抛出异常 -- 2019-1-7 09:27

# 基础测试类，负责获取测试用例的参数，请求，验证等信息后，进行测试，测试通过则返回{'status': 'ok'} ，
# 不通过则返回{'status': 'failed'} -- 2019-1-11 15:03

class tester:

    def __init__(self, test_case_list, domain, test_result_list=None, max_retries=5, global_vars=None):

        if not isinstance(test_case_list, list):
            raise ValueError('test_case_list must be a list!')

        try:
            from app import nlper
            self.nlper = nlper
        except ImportError as e:
            raise ImportError('nlp模型导入失败！<%s>' % e)

        self.test_case_list = test_case_list
        self.domain = domain
        self.session = requests.Session()

        if isinstance(max_retries, int) and max_retries > 0:
            # 设置连接重试
            a = requests.adapters.HTTPAdapter(max_retries=max_retries)
            self.session.mount('https://', a)
            self.session.mount('http://', a)

        self.test_result_list = test_result_list

        if global_vars is None:
            self.global_vars = {}

    # 异步方便返回测试启动是否成功的提示给前端
    @async_test
    def execute_all_test_and_send_report(self, testing_case_model, test_report_model,
                                         project_id, executor_nick_name, execution_mode):
        test_results = []
        for test_case in self.test_case_list:
            test_start_time = time.time()
            test_start_datetime = datetime.datetime.utcnow()
            test_result = self.execute_single_test(test_case)
            test_end_time = time.time()
            if 'lastManualTestResult' in test_case:
                test_case.pop('lastManualTestResult')
            domain = test_case["domain"] if 'domain' in test_case and isinstance(test_case["domain"], str) and \
                                            not test_case["domain"].strip() == '' else self.domain
            if 'requestProtocol' in test_case and 'route' in test_case:
                url = '%s://%s%s' % (test_case['requestProtocol'].lower(), domain, test_case['route'])
                test_case["url"] = url
            test_result["testBaseInfo"] = test_case
            test_result["testStartTime"] = test_start_datetime
            test_result["spendingTimeInSec"] = round(test_end_time - test_start_time, 3)
            test_results.append(test_result)

        self.test_result_list = test_results
        self.update_case_info(testing_case_model)
        self.send_report(test_report_model, project_id, executor_nick_name, execution_mode)

    # TODO 方便单个接口调试时同步返回结果，需重构
    def execute_all_test_for_cron_and_single_test(self):
        test_results = []
        for test_case in self.test_case_list:
            test_start_time = time.time()
            test_start_datetime = datetime.datetime.utcnow()
            test_result = self.execute_single_test(test_case)
            test_end_time = time.time()
            if 'lastManualTestResult' in test_case:
                test_case.pop('lastManualTestResult')
            domain = test_case["domain"] if 'domain' in test_case and isinstance(test_case["domain"], str) and \
                                            not test_case["domain"].strip() == '' else self.domain
            if 'requestProtocol' in test_case and 'route' in test_case:
                url = '%s://%s%s' % (test_case['requestProtocol'].lower(), domain, test_case['route'])
                test_case["url"] = url
            test_result["testBaseInfo"] = test_case
            test_result["testStartTime"] = test_start_datetime
            test_result["spendingTimeInSec"] = round(test_end_time - test_start_time, 3)
            test_results.append(test_result)
        return test_results

    def execute_single_test(self, test_case):
        returned_data = dict()
        returned_data["_id"] = test_case["_id"]
        returned_data["testConclusion"] = []
        if not isinstance(test_case, dict):
            returned_data["status"] = 'failed'
            returned_data["testConclusion"].append('测试用例结构不正确！ ')
            return returned_data

        def validate_test_case(test_case):
            compulsory_key_list = ['requestProtocol', 'route', 'requestMethod']
            return all([compulsory_key in test_case.keys() for compulsory_key in compulsory_key_list])

        if not validate_test_case(test_case):
            returned_data["status"] = 'failed'
            returned_data["testConclusion"].append('测试用例缺失必要参数！ ')
            return returned_data

        if test_case.get('isClearCookie'):
            self.session.cookies.clear()

        session = self.session

        url = None
        method = None
        json_data = None
        headers = dict()
        check_http_code = None
        check_response_data = None
        check_response_number = None
        check_response_similarity = None
        set_global_vars = None  # for example {'status': ['status']}

        domain = test_case["domain"] if 'domain' in test_case and isinstance(test_case["domain"], str) and \
                                        not test_case["domain"].strip() == '' else self.domain
        if 'requestProtocol' in test_case and 'route' in test_case:
            test_case['route'] = \
                common.resolve_global_var(pre_resolve_var=test_case['route'], global_var_dic=self.global_vars) \
                    if isinstance(test_case['route'], str) else test_case['route']
            url = '%s://%s%s' % (test_case['requestProtocol'].lower(), domain, test_case['route'])

        if 'requestMethod' in test_case:
            method = test_case['requestMethod']

        if 'presendParams' in test_case and isinstance(test_case['presendParams'], dict):
            # dict 先转 str，方便全局变量替换
            test_case['presendParams'] = str(test_case['presendParams'])

            # 全局替换
            test_case['presendParams'] = common.resolve_global_var(pre_resolve_var=test_case['presendParams'],
                                                                   global_var_dic=self.global_vars)

            # 转回 dict
            test_case['presendParams'] = ast.literal_eval(test_case['presendParams'])

            json_data = test_case['presendParams']

        if 'headers' in test_case and not test_case['headers'] in ["", None, {}, {'': ''}]:
            if isinstance(test_case['headers'], list):
                for header in test_case['headers']:
                    if not header['name'].strip() == '':
                        headers[header['name']] = \
                            common.resolve_global_var(pre_resolve_var=header['value'], global_var_dic=self.global_vars) \
                                if isinstance(header['value'], str) else headers[header['name']]
            else:
                raise TypeError('headers must be list!')

        if 'setGlobalVars' in test_case and not test_case['setGlobalVars'] in [[], {}, "", None]:
            set_global_vars = test_case['setGlobalVars']

        headers = None if headers == {} else headers

        test_case['cookies'] = []
        for key, value in session.cookies.items():
            cookie_dic = dict()
            cookie_dic['name'] = key
            cookie_dic['value'] = value
            test_case['cookies'].append(cookie_dic)

        try:

            use_json_data = len(list(filter(lambda x: str(x).lower() == 'content-type' and 'json'
                                                      in headers[x], headers.keys() if headers else {}))) > 0

            if test_case['requestMethod'].lower() == 'get':
                response = session.request(url=url, method=method, params=json_data, headers=headers, verify=False)
            else:
                response = session.request(url=url, method=method, json=json_data, headers=headers,
                                           verify=False) if use_json_data \
                    else session.request(url=url, method=method, data=json_data, headers=headers, verify=False)

        except BaseException as e:
            returned_data["status"] = 'failed'
            returned_data["testConclusion"].append('请求失败, 错误信息: <%s> ' % e)
            return returned_data

        test_case['headers'] = headers  # 重新赋值生成报告时用

        response_status_code = response.status_code
        returned_data["responseHttpStatusCode"] = response_status_code
        returned_data["responseData"] = response.text

        try:
            response_json = json.loads(response.text) if isinstance(response.text, str) \
                                                         and response.text.strip() else {}
        except BaseException as e:

            if set_global_vars and isinstance(set_global_vars, list):
                for set_global_var in set_global_vars:
                    if isinstance(set_global_var, dict) and isinstance(set_global_var.get('name'), str):
                        name = set_global_var.get('name')
                        query = set_global_var.get('query')
                        value = common.dict_get(response.text, query)
                        self.global_vars[name] = str(value) if value else value

            if 'checkHttpCode' in test_case and not test_case['checkHttpCode'] in ["", None]:
                check_http_code = test_case['checkHttpCode']

            if check_http_code and not str(response_status_code) == str(check_http_code):
                returned_data["status"] = 'failed'
                returned_data["testConclusion"].append('响应状态码错误, 期待值: <%s>, 实际值: <%s>。\t'
                                                       % (check_http_code, response_status_code))
                return returned_data

            is_check_res_data_valid = isinstance(test_case.get('checkResponseData'), list) and \
                                      len(list(filter(lambda x: str(x.get('regex')).strip() == '',
                                                      test_case.get('checkResponseData')))) < 1
            is_check_res_similarity_valid = isinstance(test_case.get('checkResponseSimilarity'), list) and \
                                            len(list(filter(lambda x: isinstance(x.get('targetSimilarity'), type(None)),
                                                            test_case.get('checkResponseSimilarity')))) < 1
            is_check_res_number_valid = isinstance(test_case.get('checkResponseNumber'), list) and \
                                        len(list(filter(lambda x: str(x.get('expressions').get('expectResult')).strip()
                                                                  == '', test_case.get('checkResponseNumber')))) < 1
            # TODO 目前默认当 is_check_res_similarity_valid 和  is_check_res_number_valid 为真时，返回格式必须可转 json ，可优化
            is_test_failed = is_check_res_data_valid or is_check_res_number_valid or is_check_res_similarity_valid

            returned_data['status'] = 'failed' if is_test_failed else 'ok'

            returned_data["testConclusion"].append('服务器返回格式不是json, 错误信息: %s, 服务器返回为: %s '
                                                   % (e, response.text)) if returned_data.get('status') and \
                                                                            returned_data.get(
                                                                                'status') == 'failed' else None
            if returned_data['status'] == 'ok':
                returned_data["testConclusion"].append('测试通过')

            return returned_data

        if set_global_vars and isinstance(set_global_vars, list):
            for set_global_var in set_global_vars:
                if isinstance(set_global_var, dict) and isinstance(set_global_var.get('name'), str):
                    name = set_global_var.get('name')
                    query = set_global_var.get('query')
                    value = common.dict_get(response_json, query)
                    self.global_vars[name] = str(value) if value else value

        if 'checkHttpCode' in test_case and not test_case['checkHttpCode'] in ["", None]:
            check_http_code = test_case['checkHttpCode']

        if 'checkResponseData' in test_case and not test_case['checkResponseData'] in [[], {}, "", None]:
            if not isinstance(test_case['checkResponseData'], list):
                raise TypeError('checkResponseData must be list！')
            for index, crd in enumerate(test_case['checkResponseData']):
                if not isinstance(crd, dict) or 'regex' not in crd or 'query' not in crd or \
                        not isinstance(crd['regex'], str) or not isinstance(crd['query'], list):
                    raise TypeError('checkResponseData is not valid!')

                # TODO 可开启/关闭 全局替换
                test_case['checkResponseData'][index]['regex'] = \
                    common.resolve_global_var(pre_resolve_var=crd['regex'], global_var_dic=self.global_vars) if \
                        crd.get('regex') and isinstance(crd.get('regex'), str) else ''  # 警告！python判断空字符串为False

            check_response_data = test_case['checkResponseData']

        if 'checkResponseSimilarity' in test_case and not test_case['checkResponseSimilarity'] in [[], {}, "", None]:
            if not isinstance(test_case['checkResponseSimilarity'], list):
                raise TypeError('checkResponseSimilarity must be list！')
            for index, crs in enumerate(test_case['checkResponseSimilarity']):
                if not isinstance(crs, dict) or 'baseText' not in crs or 'targetSimilarity' not in crs \
                        or 'compairedText' not in crs or not isinstance(crs['baseText'], str) \
                        or not isinstance(crs['compairedText'], str):
                    raise TypeError('checkResponseSimilarity is not valid!')
                test_case['checkResponseSimilarity'][index]['baseText'] = \
                    common.resolve_global_var(pre_resolve_var=crs['baseText'], global_var_dic=self.global_vars) if \
                        crs.get('baseText') and isinstance(crs.get('baseText'), str) else ''
                test_case['checkResponseSimilarity'][index]['compairedText'] = \
                    common.resolve_global_var(pre_resolve_var=crs['compairedText'], global_var_dic=self.global_vars) if \
                        crs.get('compairedText') and isinstance(crs.get('compairedText'), str) else ''
            check_response_similarity = test_case['checkResponseSimilarity']

        if 'checkResponseNumber' in test_case and not test_case['checkResponseNumber'] in [[], {}, "", None]:
            if not isinstance(test_case['checkResponseNumber'], list):
                raise TypeError('checkResponseNumber must be list！')
            for index, crn in enumerate(test_case['checkResponseNumber']):
                if not isinstance(crn, dict) or 'expressions' not in crn or \
                        not isinstance(crn['expressions'], dict):
                    raise TypeError('checkResponseNumber is not valid!')

                test_case['checkResponseNumber'][index]['expressions']['firstArg'] = \
                    common.resolve_global_var(pre_resolve_var=crn['expressions']['firstArg'],
                                              global_var_dic=self.global_vars) if \
                        crn['expressions'].get('firstArg') and isinstance(crn['expressions'].get('firstArg'),
                                                                          str) else ''

                test_case['checkResponseNumber'][index]['expressions']['secondArg'] = \
                    common.resolve_global_var(pre_resolve_var=crn['expressions']['secondArg'],
                                              global_var_dic=self.global_vars) if \
                        crn['expressions'].get('secondArg') and isinstance(crn['expressions'].get('secondArg'),
                                                                           str) else ''

                test_case['checkResponseNumber'][index]['expressions']['expectResult'] = \
                    common.resolve_global_var(pre_resolve_var=crn['expressions']['expectResult'],
                                              global_var_dic=self.global_vars) if \
                        crn['expressions'].get('expectResult') and isinstance(crn['expressions'].get('expectResult'),
                                                                              str) else ''
            check_response_number = test_case['checkResponseNumber']

        if check_http_code and not str(response_status_code) == str(check_http_code):
            returned_data["status"] = 'failed'
            returned_data["testConclusion"].append('响应状态码错误, 期待值: <%s>, 实际值: <%s>。\t'
                                                   % (check_http_code, response_status_code))
        if check_response_data:
            try:
                for crd in check_response_data:
                    regex = crd['regex']
                    if regex.strip() == '':
                        continue
                    query = crd['query']
                    real_value = common.dict_get(response_json, query)
                    if real_value is None:
                        returned_data["status"] = 'failed'
                        returned_data["testConclusion"].append('未找到正则校验的Json值(查询语句为: %s),   服务器响应为: %s'
                                                               % (query, response_json))
                        return returned_data
                    result = re.search(regex, str(real_value))  # python 将regex字符串取了r''(原生字符串)
                    if not result:
                        returned_data["status"] = 'failed'
                        returned_data["testConclusion"].append('判断响应值错误(查询语句为: %s),    响应值应满足正则: <%s>,\
                                                                    实际值: <%s> (%s)。(正则匹配时会将数据转化成string)\t'
                                                               % (query, regex, real_value, type(real_value)))
            except BaseException as e:
                returned_data["status"] = 'failed'
                returned_data["testConclusion"].append('判断响应值时报错, 错误信息: <%s>。\t' % e)

        if check_response_number:
            try:
                for crn in check_response_number:
                    expressions = crn['expressions']
                    # print(expressions)
                    if '' in expressions.values() or None in expressions.values():
                        continue
                    expressions_str, result = common.get_numbers_compared_result(expressions)
                    if not result:
                        returned_data["status"] = 'failed'
                        returned_data["testConclusion"].append('判断数值错误(判断表达式为: %s)。\t' % expressions_str)
            except BaseException as e:
                returned_data["status"] = 'failed'
                returned_data["testConclusion"].append('判断数值时报错, 错误信息: <%s>。\t ' % e)

        if hasattr(self, 'nlper') and self.nlper and check_response_similarity:
            try:
                for crs in check_response_similarity:
                    base_text = crs['baseText']
                    compaired_text = crs['compairedText']
                    target_similarity = crs['targetSimilarity']
                    if base_text.strip() == '' or compaired_text.strip() == '' or \
                            not common.can_convert_to_float(target_similarity):
                        continue
                    actual_similarity = self.nlper.get_text_similarity(base_text, compaired_text)

                    if float(actual_similarity) < float(target_similarity):
                        returned_data["status"] = 'failed'
                        returned_data["testConclusion"].append('相似度校验未达标！已对比字符串: 「%s」、「%s」, 实际相似度: 「%s」 '
                                                               '预期相似度: 「%s」。\t ' % (base_text, compaired_text,
                                                                                    actual_similarity,
                                                                                    target_similarity))
            except BaseException as e:
                returned_data["status"] = 'failed'
                returned_data["testConclusion"].append('判断相似度时报错, 模型服务器可能已宕机/断网。具体错误信息: <%s>。\t' % e)

        if returned_data["testConclusion"] == []:
            returned_data["status"] = 'ok'
            returned_data["testConclusion"].append('测试通过')
        else:
            returned_data["status"] = 'failed'
            returned_data["testConclusion"].append('测试不通过！')
        return returned_data

    def update_case_info(self, testing_case_model):
        for index, test_result in enumerate(self.test_result_list):
            test_case_id = test_result["_id"]
            test_result = common.format_response_in_dic(test_result)
            self.test_result_list[index] = test_result
            testing_case_model.update({"_id": ObjectId(test_case_id)},
                                      {'$set': {'lastManualTestResult': test_result}})

    def send_report(self, test_report_model, project_id, executor_nick_name, execution_mode):
        test_count = len(self.test_result_list)
        passed_count = len(
            list(filter(lambda x: x == 'ok', [test_result["status"] for test_result in self.test_result_list])))
        failed_count = len(
            list(filter(lambda x: x == 'failed', [test_result["status"] for test_result in self.test_result_list])))
        passed_rate = '%d' % round((passed_count / test_count) * 100, 2) + '%'

        if test_count > 0:
            for test_result in self.test_result_list:
                if 'testBaseInfo' in test_result and 'lastManualTestResult' in test_result['testBaseInfo']:
                    test_result['testBaseInfo'].pop('lastManualTestResult')

            raw_data = {
                "projectId": ObjectId(project_id),
                "testCount": test_count,
                "passCount": passed_count,
                "failedCount": failed_count,
                "passRate": passed_rate,
                "comeFrom": execution_mode,
                "executorNickName": executor_nick_name,
                "testDetail": self.test_result_list,
                "createAt": datetime.datetime.utcnow()
            }
            filtered_data = test_report_model.filter_field(raw_data, use_set_default=True)
            test_report_model.insert(
                filtered_data
            )


if __name__ == '__main__':
    pass