Python微信订餐小步伐课程视频
https://edu.caogenba.net/course/detail/36074
Python实战量化买卖业务理财体系
https://edu.caogenba.net/course/detail/35475
在HttpRunner3的示例代码中,发送HTTP哀求的代码是如许写的:
<code>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的内部界说了teststeps列表,由多个Step类的实例对象构成。
类Step初始化传入类RunRequest的方法get和post就把HTTP哀求发出去了。
这到底是怎么实现的?
先看下RunRequest的源码:
<code>class RunRequest(object): def \_\_init\_\_(self, name: Text): <code>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: <code>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哀求的Method。方法内部:
<code>self.__step_context.request = TRequest(method=MethodEnum.GET, url=url) 复制代码
有个TRequest类:
<code>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,是用来做数据验证的 ,好比这里的url指定了Url范例,假如传一个str范例,就会校验失败。简而言之,这是给代码规范用的,没有现实的业务功能。
下面有一行解释:requests.Request mode,看来这个跟requests有点关系。
回过头来看看self.__step_context.request,也就是self.__step_context对象有个request属性,它的界说是:
<code>self.__step_context = TStep(name=name) 复制代码
答案应该就在TStep中了:
<code>class TStep(BaseModel): name: Name <code>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界说是:
<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的界说如下:
<code>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哀求添加params、headers等可选项。
看到这里,仍旧不知道HTTP哀求到底发出去的,由于没有调用呀。
只能往上层找,看调用RunRequest的Step类:
<code>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做了范例校验,此中RequestWithOptionalArgs就是RunRequest的gei等方法会返回的,这倒是匹配上了。它另有个request属性。有点端倪了。
再往上层找,看HttpRunner类,有个__run_step_request的方法:
<code>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 复制代码
就是这里了,它的函数名用了双下划线开头:双下划线前缀 会让Python表明器重写属性名称,以制止子类中的定名辩论。 这也称为名称改写(name mangling),即表明器会更改变量的名称,以便在稍后扩展这个类时制止定名辩论。说人话就是,类的私有成员,只能在类的内部调用,不对外袒露 。它只在__run_step()方法中调用了1次:step_data = self.__run_step_request(step)。
中心有一段:
<code># request resp = self.__session.request(method, url, **parsed_request_dict) resp_obj = ResponseObject(resp) step.variables["response"] = resp_obj 复制代码
好家伙,self.__session.request(),跟reqeusts谁人有点像了。点进去。
一下就跳转到了httprunner.client.py,众里寻他千百度,沉默回顾,它竟然就在client 。
<code>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 `\_'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然后举行了重写。
果然,照旧用到了requests库。
参考资料:
https://github.com/httprunner/httprunner
免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!