# -*- coding: utf-8 -*- """ @File : convert2hrp.py @Time : 2023/2/13 10:29 @Author : geekbing @LastEditTime : - @LastEditors : - @Description : - """ import json import os from enum import Enum from typing import Any, Dict, List, Text, Union, Callable from urllib.parse import urlparse from pydantic import BaseModel, Field from pydantic import HttpUrl Name = Text Url = Text BaseUrl = Union[HttpUrl, Text] VariablesMapping = Dict[Text, Any] FunctionsMapping = Dict[Text, Callable] Headers = Dict[Text, Text] Cookies = Dict[Text, Text] Verify = bool Hooks = List[Union[Text, Dict[Text, Text]]] Export = List[Text] Validators = List[Dict] Env = Dict[Text, Any] class MethodEnum(Text, Enum): GET = "GET" POST = "POST" PUT = "PUT" DELETE = "DELETE" HEAD = "HEAD" OPTIONS = "OPTIONS" PATCH = "PATCH" class TConfig(BaseModel): name: Name verify: Verify = False base_url: BaseUrl = "" # Text: prepare variables in debugtalk.py, ${gen_variables(} variables: Union[VariablesMapping, Text] = {} parameters: Union[VariablesMapping, Text] = {} # setup_hooks: Hooks = [] # teardown_hooks: Hooks = [] export: Export = [] path: Text = None weight: int = 1 class TRequest(BaseModel): """requests.Request model""" # TODO: 先注释TRequest-method类型,后期再优化 method: str url: Url params: Dict[Text, Text] = {} headers: Headers = {} req_json: Union[Dict, List, Text] = Field(None) body: Union[Text, Dict[Text, Any]] = None cookies: Cookies = {} timeout: float = 120 allow_redirects: bool = True verify: Verify = False upload: Dict = {} # used for upload files class TStep(BaseModel): name: Name request: Union[TRequest, None] = None testcase: Union[Text, Callable, None] = None variables: VariablesMapping = {} setup_hooks: Hooks = [] # used to extract request's response field extract: VariablesMapping = {} # used to export session variable from referenced testcase export: Export = [] validators: Validators = Field([], alias="validate") validate_script: List[Text] = [] class TestCase(BaseModel): config: TConfig teststeps: List[TStep] class Hrp: def __init__(self, faster_req_json: Dict): self.faster_req_json = faster_req_json def parse_url(self): url = self.faster_req_json["url"] o = urlparse(url=url) baseurl = o.scheme + "://" + o.netloc return baseurl, o.path def get_headers(self): headers: Dict = self.faster_req_json.get("headers", {}) # Content-Length may be error headers.pop("Content-Length", None) return headers def get_request(self) -> TRequest: base_url, path = self.parse_url() req = TRequest( method=self.faster_req_json["method"], url=base_url + path, params=self.faster_req_json.get("params", {}), headers=self.get_headers(), body=self.faster_req_json.get("body", {}), req_json=self.faster_req_json.get("json", {}), verify=self.faster_req_json.get("verify", False), ) return req def get_step(self) -> TStep: _, path = self.parse_url() return TStep( name=path, request=self.get_request(), ) def get_config(self) -> TConfig: base_url, _ = self.parse_url() return TConfig( name=base_url, base_url=base_url, ) def get_testcase(self) -> TestCase: config = self.get_config() teststeps: List = [self.get_step()] return TestCase( config=config, teststeps=teststeps, )