• 售前

  • 售后

热门帖子
入门百科

HttpRunner3的HTTP哀求是怎么发出去的

[复制链接]
冷专问 显示全部楼层 发表于 2022-1-16 15:58:37 |阅读模式 打印 上一主题 下一主题
Python微信订餐小步伐课程视频

https://edu.caogenba.net/course/detail/36074
Python实战量化买卖业务理财体系

https://edu.caogenba.net/course/detail/35475
在HttpRunner3的示例代码中,发送HTTP哀求的代码是如许写的:
  1. <code>from httprunner import HttpRunner, Config, Step, RunRequest, RunTestCase
  2. class TestCaseBasic(HttpRunner):
  3.     config = Config("basic test with httpbin").base_url("https://httpbin.org/")
  4.     teststeps = [
  5.         Step(
  6.             RunRequest("headers")
  7.             .get("/headers")
  8.             .validate()
  9.             .assert_equal("status\_code", 200)
  10.             .assert_equal("body.headers.Host", "httpbin.org")
  11.         ),
  12.         # 省略
  13.         Step(
  14.             RunRequest("post data")
  15.             .post("/post")
  16.             .with_headers(**{"Content-Type": "application/json"})
  17.             .with_data("abc")
  18.             .validate()
  19.             .assert_equal("status\_code", 200)
  20.         ),
  21.         # 省略
  22.     ]
  23. if __name__ == "\_\_main\_\_":
  24.     TestCaseBasic().test_start()
复制代码


  • 类TestCaseBasic继续了类HttpRunner。
  • 在类TestCaseBasic的内部界说了teststeps列表,由多个Step类的实例对象构成。
  • 类Step初始化传入类RunRequest的方法get和post就把HTTP哀求发出去了。
这到底是怎么实现的?
先看下RunRequest的源码:
  1. <code>class RunRequest(object):
  2.     def \_\_init\_\_(self, name: Text):
  3.         <code>self.__step_context = TStep(name=name)
  4.     def with\_variables(self, **variables) -> "RunRequest":
  5.         self.__step_context.variables.update(variables)
  6.         return self
  7.     def setup\_hook(self, hook: Text, assign\_var\_name: Text = None) -> "RunRequest":
  8.         if assign_var_name:
  9.             self.__step_context.setup_hooks.append({assign_var_name: hook})
  10.         else:
  11.             self.__step_context.setup_hooks.append(hook)
  12.         return self
  13.     def get(self, url: Text) -> RequestWithOptionalArgs:
  14.         <code>self.__step_context.request = TRequest(method=MethodEnum.GET, url=url)
  15.         return RequestWithOptionalArgs(self.__step_context)
  16.     def post(self, url: Text) -> RequestWithOptionalArgs:
  17.         self.__step_context.request = TRequest(method=MethodEnum.POST, url=url)
  18.         return RequestWithOptionalArgs(self.__step_context)
  19.     def put(self, url: Text) -> RequestWithOptionalArgs:
  20.         self.__step_context.request = TRequest(method=MethodEnum.PUT, url=url)
  21.         return RequestWithOptionalArgs(self.__step_context)
  22.     def head(self, url: Text) -> RequestWithOptionalArgs:
  23.         self.__step_context.request = TRequest(method=MethodEnum.HEAD, url=url)
  24.         return RequestWithOptionalArgs(self.__step_context)
  25.     def delete(self, url: Text) -> RequestWithOptionalArgs:
  26.         self.__step_context.request = TRequest(method=MethodEnum.DELETE, url=url)
  27.         return RequestWithOptionalArgs(self.__step_context)
  28.     def options(self, url: Text) -> RequestWithOptionalArgs:
  29.         self.__step_context.request = TRequest(method=MethodEnum.OPTIONS, url=url)
  30.         return RequestWithOptionalArgs(self.__step_context)
  31.     def patch(self, url: Text) -> RequestWithOptionalArgs:
  32.         self.__step_context.request = TRequest(method=MethodEnum.PATCH, url=url)
  33.         return RequestWithOptionalArgs(self.__step_context)
复制代码
内里界说了get、post等HTTP哀求的Method。方法内部:
  1. <code>self.__step_context.request = TRequest(method=MethodEnum.GET, url=url)
复制代码
有个TRequest类:
  1. <code>class TRequest(BaseModel):
  2.     """requests.Request model"""
  3.     method: MethodEnum
  4.     url: Url
  5.     params: Dict[Text, Text] = {}
  6.     headers: Headers = {}
  7.     req_json: Union[Dict, List, Text] = Field(None, alias="json")
  8.     data: Union[Text, Dict[Text, Any]] = None
  9.     cookies: Cookies = {}
  10.     timeout: float = 120
  11.     allow_redirects: bool = True
  12.     verify: Verify = False
  13.     upload: Dict = {}  # used for upload files
复制代码
它继续了pydantic.BaseModel,是用来做数据验证的,好比这里的url指定了Url范例,假如传一个str范例,就会校验失败。简而言之,这是给代码规范用的,没有现实的业务功能。
下面有一行解释:requests.Request mode,看来这个跟requests有点关系。
回过头来看看self.__step_context.request,也就是self.__step_context对象有个request属性,它的界说是:
  1. <code>self.__step_context = TStep(name=name)
复制代码
答案应该就在TStep中了:
  1. <code>class TStep(BaseModel):
  2.     name: Name
  3.     <code>request: Union[TRequest, None] = None
  4.     testcase: Union[Text, Callable, None] = None
  5.     variables: VariablesMapping = {}
  6.     setup_hooks: Hooks = []
  7.     teardown_hooks: Hooks = []
  8.     # used to extract request's response field
  9.     extract: VariablesMapping = {}
  10.     # used to export session variables from referenced testcase
  11.     export: Export = []
  12.     validators: Validators = Field([], alias="validate")
  13.     validate_script: List[Text] = []
复制代码
照旧个Model,内里的request界说是:
  1. <code>request: Union[TRequest, None] = None
复制代码
又绕回TRequest了。这个Union是typing模块内里的:Union[X, Y] means either X or Y. 意思就是request的范例要么是TRequest要么是None。
在刚才get的方法中,另有一句return RequestWithOptionalArgs(self.__step_context),RequestWithOptionalArgs的界说如下:
  1. <code>class RequestWithOptionalArgs(object):
  2.     def \_\_init\_\_(self, step\_context: TStep):
  3.         self.__step_context = step_context
  4.     def with\_params(self, **params) -> "RequestWithOptionalArgs":
  5.         self.__step_context.request.params.update(params)
  6.         return self
  7.     def with\_headers(self, **headers) -> "RequestWithOptionalArgs":
  8.         self.__step_context.request.headers.update(headers)
  9.         return self
  10.     def with\_cookies(self, **cookies) -> "RequestWithOptionalArgs":
  11.         self.__step_context.request.cookies.update(cookies)
  12.         return self
  13.     def with\_data(self, data) -> "RequestWithOptionalArgs":
  14.         self.__step_context.request.data = data
  15.         return self
  16.     def with\_json(self, req\_json) -> "RequestWithOptionalArgs":
  17.         self.__step_context.request.req_json = req_json
  18.         return self
  19.     def set\_timeout(self, timeout: float) -> "RequestWithOptionalArgs":
  20.         self.__step_context.request.timeout = timeout
  21.         return self
  22.     def set\_verify(self, verify: bool) -> "RequestWithOptionalArgs":
  23.         self.__step_context.request.verify = verify
  24.         return self
  25.     def set\_allow\_redirects(self, allow\_redirects: bool) -> "RequestWithOptionalArgs":
  26.         self.__step_context.request.allow_redirects = allow_redirects
  27.         return self
  28.     def upload(self, **file\_info) -> "RequestWithOptionalArgs":
  29.         self.__step_context.request.upload.update(file_info)
  30.         return self
  31.     def teardown\_hook(
  32. self, hook: Text, assign\_var\_name: Text = None
  33. ) -> "RequestWithOptionalArgs":
  34.         if assign_var_name:
  35.             self.__step_context.teardown_hooks.append({assign_var_name: hook})
  36.         else:
  37.             self.__step_context.teardown_hooks.append(hook)
  38.         return self
  39.     def extract(self) -> StepRequestExtraction:
  40.         return StepRequestExtraction(self.__step_context)
  41.     def validate(self) -> StepRequestValidation:
  42.         return StepRequestValidation(self.__step_context)
  43.     def perform(self) -> TStep:
  44.         return self.__step_context
复制代码
可以给HTTP哀求添加params、headers等可选项。
看到这里,仍旧不知道HTTP哀求到底发出去的,由于没有调用呀。
只能往上层找,看调用RunRequest的Step类:
  1. <code>class Step(object):
  2.     def \_\_init\_\_(
  3. self,
  4. step\_context: Union[
  5. StepRequestValidation,
  6. StepRequestExtraction,
  7. RequestWithOptionalArgs,
  8. RunTestCase,
  9. StepRefCase,
  10. ],
  11. ):
  12.         self.__step_context = step_context.perform()
  13. @property
  14.     def request(self) -> TRequest:
  15.         return self.__step_context.request
  16. @property
  17.     def testcase(self) -> TestCase:
  18.         return self.__step_context.testcase
  19.     def perform(self) -> TStep:
  20.         return self.__step_context
复制代码
Step类的__init__方法也用Union做了范例校验,此中RequestWithOptionalArgs就是RunRequest的gei等方法会返回的,这倒是匹配上了。它另有个request属性。有点端倪了。
再往上层找,看HttpRunner类,有个__run_step_request的方法:
  1. <code>def \_\_run\_step\_request(self, step: TStep) -> StepData:
  2.     """run teststep: request"""
  3.     step_data = StepData(name=step.name)
  4.     # parse
  5.     prepare_upload_step(step, self.__project_meta.functions)
  6.     request_dict = step.request.dict()
  7.     request_dict.pop("upload", None)
  8.     parsed_request_dict = parse_data(
  9.         request_dict, step.variables, self.__project_meta.functions
  10.     )
  11.     parsed_request_dict["headers"].setdefault(
  12.         "HRUN-Request-ID",
  13.         f"HRUN-{self.\_\_case\_id}-{str(int(time.time() * 1000))[-6:]}",
  14.     )
  15.     step.variables["request"] = parsed_request_dict
  16.     # setup hooks
  17.     if step.setup_hooks:
  18.         self.__call_hooks(step.setup_hooks, step.variables, "setup request")
  19.     # prepare arguments
  20.     method = parsed_request_dict.pop("method")
  21.     url_path = parsed_request_dict.pop("url")
  22.     url = build_url(self.__config.base_url, url_path)
  23.     parsed_request_dict["verify"] = self.__config.verify
  24.     parsed_request_dict["json"] = parsed_request_dict.pop("req\_json", {})
  25.     # request
  26.     resp = self.__session.request(method, url, **parsed_request_dict)
  27.     resp_obj = ResponseObject(resp)
  28.     step.variables["response"] = resp_obj
  29.     # teardown hooks
  30.     if step.teardown_hooks:
  31.         self.__call_hooks(step.teardown_hooks, step.variables, "teardown request")
  32.     def log\_req\_resp\_details():
  33.         err_msg = "\n{} DETAILED REQUEST & RESPONSE {}\n".format("*" * 32, "*" * 32)
  34.         # log request
  35.         err_msg += "====== request details ======\n"
  36.         err_msg += f"url: {url}\n"
  37.         err_msg += f"method: {method}\n"
  38.         headers = parsed_request_dict.pop("headers", {})
  39.         err_msg += f"headers: {headers}\n"
  40.         for k, v in parsed_request_dict.items():
  41.             v = utils.omit_long_data(v)
  42.             err_msg += f"{k}: {repr(v)}\n"
  43.         err_msg += "\n"
  44.         # log response
  45.         err_msg += "====== response details ======\n"
  46.         err_msg += f"status\_code: {resp.status\_code}\n"
  47.         err_msg += f"headers: {resp.headers}\n"
  48.         err_msg += f"body: {repr(resp.text)}\n"
  49.         logger.error(err_msg)
  50.     # extract
  51.     extractors = step.extract
  52.     extract_mapping = resp_obj.extract(extractors)
  53.     step_data.export_vars = extract_mapping
  54.     variables_mapping = step.variables
  55.     variables_mapping.update(extract_mapping)
  56.     # validate
  57.     validators = step.validators
  58.     session_success = False
  59.     try:
  60.         resp_obj.validate(
  61.             validators, variables_mapping, self.__project_meta.functions
  62.         )
  63.         session_success = True
  64.     except ValidationFailure:
  65.         session_success = False
  66.         log_req_resp_details()
  67.         # log testcase duration before raise ValidationFailure
  68.         self.__duration = time.time() - self.__start_at
  69.         raise
  70.     finally:
  71.         self.success = session_success
  72.         step_data.success = session_success
  73.         if hasattr(self.__session, "data"):
  74.             # httprunner.client.HttpSession, not locust.clients.HttpSession
  75.             # save request & response meta data
  76.             self.__session.data.success = session_success
  77.             self.__session.data.validators = resp_obj.validation_results
  78.             # save step data
  79.             step_data.data = self.__session.data
  80.     return step_data
复制代码
就是这里了,它的函数名用了双下划线开头:双下划线前缀会让Python表明器重写属性名称,以制止子类中的定名辩论。 这也称为名称改写(name mangling),即表明器会更改变量的名称,以便在稍后扩展这个类时制止定名辩论。说人话就是,类的私有成员,只能在类的内部调用,不对外袒露。它只在__run_step()方法中调用了1次:step_data = self.__run_step_request(step)。
中心有一段:
  1. <code># request
  2. resp = self.__session.request(method, url, **parsed_request_dict)
  3. resp_obj = ResponseObject(resp)
  4. step.variables["response"] = resp_obj
复制代码
好家伙,self.__session.request(),跟reqeusts谁人有点像了。点进去。
一下就跳转到了httprunner.client.py,众里寻他千百度,沉默回顾,它竟然就在client
  1. <code>class HttpSession(requests.Session):
  2.     """
  3. Class for performing HTTP requests and holding (session-) cookies between requests (in order
  4. to be able to log in and out of websites). Each request is logged so that HttpRunner can
  5. display statistics.
  6. This is a slightly extended version of `python-request `\_'s
  7. :py:class:`requests.Session` class and mostly this class works exactly the same.
  8. """
  9.     def \_\_init\_\_(self):
  10.         super(HttpSession, self).__init__()
  11.         self.data = SessionData()
  12.     def update\_last\_req\_resp\_record(self, resp\_obj):
  13.         """
  14. update request and response info from Response() object.
  15. """
  16.         # TODO: fix
  17.         self.data.req_resps.pop()
  18.         self.data.req_resps.append(get_req_resp_record(resp_obj))
  19.     def request(self, method, url, name=None, **kwargs):
复制代码
继续了requests.Session然后举行了重写。
果然,照旧用到了requests库。
   参考资料:
  https://github.com/httprunner/httprunner

免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!

帖子地址: 

回复

使用道具 举报

分享
推广
火星云矿 | 预约S19Pro,享500抵1000!
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

草根技术分享(草根吧)是全球知名中文IT技术交流平台,创建于2021年,包含原创博客、精品问答、职业培训、技术社区、资源下载等产品服务,提供原创、优质、完整内容的专业IT技术开发社区。
  • 官方手机版

  • 微信公众号

  • 商务合作