Files
httprunner/tests/test_response.py
2022-04-30 15:06:31 +08:00

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)