一个持续迭代的开源接口测试平台

import asyncio import json import re import time from collections import defaultdict from datetime import datetime from typing import List, Any, Callable from app.core.constructor.case_constructor import TestcaseConstructor from app.core.constructor.http_constructor import HttpConstructor from app.core.constructor.python_constructor import PythonConstructor from app.core.constructor.redis_constructor import RedisConstructor from app.core.constructor.sql_constructor import SqlConstructor from app.core.msg.dingtalk import DingTalk from app.core.msg.mail import Email from app.core.paramters import parameters_parser from app.core.render import Render from app.core.ws_connection_manager import ws_manage from app.crud.auth.UserDao import UserDao from app.crud.config.AddressDao import PityGatewayDao from app.crud.config.EnvironmentDao import EnvironmentDao from app.crud.config.GConfigDao import GConfigDao from app.crud.project.ProjectDao import ProjectDao from app.crud.test_case.TestCaseAssertsDao import TestCaseAssertsDao from app.crud.test_case.TestCaseDao import TestCaseDao from app.crud.test_case.TestCaseOutParametersDao import PityTestCaseOutParametersDao from app.crud.test_case.TestPlan import PityTestPlanDao from app.crud.test_case.TestReport import TestReportDao from app.crud.test_case.TestResult import TestResultDao from app.crud.test_case.TestcaseDataDao import PityTestcaseDataDao from app.enums.ConstructorEnum import ConstructorType from app.enums.GconfigEnum import GConfigParserEnum, GconfigType from app.enums.NoticeEnum import NoticeType from app.enums.RequestBodyEnum import BodyType from app.middleware.AsyncHttpClient import AsyncRequest from app.models.constructor import Constructor from app.models.out_parameters import PityTestCaseOutParameters from app.models.project import Project from app.models.test_case import TestCase from app.models.test_plan import PityTestPlan from app.models.testcase_asserts import TestCaseAsserts from app.utils.case_logger import CaseLog from app.utils.decorator import case_log, lock from app.utils.gconfig_parser import StringGConfigParser, JSONGConfigParser, YamlGConfigParser from app.utils.json_compare import JsonCompare from app.utils.logger import Log from config import Config # construct method mapping construct_type = { ConstructorType.testcase: TestcaseConstructor, ConstructorType.sql: SqlConstructor, ConstructorType.redis: RedisConstructor, ConstructorType.py_script: PythonConstructor, ConstructorType.http: HttpConstructor, } # gconfig parser mapping gconfig_parser = { GConfigParserEnum.string: StringGConfigParser.get_data, GConfigParserEnum.json: JSONGConfigParser.get_data, GConfigParserEnum.yaml: YamlGConfigParser.get_data, } class Executor(object): log = Log("Executor") el_exp = r"\$\{(.+?)\}" pattern = re.compile(el_exp) # 需要替换全局变量的字段 fields = ['url', 'request_headers'] def __init__(self, log: CaseLog = None): # 这里是一个彩蛋, 奔驰大G LB（括弧1.3T） self.glb = None if log is None: self._logger = CaseLog() self._main = True else: self._logger = log self._main = False @property def logger(self): return self._logger @staticmethod def get_constructor_type(c: Constructor): return construct_type.get(c.type) def append(self, content, end=False): if end: self.logger.append(content, end) else: self.logger.append(content, end) @case_log async def parse_gconfig(self, data, type_, env, *fields): """ 解析全局变量 """ for f in fields: await self.parse_field(data, f, GconfigType.text(type_), env) async def load_testcase_variables(self, data, type_, *fields): """load_testcase_variables, include global variables""" for f in fields: self.append("解析{}: [{}]中的变量".format(GconfigType.text(type_), data, f)) origin_field = getattr(data, f) # if not None or "" if origin_field: rendered = Render.render(self.glb, origin_field) if rendered != origin_field: self.append("替换变量成功, [{}]:\n\n[{}] -> [{}]\n".format(f, origin_field, rendered)) setattr(data, f, rendered) @case_log async def query_gconfig(self, env: int): """加载全局变量""" gconfig_list = await GConfigDao.list_gconfig(env) gconfig_map = dict() for g in gconfig_list: parser = Executor.get_parser(g.key_type) gconfig_map[g.key] = parser(g.value) self.glb = gconfig_map @staticmethod def get_parser(key_type) -> Callable: """获取变量解析器 """ parser = gconfig_parser.get(key_type) if parser is None: raise Exception(f"全局变量类型: {key_type}不合法, 请检查!") return parser async def parse_field(self, data, field, name, env): """ 解析字段 """ try: self.append("获取{}: [{}]字段: [{}]中的el表达式".format(name, data, field)) field_origin = getattr(data, field) variables = self.get_el_expression(field_origin) for v in variables: key = v.split(".")[0] cf = await GConfigDao.async_get_gconfig_by_key(key, env) if cf is not None: # 解析变量 parse = self.get_parser(cf.key_type) new_value = parse(cf.value, v) new_field = field_origin.replace("${%s}" % v, new_value) setattr(data, field, new_field) self.append("替换全局变量成功, 字段: [{}]:\n\n[{}] -> [{}]\n".format(field, "${%s}" % v, new_value)) field_origin = new_field self.append("获取{}字段: [{}]中的el表达式".format(name, field), True) except Exception as e: Executor.log.error(f"查询全局变量失败, error: {str(e)}") raise Exception(f"查询全局变量失败, error: {str(e)}") def replace_params(self, field_name, field_origin, params: dict): new_data = dict() if not isinstance(field_origin, str): return new_data variables = self.get_el_expression(field_origin) for v in variables: key = v.split(".") if not params.get(key[0]): continue result = params for branch in key: if isinstance(result, str): # 说明需要反序列化 try: result = json.loads(result) except Exception as e: self.append(f"反序列化失败, result: {result}\nERROR: {e}") break if branch.isdigit(): # 说明路径里面的是数组 result = result[int(branch)] else: result = result.get(branch) if result is None: raise Exception(f"变量路径: {v}不存在, 请检查JSON或路径!") if field_name == "request_headers": new_value = json.loads(result) elif not isinstance(result, str): new_value = json.dumps(result, ensure_ascii=False) else: new_value = result if new_value is None: self.append("替换变量失败, 找不到对应的数据") continue new_data["${%s}" % v] = new_value return new_data async def parse_params(self, data: TestCase, params: dict): self.append("正在替换变量") try: for c in data.__table__.col