GithubBackup/httprunner
1
0
Fork 0
mirror of https://github.com/httprunner/httprunner.git synced 2026-09-07 00:17:37 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: extract content/text/json

Browse Source
This commit is contained in:
httprunner
2018-07-25 16:10:45 +08:00
parent 077cae3ca1
commit 7f640de79e
3 changed files with 118 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
httprunner/response.py
+19 -30
View File
@@ -63,7 +63,6 @@ class ResponseObject(object):
"headers.content-type" "headers.content-type"
"content.person.name.first_name" "content.person.name.first_name"
""" """
try:
# string.split(sep=None, maxsplit=-1) -> list of strings # string.split(sep=None, maxsplit=-1) -> list of strings
# e.g. "content.person.name" => ["content", "person.name"] # e.g. "content.person.name" => ["content", "person.name"]
try: try:
@@ -130,41 +129,31 @@ class ResponseObject(object):
logger.log_error(err_msg) logger.log_error(err_msg)
raise exceptions.ParamsError(err_msg) raise exceptions.ParamsError(err_msg)
# response body
elif top_query in ["content", "text", "json"]:
try: try:
top_query_content = getattr(self, top_query) body = self.json
except AttributeError:
err_msg = u"Failed to extract attribute from response object: resp_obj.{}".format(top_query)
logger.log_error(err_msg)
raise exceptions.ParamsError(err_msg)
if sub_query:
if not isinstance(top_query_content, (dict, CaseInsensitiveDict, list)):
try:
# TODO: remove compatibility for content, text
if isinstance(top_query_content, bytes):
top_query_content = top_query_content.decode("utf-8")
if isinstance(top_query_content, PreparedRequest):
top_query_content = top_query_content.__dict__
else:
top_query_content = json.loads(top_query_content)
except exceptions.JSONDecodeError: except exceptions.JSONDecodeError:
err_msg = u"Failed to extract data with delimiter!\n" body = self.text
err_msg += u"response content: {}\n".format(self.content)
err_msg += u"regex: {}\n".format(field) if not sub_query:
# extract response body
return body
if isinstance(body, (dict, list)):
# content = {"xxx": 123}, content.xxx
return utils.query_json(body, sub_query)
else:
# content = "<html>abcdefg</html>", content.xxx
err_msg = u"ParamsError: Failed to extract attribute from response body! => {}\n".format(field)
err_msg += u"response body: {}\n".format(body)
logger.log_error(err_msg) logger.log_error(err_msg)
raise exceptions.ParamsError(err_msg) raise exceptions.ParamsError(err_msg)
# e.g. key: resp_headers_content_type, sub_query = "content-type" # others
return utils.query_json(top_query_content, sub_query)
else: else:
# e.g. key: resp_status_code, resp_content err_msg = u"ParamsError: Failed to extract attribute from response! => {}\n".format(field)
return top_query_content err_msg += u"available response attributes: status_code, cookies, elapsed, headers, content, text, json."
except AttributeError:
err_msg = u"Failed to extract value from response!\n"
err_msg += u"response content: {}\n".format(self.content)
err_msg += u"extract: {}\n".format(field)
logger.log_error(err_msg) logger.log_error(err_msg)
raise exceptions.ParamsError(err_msg) raise exceptions.ParamsError(err_msg)
tests/httpbin/hooks.yml
+2 -2
View File
@@ -32,5 +32,5 @@
- eq: ["status_code", 500] - eq: ["status_code", 500]
- eq: ["headers.content-type", "html/text"] - eq: ["headers.content-type", "html/text"]
- eq: [json.headers.Host, "127.0.0.1:8888"] - eq: [json.headers.Host, "127.0.0.1:8888"]
- eq: [content.headers.Host, "127.0.0.1:3458"] - eq: [content.headers.Host, "127.0.0.1:8888"]
- eq: [text.headers.Host, "127.0.0.1:3458"] - eq: [text.headers.Host, "127.0.0.1:8888"]
tests/test_response.py
+34 -7
View File
@@ -1,6 +1,6 @@
import requests import requests
from httprunner import exceptions, response, utils from httprunner import exceptions, response, utils
from httprunner.compat import bytes from httprunner.compat import bytes, str
from tests.base import ApiServerUnittest from tests.base import ApiServerUnittest
@@ -118,11 +118,13 @@ class TestResponse(ApiServerUnittest):
extract_binds_list = [ extract_binds_list = [
{"resp_headers": "headers"}, {"resp_headers": "headers"},
{"resp_headers_content_type": "headers.Content-Type"} {"resp_headers_content_type": "headers.Content-Type"},
{"resp_headers_content_type_lowercase": "headers.content-type"}
] ]
extract_binds_dict = resp_obj.extract_response(extract_binds_list) extract_binds_dict = resp_obj.extract_response(extract_binds_list)
self.assertIn("Content-Type", extract_binds_dict["resp_headers"]) self.assertIn("Content-Type", extract_binds_dict["resp_headers"])
self.assertIn("text/html", extract_binds_dict["resp_headers_content_type"]) self.assertIn("text/html", extract_binds_dict["resp_headers_content_type"])
self.assertIn("text/html", extract_binds_dict["resp_headers_content_type_lowercase"])
extract_binds_list = [ extract_binds_list = [
{"resp_headers_xxx": "headers.xxx"} {"resp_headers_xxx": "headers.xxx"}
@@ -130,7 +132,7 @@ class TestResponse(ApiServerUnittest):
with self.assertRaises(exceptions.ParamsError): with self.assertRaises(exceptions.ParamsError):
resp_obj.extract_response(extract_binds_list) resp_obj.extract_response(extract_binds_list)
def test_extract_response_json(self): def test_extract_response_body_json(self):
resp = requests.post( resp = requests.post(
url="http://127.0.0.1:3458/anything", url="http://127.0.0.1:3458/anything",
json={ json={
@@ -190,10 +192,6 @@ class TestResponse(ApiServerUnittest):
resp_obj = response.ResponseObject(resp) resp_obj = response.ResponseObject(resp)
extract_binds_dict = resp_obj.extract_response(extract_binds_list) extract_binds_dict = resp_obj.extract_response(extract_binds_list)
self.assertEqual(
extract_binds_dict["resp_status_code"],
200
)
self.assertEqual( self.assertEqual(
extract_binds_dict["resp_headers_content_type"], extract_binds_dict["resp_headers_content_type"],
"application/json" "application/json"
@@ -219,6 +217,35 @@ class TestResponse(ApiServerUnittest):
"Shenzhen" "Shenzhen"
) )
def test_extract_response_body_html(self):
resp = requests.get(url="http://127.0.0.1:3458/")
resp_obj = response.ResponseObject(resp)
extract_binds_list = [
{"resp_content": "content"}
]
extract_binds_dict = resp_obj.extract_response(extract_binds_list)
self.assertIsInstance(extract_binds_dict["resp_content"], str)
self.assertIn("python-requests.org", extract_binds_dict["resp_content"])
extract_binds_list = [
{"resp_content": "content.xxx"}
]
with self.assertRaises(exceptions.ParamsError):
resp_obj.extract_response(extract_binds_list)
def test_extract_response_others(self):
resp = requests.get(url="http://127.0.0.1:3458/status/200")
resp_obj = response.ResponseObject(resp)
extract_binds_list = [
{"resp_others_encoding": "encoding"},
{"resp_others_history": "history"}
]
with self.assertRaises(exceptions.ParamsError):
resp_obj.extract_response(extract_binds_list)
def test_extract_response_fail(self): def test_extract_response_fail(self):
resp = requests.post( resp = requests.post(
url="http://127.0.0.1:3458/anything", url="http://127.0.0.1:3458/anything",