# -*- coding: utf-8 -*- """ @File : generator.py @Time : 2023/9/11 15:47 @Author : geekbing @LastEditTime : - @LastEditors : - @Description : 根据RequestInfo生成测试用例 """ import logging import json import re from collections import defaultdict from enum import IntEnum from typing import Any, Dict, List, Tuple, Union, Set from urllib.parse import urlparse, parse_qs from apps.exceptions.convert import GenerateError from apps.schema.request import RequestInfo from apps.schema.api_schema import APIBody, APISchema from apps.schema.testcase_schema import RecordCaseSchema, APIBodySchema from lunarlink.models import API from lunarlink.utils.enums.RequestBodyEnum import BodyType from lunarlink.utils.parser import Format logger = logging.getLogger(__name__) class CaseTag(IntEnum): INTEGRATION_CASE = 2 # 忽略的字段 ignored_headers = { "content-type", "connection", "date", "content-length", "host", "access-control-allow-credentials", "access-control-allow-origin", "user-agent", "server", } class CaseGenerator: """ 根据RequestInfo数组生成测试用例 """ @staticmethod def ignore(key: str): """ 检查指定的HTTP头部字段名称是否在预定义的忽略列表中。 :param key: 忽略字段 headers.Origin :return: 如果字段应被忽略，则返回True；否则返回False。 """ for ig in ignored_headers: if key.lower().endswith(ig): return True return False @staticmethod def get_body_type(headers: Dict): headers = {k.lower(): v for k, v in headers.items()} content_type = headers.get("content-type", "").lower() if "json" in content_type: return BodyType.json if "x-www-form" in content_type: return BodyType.x_form if "form" in content_type: return BodyType.form return BodyType.none @staticmethod def extract_variables(input_string: str): """ 使用正则表达式提取 ${http_res_INDEX_variable::path} 格式的变量、索引和路径 :param input_string: :return: 返回http_res_INDEX, http_res_INDEX_variable, path """ if isinstance(input_string, str): matches = re.findall(r"\$http_res_(\d+)_(.*?)::(.+)", input_string) else: matches = [] if matches: idx, var_name, path = matches[0] full_variable = "http_res_" + idx + "_" + var_name return "http_res_" + idx, full_variable, path return None @staticmethod def generate_case( length: int, project_id: int, case_dir: int, api_dir: int, config: Dict, case_name: str, requests: List[RequestInfo], user: int, ) -> Tuple[RecordCaseSchema, List]: """ 根据录制接口生成测试用例 :param api_dir: 接口目录 :param length: :param project_id: :param case_dir: 用例目录 :param config: :param case_name: :param requests: :param user: 当前用户 :return: """ record_case_body = [config] api_info_map = {} # 先将 requests 转成 faster 格式 api for r in range(len(requests)): api_info = APIBodySchema() # 初始化 faster 格式的 api 模板 api_name = f"http_res_{r + 1}" api_info.name = api_name api_info.method = requests[r].request_method # 请求头 headers = requests[r].request_headers CaseGenerator.merge_headers(headers, api_info) # 请求参数为: form、json requests_body = requests[r].body if requests_body: CaseGenerator.merge_body(headers, api_info, requests_body) # 请求参数为: params CaseGenerator.merge_params(api_info, requests[r].url) api_info_map.update({api_name: api_info.dict(by_alias=True)}) # 保持别名 # 遍历所有接口，将 htt_res_1_token 的值 content.token 加入 extract for api_name, request_data in api_info_map.items(): for headers_key, headers_value in request_data["header"]["header"].items(): CaseGenerator.append_extract( request_data, api_info_map, headers_key, headers_value, "headers" ) for k, v in request_data["request"]["form"]["data"].items(): CaseGenerator.append_extract(request_data, api_info_map, k, v, "form") for k, v in request_data["request"]["json"].items(): CaseGenerator.append_extract(request_data, api_info_map, k, v, "json") for k, v in request_data["request"]["params"]["params"].items(): CaseGenerator.append_extract(request_data, api_info_map, k, v, "params") api_instances = [] for _, v in api_info_map.items(): api = Format(v) api.parse() if api_dir: api_body = APIBody(**api.testcase) api_schema = APISchema( name=api.name, body=api_body, url=api.url, method=api.method, project_id=project_id, relation=api_dir, creator_id=user, ).dict(by_alias=True) api_instances.append(API(**api_schema)) record_case_body.append({"body": api.testcase}) return ( RecordCaseSchema( length=length, project_id=project_id, relation=case_dir, name=case_name, tag=CaseTag.INTEGRATION_CASE.value, body=record_case_body, ), api_instances, ) @staticmethod def merge_headers(headers: Dict, api_info: APIBodySchema): """ 将录制接口中的headers格式化后加入fastapi的headers中 :param headers: 录制接口中的headers :param api_info: fastapi接口数据 :return: """ api_info.header.header.update(headers) api_info.header.desc.update({key: "" for key in headers.keys()}) @staticmethod def merge_body(headers: Dict, api_info: APIBodySchema, body: str): """ 将录制接口中的body格式化后加入fastapi的body中 :param headers: 录制接口中的headers :param api_info: fastapi接口数据 :param body: 接口请求body :return: """ try: body = json.loads(body) except json.JSONDecodeError: pass except Exception as e: logger.error(f"转换body变量失败: {e}") else: body_type = CaseGenerator.get_body_type(headers) if body_type == BodyType.x_form: api_info.request.form.data.update(body) api_info.request.form.desc.update({key: "" for key in body.keys()}) elif body_type == BodyType.json: api_info.request.json_data.update(body) @staticmethod def merge_params(api_info: APIBodySchema, url: str): """ 将录制接口中的 query 参数提取后加入fastapi的params中 :param api_info: fastapi接口数据 :param url: 接口请求url :return: """ parsed_url = urlparse(url) api_info.url = parsed_url.path query_params = parse_qs(parsed_url.query) params = {k: v[0] for k, v in query_params.items()} api_info.request.params.params.update(params) api_info.request.params.desc.update({key: "" for key in params.keys()}) @staticmethod def append_extract( request_data: Dict, api_info_map: Dict, item_key, item_value, request_type: str ): """ 提取引用变量，将接口中引用此变量的请求数据替换为变量值 :param request_data: 接口请求数据, 保存格式为{"header": {...}, "request": {...}} :param api_info_map: 全部接口的请求信息和接口名的映射，保存格式为{"api_name": {"header": {...}, "request": {...}} :param item_key: 键名 :param item_value: 键值 :param request_type: 消息请求类型（headers, json, form, params） :return: """ variables = CaseGenerator.extract_variables(item_value) if variables: api_name = variables[0] extract_var_name = variables[1] extract_var_path = variables[2] if request_type == "headers": CaseGenerator.replace_faster_headers(request_data, item_key, variables) elif request_type == "form": CaseGenerator.replace_faster_form(request_data, item_key, variables) elif request_type == "json": CaseGenerator.replace_faster_json(request_data, item_key, variables) elif request_type == "params": CaseGenerator.replace_faster_params(request_data, item_key, variables) if api_name in api_info_map: if {extract_var_name: extract_var_path} not in api_info_map[api_name][ "extract" ]["extract"]: api_info_map[api_name]["extract"]["extract"].append( {extract_var_name: extract_var_path} ) api_info_map[api_name]["extract"]["desc"].update( {extract_var_name: ""} ) @staticmethod def replace_faster_form(request_data: Dict, var_name: str, variables: Tuple): """ 替换fastapi格式请求form参数 :param request_data: :param var_name: :param variables: :return: """ request_data["request"]["form"]["data"].update({var_name: f"${variables[1]}"}) @staticmethod def replace_faster_json(request_data: Dict, var_name: str, variables: Tuple): """ 替换fastapi格式请求json参数 :param request_data: :param var_name: :param variables: :return: """ request_data["request"]["json"].update({var_name: f"${variables[1]}"}) @staticmethod def replace_faster_params(request_data: Dict, var_name: str, variables: Tuple): """ 替换fastapi格式请求params查询参数 :param request_data: :param var_name: :param variables: :return: """ request_data["request"]["params"]["params"].update( {var_name: f"${variables[1]}"} ) @staticmethod def replace_faster_headers(request_data: Dict, var_name: str, variables: Tuple): request_data["header"]["header"].update({var_name: f"${variables[1]}"}) @staticmethod def extract_field(requests: List[RequestInfo]) -> List[str]: """遍历接口，提取其中的变量并替换 :param requests: 录制流量接口信息 :return: """ value_to_path_map: Dict = defaultdict(list) # 变量值到路径的映射 {变量值: [变量路径, ...], ...} replaced = [] # 替换记录 for i, item in enumerate(requests): if "Content-Length" in item.request_headers: item.request_headers.pop("Content-Length") if "Content-Type" in item.response_headers: item.response_headers.pop("Content-Type") # 记录变量 CaseGenerator.record_vars( request=item, value_to_path_map=value_to_path_map, var_name=f"http_res_{i + 1}", ) if i > 0: # 接口变量替换 CaseGenerator.replace_vars(item, value_to_path_map, replaced) return replaced @staticmethod def replace_vars(request: RequestInfo, value_to_path_map: Dict, replaced: List): """ 替换变量 :param request: 录制流量接口 :param value_to_path_map: 记录变量值, {变量值: [变量路径, ...], ...} :param replaced: 替换记录 :return: """ # 提取响应内容和响应头部中的所有值 response_values = CaseGenerator.extract_response_values(request) CaseGenerator.replace_url(request, value_to_path_map, replaced, response_values) CaseGenerator.replace_headers( request, value_to_path_map, replaced, response_values, ) CaseGenerator.replace_body( request, value_to_path_map, replaced, response_values, ) @staticmethod def extract_response_values(request: RequestInfo) -> Set: """ 提取响应内容和响应头部中的所有值。 :param request: 录制流量接口 :return: 响应值集合 """ response_values = set() # 假设响应内容是 JSON 格式的字符串，提取所有值 if request.response_content: try: response_data = json.loads(request.response_content) response_values.update( CaseGenerator.get_values_from_json(response_data) ) except json.JSONDecodeError: pass # 添加响应头部中的所有值，除了忽略的头部键 for key, value in request.response_headers.items(): if key.lower() not in ignored_headers: response_values.add(value) return response_values @staticmethod def get_values_from_json(data: Union[Dict, List]) -> Set: """ 从 JSON 数据中递归提取所有值。 :param data: JSON 数据，可以是字典或列表 :return: 包含所有值的集合 """ values = set() if isinstance(data, dict): for value in data.values(): values.update(CaseGenerator.get_values_from_json(value)) elif isinstance(data, list): for item in data: values.update(CaseGenerator.get_values_from_json(item)) else: values.add(data) return values @staticmethod def parse_url(url: str): """ 解析完整的URL，提取协议、域名路径和查询参数。 :param url: 要解析的完整URL字符串。 :return: 一个包含协议、域名路径和查询参数列表的元组。 """ url_parts = url.split("?") base_url = url_parts[0] query_params = [param for param in url_parts[1].split("&") if param] if len(url_parts) > 1 else [] protocol, path_with_domain = base_url.split("//") return protocol, path_with_domain, query_params @staticmethod def replace_path_segments( path_with_domain: str, value_to_path_map: Dict, replaced: List, response_values: Set, ): """ 替换URL路径中的各个段落。 :param replaced: :param path_with_domain: 域名和路径组合的字符串。 :param value_to_path_map: 一个映射字典，用于替换路径中的变量。 :param replaced: 记录替换详情的列表 :param response_values: request_headers和request_content响应值集合 :return: 替换后的路径段落列表。 """ new_url = [] domain_and_path_list = path_with_domain.split("/") for segment in domain_and_path_list: # 检查当前字段是否是自身响应中的值，如果是则跳过 if segment.lower() in value_to_path_map and segment not in response_values: new_segment = value_to_path_map[segment.lower()][0] new_url.append(new_segment) replaced.append(f"{segment} => ${new_segment}") else: new_url.append(segment) return new_url @staticmethod def replace_query_parameters( query_params: List[str], value_to_path_map: Dict, replaced: List, response_values: Set, ): """ 替换查询参数中的值 :param query_params: 查询参数列表。 :param value_to_path_map: 一个映射字典，用于替换查询参数的值。 :param replaced: 替换后的路径段落列表。 :return: 替换后的查询参数字符串列表。 :param response_values: request_headers和request_content响应值集合 """ new_query = [] for param in query_params: param_name, param_value = param.split("=") # 检查当前查询参数值是否是自身响应中的值，如果是则跳过 if ( param_value.lower() in value_to_path_map and param_value not in response_values ): new_param_value = value_to_path_map[param_value.lower()][0] new_query.append(f"{param_name}=${new_param_value}") replaced.append(f"{param_value} => ${new_param_value}") else: new_query.append(param) return new_query @staticmethod def reconstruct_url(protocol: str, new_url: List[str], new_query: List[str]): """ 重构URL，将协议、替换后的路径和查询参数组合成完整的URL。 :param protocol: URL的协议部分，如 http 或 https。 :param new_url: 经过替换的路径段落列表。 :param new_query: 经过替换的查询参数字符串列表。 :return: 重构后的完整URL字符串。 """ return f"{protocol}//{'/'.join(new_url)}{'?' + '&'.join(new_query) if new_query else ''}" @staticmethod def replace_url( request: RequestInfo, value_to_path_map: Dict, replaced: List, response_values: Set, ): """ 替换请求对象中的URL路径和查询参数。 :param request: 包含原始URL的请求对象。 :param value_to_path_map: 一个映射字典，用于替换URL的路径和查询参数中的值。 :param replaced: 记录替换详情的列表。 :param response_values: request_headers和request_content响应值集合 :return: None。函数直接修改传入的请求对象中的URL属性。 """ protocol, path_with_domain, query_params = CaseGenerator.parse_url(request.url) new_url = CaseGenerator.replace_path_segments( path_with_domain, value_to_path_map, replaced, response_values, ) new_query = CaseGenerator.replace_query_parameters( query_params, value_to_path_map, replaced, response_values, ) request.url = CaseGenerator.reconstruct_url(protocol, new_url, new_query) @staticmethod def replace_headers( request: RequestInfo, value_to_path_map: Dict, replaced: List, response_values: Set, ): """ 替换请求接口中的request_headers。 :param request: 包含原始request_headers的请求对象。 :param value_to_path_map: 一个映射字典，用于替换request_headers的值。 :param replaced: 记录替换详情的列表。 :param response_values: 从响应内容和头部中提取的所有值的集合。 :return: None。函数直接修改传入的请求对象中的request_headers属性。 """ for header_key, header_value in list(request.request_headers.items()): if header_value not in response_values: replacement = value_to_path_map.get(header_value) if replacement: new_header_value = replacement[0] request.request_headers[header_key] = f"${new_header_value}" replaced.append(f"{header_key} => ${new_header_value}") @staticmethod def replace_body( request: RequestInfo, value_to_path_map: Dict, replaced: List, response_values: Set, ): """ 替换请求接口中的body。 :param request: 包含原始body的请求对象。 :param value_to_path_map: 一个映射字典，用于替换body的值。，{变量值: [变量路径, ...],...} :param replaced: 记录替换详情的列表。 :param response_values: 从响应内容和头部中提取的所有值的集合。 :return: None。函数直接修改传入的请求对象中的body属性。 """ if request.body: try: request_body = json.loads(request.body) replaced_non_str_variables = [] # 非字符串变量替换路径 CaseGenerator.dfs_replace( request_body, value_to_path_map, replaced_non_str_variables, replaced, response_values, ) result = json.dumps(request_body, ensure_ascii=False) for v in replaced_non_str_variables: result = result.replace(f"'{v}'", f"{v}") request.body = result except json.JSONDecodeError: pass except Exception as e: logger.error(f"转换body变量失败: {e}") @staticmethod def _dfs_replace_dict( request_dict: Dict, value_to_path_map: Dict, replaced_non_str_variables: List, replaced: List, response_values: Set, ) -> None: """ 递归地替换字典中的变量。 :param request_dict: 要处理的请求体字典。 :param value_to_path_map: 存储变量替换信息的字典。 :param replaced_non_str_variables: 存储被替换的非字符串变量。 :param replaced: 存储所有替换操作的列表。 :param response_values: 从响应内容和头部中提取的所有值的集。 :return: """ for key, value in request_dict.items(): is_str, new_value = CaseGenerator.dfs_replace( value, value_to_path_map, replaced_non_str_variables, replaced, response_values, ) # 如果得到的替换值不是None，并且这个值不在响应值中，进行替换 if new_value is not None and new_value not in response_values: request_dict[key] = f"${new_value}" if not is_str: replaced_non_str_variables.append(f"${new_value}") @staticmethod def _dfs_replace_list( request_list: List, value_to_path_map: Dict, replaced_non_str_variables: List, replaced: List, response_values: Set, ) -> None: """ 递归地替换列表中的变量。 :param request_list: 要处理的请求体列表。 :param value_to_path_map: 存储变量替换信息的字典。 :param replaced_non_str_variables: 存储被替换的非字符串变量。 :param replaced: 存储所有替换操作的列表。 :param response_values: 响应体中的值，这些值不应被替换。 :return: """ for i, item in enumerate(request_list): is_str, new_value = CaseGenerator.dfs_replace( item, value_to_path_map, replaced_non_str_variables, replaced, response_values, ) if new_value is not None and new_value not in response_values: request_list[i] = f"${new_value}" if not is_str: replaced_non_str_variables.append(f"${new_value}") @staticmethod def _dfs_replace_value(value, value_to_path_map, replaced, response_values): """ 替换基本数据类型的值。 :param value: 要替换的值。 :param value_to_path_map: 存储变量替换信息的字典。 :param replaced: 存储所有替换操作的列表。 :param response_values: 响应体中的值，这些值不应被替换。 :return: 替换结果和是否为字符串的标志。 """ # 将非字符串类型的值转换为字符串以进行比较 value_str = str(value) if not isinstance(value, str) else value # 如果这个值在映射中并且不在响应值中，使用映射中的新值进行替换 if value_str in value_to_path_map and value_str not in response_values: new_value = value_to_path_map[value_str][0] replaced.append(f"{value_str} => ${new_value}") # 返回替换状态和新值，非字符串类型的变量也记录在replaced_non_str_variables中 return not isinstance(value, str), new_value return None, None @staticmethod def dfs_replace( request_body: Any, value_to_path_map: Dict, replaced_non_str_variables: List, replaced: List, response_values: Set, ): """ 对请求体进行深度优先遍历，替换其中的变量。 :param request_body: 请求体，可能是字典、列表或基本数据类型。 :param value_to_path_map: 存储变量替换信息的字典。 :param replaced_non_str_variables: 存储被替换的非字符串变量。 :param replaced: 存储所有替换操作的列表。 :param response_values: 从响应内容和头部中提取的所有值的集。 :return: """ # 处理字典类型的请求体 if isinstance(request_body, dict): CaseGenerator._dfs_replace_dict( request_body, value_to_path_map, replaced_non_str_variables, replaced, response_values, ) # 处理列表类型的请求体 elif isinstance(request_body, list): CaseGenerator._dfs_replace_list( request_body, value_to_path_map, replaced_non_str_variables, replaced, response_values, ) # 处理基本数据类型的请求体 else: return CaseGenerator._dfs_replace_value( request_body, value_to_path_map, replaced, response_values, ) @staticmethod def record_vars(request: RequestInfo, value_to_path_map: Dict, var_name: str): """ 记录变量 :param request: 录制流量的接口信息 :param value_to_path_map: 变量值到路径的映射 {变量值: [变量路径, ...], ...} :param var_name: http_res_{i + 1} :return: """ CaseGenerator.split_headers( request, value_to_path_map, var_name, ) CaseGenerator.split_body( request, value_to_path_map, var_name, ) @staticmethod def split_headers( request: RequestInfo, value_to_path_map: Dict, var_name: str = "", header_path_prefix: str = "headers", ): """ 分析和记录请求头中的变量及其对应的路径。 :param request: 包含响应头的请求信息对象。 :param value_to_path_map: 一个映射，将变量值映射到其在请求中的路径。 :param var_name: 用于变量提取的基本名称。 :param header_path_prefix: 用于变量路径的前缀。 :return: """ try: CaseGenerator.dfs( request.response_headers, var_name, header_path_prefix, value_to_path_map, headers=True, ) except Exception as e: raise GenerateError(f"解析接口headers变量出错：{e}") @staticmethod def split_body( request: RequestInfo, value_to_path_map: Dict, var_name: str = "", content_path_prefix: str = "content", ): """ 分析和记录请求体中的变量及其对应的路径。 :param request: 包含响应内容的请求信息对象。 :param value_to_path_map: 一个映射，将变量值映射到其在请求中的路径。 :param var_name: 用于变量提取的基本名称。 :param content_path_prefix: 用于变量路径的前缀。 :return: """ if request.body: try: body = json.loads(request.response_content) CaseGenerator.dfs( body=body, var_name=var_name, var_path=content_path_prefix, value_to_path_map=value_to_path_map, ) except json.JSONDecodeError: # body不是JSON，跳过 pass except Exception as e: raise GenerateError(f"解析接口body变量出错：{e}") @staticmethod def dfs_dict( body_dict, var_name, var_path, value_to_path_map, headers, ): """ 递归地遍历响应的头部或内容，提取并记录变量及其路径。 :param body_dict: :param var_name: :param var_path: :param value_to_path_map: :param headers: :return: """ for key, value in body_dict.items(): c_name = f"{var_name}_{key}" c_path = f"{var_path}.{key}" CaseGenerator.dfs(value, c_name, c_path, value_to_path_map, headers) @staticmethod def dfs_list(body_list, var_name, var_path, value_to_path_map, headers): """ 递归地遍历响应的头部或内容，提取并记录变量及其路径。 :param body_list: :param var_name: :param var_path: :param value_to_path_map: :param headers: :return: """ for i, item in enumerate(body_list): c_name = f"{var_name}_{i}" c_path = f"{var_path}.{i}" CaseGenerator.dfs(item, c_name, c_path, value_to_path_map, headers) @staticmethod def process_basic_type( value, var_name, var_path, value_to_path_map, headers, ): """ 处理基本数据类型的值。 :param value: :param var_name: :param var_path: :param value_to_path_map: :param headers: :return: """ if not headers or not CaseGenerator.ignore(var_path): var_name_path = f"{var_name}::{var_path}" # 如果是bool值，需要特殊处理一下，因为Python get False/True会变成get 0 1 if value is not None: if isinstance(value, bool): value_to_path_map[str(value)].append(var_name_path) else: value_to_path_map[value].append(var_name_path) @staticmethod def dfs( body: Any, var_name: str, var_path: str, value_to_path_map: Dict, headers: bool = False, ): """ 递归地遍历响应的头部或内容，提取并记录变量及其路径。 :param body: 可能是响应头或响应内容的部分，可以是字典或列表。 :param var_name: 当前变量的名称。 :param var_path: 当前变量的提取路径。 :param value_to_path_map: 变量到路径的映射。 :param headers: 指示当前处理的是否是响应头。 :return: """ if isinstance(body, list): CaseGenerator.dfs_list( body, var_name, var_path, value_to_path_map, headers, ) elif isinstance(body, dict): CaseGenerator.dfs_dict( body, var_name, var_path, value_to_path_map, headers, ) else: CaseGenerator.process_basic_type( body, var_name, var_path, value_to_path_map, headers, )