詳解HttpRunner3的HTTP請(qǐng)是如何發(fā)出

更新時(shí)間：2023年07月23日 14:43:55 作者：python亦希

這篇文章主要為大家介紹了HttpRunner3的HTTP請(qǐng)是如何發(fā)出詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

HttpRunner3示例代碼發(fā)送HTTP請(qǐng)求

在HttpRunner3的示例代碼中，發(fā)送HTTP請(qǐng)求的代碼是這樣寫的：

from httprunner import HttpRunner, Config, Step, RunRequest, RunTestCase
class TestCaseBasic(HttpRunner):
    config = Config("basic test with httpbin").base_url("https://httpbin.org/")
    teststeps = [
        Step(
            RunRequest("headers")
            .get("/headers")
            .validate()
            .assert_equal("status_code", 200)
            .assert_equal("body.headers.Host", "httpbin.org")
        ),
        # 省略
        Step(
            RunRequest("post data")
            .post("/post")
            .with_headers(**{"Content-Type": "application/json"})
            .with_data("abc")
            .validate()
            .assert_equal("status_code", 200)
        ),
        # 省略
    ]
if __name__ == "__main__":
    TestCaseBasic().test_start()

類TestCaseBasic繼承了類HttpRunner

在類TestCaseBasic的內(nèi)部定義了teststeps列表，由多個(gè)Step類的實(shí)例對(duì)象組成。
類Step初始化傳入類RunRequest的方法get和post就把HTTP請(qǐng)求發(fā)出去了。

如何實(shí)現(xiàn)解析

先看下RunRequest的源碼：

class RunRequest(object):
    def __init__(self, name: Text):
        self.__step_context = TStep(name=name)
    def with_variables(self, **variables) -> "RunRequest":
        self.__step_context.variables.update(variables)
        return self
    def setup_hook(self, hook: Text, assign_var_name: Text = None) -> "RunRequest":
        if assign_var_name:
            self.__step_context.setup_hooks.append({assign_var_name: hook})
        else:
            self.__step_context.setup_hooks.append(hook)
        return self
    def get(self, url: Text) -> RequestWithOptionalArgs:
        self.__step_context.request = TRequest(method=MethodEnum.GET, url=url)
        return RequestWithOptionalArgs(self.__step_context)
    def post(self, url: Text) -> RequestWithOptionalArgs:
        self.__step_context.request = TRequest(method=MethodEnum.POST, url=url)
        return RequestWithOptionalArgs(self.__step_context)
    def put(self, url: Text) -> RequestWithOptionalArgs:
        self.__step_context.request = TRequest(method=MethodEnum.PUT, url=url)
        return RequestWithOptionalArgs(self.__step_context)
    def head(self, url: Text) -> RequestWithOptionalArgs:
        self.__step_context.request = TRequest(method=MethodEnum.HEAD, url=url)
        return RequestWithOptionalArgs(self.__step_context)
    def delete(self, url: Text) -> RequestWithOptionalArgs:
        self.__step_context.request = TRequest(method=MethodEnum.DELETE, url=url)
        return RequestWithOptionalArgs(self.__step_context)
    def options(self, url: Text) -> RequestWithOptionalArgs:
        self.__step_context.request = TRequest(method=MethodEnum.OPTIONS, url=url)
        return RequestWithOptionalArgs(self.__step_context)
    def patch(self, url: Text) -> RequestWithOptionalArgs:
        self.__step_context.request = TRequest(method=MethodEnum.PATCH, url=url)
        return RequestWithOptionalArgs(self.__step_context)

里面定義了get、post等HTTP請(qǐng)求的Method。

方法內(nèi)部

self.__step_context.request = TRequest(method=MethodEnum.GET, url=url)

有個(gè)TRequest類：

class TRequest(BaseModel):
    """requests.Request model"""

    method: MethodEnum
    url: Url
    params: Dict[Text, Text] = {}
    headers: Headers = {}
    req_json: Union[Dict, List, Text] = Field(None, alias="json")
    data: Union[Text, Dict[Text, Any]] = None
    cookies: Cookies = {}
    timeout: float = 120
    allow_redirects: bool = True
    verify: Verify = False
    upload: Dict = {}  # used for upload files

它繼承了pydantic.BaseModel，是用來做數(shù)據(jù)驗(yàn)證的，比如這里的url指定了Url類型，如果傳一個(gè)str類型，就會(huì)校驗(yàn)失敗。簡(jiǎn)而言之，這是給代碼規(guī)范用的，沒有實(shí)際的業(yè)務(wù)功能。

下面有一行注釋：requests.Request mode，看來這個(gè)跟requests有點(diǎn)關(guān)系。

回過頭來看看??self.__step_context.request???，也就是??self.__step_context??對(duì)象有個(gè)request屬性，它的定義是：

self.__step_context = TStep(name=name)

答案應(yīng)該就在TStep中了：

class TStep(BaseModel):
    name: Name
    request: Union[TRequest, None] = None
    testcase: Union[Text, Callable, None] = None
    variables: VariablesMapping = {}
    setup_hooks: Hooks = []
    teardown_hooks: Hooks = []
    # used to extract request's response field
    extract: VariablesMapping = {}
    # used to export session variables from referenced testcase
    export: Export = []
    validators: Validators = Field([], alias="validate")
    validate_script: List[Text] = []

Model里request定義

還是個(gè)Model，里面的request定義是：

request: Union[TRequest, None] = None

又繞回TRequest了。這個(gè)Union是typing模塊里面的：Union[X, Y] means either X or Y. 意思就是request的類型要么是TRequest要么是None。

在剛才get的方法中，還有一句??return RequestWithOptionalArgs(self.__step_context)??，RequestWithOptionalArgs的定義如下：

class RequestWithOptionalArgs(object):
    def __init__(self, step_context: TStep):
        self.__step_context = step_context
    def with_params(self, **params) -> "RequestWithOptionalArgs":
        self.__step_context.request.params.update(params)
        return self
    def with_headers(self, **headers) -> "RequestWithOptionalArgs":
        self.__step_context.request.headers.update(headers)
        return self
    def with_cookies(self, **cookies) -> "RequestWithOptionalArgs":
        self.__step_context.request.cookies.update(cookies)
        return self
    def with_data(self, data) -> "RequestWithOptionalArgs":
        self.__step_context.request.data = data
        return self
    def with_json(self, req_json) -> "RequestWithOptionalArgs":
        self.__step_context.request.req_json = req_json
        return self
    def set_timeout(self, timeout: float) -> "RequestWithOptionalArgs":
        self.__step_context.request.timeout = timeout
        return self
    def set_verify(self, verify: bool) -> "RequestWithOptionalArgs":
        self.__step_context.request.verify = verify
        return self
    def set_allow_redirects(self, allow_redirects: bool) -> "RequestWithOptionalArgs":
        self.__step_context.request.allow_redirects = allow_redirects
        return self
    def upload(self, **file_info) -> "RequestWithOptionalArgs":
        self.__step_context.request.upload.update(file_info)
        return self
    def teardown_hook(
        self, hook: Text, assign_var_name: Text = None
    ) -> "RequestWithOptionalArgs":
        if assign_var_name:
            self.__step_context.teardown_hooks.append({assign_var_name: hook})
        else:
            self.__step_context.teardown_hooks.append(hook)
        return self
    def extract(self) -> StepRequestExtraction:
        return StepRequestExtraction(self.__step_context)
    def validate(self) -> StepRequestValidation:
        return StepRequestValidation(self.__step_context)
    def perform(self) -> TStep:
        return self.__step_context

可以給HTTP請(qǐng)求添加params、headers等可選項(xiàng)。

看到這里，仍然不知道HTTP請(qǐng)求到底發(fā)出去的，因?yàn)闆]有調(diào)用呀。

調(diào)用RunRequest的Step類

只能往上層找，看調(diào)用RunRequest的Step類：

class Step(object):
    def __init__(
        self,
        step_context: Union[
            StepRequestValidation,
            StepRequestExtraction,
            RequestWithOptionalArgs,
            RunTestCase,
            StepRefCase,
        ],
    ):
        self.__step_context = step_context.perform()
    @property
    def request(self) -> TRequest:
        return self.__step_context.request
    @property
    def testcase(self) -> TestCase:
        return self.__step_context.testcase
    def perform(self) -> TStep:
        return self.__step_context

Step類的??__init__??方法也用Union做了類型校驗(yàn)，其中RequestWithOptionalArgs就是RunRequest的gei等方法會(huì)返回的，這倒是匹配上了。它還有個(gè)request屬性。有點(diǎn)眉目了。

看HttpRunner類

再往上層找，看HttpRunner類，有個(gè)??__run_step_request??的方法：

def __run_step_request(self, step: TStep) -> StepData:
    """run teststep: request"""
    step_data = StepData(name=step.name)
    # parse
    prepare_upload_step(step, self.__project_meta.functions)
    request_dict = step.request.dict()
    request_dict.pop("upload", None)
    parsed_request_dict = parse_data(
        request_dict, step.variables, self.__project_meta.functions
    )
    parsed_request_dict["headers"].setdefault(
        "HRUN-Request-ID",
        f"HRUN-{self.__case_id}-{str(int(time.time() * 1000))[-6:]}",
    )
    step.variables["request"] = parsed_request_dict
    # setup hooks
    if step.setup_hooks:
        self.__call_hooks(step.setup_hooks, step.variables, "setup request")
    # prepare arguments
    method = parsed_request_dict.pop("method")
    url_path = parsed_request_dict.pop("url")
    url = build_url(self.__config.base_url, url_path)
    parsed_request_dict["verify"] = self.__config.verify
    parsed_request_dict["json"] = parsed_request_dict.pop("req_json", {})
    # request
    resp = self.__session.request(method, url, **parsed_request_dict)
    resp_obj = ResponseObject(resp)
    step.variables["response"] = resp_obj
    # teardown hooks
    if step.teardown_hooks:
        self.__call_hooks(step.teardown_hooks, step.variables, "teardown request")
    def log_req_resp_details():
        err_msg = "\n{} DETAILED REQUEST & RESPONSE {}\n".format("*" * 32, "*" * 32)
        # log request
        err_msg += "====== request details ======\n"
        err_msg += f"url: {url}\n"
        err_msg += f"method: {method}\n"
        headers = parsed_request_dict.pop("headers", {})
        err_msg += f"headers: {headers}\n"
        for k, v in parsed_request_dict.items():
            v = utils.omit_long_data(v)
            err_msg += f"{k}: {repr(v)}\n"
        err_msg += "\n"
        # log response
        err_msg += "====== response details ======\n"
        err_msg += f"status_code: {resp.status_code}\n"
        err_msg += f"headers: {resp.headers}\n"
        err_msg += f"body: {repr(resp.text)}\n"
        logger.error(err_msg)
    # extract
    extractors = step.extract
    extract_mapping = resp_obj.extract(extractors)
    step_data.export_vars = extract_mapping
    variables_mapping = step.variables
    variables_mapping.update(extract_mapping)
    # validate
    validators = step.validators
    session_success = False
    try:
        resp_obj.validate(
            validators, variables_mapping, self.__project_meta.functions
        )
        session_success = True
    except ValidationFailure:
        session_success = False
        log_req_resp_details()
        # log testcase duration before raise ValidationFailure
        self.__duration = time.time() - self.__start_at
        raise
    finally:
        self.success = session_success
        step_data.success = session_success
        if hasattr(self.__session, "data"):
            # httprunner.client.HttpSession, not locust.clients.HttpSession
            # save request & response meta data
            self.__session.data.success = session_success
            self.__session.data.validators = resp_obj.validation_results
            # save step data
            step_data.data = self.__session.data
    return step_data

就是這里了，它的函數(shù)名用了雙下劃線開頭：雙下劃線前綴會(huì)讓Python解釋器重寫屬性名稱，以避免子類中的命名沖突。這也稱為名稱改寫（name mangling），即解釋器會(huì)更改變量的名稱，以便在稍后擴(kuò)展這個(gè)類時(shí)避免命名沖突。說人話就是，類的私有成員，只能在類的內(nèi)部調(diào)用，不對(duì)外暴露。

它只在??__run_step()???方法中調(diào)用了1次：??step_data = self.__run_step_request(step)??。

中間有一段：

# request
resp = self.__session.request(method, url, **parsed_request_dict)
resp_obj = ResponseObject(resp)
step.variables["response"] = resp_obj

好家伙，??self.__session.request()??，跟reqeusts那個(gè)有點(diǎn)像了。點(diǎn)進(jìn)去。

一下就跳轉(zhuǎn)到了??httprunner.client.py??，眾里尋他千百度，默然回首，它竟然就在client。

class HttpSession(requests.Session):
    """
    Class for performing HTTP requests and holding (session-) cookies between requests (in order
    to be able to log in and out of websites). Each request is logged so that HttpRunner can
    display statistics.
    This is a slightly extended version of `python-request <http://python-requests.org>`_'s
    :py:class:`requests.Session` class and mostly this class works exactly the same.
    """
    def __init__(self):
        super(HttpSession, self).__init__()
        self.data = SessionData()
    def update_last_req_resp_record(self, resp_obj):
        """
        update request and response info from Response() object.
        """
        # TODO: fix
        self.data.req_resps.pop()
        self.data.req_resps.append(get_req_resp_record(resp_obj))
    def request(self, method, url, name=None, **kwargs):

繼承了requests.Session然后進(jìn)行了重寫。

果然，還是用到了requests庫(kù)。

以上就是詳解HttpRunner3的HTTP請(qǐng)是如何發(fā)出的詳細(xì)內(nèi)容，更多關(guān)于HttpRunner3 HTTP請(qǐng)求發(fā)出的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: