mirror of
https://github.com/httprunner/httprunner.git
synced 2026-06-23 16:43:49 +08:00
318 lines
13 KiB
Python
318 lines
13 KiB
Python
import requests
|
|
|
|
from httprunner import exceptions, response
|
|
from httprunner.compat import basestring, bytes
|
|
from tests.api_server import HTTPBIN_SERVER
|
|
from tests.base import ApiServerUnittest
|
|
|
|
|
|
class TestResponse(ApiServerUnittest):
|
|
def test_parse_response_object_json(self):
|
|
url = "http://127.0.0.1:5000/api/users"
|
|
resp = requests.get(url)
|
|
resp_obj = response.ResponseObject(resp)
|
|
self.assertTrue(hasattr(resp_obj, "status_code"))
|
|
self.assertTrue(hasattr(resp_obj, "headers"))
|
|
self.assertTrue(hasattr(resp_obj, "content"))
|
|
self.assertIn("Content-Type", resp_obj.headers)
|
|
self.assertIn("Content-Length", resp_obj.headers)
|
|
self.assertIn("success", resp_obj.json)
|
|
|
|
def test_parse_response_object_content(self):
|
|
url = "http://127.0.0.1:5000/"
|
|
resp = requests.get(url)
|
|
resp_obj = response.ResponseObject(resp)
|
|
self.assertEqual(bytes, type(resp_obj.content))
|
|
|
|
def test_extract_response_status_code(self):
|
|
resp = requests.get(url="{}/status/200".format(HTTPBIN_SERVER))
|
|
resp_obj = response.ResponseObject(resp)
|
|
|
|
extract_binds_list = [{"resp_status_code": "status_code"}]
|
|
extract_binds_dict = resp_obj.extract_response(extract_binds_list)
|
|
|
|
self.assertEqual(extract_binds_dict["resp_status_code"], 200)
|
|
|
|
extract_binds_list = [{"resp_status_code": "status_code.xx"}]
|
|
with self.assertRaises(exceptions.ParamsError):
|
|
resp_obj.extract_response(extract_binds_list)
|
|
|
|
def test_extract_response_encoding_ok_reason_url(self):
|
|
resp = requests.get(url="{}/status/200".format(HTTPBIN_SERVER))
|
|
resp_obj = response.ResponseObject(resp)
|
|
|
|
extract_binds_list = [
|
|
{"resp_encoding": "encoding"},
|
|
{"resp_ok": "ok"},
|
|
{"resp_reason": "reason"},
|
|
{"resp_url": "url"},
|
|
]
|
|
extract_binds_dict = resp_obj.extract_response(extract_binds_list)
|
|
|
|
self.assertEqual(extract_binds_dict["resp_encoding"], "utf-8")
|
|
self.assertEqual(extract_binds_dict["resp_ok"], True)
|
|
self.assertEqual(extract_binds_dict["resp_reason"], "OK")
|
|
self.assertEqual(
|
|
extract_binds_dict["resp_url"], "{}/status/200".format(HTTPBIN_SERVER)
|
|
)
|
|
|
|
extract_binds_list = [{"resp_encoding": "encoding.xx"}]
|
|
with self.assertRaises(exceptions.ParamsError):
|
|
resp_obj.extract_response(extract_binds_list)
|
|
|
|
extract_binds_list = [{"resp_ok": "ok.xx"}]
|
|
with self.assertRaises(exceptions.ParamsError):
|
|
resp_obj.extract_response(extract_binds_list)
|
|
|
|
extract_binds_list = [{"resp_reason": "reason.xx"}]
|
|
with self.assertRaises(exceptions.ParamsError):
|
|
resp_obj.extract_response(extract_binds_list)
|
|
|
|
extract_binds_list = [{"resp_url": "url.xx"}]
|
|
with self.assertRaises(exceptions.ParamsError):
|
|
resp_obj.extract_response(extract_binds_list)
|
|
|
|
def test_extract_response_cookies(self):
|
|
resp = requests.get(
|
|
url="{}/cookies".format(HTTPBIN_SERVER),
|
|
headers={"accept": "application/json"},
|
|
)
|
|
resp_obj = response.ResponseObject(resp)
|
|
|
|
extract_binds_list = [{"resp_cookies": "cookies"}]
|
|
extract_binds_dict = resp_obj.extract_response(extract_binds_list)
|
|
self.assertEqual(extract_binds_dict["resp_cookies"], {})
|
|
|
|
extract_binds_list = [{"resp_cookies": "cookies.xx"}]
|
|
with self.assertRaises(exceptions.ExtractFailure):
|
|
resp_obj.extract_response(extract_binds_list)
|
|
|
|
def test_extract_response_elapsed(self):
|
|
resp = requests.post(
|
|
url="{}/anything".format(HTTPBIN_SERVER),
|
|
json={
|
|
"success": False,
|
|
"person": {
|
|
"name": {
|
|
"first_name": "Leo",
|
|
"last_name": "Lee",
|
|
},
|
|
"age": 29,
|
|
"cities": ["Guangzhou", "Shenzhen"],
|
|
},
|
|
},
|
|
)
|
|
resp_obj = response.ResponseObject(resp)
|
|
|
|
extract_binds_list = [{"resp_elapsed": "elapsed"}]
|
|
with self.assertRaises(exceptions.ParamsError):
|
|
resp_obj.extract_response(extract_binds_list)
|
|
|
|
extract_binds_list = [
|
|
{"resp_elapsed_microseconds": "elapsed.microseconds"},
|
|
{"resp_elapsed_seconds": "elapsed.seconds"},
|
|
{"resp_elapsed_days": "elapsed.days"},
|
|
{"resp_elapsed_total_seconds": "elapsed.total_seconds"},
|
|
]
|
|
extract_binds_dict = resp_obj.extract_response(extract_binds_list)
|
|
self.assertGreater(extract_binds_dict["resp_elapsed_microseconds"], 1000)
|
|
self.assertLess(extract_binds_dict["resp_elapsed_seconds"], 60)
|
|
self.assertEqual(extract_binds_dict["resp_elapsed_days"], 0)
|
|
self.assertGreater(extract_binds_dict["resp_elapsed_total_seconds"], 0)
|
|
|
|
extract_binds_list = [{"resp_elapsed": "elapsed.years"}]
|
|
with self.assertRaises(exceptions.ParamsError):
|
|
resp_obj.extract_response(extract_binds_list)
|
|
|
|
def test_extract_response_headers(self):
|
|
resp = requests.get(url="{}/status/200".format(HTTPBIN_SERVER))
|
|
resp_obj = response.ResponseObject(resp)
|
|
|
|
extract_binds_list = [
|
|
{"resp_headers": "headers"},
|
|
{"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)
|
|
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_lowercase"]
|
|
)
|
|
|
|
extract_binds_list = [{"resp_headers_xxx": "headers.xxx"}]
|
|
with self.assertRaises(exceptions.ExtractFailure):
|
|
resp_obj.extract_response(extract_binds_list)
|
|
|
|
def test_extract_response_body_json(self):
|
|
resp = requests.post(
|
|
url="{}/anything".format(HTTPBIN_SERVER),
|
|
json={
|
|
"success": False,
|
|
"person": {
|
|
"name": {
|
|
"first_name": "Leo",
|
|
"last_name": "Lee",
|
|
},
|
|
"age": 29,
|
|
"cities": ["Guangzhou", "Shenzhen"],
|
|
},
|
|
},
|
|
)
|
|
# resp.json()
|
|
# {
|
|
# "args": {},
|
|
# "data": "{\"success\": false, \"person\": {\"name\": {\"first_name\": \"Leo\", \"last_name\": \"Lee\"}, \"age\": 29, \"cities\": [\"Guangzhou\", \"Shenzhen\"]}}",
|
|
# "files": {},
|
|
# "form": {},
|
|
# "headers": {
|
|
# "Accept": "*/*",
|
|
# "Accept-Encoding": "gzip, deflate",
|
|
# "Connection": "keep-alive",
|
|
# "Content-Length": "129",
|
|
# "Content-Type": "application/json",
|
|
# "Host": HTTPBIN_SERVER,
|
|
# "User-Agent": "python-requests/2.18.4"
|
|
# },
|
|
# "json": {
|
|
# "person": {
|
|
# "age": 29,
|
|
# "cities": [
|
|
# "Guangzhou",
|
|
# "Shenzhen"
|
|
# ],
|
|
# "name": {
|
|
# "first_name": "Leo",
|
|
# "last_name": "Lee"
|
|
# }
|
|
# },
|
|
# "success": false
|
|
# },
|
|
# "method": "POST",
|
|
# "origin": "127.0.0.1",
|
|
# "url": "{}/anything".format(HTTPBIN_SERVER)
|
|
# }
|
|
|
|
extract_binds_list = [
|
|
{"resp_headers_content_type": "headers.content-type"},
|
|
{"resp_content_body_success": "json.json.success"},
|
|
{"resp_content_content_success": "content.json.success"},
|
|
{"resp_content_text_success": "text.json.success"},
|
|
{"resp_content_person_first_name": "content.json.person.name.first_name"},
|
|
{"resp_content_cities_1": "content.json.person.cities.1"},
|
|
]
|
|
resp_obj = response.ResponseObject(resp)
|
|
extract_binds_dict = resp_obj.extract_response(extract_binds_list)
|
|
|
|
self.assertEqual(
|
|
extract_binds_dict["resp_headers_content_type"], "application/json"
|
|
)
|
|
self.assertEqual(extract_binds_dict["resp_content_body_success"], False)
|
|
self.assertEqual(extract_binds_dict["resp_content_content_success"], False)
|
|
self.assertEqual(extract_binds_dict["resp_content_text_success"], False)
|
|
self.assertEqual(extract_binds_dict["resp_content_person_first_name"], "Leo")
|
|
self.assertEqual(extract_binds_dict["resp_content_cities_1"], "Shenzhen")
|
|
|
|
def test_extract_response_body_html(self):
|
|
resp = requests.get(url=HTTPBIN_SERVER)
|
|
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"], basestring)
|
|
self.assertIn("httpbin.org", extract_binds_dict["resp_content"])
|
|
|
|
extract_binds_list = [{"resp_content": "content.xxx"}]
|
|
with self.assertRaises(exceptions.ExtractFailure):
|
|
resp_obj.extract_response(extract_binds_list)
|
|
|
|
def test_extract_response_others(self):
|
|
resp = requests.get(url="{}/status/200".format(HTTPBIN_SERVER))
|
|
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):
|
|
resp = requests.post(
|
|
url="{}/anything".format(HTTPBIN_SERVER),
|
|
json={
|
|
"success": False,
|
|
"person": {
|
|
"name": {
|
|
"first_name": "Leo",
|
|
"last_name": "Lee",
|
|
},
|
|
"age": 29,
|
|
"cities": ["Guangzhou", "Shenzhen"],
|
|
},
|
|
},
|
|
)
|
|
|
|
extract_binds_list = [{"resp_content_dict_key_error": "content.not_exist"}]
|
|
resp_obj = response.ResponseObject(resp)
|
|
|
|
with self.assertRaises(exceptions.ExtractFailure):
|
|
resp_obj.extract_response(extract_binds_list)
|
|
|
|
extract_binds_list = [
|
|
{"resp_content_list_index_error": "content.person.cities.3"}
|
|
]
|
|
resp_obj = response.ResponseObject(resp)
|
|
|
|
with self.assertRaises(exceptions.ExtractFailure):
|
|
resp_obj.extract_response(extract_binds_list)
|
|
|
|
def test_extract_response_json_string(self):
|
|
resp = requests.post(url="{}/anything".format(HTTPBIN_SERVER), data="abc")
|
|
|
|
extract_binds_list = [{"resp_content_body": "content.data"}]
|
|
resp_obj = response.ResponseObject(resp)
|
|
|
|
extract_binds_dict = resp_obj.extract_response(extract_binds_list)
|
|
self.assertEqual(extract_binds_dict["resp_content_body"], "abc")
|
|
|
|
def test_extract_text_response(self):
|
|
resp = requests.post(
|
|
url="{}/anything".format(HTTPBIN_SERVER), data="LB123abcRB789"
|
|
)
|
|
|
|
extract_binds_list = [
|
|
{"resp_content_key1": "LB123(.*)RB789"},
|
|
{"resp_content_key2": "LB[\d]*(.*)RB[\d]*"},
|
|
{"resp_content_key3": "LB[\d]*(.*)9"},
|
|
]
|
|
resp_obj = response.ResponseObject(resp)
|
|
|
|
extract_binds_dict = resp_obj.extract_response(extract_binds_list)
|
|
self.assertEqual(extract_binds_dict["resp_content_key1"], "abc")
|
|
self.assertEqual(extract_binds_dict["resp_content_key2"], "abc")
|
|
self.assertEqual(extract_binds_dict["resp_content_key3"], "abcRB78")
|
|
|
|
def test_extract_text_response_exception(self):
|
|
resp = requests.post(
|
|
url="{}/anything".format(HTTPBIN_SERVER), data="LB123abcRB789"
|
|
)
|
|
extract_binds_list = [{"resp_content_key1": "LB123.*RB789"}]
|
|
resp_obj = response.ResponseObject(resp)
|
|
with self.assertRaises(exceptions.ParamsError):
|
|
resp_obj.extract_response(extract_binds_list)
|
|
|
|
def test_extract_response_empty(self):
|
|
resp = requests.post(url="{}/anything".format(HTTPBIN_SERVER), data="abc")
|
|
|
|
extract_binds_list = [{"resp_content_body": "content.data"}]
|
|
resp_obj = response.ResponseObject(resp)
|
|
extract_binds_dict = resp_obj.extract_response(extract_binds_list)
|
|
self.assertEqual(extract_binds_dict["resp_content_body"], "abc")
|
|
|
|
extract_binds_list = [{"resp_content_body": "content.data.def"}]
|
|
resp_obj = response.ResponseObject(resp)
|
|
with self.assertRaises(exceptions.ExtractFailure):
|
|
resp_obj.extract_response(extract_binds_list)
|