詳解HttpRunner3的HTTP請是如何發(fā)出
HttpRunner3示例代碼發(fā)送HTTP請求
在HttpRunner3的示例代碼中,發(fā)送HTTP請求的代碼是這樣寫的:
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請求發(fā)出去了。
如何實現(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請求的Method。
方法內部
self.__step_context.request = TRequest(method=MethodEnum.GET, url=url)
有個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ù)據驗證的,比如這里的url指定了Url類型,如果傳一個str類型,就會校驗失敗。簡而言之,這是給代碼規(guī)范用的,沒有實際的業(yè)務功能。
下面有一行注釋:requests.Request mode,看來這個跟requests有點關系。
回過頭來看看??self.__step_context.request???,也就是??self.__step_context??對象有個request屬性,它的定義是:
self.__step_context = TStep(name=name)
答案應該就在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定義
還是個Model,里面的request定義是:
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的定義如下:
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請求到底發(fā)出去的,因為沒有調用呀。
調用RunRequest的Step類
只能往上層找,看調用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_contextStep類的??__init__??方法也用Union做了類型校驗,其中RequestWithOptionalArgs就是RunRequest的gei等方法會返回的,這倒是匹配上了。它還有個request屬性。有點眉目了。
看HttpRunner類
再往上層找,看HttpRunner類,有個??__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ù)名用了雙下劃線開頭:雙下劃線前綴會讓Python解釋器重寫屬性名稱,以避免子類中的命名沖突。 這也稱為名稱改寫(name mangling),即解釋器會更改變量的名稱,以便在稍后擴展這個類時避免命名沖突。說人話就是,類的私有成員,只能在類的內部調用,不對外暴露。
它只在??__run_step()???方法中調用了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那個有點像了。點進去。
一下就跳轉到了??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然后進行了重寫。
果然,還是用到了requests庫。
以上就是詳解HttpRunner3的HTTP請是如何發(fā)出的詳細內容,更多關于HttpRunner3 HTTP請求發(fā)出的資料請關注腳本之家其它相關文章!
相關文章
Python 格式化輸出_String Formatting_控制小數(shù)點位數(shù)的實例詳解
在本篇文章里小編給大家整理了關于Python 格式化輸出_String Formatting_控制小數(shù)點位數(shù)的實例內容,需要的朋友們參考下。2020-02-02
Python判斷一個三位數(shù)是否為水仙花數(shù)的示例
今天小編就為大家分享一篇Python判斷一個三位數(shù)是否為水仙花數(shù)的示例,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧2018-11-11
Python實現(xiàn)給qq郵箱發(fā)送郵件的方法
這篇文章主要介紹了Python實現(xiàn)給qq郵箱發(fā)送郵件的方法,涉及Python郵件發(fā)送的相關技巧,需要的朋友可以參考下2015-05-05
解決出現(xiàn)Incorrect integer value: '''' for column ''id'' at row 1
這篇文章主要介紹了解決出現(xiàn)Incorrect integer value: '' for column 'id' at row 1的問題的相關資料,希望通過本文能幫助到大家,讓大家遇到這樣的問題及時的解決,需要的朋友可以參考下2017-10-10
在linux系統(tǒng)中安裝python3.8.1?并卸載?python3.6.2?更新python3引導到3.8.1的
這篇文章主要介紹了如何在linux系統(tǒng)中安裝python3.8.1?并卸載?python3.6.2?更新python3引導到3.8.1,本文分步驟給大家介紹的非常詳細,對大家的學習或工作具有一定的參考借鑒價值,需要的朋友可以參考下2023-11-11
利用Tensorflow構建和訓練自己的CNN來做簡單的驗證碼識別方式
今天小編就為大家分享一篇利用Tensorflow構建和訓練自己的CNN來做簡單的驗證碼識別方式,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧2020-01-01

