GithubBackup/httprunner
1
0
Fork 0
mirror of https://github.com/httprunner/httprunner.git synced 2026-05-12 11:29:48 +08:00
Code Issues Packages Projects Releases Wiki Activity

change: format code with balck

Browse Source
This commit is contained in:
debugtalk
2020-05-14 14:21:35 +08:00
parent a95d4137c1
commit 1c97e0bbfe
45 changed files with 664 additions and 844 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
httprunner/api.py
View File

@@ -36,10 +36,7 @@ class HttpRunner(object):
"""
self.exception_stage = "initialize HttpRunner()"
kwargs = {
"failfast": True,
"resultclass": report.HtmlTestResult
}
kwargs = {"failfast": True, "resultclass": report.HtmlTestResult}
logger.remove()
log_level = log_level.upper()
@@ -57,6 +54,7 @@ class HttpRunner(object):
def _add_test(test_runner: TestCaseRunner):
""" add test to testcase.
"""
def test(self):
try:
test_runner.run()
@@ -79,7 +77,7 @@ class HttpRunner(object):
test_runner = TestCaseRunner().init(testcase)
TestSequense = type('TestSequense', (unittest.TestCase,), {})
TestSequense = type("TestSequense", (unittest.TestCase,), {})
test_method = _add_test(test_runner)
setattr(TestSequense, "test_method_name", test_method)
@@ -89,7 +87,9 @@ class HttpRunner(object):
return prepared_testcases
def _run_suite(self, prepared_testcases: List[unittest.TestSuite]) -> List[TestCaseSummary]:
def _run_suite(
self, prepared_testcases: List[unittest.TestSuite]
) -> List[TestCaseSummary]:
""" run prepared testcases
"""
tests_results: List[TestCaseSummary] = []
@@ -131,9 +131,7 @@ class HttpRunner(object):
"""
testsuite_summary = TestSuiteSummary(
success=True,
platform=report.get_platform(),
testcases=[]
success=True, platform=report.get_platform(), testcases=[]
)
testsuite_summary.stat.total = len(tests_results)
testsuite_summary.stat.success = 0
@@ -148,11 +146,16 @@ class HttpRunner(object):
testsuite_summary.success &= testcase_summary.success
testsuite_summary.testcases.append(testcase_summary)
total_duration = tests_results[-1].time.start_at + tests_results[-1].time.duration \
- tests_results[0].time.start_at
total_duration = (
tests_results[-1].time.start_at
+ tests_results[-1].time.duration
- tests_results[0].time.start_at
)
testsuite_summary.time.start_at = tests_results[0].time.start_at
testsuite_summary.time.start_at_iso_format = tests_results[0].time.start_at_iso_format
testsuite_summary.time.start_at_iso_format = tests_results[
0
].time.start_at_iso_format
testsuite_summary.time.duration = total_duration
return testsuite_summary
@@ -166,7 +169,7 @@ class HttpRunner(object):
if self.save_tests:
utils.dump_json_file(
tests_mapping,
utils.prepare_log_file_abs_path(self.test_path, "loaded.json")
utils.prepare_log_file_abs_path(self.test_path, "loaded.json"),
)
# prepare testcases
@@ -187,13 +190,12 @@ class HttpRunner(object):
if self.save_tests:
utils.dump_json_file(
self._summary.dict(),
utils.prepare_log_file_abs_path(self.test_path, "summary.json")
utils.prepare_log_file_abs_path(self.test_path, "summary.json"),
)
# save variables and export data
vars_out = self.get_vars_out()
utils.dump_json_file(
vars_out,
utils.prepare_log_file_abs_path(self.test_path, "io.json")
vars_out, utils.prepare_log_file_abs_path(self.test_path, "io.json")
)
return self._summary
@@ -266,7 +268,9 @@ class HttpRunner(object):
if loader.is_test_path(path_or_tests):
return self.run_path(path_or_tests, dot_env_path, mapping)
project_working_directory = path_or_tests.get("project_meta", {}).get("PWD", os.getcwd())
project_working_directory = path_or_tests.get("project_meta", {}).get(
"PWD", os.getcwd()
)
loader.init_pwd(project_working_directory)
return self.run_tests(path_or_tests)

21
httprunner/api_test.py
View File

@@ -4,12 +4,25 @@ from httprunner.api import HttpRunner
class TestHttpRunner(unittest.TestCase):
def setUp(self):
self.runner = HttpRunner()
def test_run_testcase_by_path(self):
summary = self.runner.run_path("examples/postman_echo/request_methods/")
def test_run_testcase_by_path_request_only(self):
summary = self.runner.run_path(
"examples/postman_echo/request_methods/request_with_variables.yml"
)
self.assertTrue(summary.success)
self.assertEqual(summary.testcases[0].name, "request methods testcase with variables")
self.assertEqual(
summary.testcases[0].name, "request methods testcase with variables"
)
self.assertGreater(summary.stat.total, 1)
def test_run_testcase_by_path_ref_testcase(self):
summary = self.runner.run_path(
"examples/postman_echo/request_methods/request_with_testcase_reference.yml"
)
self.assertTrue(summary.success)
self.assertEqual(
summary.testcases[0].name, "request methods testcase with variables"
)
self.assertGreater(summary.stat.total, 1)

8
httprunner/app/main.py
View File

@@ -8,13 +8,7 @@ app = FastAPI()
@app.get("/hrun/version")
async def get_hrun_version():
return {
"code": 0,
"message": "success",
"result": {
"HttpRunner": __version__
}
}
return {"code": 0, "message": "success", "result": {"HttpRunner": __version__}}
app.include_router(deps.router)

11
httprunner/app/routers/debug.py
View File

@@ -9,11 +9,7 @@ runner = HttpRunner()
@router.post("/hrun/debug/testcase", tags=["debug"])
async def debug_single_testcase(project_meta: ProjectMeta, testcase: TestCase):
resp = {
"code": 0,
"message": "success",
"result": {}
}
resp = {"code": 0, "message": "success", "result": {}}
project_meta_json = project_meta.dict(by_alias=True)
if project_meta.debugtalk_py:
@@ -27,10 +23,7 @@ async def debug_single_testcase(project_meta: ProjectMeta, testcase: TestCase):
project_meta_json["functions"][func_name] = locals()[func_name]
testcase_json = testcase.dict(by_alias=True)
tests_mapping = {
"project_meta": project_meta_json,
"testcases": [testcase_json]
}
tests_mapping = {"project_meta": project_meta_json, "testcases": [testcase_json]}
summary = runner.run_tests(tests_mapping)
if not summary["success"]:

13
httprunner/app/routers/debug_test.py
View File

@@ -8,13 +8,12 @@ client = TestClient(app)
class TestDebug(unittest.TestCase):
def test_debug_single_testcase(self):
json_data = {
"project_meta": {
"debugtalk_py": "\ndef hello(name):\n print(f'hello, {name}')\n",
"variables": {},
"env": {}
"env": {},
},
"testcase": {
"config": {
@@ -24,7 +23,7 @@ class TestDebug(unittest.TestCase):
"variables": {},
"setup_hooks": [],
"teardown_hooks": [],
"export": []
"export": [],
},
"teststeps": [
{
@@ -38,13 +37,13 @@ class TestDebug(unittest.TestCase):
"cookies": {},
"timeout": 30,
"allow_redirects": True,
"verify": False
"verify": False,
},
"extract": {},
"validate": []
"validate": [],
}
]
}
],
},
}
response = client.post("/hrun/debug/testcase", json=json_data)
assert response.status_code == 200

8
httprunner/app/routers/debugtalk.py
View File

@@ -23,15 +23,11 @@ def stdout_io(stdout=None):
async def debug_python(request: Request):
body = await request.body()
if request.headers.get('content-transfer-encoding') == "base64":
if request.headers.get("content-transfer-encoding") == "base64":
# TODO: decode base64
pass
resp = {
"code": 0,
"message": "success",
"result": ""
}
resp = {"code": 0, "message": "success", "result": ""}
try:
with stdout_io() as s:
exec(body, globals())

12
httprunner/app/routers/deps.py
View File

@@ -10,11 +10,7 @@ router = APIRouter()
@router.get("/hrun/deps", tags=["deps"])
async def get_installed_dependenies():
resp = {
"code": 0,
"message": "success",
"result": {}
}
resp = {"code": 0, "message": "success", "result": {}}
for p in pkg_resources.working_set:
resp["result"][p.project_name] = p.version
@@ -23,11 +19,7 @@ async def get_installed_dependenies():
@router.post("/hrun/deps", tags=["deps"])
async def install_dependenies(deps: List[str]):
resp = {
"code": 0,
"message": "success",
"result": {}
}
resp = {"code": 0, "message": "success", "result": {}}
for dep in deps:
try:
p = subprocess.run(["pip", "install", dep])

6
httprunner/builtin/functions.py
View File

@@ -13,8 +13,9 @@ from httprunner.exceptions import ParamsError
def gen_random_string(str_len):
""" generate random string with specified length
"""
return ''.join(
random.choice(string.ascii_letters + string.digits) for _ in range(str_len))
return "".join(
random.choice(string.ascii_letters + string.digits) for _ in range(str_len)
)
def get_timestamp(str_len=13):
@@ -36,4 +37,3 @@ def sleep(n_secs):
""" sleep n seconds
"""
time.sleep(n_secs)

63
httprunner/cli.py
View File

@@ -6,6 +6,7 @@ if len(sys.argv) >= 2 and sys.argv[1] == "locusts":
# monkey patch ssl at beginning to avoid RecursionError when running locust.
try:
from gevent import monkey
monkey.patch_ssl()
from locust.main import main as _
except ImportError:
@@ -27,42 +28,42 @@ from httprunner.ext.locusts import init_parser_locusts, main_locusts
def init_parser_run(subparsers):
sub_parser_run = subparsers.add_parser(
"run", help="Run HttpRunner testcases.")
sub_parser_run = subparsers.add_parser("run", help="Run HttpRunner testcases.")
sub_parser_run.add_argument(
'testfile_paths', nargs='*',
help="Specify api/testcase/testsuite file paths to run.")
"testfile_paths",
nargs="*",
help="Specify api/testcase/testsuite file paths to run.",
)
sub_parser_run.add_argument(
'--log-level', default='INFO',
help="Specify logging level, default is INFO.")
"--log-level", default="INFO", help="Specify logging level, default is INFO."
)
sub_parser_run.add_argument("--log-file", help="Write logs to specified file path.")
sub_parser_run.add_argument(
'--log-file',
help="Write logs to specified file path.")
"--dot-env-path",
help="Specify .env file path, which is useful for keeping sensitive data.",
)
sub_parser_run.add_argument(
'--dot-env-path',
help="Specify .env file path, which is useful for keeping sensitive data.")
"--report-template", help="Specify report template path."
)
sub_parser_run.add_argument("--report-dir", help="Specify report save directory.")
sub_parser_run.add_argument(
'--report-template',
help="Specify report template path.")
"--report-file",
help="Specify report file path, this has higher priority than specifying report dir.",
)
sub_parser_run.add_argument(
'--report-dir',
help="Specify report save directory.")
sub_parser_run.add_argument(
'--report-file',
help="Specify report file path, this has higher priority than specifying report dir.")
sub_parser_run.add_argument(
'--save-tests', action='store_true', default=False,
help="Save loaded/parsed/vars_out/summary json data to JSON files.")
"--save-tests",
action="store_true",
default=False,
help="Save loaded/parsed/vars_out/summary json data to JSON files.",
)
return sub_parser_run
def main_run(args):
runner = HttpRunner(
save_tests=args.save_tests,
log_level=args.log_level,
log_file=args.log_file
save_tests=args.save_tests, log_level=args.log_level, log_file=args.log_file
)
err_code = 0
@@ -73,11 +74,13 @@ def main_run(args):
runner.gen_html_report(
report_template=args.report_template,
report_dir=report_dir,
report_file=args.report_file
report_file=args.report_file,
)
err_code |= (0 if testsuite_summary and testsuite_summary.success else 1)
err_code |= 0 if testsuite_summary and testsuite_summary.success else 1
except Exception as ex:
logger.error(f"!!!!!!!!!! exception stage: {runner.exception_stage} !!!!!!!!!!\n{str(ex)}")
logger.error(
f"!!!!!!!!!! exception stage: {runner.exception_stage} !!!!!!!!!!\n{str(ex)}"
)
err_code = 1
sys.exit(err_code)
@@ -88,10 +91,10 @@ def main():
"""
parser = argparse.ArgumentParser(description=__description__)
parser.add_argument(
'-V', '--version', dest='version', action='store_true',
help="show version")
"-V", "--version", dest="version", action="store_true", help="show version"
)
subparsers = parser.add_subparsers(help='sub-command help')
subparsers = parser.add_subparsers(help="sub-command help")
sub_parser_run = init_parser_run(subparsers)
sub_parser_scaffold = init_parser_scaffold(subparsers)
sub_parser_har2case = init_har2case_parser(subparsers)
@@ -153,5 +156,5 @@ def main_hrun_alias():
main()
if __name__ == '__main__':
if __name__ == "__main__":
main()

3
httprunner/cli_test.py
View File

@@ -6,7 +6,6 @@ from httprunner.cli import main
class TestCli(unittest.TestCase):
def setUp(self):
self.captured_output = io.StringIO()
sys.stdout = self.captured_output
@@ -23,6 +22,7 @@ class TestCli(unittest.TestCase):
self.assertEqual(cm.exception.code, 0)
from httprunner import __version__
self.assertIn(__version__, self.captured_output.getvalue().strip())
def test_show_help(self):
@@ -34,4 +34,5 @@ class TestCli(unittest.TestCase):
self.assertEqual(cm.exception.code, 0)
from httprunner import __description__
self.assertIn(__description__, self.captured_output.getvalue().strip())

29
httprunner/client.py
View File

@@ -4,8 +4,12 @@ import requests
import urllib3
from loguru import logger
from requests import Request, Response
from requests.exceptions import (InvalidSchema, InvalidURL, MissingSchema,
RequestException)
from requests.exceptions import (
InvalidSchema,
InvalidURL,
MissingSchema,
RequestException,
)
from httprunner.schema import RequestData, ResponseData
from httprunner.schema import SessionData, ReqRespData
@@ -15,9 +19,8 @@ urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
class ApiResponse(Response):
def raise_for_status(self):
if hasattr(self, 'error') and self.error:
if hasattr(self, "error") and self.error:
raise self.error
Response.raise_for_status(self)
@@ -25,6 +28,7 @@ class ApiResponse(Response):
def get_req_resp_record(resp_obj: Response) -> ReqRespData:
""" get request and response info from Response() object.
"""
def log_print(req_or_resp, r_type):
msg = f"\n================== {r_type} details ==================\n"
for key, value in req_or_resp.dict().items():
@@ -36,14 +40,12 @@ def get_req_resp_record(resp_obj: Response) -> ReqRespData:
method=resp_obj.request.method,
url=resp_obj.request.url,
headers=dict(resp_obj.request.headers),
body=resp_obj.request.body
body=resp_obj.request.body,
)
request_body = resp_obj.request.body
if request_body:
request_content_type = lower_dict_keys(
request_data.headers
).get("content-type")
request_content_type = lower_dict_keys(request_data.headers).get("content-type")
if request_content_type and "multipart/form-data" in request_content_type:
# upload file type
request_data.body = "upload file stream (OMITTED)"
@@ -76,16 +78,13 @@ def get_req_resp_record(resp_obj: Response) -> ReqRespData:
encoding=resp_obj.encoding,
headers=resp_headers,
content_type=content_type,
body=response_body
body=response_body,
)
# log response details in debug mode
log_print(response_data, "response")
req_resp_data = ReqRespData(
request=request_data,
response=response_data
)
req_resp_data = ReqRespData(request=request_data, response=response_data)
return req_resp_data
@@ -98,6 +97,7 @@ class HttpSession(requests.Session):
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()
@@ -173,8 +173,7 @@ class HttpSession(requests.Session):
# record request and response histories, include 30X redirection
response_list = response.history + [response]
self.data.req_resps = [
get_req_resp_record(resp_obj)
for resp_obj in response_list
get_req_resp_record(resp_obj) for resp_obj in response_list
]
try:

30
httprunner/ext/har2case/__init__.py
View File

@@ -20,19 +20,27 @@ def init_har2case_parser(subparsers):
""" HAR converter: parse command line options and run commands.
"""
parser = subparsers.add_parser(
"har2case", help="Convert HAR(HTTP Archive) to YAML/JSON testcases for HttpRunner.")
parser.add_argument('har_source_file', nargs='?',
help="Specify HAR source file")
"har2case",
help="Convert HAR(HTTP Archive) to YAML/JSON testcases for HttpRunner.",
)
parser.add_argument("har_source_file", nargs="?", help="Specify HAR source file")
parser.add_argument(
'-2y', '--to-yml', '--to-yaml',
dest='to_yaml', action='store_true',
help="Convert to YAML format, if not specified, convert to JSON format by default.")
"-2y",
"--to-yml",
"--to-yaml",
dest="to_yaml",
action="store_true",
help="Convert to YAML format, if not specified, convert to JSON format by default.",
)
parser.add_argument(
'--filter', help="Specify filter keyword, only url include filter string will be converted.")
"--filter",
help="Specify filter keyword, only url include filter string will be converted.",
)
parser.add_argument(
'--exclude',
"--exclude",
help="Specify exclude keyword, url that includes exclude string will be ignored, "
"multiple keywords can be joined with '|'")
"multiple keywords can be joined with '|'",
)
return parser
@@ -48,8 +56,6 @@ def main_har2case(args):
sys.exit(1)
output_file_type = "YML" if args.to_yaml else "JSON"
HarParser(
har_source_file, args.filter, args.exclude
).gen_testcase(output_file_type)
HarParser(har_source_file, args.filter, args.exclude).gen_testcase(output_file_type)
return 0

32
httprunner/ext/har2case/core.py
View File

@@ -28,12 +28,11 @@ IGNORE_REQUEST_HEADERS = [
":authority",
":method",
":scheme",
":path"
":path",
]
class HarParser(object):
def __init__(self, har_file_path, filter_str=None, exclude_str=None):
self.har_file_path = har_file_path
self.filter_str = filter_str
@@ -76,7 +75,7 @@ class HarParser(object):
parsed_object = urlparse.urlparse(url)
if request_params:
parsed_object = parsed_object._replace(query='')
parsed_object = parsed_object._replace(query="")
teststep_dict["request"]["url"] = parsed_object.geturl()
teststep_dict["request"]["params"] = request_params
else:
@@ -241,7 +240,7 @@ class HarParser(object):
encoding = resp_content_dict.get("encoding")
if encoding and encoding == "base64":
content = base64.b64decode(text).decode('utf-8')
content = base64.b64decode(text).decode("utf-8")
else:
content = text
@@ -249,7 +248,9 @@ class HarParser(object):
resp_content_json = json.loads(content)
except JSONDecodeError:
logger.warning(
"response content can not be loaded as json: {}".format(content.encode("utf-8"))
"response content can not be loaded as json: {}".format(
content.encode("utf-8")
)
)
return
@@ -285,11 +286,7 @@ class HarParser(object):
}
"""
teststep_dict = {
"name": "",
"request": {},
"validate": []
}
teststep_dict = {"name": "", "request": {}, "validate": []}
self.__make_request_url(teststep_dict, entry_json)
self.__make_request_method(teststep_dict, entry_json)
@@ -302,16 +299,14 @@ class HarParser(object):
def _prepare_config(self):
""" prepare config block.
"""
return {
"name": "testcase description",
"variables": {}
}
return {"name": "testcase description", "variables": {}}
def _prepare_teststeps(self):
""" make teststep list.
teststeps list are parsed from HAR log entries list.
"""
def is_exclude(url, exclude_str):
exclude_str_list = exclude_str.split("|")
for exclude_str in exclude_str_list:
@@ -330,9 +325,7 @@ class HarParser(object):
if is_exclude(url, self.exclude_str):
continue
teststeps.append(
self._prepare_teststep(entry_json)
)
teststeps.append(self._prepare_teststep(entry_json))
return teststeps
@@ -344,10 +337,7 @@ class HarParser(object):
config = self._prepare_config()
teststeps = self._prepare_teststeps()
testcase = {
"config": config,
"teststeps": teststeps
}
testcase = {"config": config, "teststeps": teststeps}
return testcase
def gen_testcase(self, file_type="JSON"):

91
httprunner/ext/har2case/core_test.py
View File

@@ -6,7 +6,6 @@ from httprunner.ext.har2case.utils_test import TestUtils
class TestHar(TestUtils):
def setUp(self):
self.har_path = os.path.join(os.path.dirname(__file__), "data", "demo.har")
self.har_parser = HarParser(self.har_path)
@@ -22,18 +21,10 @@ class TestHar(TestUtils):
validator["eq"][0]: validator["eq"][1]
for validator in teststep_dict["validate"]
}
self.assertEqual(
validators_mapping["status_code"], 200
)
self.assertEqual(
validators_mapping["content.IsSuccess"], True
)
self.assertEqual(
validators_mapping["content.Code"], 200
)
self.assertEqual(
validators_mapping["content.Message"], None
)
self.assertEqual(validators_mapping["status_code"], 200)
self.assertEqual(validators_mapping["content.IsSuccess"], True)
self.assertEqual(validators_mapping["content.Code"], 200)
self.assertEqual(validators_mapping["content.Message"], None)
def test_prepare_teststeps(self):
teststeps = self.har_parser._prepare_teststeps()
@@ -43,16 +34,14 @@ class TestHar(TestUtils):
self.assertIn("validate", teststeps[0])
def test_gen_testcase_yaml(self):
yaml_file = os.path.join(
os.path.dirname(__file__), "data", "demo.yaml")
yaml_file = os.path.join(os.path.dirname(__file__), "data", "demo.yaml")
self.har_parser.gen_testcase(file_type="YAML")
self.assertTrue(os.path.isfile(yaml_file))
os.remove(yaml_file)
def test_gen_testcase_json(self):
json_file = os.path.join(
os.path.dirname(__file__), "data", "demo.json")
json_file = os.path.join(os.path.dirname(__file__), "data", "demo.json")
self.har_parser.gen_testcase(file_type="JSON")
self.assertTrue(os.path.isfile(json_file))
@@ -64,7 +53,7 @@ class TestHar(TestUtils):
teststeps = har_parser._prepare_teststeps()
self.assertEqual(
teststeps[0]["request"]["url"],
"https://httprunner.top/api/v1/Account/Login"
"https://httprunner.top/api/v1/Account/Login",
)
filter_str = "debugtalk"
@@ -78,7 +67,7 @@ class TestHar(TestUtils):
teststeps = har_parser._prepare_teststeps()
self.assertEqual(
teststeps[0]["request"]["url"],
"https://httprunner.top/api/v1/Account/Login"
"https://httprunner.top/api/v1/Account/Login",
)
exclude_str = "httprunner"
@@ -98,20 +87,13 @@ class TestHar(TestUtils):
self.assertEqual(teststeps, [])
def test_make_request_data_params(self):
testcase_dict = {
"name": "",
"request": {},
"validate": []
}
testcase_dict = {"name": "", "request": {}, "validate": []}
entry_json = {
"request": {
"method": "POST",
"postData": {
"mimeType": "application/x-www-form-urlencoded; charset=utf-8",
"params": [
{"name": "a", "value": 1},
{"name": "b", "value": "2"}
]
"params": [{"name": "a", "value": 1}, {"name": "b", "value": "2"}],
},
}
}
@@ -120,53 +102,32 @@ class TestHar(TestUtils):
self.assertEqual(testcase_dict["request"]["data"]["b"], "2")
def test_make_request_data_json(self):
testcase_dict = {
"name": "",
"request": {},
"validate": []
}
testcase_dict = {"name": "", "request": {}, "validate": []}
entry_json = {
"request": {
"method": "POST",
"postData": {
"mimeType": "application/json; charset=utf-8",
"text": "{\"a\":\"1\",\"b\":\"2\"}"
"text": '{"a":"1","b":"2"}',
},
}
}
self.har_parser._make_request_data(testcase_dict, entry_json)
self.assertEqual(
testcase_dict["request"]["json"],
{'a': '1', 'b': '2'}
)
self.assertEqual(testcase_dict["request"]["json"], {"a": "1", "b": "2"})
def test_make_request_data_text_empty(self):
testcase_dict = {
"name": "",
"request": {},
"validate": []
}
testcase_dict = {"name": "", "request": {}, "validate": []}
entry_json = {
"request": {
"method": "POST",
"postData": {
"mimeType": "application/json; charset=utf-8",
"text": ""
},
"postData": {"mimeType": "application/json; charset=utf-8", "text": ""},
}
}
self.har_parser._make_request_data(testcase_dict, entry_json)
self.assertEqual(
testcase_dict["request"]["data"],
""
)
self.assertEqual(testcase_dict["request"]["data"], "")
def test_make_validate(self):
testcase_dict = {
"name": "",
"request": {},
"validate": []
}
testcase_dict = {"name": "", "request": {}, "validate": []}
entry_json = {
"request": {},
"response": {
@@ -174,7 +135,7 @@ class TestHar(TestUtils):
"headers": [
{
"name": "Content-Type",
"value": "application/json; charset=utf-8"
"value": "application/json; charset=utf-8",
},
],
"content": {
@@ -182,23 +143,21 @@ class TestHar(TestUtils):
"mimeType": "application/json; charset=utf-8",
# raw response content text is application/jose type
"text": "ZXlKaGJHY2lPaUpTVTBFeFh6VWlMQ0psYm1NaU9pSkJNVEk0UTBKRExV",
"encoding": "base64"
}
}
"encoding": "base64",
},
},
}
self.har_parser._make_validate(testcase_dict, entry_json)
self.assertEqual(
testcase_dict["validate"][0],
{"eq": ["status_code", 200]}
)
self.assertEqual(testcase_dict["validate"][0], {"eq": ["status_code", 200]})
self.assertEqual(
testcase_dict["validate"][1],
{"eq": ["headers.Content-Type", "application/json; charset=utf-8"]}
{"eq": ["headers.Content-Type", "application/json; charset=utf-8"]},
)
def test_make_testcase(self):
har_path = os.path.join(
os.path.dirname(__file__), "data", "demo-quickstart.har")
os.path.dirname(__file__), "data", "demo-quickstart.har"
)
har_parser = HarParser(har_path)
testcase = har_parser._make_testcase()
self.assertIsInstance(testcase, dict)

20
httprunner/ext/har2case/utils.py
View File

@@ -50,10 +50,9 @@ def x_www_form_urlencoded(post_data):
"""
if isinstance(post_data, dict):
return "&".join([
u"{}={}".format(key, value)
for key, value in post_data.items()
])
return "&".join(
[u"{}={}".format(key, value) for key, value in post_data.items()]
)
else:
return post_data
@@ -98,10 +97,7 @@ def convert_list_to_dict(origin_list):
{"v": "1", "w": "2"}
"""
return {
item["name"]: item.get("value")
for item in origin_list
}
return {item["name"]: item.get("value") for item in origin_list}
def dump_yaml(testcase, yaml_file):
@@ -109,8 +105,10 @@ def dump_yaml(testcase, yaml_file):
"""
logging.info("dump testcase to YAML format.")
with io.open(yaml_file, 'w', encoding="utf-8") as outfile:
yaml.dump(testcase, outfile, allow_unicode=True, default_flow_style=False, indent=4)
with io.open(yaml_file, "w", encoding="utf-8") as outfile:
yaml.dump(
testcase, outfile, allow_unicode=True, default_flow_style=False, indent=4
)
logging.info("Generate YAML testcase successfully: {}".format(yaml_file))
@@ -120,7 +118,7 @@ def dump_json(testcase, json_file):
"""
logging.info("dump testcase to JSON format.")
with io.open(json_file, 'w', encoding="utf-8") as outfile:
with io.open(json_file, "w", encoding="utf-8") as outfile:
my_json_str = json.dumps(testcase, ensure_ascii=False, indent=4)
if isinstance(my_json_str, bytes):
my_json_str = my_json_str.decode("utf-8")

19
httprunner/ext/har2case/utils_test.py
View File

@@ -6,11 +6,11 @@ from httprunner.ext.har2case import utils
class TestUtils(unittest.TestCase):
@staticmethod
def create_har_file(file_name, content):
file_path = os.path.join(
os.path.dirname(__file__), "data", "{}.har".format(file_name))
os.path.dirname(__file__), "data", "{}.har".format(file_name)
)
with open(file_path, "w") as f:
f.write(json.dumps(content))
@@ -24,7 +24,9 @@ class TestUtils(unittest.TestCase):
self.assertIn("response", log_entries[0])
def test_load_har_log_key_error(self):
empty_json_file_path = TestUtils.create_har_file(file_name="empty_json", content={})
empty_json_file_path = TestUtils.create_har_file(
file_name="empty_json", content={}
)
with self.assertRaises(SystemExit):
utils.load_har_log_entries(empty_json_file_path)
os.remove(empty_json_file_path)
@@ -35,21 +37,14 @@ class TestUtils(unittest.TestCase):
utils.load_har_log_entries(empty_file_path)
os.remove(empty_file_path)
# def test_x_www_form_urlencoded(self):
# origin_dict = {"a":1, "b": "2"}
# self.assertIn("a=1", utils.x_www_form_urlencoded(origin_dict))
# self.assertIn("b=2", utils.x_www_form_urlencoded(origin_dict))
def test_convert_list_to_dict(self):
origin_list = [
{"name": "v", "value": "1"},
{"name": "w", "value": "2"}
]
self.assertEqual(
utils.convert_list_to_dict(origin_list),
{"v": "1", "w": "2"}
)
origin_list = [{"name": "v", "value": "1"}, {"name": "w", "value": "2"}]
self.assertEqual(utils.convert_list_to_dict(origin_list), {"v": "1", "w": "2"})
def test_convert_x_www_form_urlencoded_to_dict(self):
origin_str = "a=1&b=2"

34
httprunner/ext/locusts/__init__.py
View File

@@ -4,27 +4,39 @@ import sys
from loguru import logger
from httprunner import __version__
from httprunner.ext.locusts.core import start_locust_main, parse_locustfile, quick_run_locusts, start_master, \
start_slaves
from httprunner.ext.locusts.core import (
start_locust_main,
parse_locustfile,
quick_run_locusts,
start_master,
start_slaves,
)
CPU_COUNT = multiprocessing.cpu_count()
def init_parser_locusts(subparsers):
sub_parser_locusts = subparsers.add_parser(
"locusts", help="Run load test with locust.")
"locusts", help="Run load test with locust."
)
sub_parser_locusts.add_argument(
'--locust-help', action='store_true', default=False,
help="Show locust help.")
sub_parser_locusts.add_argument('test_file', nargs='?',
help="Specify YAML/JSON testcase file.")
"--locust-help", action="store_true", default=False, help="Show locust help."
)
sub_parser_locusts.add_argument(
"--master", action='store_true', default=False, help="Start locust master.")
"test_file", nargs="?", help="Specify YAML/JSON testcase file."
)
sub_parser_locusts.add_argument(
"--slaves", type=int, help="Specify locust slave number.")
"--master", action="store_true", default=False, help="Start locust master."
)
sub_parser_locusts.add_argument(
"--quickstart", action='store_true', default=False,
help=f"Start locust master with {CPU_COUNT} slaves.")
"--slaves", type=int, help="Specify locust slave number."
)
sub_parser_locusts.add_argument(
"--quickstart",
action="store_true",
default=False,
help=f"Start locust master with {CPU_COUNT} slaves.",
)
return sub_parser_locusts

20
httprunner/ext/locusts/core.py
View File

@@ -18,7 +18,7 @@ def parse_locustfile(file_path):
file_suffix = os.path.splitext(file_path)[1]
if file_suffix == ".py":
locustfile_path = file_path
elif file_suffix in ['.yaml', '.yml', '.json']:
elif file_suffix in [".yaml", ".yml", ".json"]:
locustfile_path = gen_locustfile(file_path)
else:
# '' or other suffix
@@ -31,16 +31,17 @@ def parse_locustfile(file_path):
def gen_locustfile(testcase_file_path):
""" generate locustfile from template.
"""
locustfile_path = 'locustfile.py'
locustfile_path = "locustfile.py"
template_path = os.path.join(
os.path.dirname(os.path.realpath(__file__)),
"locustfile_template.py"
os.path.dirname(os.path.realpath(__file__)), "locustfile_template.py"
)
with io.open(template_path, encoding='utf-8') as template:
with io.open(locustfile_path, 'w', encoding='utf-8') as locustfile:
with io.open(template_path, encoding="utf-8") as template:
with io.open(locustfile_path, "w", encoding="utf-8") as locustfile:
template_content = template.read()
template_content = template_content.replace("$TESTCASE_FILE", testcase_file_path)
template_content = template_content.replace(
"$TESTCASE_FILE", testcase_file_path
)
locustfile.write(template_content)
return locustfile_path
@@ -49,6 +50,7 @@ def gen_locustfile(testcase_file_path):
def start_locust_main():
logger.info(f"run command: {sys.argv}")
from locust.main import main
main()
@@ -94,7 +96,5 @@ def quick_run_locusts(slave_num):
logger.info(f"Start locust master with {slave_num} slaves ...")
processes = init_slave_processes(slave_num)
processes.append(
multiprocessing.Process(target=start_master, args=(sys.argv,))
)
processes.append(multiprocessing.Process(target=start_master, args=(sys.argv,)))
[process.join() for process in processes]

6
httprunner/ext/locusts/locustfile_template.py
View File

@@ -9,8 +9,8 @@ from httprunner.ext.locusts.utils import prepare_locust_tests
from httprunner.runner import Runner
logging.getLogger().setLevel(logging.CRITICAL)
logging.getLogger('locust.main').setLevel(logging.INFO)
logging.getLogger('locust.runners').setLevel(logging.INFO)
logging.getLogger("locust.main").setLevel(logging.INFO)
logging.getLogger("locust.runners").setLevel(logging.INFO)
class WebPageTasks(TaskSet):
@@ -28,7 +28,7 @@ class WebPageTasks(TaskSet):
request_type=self.test_runner.exception_request_type,
name=self.test_runner.exception_name,
response_time=0,
exception=ex
exception=ex,
)

6
httprunner/ext/locusts/utils_test.py
View File

@@ -5,15 +5,13 @@ from httprunner.ext.locusts.utils import prepare_locust_tests
class TestLocust(unittest.TestCase):
def test_prepare_locust_tests(self):
path = os.path.join(
os.path.dirname(__file__), "data", "demo_locusts.yml")
path = os.path.join(os.path.dirname(__file__), "data", "demo_locusts.yml")
locust_tests = prepare_locust_tests(path)
self.assertEqual(len(locust_tests), 2 + 3)
name_list = [
"create user 1000 and check result.",
"create user 1001 and check result."
"create user 1001 and check result.",
]
self.assertIn(locust_tests[0]["config"]["name"], name_list)
self.assertIn(locust_tests[4]["config"]["name"], name_list)

39
httprunner/ext/scaffold/__init__.py
View File

@@ -6,8 +6,11 @@ from loguru import logger
def init_parser_scaffold(subparsers):
sub_parser_scaffold = subparsers.add_parser(
"startproject", help="Create a new project with template structure.")
sub_parser_scaffold.add_argument("project_name", type=str, nargs="?", help="Specify new project name.")
"startproject", help="Create a new project with template structure."
)
sub_parser_scaffold.add_argument(
"project_name", type=str, nargs="?", help="Specify new project name."
)
return sub_parser_scaffold
@@ -15,7 +18,9 @@ def create_scaffold(project_name):
""" create scaffold with specified project name.
"""
if os.path.isdir(project_name):
logger.warning(f"Folder {project_name} exists, please specify a new folder name.")
logger.warning(
f"Folder {project_name} exists, please specify a new folder name."
)
return
logger.info(f"Start to create new project: {project_name}")
@@ -27,7 +32,7 @@ def create_scaffold(project_name):
logger.info(msg)
def create_file(path, file_content=""):
with open(path, 'w') as f:
with open(path, "w") as f:
f.write(file_content)
msg = f"created file: {path}"
logger.info(msg)
@@ -92,24 +97,16 @@ testcases:
variables:
device_sn: $device_sn
"""
ignore_content = "\n".join([
".env",
"reports/*",
"__pycache__/*",
"*.pyc",
".python-version",
"logs/*"
])
ignore_content = "\n".join(
[".env", "reports/*", "__pycache__/*", "*.pyc", ".python-version", "logs/*"]
)
demo_debugtalk_content = """
import time
def sleep(n_secs):
time.sleep(n_secs)
"""
demo_env_content = "\n".join([
"USERNAME=leolee",
"PASSWORD=123456"
])
demo_env_content = "\n".join(["USERNAME=leolee", "PASSWORD=123456"])
create_folder(project_name)
create_folder(os.path.join(project_name, "api"))
@@ -117,8 +114,14 @@ def sleep(n_secs):
create_folder(os.path.join(project_name, "testsuites"))
create_folder(os.path.join(project_name, "reports"))
create_file(os.path.join(project_name, "api", "demo_api.yml"), demo_api_content)
create_file(os.path.join(project_name, "testcases", "demo_testcase.yml"), demo_testcase_content)
create_file(os.path.join(project_name, "testsuites", "demo_testsuite.yml"), demo_testsuite_content)
create_file(
os.path.join(project_name, "testcases", "demo_testcase.yml"),
demo_testcase_content,
)
create_file(
os.path.join(project_name, "testsuites", "demo_testsuite.yml"),
demo_testsuite_content,
)
create_file(os.path.join(project_name, "debugtalk.py"), demo_debugtalk_content)
create_file(os.path.join(project_name, ".env"), demo_env_content)
create_file(os.path.join(project_name, ".gitignore"), ignore_content)

1
httprunner/ext/scaffold/scaffold_test.py
View File

@@ -6,7 +6,6 @@ from httprunner.ext.scaffold import create_scaffold
class TestUtils(unittest.TestCase):
def test_create_scaffold(self):
project_name = "projectABC"
create_scaffold(project_name)

7
httprunner/ext/uploader/__init__.py
View File

@@ -96,7 +96,9 @@ def prepare_upload_test(test_dict):
test_dict["variables"]["m_encoder"] = "${multipart_encoder(" + params_str + ")}"
test_dict["request"].setdefault("headers", {})
test_dict["request"]["headers"]["Content-Type"] = "${multipart_content_type($m_encoder)}"
test_dict["request"]["headers"][
"Content-Type"
] = "${multipart_content_type($m_encoder)}"
test_dict["request"]["data"] = "$m_encoder"
@@ -122,6 +124,7 @@ def multipart_encoder(**kwargs):
else:
# value is not absolute file path, check if it is relative file path
from httprunner.loader import get_pwd
_file_path = os.path.join(get_pwd(), value)
is_exists_file = os.path.isfile(_file_path)
@@ -130,7 +133,7 @@ def multipart_encoder(**kwargs):
filename = os.path.basename(_file_path)
mime_type = get_filetype(_file_path)
# TODO: fix ResourceWarning for unclosed file
file_handler = open(_file_path, 'rb')
file_handler = open(_file_path, "rb")
fields_dict[key] = (filename, file_handler, mime_type)
else:
fields_dict[key] = value

8
httprunner/loader/__init__.py
View File

@@ -11,8 +11,10 @@ HttpRunner loader
from httprunner.loader.buildup import load_cases, load_project_data
from httprunner.loader.check import is_test_path
from httprunner.loader.load import load_csv_file, load_builtin_functions
from httprunner.loader.locate import get_project_working_directory as get_pwd, \
init_project_working_directory as init_pwd
from httprunner.loader.locate import (
get_project_working_directory as get_pwd,
init_project_working_directory as init_pwd,
)
__all__ = [
"is_test_path",
@@ -21,5 +23,5 @@ __all__ = [
"load_csv_file",
"load_builtin_functions",
"load_project_data",
"load_cases"
"load_cases",
]

43
httprunner/loader/buildup.py
View File

@@ -4,14 +4,18 @@ import os
from loguru import logger
from httprunner import exceptions, utils
from httprunner.loader.load import load_module_functions, load_file, load_dot_env_file, \
load_folder_files
from httprunner.loader.locate import init_project_working_directory, get_project_working_directory
from httprunner.loader.load import (
load_module_functions,
load_file,
load_dot_env_file,
load_folder_files,
)
from httprunner.loader.locate import (
init_project_working_directory,
get_project_working_directory,
)
tests_def_mapping = {
"api": {},
"testcases": {}
}
tests_def_mapping = {"api": {}, "testcases": {}}
def load_debugtalk_functions():
@@ -71,10 +75,7 @@ def __extend_with_testcase_ref(raw_testinfo):
if testcase_path not in tests_def_mapping["testcases"]:
# make compatible with Windows/Linux
pwd = get_project_working_directory()
testcase_path = os.path.join(
pwd,
*testcase_path.split("/")
)
testcase_path = os.path.join(pwd, *testcase_path.split("/"))
loaded_testcase = load_file(testcase_path)
# TODO: validate with pydantic
@@ -82,7 +83,8 @@ def __extend_with_testcase_ref(raw_testinfo):
testcase_dict = load_testcase(loaded_testcase)
else:
raise exceptions.FileFormatError(
f"Invalid format testcase: {testcase_path}")
f"Invalid format testcase: {testcase_path}"
)
tests_def_mapping["testcases"][testcase_path] = testcase_dict
else:
@@ -174,10 +176,7 @@ def load_testcase(raw_testcase):
"""
raw_teststeps = raw_testcase.pop("teststeps")
raw_testcase["teststeps"] = [
load_teststep(teststep)
for teststep in raw_teststeps
]
raw_testcase["teststeps"] = [load_teststep(teststep) for teststep in raw_teststeps]
return raw_testcase
@@ -305,7 +304,9 @@ def load_project_data(test_path, dot_env_path=None):
environments and debugtalk.py functions.
"""
debugtalk_path, project_working_directory = init_project_working_directory(test_path)
debugtalk_path, project_working_directory = init_project_working_directory(
test_path
)
project_meta = {}
@@ -325,7 +326,9 @@ def load_project_data(test_path, dot_env_path=None):
# locate PWD and load debugtalk.py functions
project_meta["PWD"] = project_working_directory
project_meta["functions"] = debugtalk_functions
project_meta["test_path"] = os.path.abspath(test_path)[len(project_working_directory) + 1:]
project_meta["test_path"] = os.path.abspath(test_path)[
len(project_working_directory) + 1 :
]
return project_meta
@@ -384,9 +387,7 @@ def load_cases(path: str, dot_env_path: str = None):
"""
tests_mapping = {
"project_meta": load_project_data(path, dot_env_path)
}
tests_mapping = {"project_meta": load_project_data(path, dot_env_path)}
def __load_file_content(path):
loaded_content = None

131
httprunner/loader/buildup_test.py
View File

@@ -1,4 +1,3 @@
import os
import unittest
@@ -7,14 +6,15 @@ from httprunner.loader import buildup
class TestModuleLoader(unittest.TestCase):
def test_filter_module_functions(self):
module_functions = buildup.load_module_functions(buildup)
self.assertIn("load_module_functions", module_functions)
self.assertNotIn("is_py3", module_functions)
def test_load_debugtalk_module(self):
project_meta = buildup.load_project_data(os.path.join(os.getcwd(), "httprunner"))
project_meta = buildup.load_project_data(
os.path.join(os.getcwd(), "httprunner")
)
self.assertNotIn("alter_response", project_meta["functions"])
project_meta = buildup.load_project_data(os.path.join(os.getcwd(), "tests"))
@@ -28,33 +28,23 @@ class TestModuleLoader(unittest.TestCase):
project_meta = buildup.load_project_data("tests/data/demo_testcase.yml")
project_working_directory = project_meta["PWD"]
debugtalk_functions = project_meta["functions"]
self.assertEqual(
project_working_directory,
os.path.join(os.getcwd(), "tests")
)
self.assertEqual(project_working_directory, os.path.join(os.getcwd(), "tests"))
self.assertIn("gen_md5", debugtalk_functions)
project_meta = buildup.load_project_data("tests/base.py")
project_working_directory = project_meta["PWD"]
debugtalk_functions = project_meta["functions"]
self.assertEqual(
project_working_directory,
os.path.join(os.getcwd(), "tests")
)
self.assertEqual(project_working_directory, os.path.join(os.getcwd(), "tests"))
self.assertIn("gen_md5", debugtalk_functions)
project_meta = buildup.load_project_data("httprunner/__init__.py")
project_working_directory = project_meta["PWD"]
debugtalk_functions = project_meta["functions"]
self.assertEqual(
project_working_directory,
os.getcwd()
)
self.assertEqual(project_working_directory, os.getcwd())
self.assertEqual(debugtalk_functions, {})
class TestSuiteLoader(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.project_meta = buildup.load_project_data(os.path.join(os.getcwd(), "tests"))
@@ -64,15 +54,10 @@ class TestSuiteLoader(unittest.TestCase):
raw_test = {
"name": "create user (override).",
"api": "api/create_user.yml",
"variables": [
{"uid": "999"}
]
"variables": [{"uid": "999"}],
}
teststep = buildup.load_teststep(raw_test)
self.assertEqual(
"create user (override).",
teststep["name"]
)
self.assertEqual("create user (override).", teststep["name"])
self.assertIn("api_def", teststep)
api_def = teststep["api_def"]
self.assertEqual(api_def["name"], "create user")
@@ -82,15 +67,10 @@ class TestSuiteLoader(unittest.TestCase):
raw_test = {
"name": "setup and reset all (override).",
"testcase": "testcases/setup.yml",
"variables": [
{"device_sn": "$device_sn"}
]
"variables": [{"device_sn": "$device_sn"}],
}
testcase = buildup.load_teststep(raw_test)
self.assertEqual(
"setup and reset all (override).",
testcase["name"]
)
self.assertEqual("setup and reset all (override).", testcase["name"])
tests = testcase["testcase_def"]["teststeps"]
self.assertEqual(len(tests), 2)
self.assertEqual(tests[0]["name"], "get token (setup)")
@@ -106,7 +86,7 @@ class TestSuiteLoader(unittest.TestCase):
def test_load_test_file_testcase(self):
for loaded_content in [
buildup.load_test_file("tests/testcases/setup.yml"),
buildup.load_test_file("tests/testcases/setup.json")
buildup.load_test_file("tests/testcases/setup.json"),
]:
self.assertEqual(loaded_content["type"], "testcase")
self.assertIn("path", loaded_content)
@@ -118,113 +98,100 @@ class TestSuiteLoader(unittest.TestCase):
def test_load_test_file_testsuite(self):
for loaded_content in [
buildup.load_test_file("tests/testsuites/create_users.yml"),
buildup.load_test_file("tests/testsuites/create_users.json")
buildup.load_test_file("tests/testsuites/create_users.json"),
]:
self.assertEqual(loaded_content["type"], "testsuite")
testcases = loaded_content["testcases"]
self.assertEqual(len(testcases), 2)
self.assertIn('create user 1000 and check result.', testcases)
self.assertIn('testcase_def', testcases["create user 1000 and check result."])
self.assertIn("create user 1000 and check result.", testcases)
self.assertIn(
"testcase_def", testcases["create user 1000 and check result."]
)
self.assertEqual(
testcases["create user 1000 and check result."]["testcase_def"]["config"]["name"],
"create user and check result."
testcases["create user 1000 and check result."]["testcase_def"][
"config"
]["name"],
"create user and check result.",
)
def test_load_tests_api_file(self):
path = os.path.join(
os.getcwd(), 'tests/api/create_user.yml')
path = os.path.join(os.getcwd(), "tests/api/create_user.yml")
tests_mapping = loader.load_cases(path)
api_list = tests_mapping["apis"]
self.assertEqual(len(api_list), 1)
self.assertEqual(api_list[0]["request"]["url"], "/api/users/$uid")
def test_load_tests_testcase_file_2(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testcase.yml')
testcase_file_path = os.path.join(os.getcwd(), "tests/data/demo_testcase.yml")
tests_mapping = loader.load_cases(testcase_file_path)
testcases = tests_mapping["testcases"]
self.assertIsInstance(testcases, list)
self.assertEqual(testcases[0]["config"]["name"], '123t$var_a')
self.assertIn(
"sum_two",
tests_mapping["project_meta"]["functions"]
self.assertEqual(testcases[0]["config"]["name"], "123t$var_a")
self.assertIn("sum_two", tests_mapping["project_meta"]["functions"])
self.assertEqual(
testcases[0]["config"]["variables"]["var_c"], "${sum_two($var_a, $var_b)}"
)
self.assertEqual(
testcases[0]["config"]["variables"]["var_c"],
"${sum_two($var_a, $var_b)}"
)
self.assertEqual(
testcases[0]["config"]["variables"]["PROJECT_KEY"],
"${ENV(PROJECT_KEY)}"
testcases[0]["config"]["variables"]["PROJECT_KEY"], "${ENV(PROJECT_KEY)}"
)
def test_load_tests_testcase_file_with_api_ref(self):
path = os.path.join(
os.getcwd(), 'tests/data/demo_testcase_layer.yml')
path = os.path.join(os.getcwd(), "tests/data/demo_testcase_layer.yml")
tests_mapping = loader.load_cases(path)
project_meta = tests_mapping["project_meta"]
testcases_list = tests_mapping["testcases"]
self.assertIn('device_sn', testcases_list[0]["config"]["variables"])
self.assertIn("device_sn", testcases_list[0]["config"]["variables"])
self.assertIn("gen_md5", project_meta["functions"])
self.assertIn("base_url", testcases_list[0]["config"])
test_dict0 = testcases_list[0]["teststeps"][0]
self.assertEqual(
"get token with $user_agent, $app_version",
test_dict0["name"]
)
self.assertEqual("get token with $user_agent, $app_version", test_dict0["name"])
self.assertIn("/api/get-token", test_dict0["api_def"]["request"]["url"])
self.assertIn(
{'eq': ['status_code', 200]},
test_dict0["validate"]
)
self.assertIn({"eq": ["status_code", 200]}, test_dict0["validate"])
def test_load_tests_testsuite_file_with_testcase_ref(self):
path = os.path.join(
os.getcwd(), 'tests/testsuites/create_users.yml')
path = os.path.join(os.getcwd(), "tests/testsuites/create_users.yml")
tests_mapping = loader.load_cases(path)
project_meta = tests_mapping["project_meta"]
testsuites_list = tests_mapping["testsuites"]
self.assertEqual("create users with uid", testsuites_list[0]["config"]["name"])
self.assertEqual(
"create users with uid",
testsuites_list[0]["config"]["name"]
)
self.assertEqual(
'${gen_random_string(15)}',
testsuites_list[0]["config"]["variables"]['device_sn']
"${gen_random_string(15)}",
testsuites_list[0]["config"]["variables"]["device_sn"],
)
self.assertIn(
"create user 1000 and check result.",
testsuites_list[0]["testcases"]
"create user 1000 and check result.", testsuites_list[0]["testcases"]
)
self.assertEqual(
testsuites_list[0]["testcases"]["create user 1000 and check result."]["testcase_def"]["config"]["name"],
"create user and check result."
testsuites_list[0]["testcases"]["create user 1000 and check result."][
"testcase_def"
]["config"]["name"],
"create user and check result.",
)
def test_load_tests_folder_path(self):
# absolute folder path
path = os.path.join(os.getcwd(), 'tests/data')
path = os.path.join(os.getcwd(), "tests/data")
tests_mapping = loader.load_cases(path)
testcase_list_1 = tests_mapping["testcases"]
self.assertGreater(len(testcase_list_1), 4)
# relative folder path
path = 'tests/data/'
path = "tests/data/"
tests_mapping = loader.load_cases(path)
testcase_list_2 = tests_mapping["testcases"]
self.assertEqual(len(testcase_list_1), len(testcase_list_2))
def test_load_tests_path_not_exist(self):
# absolute folder path
path = os.path.join(os.getcwd(), 'tests/data_not_exist')
path = os.path.join(os.getcwd(), "tests/data_not_exist")
with self.assertRaises(exceptions.FileNotFound):
loader.load_cases(path)
# relative folder path
path = 'tests/data_not_exist'
path = "tests/data_not_exist"
with self.assertRaises(exceptions.FileNotFound):
loader.load_cases(path)
@@ -232,11 +199,5 @@ class TestSuiteLoader(unittest.TestCase):
buildup.load_project_data(os.path.join(os.getcwd(), "tests"))
self.assertIn("gen_md5", self.project_meta["functions"])
self.assertEqual(self.project_meta["env"]["PROJECT_KEY"], "ABCDEFGH")
self.assertEqual(
os.path.basename(self.project_meta["PWD"]),
"tests"
)
self.assertEqual(
os.path.basename(self.project_meta["test_path"]),
"" # FIXME
)
self.assertEqual(os.path.basename(self.project_meta["PWD"]), "tests")
self.assertEqual(os.path.basename(self.project_meta["test_path"]), "") # FIXME

2
httprunner/loader/check.py
View File

@@ -30,7 +30,7 @@ def is_test_path(path):
if os.path.isfile(path):
# path is a file
file_suffix = os.path.splitext(path)[1].lower()
if file_suffix not in ['.json', '.yaml', '.yml']:
if file_suffix not in [".json", ".yaml", ".yml"]:
# path is not json/yaml file
return False
else:

17
httprunner/loader/load.py
View File

@@ -14,7 +14,7 @@ from httprunner.loader.locate import get_project_working_directory
try:
# PyYAML version >= 5.1
# ref: https://github.com/yaml/pyyaml/wiki/PyYAML-yaml.load(input)-Deprecation
yaml.warnings({'YAMLLoadWarning': False})
yaml.warnings({"YAMLLoadWarning": False})
except AttributeError:
pass
@@ -22,7 +22,7 @@ except AttributeError:
def _load_yaml_file(yaml_file):
""" load yaml file and check file content format
"""
with io.open(yaml_file, 'r', encoding='utf-8') as stream:
with io.open(yaml_file, "r", encoding="utf-8") as stream:
try:
yaml_content = yaml.load(stream)
except yaml.YAMLError as ex:
@@ -35,7 +35,7 @@ def _load_yaml_file(yaml_file):
def _load_json_file(json_file):
""" load json file and check file content format
"""
with io.open(json_file, encoding='utf-8') as data_file:
with io.open(json_file, encoding="utf-8") as data_file:
try:
json_content = json.load(data_file)
except json.JSONDecodeError:
@@ -81,7 +81,7 @@ def load_csv_file(csv_file):
csv_content_list = []
with io.open(csv_file, encoding='utf-8') as csvfile:
with io.open(csv_file, encoding="utf-8") as csvfile:
reader = csv.DictReader(csvfile)
for row in reader:
csv_content_list.append(row)
@@ -94,9 +94,9 @@ def load_file(file_path):
raise exceptions.FileNotFound(f"{file_path} does not exist.")
file_suffix = os.path.splitext(file_path)[1].lower()
if file_suffix == '.json':
if file_suffix == ".json":
return _load_json_file(file_path)
elif file_suffix in ['.yaml', '.yml']:
elif file_suffix in [".yaml", ".yml"]:
return _load_yaml_file(file_path)
elif file_suffix == ".csv":
return load_csv_file(file_path)
@@ -132,7 +132,7 @@ def load_folder_files(folder_path, recursive=True):
filenames_list = []
for filename in filenames:
if not filename.endswith(('.yml', '.yaml', '.json')):
if not filename.endswith((".yml", ".yaml", ".json")):
continue
filenames_list.append(filename)
@@ -172,7 +172,7 @@ def load_dot_env_file(dot_env_path):
logger.info(f"Loading environment variables from {dot_env_path}")
env_variables_mapping = {}
with io.open(dot_env_path, 'r', encoding='utf-8') as fp:
with io.open(dot_env_path, "r", encoding="utf-8") as fp:
for line in fp:
# maxsplit=1
if "=" in line:
@@ -216,4 +216,3 @@ def load_builtin_functions():
""" load builtin module functions
"""
return load_module_functions(builtin)

57
httprunner/loader/load_test.py
View File

@@ -7,11 +7,10 @@ from httprunner.loader.buildup import load_test_file
class TestFileLoader(unittest.TestCase):
def test_load_yaml_file_file_format_error(self):
yaml_tmp_file = "tests/data/tmp.yml"
# create empty yaml file
with open(yaml_tmp_file, 'w') as f:
with open(yaml_tmp_file, "w") as f:
f.write("")
with self.assertRaises(exceptions.FileFormatError):
@@ -20,7 +19,7 @@ class TestFileLoader(unittest.TestCase):
os.remove(yaml_tmp_file)
# create invalid format yaml file
with open(yaml_tmp_file, 'w') as f:
with open(yaml_tmp_file, "w") as f:
f.write("abc")
with self.assertRaises(exceptions.FileFormatError):
@@ -31,7 +30,7 @@ class TestFileLoader(unittest.TestCase):
def test_load_json_file_file_format_error(self):
json_tmp_file = "tests/data/tmp.json"
# create empty file
with open(json_tmp_file, 'w') as f:
with open(json_tmp_file, "w") as f:
f.write("")
with self.assertRaises(exceptions.FileFormatError):
@@ -40,7 +39,7 @@ class TestFileLoader(unittest.TestCase):
os.remove(json_tmp_file)
# create empty json file
with open(json_tmp_file, 'w') as f:
with open(json_tmp_file, "w") as f:
f.write("{}")
with self.assertRaises(exceptions.FileFormatError):
@@ -49,7 +48,7 @@ class TestFileLoader(unittest.TestCase):
os.remove(json_tmp_file)
# create invalid format json file
with open(json_tmp_file, 'w') as f:
with open(json_tmp_file, "w") as f:
f.write("abc")
with self.assertRaises(exceptions.FileFormatError):
@@ -58,40 +57,38 @@ class TestFileLoader(unittest.TestCase):
os.remove(json_tmp_file)
def test_load_testcases_bad_filepath(self):
testcase_file_path = os.path.join(os.getcwd(), 'tests/data/demo')
testcase_file_path = os.path.join(os.getcwd(), "tests/data/demo")
with self.assertRaises(exceptions.FileNotFound):
load.load_file(testcase_file_path)
def test_load_csv_file_one_parameter(self):
csv_file_path = os.path.join(
os.getcwd(), 'tests/data/user_agent.csv')
csv_file_path = os.path.join(os.getcwd(), "tests/data/user_agent.csv")
csv_content = load.load_file(csv_file_path)
self.assertEqual(
csv_content,
[
{'user_agent': 'iOS/10.1'},
{'user_agent': 'iOS/10.2'},
{'user_agent': 'iOS/10.3'}
]
{"user_agent": "iOS/10.1"},
{"user_agent": "iOS/10.2"},
{"user_agent": "iOS/10.3"},
],
)
def test_load_csv_file_multiple_parameters(self):
csv_file_path = os.path.join(
os.getcwd(), 'tests/data/account.csv')
csv_file_path = os.path.join(os.getcwd(), "tests/data/account.csv")
csv_content = load.load_file(csv_file_path)
self.assertEqual(
csv_content,
[
{'username': 'test1', 'password': '111111'},
{'username': 'test2', 'password': '222222'},
{'username': 'test3', 'password': '333333'}
]
{"username": "test1", "password": "111111"},
{"username": "test2", "password": "222222"},
{"username": "test3", "password": "333333"},
],
)
def test_load_folder_files(self):
folder = os.path.join(os.getcwd(), 'tests')
file1 = os.path.join(os.getcwd(), 'tests', 'test_utils.py')
file2 = os.path.join(os.getcwd(), 'tests', 'api', 'reset_all.yml')
folder = os.path.join(os.getcwd(), "tests")
file1 = os.path.join(os.getcwd(), "tests", "test_utils.py")
file2 = os.path.join(os.getcwd(), "tests", "api", "reset_all.yml")
files = load.load_folder_files(folder, recursive=False)
self.assertEqual(files, [])
@@ -107,25 +104,21 @@ class TestFileLoader(unittest.TestCase):
self.assertEqual([], files)
def test_load_dot_env_file(self):
dot_env_path = os.path.join(
os.getcwd(), "tests", ".env"
)
dot_env_path = os.path.join(os.getcwd(), "tests", ".env")
env_variables_mapping = load.load_dot_env_file(dot_env_path)
self.assertIn("PROJECT_KEY", env_variables_mapping)
self.assertEqual(env_variables_mapping["UserName"], "debugtalk")
def test_load_custom_dot_env_file(self):
dot_env_path = os.path.join(
os.getcwd(), "tests", "data", "test.env"
)
dot_env_path = os.path.join(os.getcwd(), "tests", "data", "test.env")
env_variables_mapping = load.load_dot_env_file(dot_env_path)
self.assertIn("PROJECT_KEY", env_variables_mapping)
self.assertEqual(env_variables_mapping["UserName"], "test")
self.assertEqual(env_variables_mapping["content_type"], "application/json; charset=UTF-8")
self.assertEqual(
env_variables_mapping["content_type"], "application/json; charset=UTF-8"
)
def test_load_env_path_not_exist(self):
dot_env_path = os.path.join(
os.getcwd(), "tests", "data",
)
dot_env_path = os.path.join(os.getcwd(), "tests", "data",)
env_variables_mapping = load.load_dot_env_file(dot_env_path)
self.assertEqual(env_variables_mapping, {})

14
httprunner/loader/locate_test.py
View File

@@ -1,4 +1,3 @@
import os
import unittest
@@ -7,7 +6,6 @@ from httprunner.loader import locate
class TestLoaderLocate(unittest.TestCase):
def test_locate_file(self):
with self.assertRaises(exceptions.FileNotFound):
locate.locate_file(os.getcwd(), "debugtalk.py")
@@ -18,23 +16,21 @@ class TestLoaderLocate(unittest.TestCase):
start_path = os.path.join(os.getcwd(), "tests")
self.assertEqual(
locate.locate_file(start_path, "debugtalk.py"),
os.path.join(
os.getcwd(), "tests/debugtalk.py"
)
os.path.join(os.getcwd(), "tests/debugtalk.py"),
)
self.assertEqual(
locate.locate_file("tests/", "debugtalk.py"),
os.path.join(os.getcwd(), "tests", "debugtalk.py")
os.path.join(os.getcwd(), "tests", "debugtalk.py"),
)
self.assertEqual(
locate.locate_file("tests", "debugtalk.py"),
os.path.join(os.getcwd(), "tests", "debugtalk.py")
os.path.join(os.getcwd(), "tests", "debugtalk.py"),
)
self.assertEqual(
locate.locate_file("tests/base.py", "debugtalk.py"),
os.path.join(os.getcwd(), "tests", "debugtalk.py")
os.path.join(os.getcwd(), "tests", "debugtalk.py"),
)
self.assertEqual(
locate.locate_file("tests/data/demo_testcase.yml", "debugtalk.py"),
os.path.join(os.getcwd(), "tests", "debugtalk.py")
os.path.join(os.getcwd(), "tests", "debugtalk.py"),
)

56
httprunner/parser.py
View File

@@ -68,9 +68,7 @@ def regex_findall_variables(content: Text) -> List[Text]:
try:
vars_list = []
for var_tuple in variable_regex_compile.findall(content):
vars_list.append(
var_tuple[0] or var_tuple[1]
)
vars_list.append(var_tuple[0] or var_tuple[1])
return vars_list
except TypeError:
return []
@@ -160,20 +158,17 @@ def parse_function_params(params: Text) -> Dict:
{'args': [1, 2], 'kwargs': {'a':3, 'b':4}}
"""
function_meta = {
"args": [],
"kwargs": {}
}
function_meta = {"args": [], "kwargs": {}}
params_str = params.strip()
if params_str == "":
return function_meta
args_list = params_str.split(',')
args_list = params_str.split(",")
for arg in args_list:
arg = arg.strip()
if '=' in arg:
key, value = arg.split('=')
if "=" in arg:
key, value = arg.split("=")
function_meta["kwargs"][key.strip()] = parse_string_value(value.strip())
else:
function_meta["args"].append(parse_string_value(arg))
@@ -181,7 +176,9 @@ def parse_function_params(params: Text) -> Dict:
return function_meta
def get_mapping_variable(variable_name: Text, variables_mapping: VariablesMapping) -> Any:
def get_mapping_variable(
variable_name: Text, variables_mapping: VariablesMapping
) -> Any:
""" get variable from variables_mapping.
Args:
@@ -199,10 +196,14 @@ def get_mapping_variable(variable_name: Text, variables_mapping: VariablesMappin
try:
return variables_mapping[variable_name]
except KeyError:
raise exceptions.VariableNotFound(f"{variable_name} not found in {variables_mapping}")
raise exceptions.VariableNotFound(
f"{variable_name} not found in {variables_mapping}"
)
def get_mapping_function(function_name: Text, functions_mapping: FunctionsMapping) -> Callable:
def get_mapping_function(
function_name: Text, functions_mapping: FunctionsMapping
) -> Callable:
""" get function from functions_mapping,
if not found, then try to check if builtin function.
@@ -229,6 +230,7 @@ def get_mapping_function(function_name: Text, functions_mapping: FunctionsMappin
elif function_name in ["multipart_encoder", "multipart_content_type"]:
# extension for upload test
from httprunner.ext import uploader
return getattr(uploader, function_name)
try:
@@ -248,9 +250,10 @@ def get_mapping_function(function_name: Text, functions_mapping: FunctionsMappin
def parse_string(
raw_string: Text,
variables_mapping: VariablesMapping,
functions_mapping: FunctionsMapping) -> Any:
raw_string: Text,
variables_mapping: VariablesMapping,
functions_mapping: FunctionsMapping,
) -> Any:
""" parse string content with variables and functions mapping.
Args:
@@ -344,9 +347,10 @@ def parse_string(
def parse_data(
raw_data: Any,
variables_mapping: VariablesMapping = None,
functions_mapping: FunctionsMapping = None) -> Any:
raw_data: Any,
variables_mapping: VariablesMapping = None,
functions_mapping: FunctionsMapping = None,
) -> Any:
""" parse raw data with evaluated variables mapping.
Notice: variables_mapping should not contain any variable or function.
"""
@@ -359,8 +363,7 @@ def parse_data(
elif isinstance(raw_data, (list, set, tuple)):
return [
parse_data(item, variables_mapping, functions_mapping)
for item in raw_data
parse_data(item, variables_mapping, functions_mapping) for item in raw_data
]
elif isinstance(raw_data, dict):
@@ -378,8 +381,8 @@ def parse_data(
def parse_variables_mapping(
variables_mapping: VariablesMapping,
functions_mapping: FunctionsMapping = None) -> VariablesMapping:
variables_mapping: VariablesMapping, functions_mapping: FunctionsMapping = None
) -> VariablesMapping:
parsed_variables: VariablesMapping = {}
@@ -401,9 +404,7 @@ def parse_variables_mapping(
# check if reference variable not in variables_mapping
not_defined_variables = [
v_name
for v_name in variables
if v_name not in variables_mapping
v_name for v_name in variables if v_name not in variables_mapping
]
if not_defined_variables:
# e.g. {"varA": "123$varB", "varB": "456$varC"}
@@ -412,7 +413,8 @@ def parse_variables_mapping(
try:
parsed_value = parse_data(
var_value, parsed_variables, functions_mapping)
var_value, parsed_variables, functions_mapping
)
except exceptions.VariableNotFound:
continue

345
httprunner/parser_test.py
View File

@@ -6,27 +6,15 @@ from httprunner.exceptions import VariableNotFound, FunctionNotFound
class TestParserBasic(unittest.TestCase):
def test_parse_variables_mapping(self):
variables = {
"varA": "$varB",
"varB": "$varC",
"varC": "123",
"a": 1,
"b": 2
}
variables = {"varA": "$varB", "varB": "$varC", "varC": "123", "a": 1, "b": 2}
parsed_variables = parser.parse_variables_mapping(variables)
print(parsed_variables)
self.assertEqual(parsed_variables["varA"], "123")
self.assertEqual(parsed_variables["varB"], "123")
def test_parse_variables_mapping_exception(self):
variables = {
"varA": "$varB",
"varB": "$varC",
"a": 1,
"b": 2
}
variables = {"varA": "$varB", "varB": "$varC", "a": 1, "b": 2}
with self.assertRaises(VariableNotFound):
parser.parse_variables_mapping(variables)
@@ -38,125 +26,77 @@ class TestParserBasic(unittest.TestCase):
self.assertEqual(parser.parse_string_value("${func}"), "${func}")
def test_extract_variables(self):
self.assertEqual(
parser.extract_variables("$var"),
{"var"}
)
self.assertEqual(
parser.extract_variables("$var123"),
{"var123"}
)
self.assertEqual(
parser.extract_variables("$var_name"),
{"var_name"}
)
self.assertEqual(
parser.extract_variables("var"),
set()
)
self.assertEqual(
parser.extract_variables("a$var"),
{"var"}
)
self.assertEqual(
parser.extract_variables("$v ar"),
{"v"}
)
self.assertEqual(
parser.extract_variables(" "),
set()
)
self.assertEqual(
parser.extract_variables("$abc*"),
{"abc"}
)
self.assertEqual(
parser.extract_variables("${func()}"),
set()
)
self.assertEqual(
parser.extract_variables("${func(1,2)}"),
set()
)
self.assertEqual(parser.extract_variables("$var"), {"var"})
self.assertEqual(parser.extract_variables("$var123"), {"var123"})
self.assertEqual(parser.extract_variables("$var_name"), {"var_name"})
self.assertEqual(parser.extract_variables("var"), set())
self.assertEqual(parser.extract_variables("a$var"), {"var"})
self.assertEqual(parser.extract_variables("$v ar"), {"v"})
self.assertEqual(parser.extract_variables(" "), set())
self.assertEqual(parser.extract_variables("$abc*"), {"abc"})
self.assertEqual(parser.extract_variables("${func()}"), set())
self.assertEqual(parser.extract_variables("${func(1,2)}"), set())
self.assertEqual(
parser.extract_variables("${gen_md5($TOKEN, $data, $random)}"),
{"TOKEN", "data", "random"}
{"TOKEN", "data", "random"},
)
def test_parse_function_params(self):
self.assertEqual(parser.parse_function_params(""), {"args": [], "kwargs": {}})
self.assertEqual(parser.parse_function_params("5"), {"args": [5], "kwargs": {}})
self.assertEqual(
parser.parse_function_params(""),
{'args': [], 'kwargs': {}}
)
self.assertEqual(
parser.parse_function_params("5"),
{'args': [5], 'kwargs': {}}
)
self.assertEqual(
parser.parse_function_params("1, 2"),
{'args': [1, 2], 'kwargs': {}}
parser.parse_function_params("1, 2"), {"args": [1, 2], "kwargs": {}}
)
self.assertEqual(
parser.parse_function_params("a=1, b=2"),
{'args': [], 'kwargs': {'a': 1, 'b': 2}}
{"args": [], "kwargs": {"a": 1, "b": 2}},
)
self.assertEqual(
parser.parse_function_params("a= 1, b =2"),
{'args': [], 'kwargs': {'a': 1, 'b': 2}}
{"args": [], "kwargs": {"a": 1, "b": 2}},
)
self.assertEqual(
parser.parse_function_params("1, 2, a=3, b=4"),
{'args': [1, 2], 'kwargs': {'a': 3, 'b': 4}}
{"args": [1, 2], "kwargs": {"a": 3, "b": 4}},
)
self.assertEqual(
parser.parse_function_params("$request, 123"),
{'args': ["$request", 123], 'kwargs': {}}
)
self.assertEqual(
parser.parse_function_params(" "),
{'args': [], 'kwargs': {}}
{"args": ["$request", 123], "kwargs": {}},
)
self.assertEqual(parser.parse_function_params(" "), {"args": [], "kwargs": {}})
self.assertEqual(
parser.parse_function_params("hello world, a=3, b=4"),
{'args': ["hello world"], 'kwargs': {'a': 3, 'b': 4}}
{"args": ["hello world"], "kwargs": {"a": 3, "b": 4}},
)
self.assertEqual(
parser.parse_function_params("$request, 12 3"),
{'args': ["$request", '12 3'], 'kwargs': {}}
{"args": ["$request", "12 3"], "kwargs": {}},
)
def test_extract_functions(self):
self.assertEqual(parser.regex_findall_functions("${func()}"), [("func", "")])
self.assertEqual(parser.regex_findall_functions("${func(5)}"), [("func", "5")])
self.assertEqual(
parser.regex_findall_functions("${func()}"),
[("func", "")]
)
self.assertEqual(
parser.regex_findall_functions("${func(5)}"),
[("func", "5")]
)
self.assertEqual(
parser.regex_findall_functions("${func(a=1, b=2)}"),
[("func", "a=1, b=2")]
parser.regex_findall_functions("${func(a=1, b=2)}"), [("func", "a=1, b=2")]
)
self.assertEqual(
parser.regex_findall_functions("${func(1, $b, c=$x, d=4)}"),
[("func", "1, $b, c=$x, d=4")]
[("func", "1, $b, c=$x, d=4")],
)
self.assertEqual(
parser.regex_findall_functions("/api/1000?_t=${get_timestamp()}"),
[("get_timestamp", "")]
[("get_timestamp", "")],
)
self.assertEqual(
parser.regex_findall_functions("/api/${add(1, 2)}"),
[("add", "1, 2")]
parser.regex_findall_functions("/api/${add(1, 2)}"), [("add", "1, 2")]
)
self.assertEqual(
parser.regex_findall_functions("/api/${add(1, 2)}?_t=${get_timestamp()}"),
[('add', '1, 2'), ('get_timestamp', '')]
[("add", "1, 2"), ("get_timestamp", "")],
)
self.assertEqual(
parser.regex_findall_functions("abc${func(1, 2, a=3, b=4)}def"),
[('func', '1, 2, a=3, b=4')]
[("func", "1, 2, a=3, b=4")],
)
def test_parse_data_string_with_variables(self):
@@ -166,67 +106,35 @@ class TestParserBasic(unittest.TestCase):
"var_3": 123,
"var_4": {"a": 1},
"var_5": True,
"var_6": None
"var_6": None,
}
self.assertEqual(parser.parse_data("$var_1", variables_mapping), "abc")
self.assertEqual(parser.parse_data("${var_1}", variables_mapping), "abc")
self.assertEqual(parser.parse_data("var_1", variables_mapping), "var_1")
self.assertEqual(parser.parse_data("$var_1#XYZ", variables_mapping), "abc#XYZ")
self.assertEqual(
parser.parse_data("$var_1", variables_mapping),
"abc"
parser.parse_data("${var_1}#XYZ", variables_mapping), "abc#XYZ"
)
self.assertEqual(
parser.parse_data("${var_1}", variables_mapping),
"abc"
parser.parse_data("/$var_1/$var_2/var3", variables_mapping), "/abc/def/var3"
)
self.assertEqual(parser.parse_data("$var_3", variables_mapping), 123)
self.assertEqual(parser.parse_data("$var_4", variables_mapping), {"a": 1})
self.assertEqual(parser.parse_data("$var_5", variables_mapping), True)
self.assertEqual(parser.parse_data("abc$var_5", variables_mapping), "abcTrue")
self.assertEqual(
parser.parse_data("var_1", variables_mapping),
"var_1"
)
self.assertEqual(
parser.parse_data("$var_1#XYZ", variables_mapping),
"abc#XYZ"
)
self.assertEqual(
parser.parse_data("${var_1}#XYZ", variables_mapping),
"abc#XYZ"
)
self.assertEqual(
parser.parse_data("/$var_1/$var_2/var3", variables_mapping),
"/abc/def/var3"
)
self.assertEqual(
parser.parse_data("$var_3", variables_mapping),
123
)
self.assertEqual(
parser.parse_data("$var_4", variables_mapping),
{"a": 1}
)
self.assertEqual(
parser.parse_data("$var_5", variables_mapping),
True
)
self.assertEqual(
parser.parse_data("abc$var_5", variables_mapping),
"abcTrue"
)
self.assertEqual(
parser.parse_data("abc$var_4", variables_mapping),
"abc{'a': 1}"
)
self.assertEqual(
parser.parse_data("$var_6", variables_mapping),
None
parser.parse_data("abc$var_4", variables_mapping), "abc{'a': 1}"
)
self.assertEqual(parser.parse_data("$var_6", variables_mapping), None)
with self.assertRaises(VariableNotFound):
parser.parse_data("/api/$SECRET_KEY", variables_mapping)
self.assertEqual(
parser.parse_data(["$var_1", "$var_2"], variables_mapping),
["abc", "def"]
parser.parse_data(["$var_1", "$var_2"], variables_mapping), ["abc", "def"]
)
self.assertEqual(
parser.parse_data({"$var_1": "$var_2"}, variables_mapping),
{"abc": "def"}
parser.parse_data({"$var_1": "$var_2"}, variables_mapping), {"abc": "def"}
)
# format: $var
@@ -286,52 +194,56 @@ class TestParserBasic(unittest.TestCase):
}
self.assertEqual(
parser.parse_data("/$var_1/$var_2/$var_1", variables_mapping),
"/abc/def/abc"
"/abc/def/abc",
)
variables_mapping = {
"userid": 100,
"data": 1498
}
variables_mapping = {"userid": 100, "data": 1498}
content = "/users/$userid/training/$data?userId=$userid&data=$data"
self.assertEqual(
parser.parse_data(content, variables_mapping),
"/users/100/training/1498?userId=100&data=1498"
"/users/100/training/1498?userId=100&data=1498",
)
variables_mapping = {
"user": 100,
"userid": 1000,
"data": 1498
}
variables_mapping = {"user": 100, "userid": 1000, "data": 1498}
content = "/users/$user/$userid/$data?userId=$userid&data=$data"
self.assertEqual(
parser.parse_data(content, variables_mapping),
"/users/100/1000/1498?userId=1000&data=1498"
"/users/100/1000/1498?userId=1000&data=1498",
)
def test_parse_data_string_with_functions(self):
import random, string
functions_mapping = {
"gen_random_string": lambda str_len: ''.join(random.choice(string.ascii_letters + string.digits) \
for _ in range(str_len))
"gen_random_string": lambda str_len: "".join(
random.choice(string.ascii_letters + string.digits)
for _ in range(str_len)
)
}
result = parser.parse_data("${gen_random_string(5)}", functions_mapping=functions_mapping)
result = parser.parse_data(
"${gen_random_string(5)}", functions_mapping=functions_mapping
)
self.assertEqual(len(result), 5)
add_two_nums = lambda a, b=1: a + b
functions_mapping["add_two_nums"] = add_two_nums
self.assertEqual(
parser.parse_data("${add_two_nums(1)}", functions_mapping=functions_mapping),
2
parser.parse_data(
"${add_two_nums(1)}", functions_mapping=functions_mapping
),
2,
)
self.assertEqual(
parser.parse_data("${add_two_nums(1, 2)}", functions_mapping=functions_mapping),
3
parser.parse_data(
"${add_two_nums(1, 2)}", functions_mapping=functions_mapping
),
3,
)
self.assertEqual(
parser.parse_data("/api/${add_two_nums(1, 2)}", functions_mapping=functions_mapping),
"/api/3"
parser.parse_data(
"/api/${add_two_nums(1, 2)}", functions_mapping=functions_mapping
),
"/api/3",
)
with self.assertRaises(FunctionNotFound):
@@ -343,28 +255,38 @@ class TestParserBasic(unittest.TestCase):
"var_3": 123,
"var_4": {"a": 1},
"var_5": True,
"var_6": None
}
functions_mapping = {
"func1": lambda x, y: str(x) + str(y)
"var_6": None,
}
functions_mapping = {"func1": lambda x, y: str(x) + str(y)}
value = parser.parse_data("${func1($var_1, $var_3)}", variables_mapping, functions_mapping)
value = parser.parse_data(
"${func1($var_1, $var_3)}", variables_mapping, functions_mapping
)
self.assertEqual(value, "abc123")
value = parser.parse_data("ABC${func1($var_1, $var_3)}DE", variables_mapping, functions_mapping)
value = parser.parse_data(
"ABC${func1($var_1, $var_3)}DE", variables_mapping, functions_mapping
)
self.assertEqual(value, "ABCabc123DE")
value = parser.parse_data("ABC${func1($var_1, $var_3)}$var_5", variables_mapping, functions_mapping)
value = parser.parse_data(
"ABC${func1($var_1, $var_3)}$var_5", variables_mapping, functions_mapping
)
self.assertEqual(value, "ABCabc123True")
value = parser.parse_data("ABC${func1($var_1, $var_3)}DE$var_4", variables_mapping, functions_mapping)
value = parser.parse_data(
"ABC${func1($var_1, $var_3)}DE$var_4", variables_mapping, functions_mapping
)
self.assertEqual(value, "ABCabc123DE{'a': 1}")
value = parser.parse_data("ABC$var_5${func1($var_1, $var_3)}", variables_mapping, functions_mapping)
value = parser.parse_data(
"ABC$var_5${func1($var_1, $var_3)}", variables_mapping, functions_mapping
)
self.assertEqual(value, "ABCTrueabc123")
value = parser.parse_data("ABC${ord(a)}DEF${len(abcd)}", variables_mapping, functions_mapping)
value = parser.parse_data(
"ABC${ord(a)}DEF${len(abcd)}", variables_mapping, functions_mapping
)
self.assertEqual(value, "ABC97DEF4")
def test_parse_data_func_var_duplicate(self):
@@ -374,22 +296,26 @@ class TestParserBasic(unittest.TestCase):
"var_3": 123,
"var_4": {"a": 1},
"var_5": True,
"var_6": None
}
functions_mapping = {
"func1": lambda x, y: str(x) + str(y)
"var_6": None,
}
functions_mapping = {"func1": lambda x, y: str(x) + str(y)}
value = parser.parse_data(
"ABC${func1($var_1, $var_3)}--${func1($var_1, $var_3)}",
variables_mapping, functions_mapping)
variables_mapping,
functions_mapping,
)
self.assertEqual(value, "ABCabc123--abc123")
value = parser.parse_data("ABC${func1($var_1, $var_3)}$var_1", variables_mapping, functions_mapping)
value = parser.parse_data(
"ABC${func1($var_1, $var_3)}$var_1", variables_mapping, functions_mapping
)
self.assertEqual(value, "ABCabc123abc")
value = parser.parse_data(
"ABC${func1($var_1, $var_3)}$var_1--${func1($var_1, $var_3)}$var_1",
variables_mapping, functions_mapping)
variables_mapping,
functions_mapping,
)
self.assertEqual(value, "ABCabc123abc--abc123abc")
def test_parse_data_func_abnormal(self):
@@ -399,20 +325,22 @@ class TestParserBasic(unittest.TestCase):
"var_3": 123,
"var_4": {"a": 1},
"var_5": True,
"var_6": None
}
functions_mapping = {
"func1": lambda x, y: str(x) + str(y)
"var_6": None,
}
functions_mapping = {"func1": lambda x, y: str(x) + str(y)}
# {
value = parser.parse_data("ABC$var_1{", variables_mapping, functions_mapping)
self.assertEqual(value, "ABCabc{")
value = parser.parse_data("{ABC$var_1{}a}", variables_mapping, functions_mapping)
value = parser.parse_data(
"{ABC$var_1{}a}", variables_mapping, functions_mapping
)
self.assertEqual(value, "{ABCabc{}a}")
value = parser.parse_data("AB{C$var_1{}a}", variables_mapping, functions_mapping)
value = parser.parse_data(
"AB{C$var_1{}a}", variables_mapping, functions_mapping
)
self.assertEqual(value, "AB{Cabc{}a}")
# }
@@ -452,27 +380,21 @@ class TestParserBasic(unittest.TestCase):
def test_parse_data_request(self):
content = {
'request': {
'url': '/api/users/$uid',
'method': "$method",
'headers': {'token': '$token'},
'data': {
"request": {
"url": "/api/users/$uid",
"method": "$method",
"headers": {"token": "$token"},
"data": {
"null": None,
"true": True,
"false": False,
"empty_str": "",
"value": "abc${add_one(3)}def"
}
"value": "abc${add_one(3)}def",
},
}
}
variables_mapping = {
"uid": 1000,
"method": "POST",
"token": "abc123"
}
functions_mapping = {
"add_one": lambda x: x + 1
}
variables_mapping = {"uid": 1000, "method": "POST", "token": "abc123"}
functions_mapping = {"add_one": lambda x: x + 1}
result = parser.parse_data(content, variables_mapping, functions_mapping)
self.assertEqual("/api/users/1000", result["request"]["url"])
self.assertEqual("abc123", result["request"]["headers"]["token"])
@@ -488,11 +410,11 @@ class TestParserBasic(unittest.TestCase):
"uid": "1000",
"random": "A2dEx",
"authorization": "a83de0ff8d2e896dbd8efb81ba14e17d",
"data": {"name": "user", "password": "123456"}
"data": {"name": "user", "password": "123456"},
}
functions = {
"add_two_nums": lambda a, b=1: a + b,
"get_timestamp": lambda: int(time.time() * 1000)
"get_timestamp": lambda: int(time.time() * 1000),
}
testcase_template = {
"url": "http://127.0.0.1:5000/api/users/$uid/${add_two_nums(1,2)}",
@@ -501,28 +423,17 @@ class TestParserBasic(unittest.TestCase):
"Content-Type": "application/json",
"authorization": "$authorization",
"random": "$random",
"sum": "${add_two_nums(1, 2)}"
"sum": "${add_two_nums(1, 2)}",
},
"body": "$data"
"body": "$data",
}
parsed_testcase = parser.parse_data(testcase_template, variables, functions)
self.assertEqual(
parsed_testcase["url"],
"http://127.0.0.1:5000/api/users/1000/3"
parsed_testcase["url"], "http://127.0.0.1:5000/api/users/1000/3"
)
self.assertEqual(
parsed_testcase["headers"]["authorization"],
variables["authorization"]
)
self.assertEqual(
parsed_testcase["headers"]["random"],
variables["random"]
)
self.assertEqual(
parsed_testcase["body"],
variables["data"]
)
self.assertEqual(
parsed_testcase["headers"]["sum"],
3
parsed_testcase["headers"]["authorization"], variables["authorization"]
)
self.assertEqual(parsed_testcase["headers"]["random"], variables["random"])
self.assertEqual(parsed_testcase["body"], variables["data"])
self.assertEqual(parsed_testcase["headers"]["sum"], 3)

2
httprunner/report/__init__.py
View File

@@ -16,5 +16,5 @@ __all__ = [
"get_summary",
"stringify_summary",
"HtmlTestResult",
"gen_html_report"
"gen_html_report",
]

5
httprunner/report/html/__init__.py
View File

@@ -9,7 +9,4 @@ HttpRunner html report
from httprunner.report.html.result import HtmlTestResult
from httprunner.report.html.gen_report import gen_html_report
__all__ = [
"HtmlTestResult",
"gen_html_report"
]
__all__ = ["HtmlTestResult", "gen_html_report"]

22
httprunner/report/html/gen_report.py
View File

@@ -8,7 +8,12 @@ from httprunner.exceptions import SummaryEmpty
from httprunner.schema import TestSuiteSummary
def gen_html_report(testsuite_summary: TestSuiteSummary, report_template=None, report_dir=None, report_file=None):
def gen_html_report(
testsuite_summary: TestSuiteSummary,
report_template=None,
report_dir=None,
report_file=None,
):
""" render html report with specified report name and template
Args:
@@ -24,8 +29,7 @@ def gen_html_report(testsuite_summary: TestSuiteSummary, report_template=None, r
if not report_template:
report_template = os.path.join(
os.path.abspath(os.path.dirname(__file__)),
"template.html"
os.path.abspath(os.path.dirname(__file__)), "template.html"
)
logger.debug("No html report template specified, use default.")
else:
@@ -39,22 +43,22 @@ def gen_html_report(testsuite_summary: TestSuiteSummary, report_template=None, r
else:
report_dir = report_dir or os.path.join(os.getcwd(), "reports")
# fix #826: Windows does not support file name include ":"
report_file_name = "{}.html".format(testsuite_summary.time.start_at_iso_format.replace(":", "").replace("-", ""))
report_file_name = "{}.html".format(
testsuite_summary.time.start_at_iso_format.replace(":", "").replace("-", "")
)
if not os.path.isdir(report_dir):
os.makedirs(report_dir)
report_path = os.path.join(report_dir, report_file_name)
with io.open(report_template, "r", encoding='utf-8') as fp_r:
with io.open(report_template, "r", encoding="utf-8") as fp_r:
template_content = fp_r.read()
with io.open(report_path, 'w', encoding='utf-8') as fp_w:
with io.open(report_path, "w", encoding="utf-8") as fp_w:
rendered_content = Template(
template_content,
extensions=["jinja2.ext.loopcontrols"]
template_content, extensions=["jinja2.ext.loopcontrols"]
).render(testsuite_summary.dict())
fp_w.write(rendered_content)
logger.info(f"Generated Html report: {report_path}")
return report_path

15
httprunner/report/html/result.py
View File

@@ -8,6 +8,7 @@ class HtmlTestResult(unittest.TextTestResult):
""" A html result class that can generate formatted html results, used by TextTestRunner.
Each testcase is corresponding to one HtmlTestResult instance
"""
def __init__(self, stream, descriptions, verbosity):
super(HtmlTestResult, self).__init__(stream, descriptions, verbosity)
self.name = ""
@@ -15,7 +16,7 @@ class HtmlTestResult(unittest.TextTestResult):
self.attachment = ""
self.step_datas = None
def _record_test(self, test, status, attachment=''):
def _record_test(self, test, status, attachment=""):
self.name = test.shortDescription()
self.status = status
self.attachment = attachment
@@ -31,32 +32,32 @@ class HtmlTestResult(unittest.TextTestResult):
def addSuccess(self, test):
super(HtmlTestResult, self).addSuccess(test)
self._record_test(test, 'success')
self._record_test(test, "success")
print("")
def addError(self, test, err):
super(HtmlTestResult, self).addError(test, err)
self._record_test(test, 'error', self._exc_info_to_string(err, test))
self._record_test(test, "error", self._exc_info_to_string(err, test))
print("")
def addFailure(self, test, err):
super(HtmlTestResult, self).addFailure(test, err)
self._record_test(test, 'failure', self._exc_info_to_string(err, test))
self._record_test(test, "failure", self._exc_info_to_string(err, test))
print("")
def addSkip(self, test, reason):
super(HtmlTestResult, self).addSkip(test, reason)
self._record_test(test, 'skipped', reason)
self._record_test(test, "skipped", reason)
print("")
def addExpectedFailure(self, test, err):
super(HtmlTestResult, self).addExpectedFailure(test, err)
self._record_test(test, 'ExpectedFailure', self._exc_info_to_string(err, test))
self._record_test(test, "ExpectedFailure", self._exc_info_to_string(err, test))
print("")
def addUnexpectedSuccess(self, test):
super(HtmlTestResult, self).addUnexpectedSuccess(test)
self._record_test(test, 'UnexpectedSuccess')
self._record_test(test, "UnexpectedSuccess")
print("")
@property

2
httprunner/report/html/template.html
View File

@@ -210,7 +210,7 @@
<div class="content">
<h3>Name: {{ session_data.name }}</h3>
{% for req_resp in session_data.req_resp %}
{% for req_resp in session_data.req_resps %}
{% if loop.index > 1 %}
<div class="separator">==================================== redirect to ====================================</div>

22
httprunner/report/report.py
View File

@@ -12,40 +12,34 @@ def prepare_event_kwargs(event_name, params):
""" prepare report event kwargs"""
kwargs = {
"headers": {
'content-type': 'application/json'
},
"headers": {"content-type": "application/json"},
"json": {
"user": {
"user_unique_id": str(uuid.getnode())
},
"user": {"user_unique_id": str(uuid.getnode())},
"header": {
"app_id": 173519,
"os_name": platform.system(),
"os_version": platform.release(),
"app_version": __version__ # HttpRunner version
"app_version": __version__, # HttpRunner version
},
"events": [
{
"event": event_name,
"params": json.dumps(params),
"time": int(time.time())
"time": int(time.time()),
}
],
"verbose": 1
}
"verbose": 1,
},
}
return kwargs
def report_event(event_name, success=True):
params = {
"success": 1 if success else 0
}
params = {"success": 1 if success else 0}
kwargs = prepare_event_kwargs(event_name, params)
resp = requests.post("http://mcs.snssdk.com/v1/json", **kwargs)
print("resp---", resp.json())
if __name__ == '__main__':
if __name__ == "__main__":
report_event("loader")

27
httprunner/report/stringify.py
View File

@@ -1,11 +1,12 @@
import json
from base64 import b64encode
from collections import Iterable
from typing import List
from jinja2 import escape
from requests.cookies import RequestsCookieJar
from httprunner.schema import TestSuiteSummary
from httprunner.schema import TestSuiteSummary, SessionData
def dumps_json(value):
@@ -125,8 +126,7 @@ def __stringify_response(response_data):
if key == "body" and "image" in response_data["content_type"]:
# display image
value = "data:{};base64,{}".format(
response_data["content_type"],
b64encode(value).decode(encoding)
response_data["content_type"], b64encode(value).decode(encoding)
)
else:
value = escape(value.decode(encoding))
@@ -152,8 +152,19 @@ def stringify_summary(testsuite_summary: TestSuiteSummary):
testcase_summary.name = f"testcase {index}"
step_datas = testcase_summary.step_datas
for session_data in step_datas:
req_resp_list = session_data.req_resp
for req_resp in req_resp_list:
__stringify_request(req_resp["request"])
__stringify_response(req_resp["response"])
for step_data in step_datas:
if isinstance(step_data.data, list):
# List[SessionData]
session_data_list: List[SessionData] = step_data.data
for session_data in session_data_list:
req_resp_list = session_data.req_resps
for req_resp in req_resp_list:
__stringify_request(req_resp.request)
__stringify_response(req_resp.response)
else:
# SessionData
session_data: SessionData = step_data.data
req_resp_list = session_data.req_resps
for req_resp in req_resp_list:
__stringify_request(req_resp.request)
__stringify_response(req_resp.response)

15
httprunner/report/summarize.py
View File

@@ -10,10 +10,9 @@ def get_platform():
return {
"httprunner_version": __version__,
"python_version": "{} {}".format(
platform.python_implementation(),
platform.python_version()
platform.python_implementation(), platform.python_version()
),
"platform": platform.platform()
"platform": platform.platform(),
}
@@ -33,8 +32,10 @@ def aggregate_stat(origin_stat, new_stat):
origin_stat["start_at"] = min(origin_stat["start_at"], new_stat["start_at"])
elif key == "duration":
# duration = max_end_time - min_start_time
max_end_time = max(origin_stat["start_at"] + origin_stat["duration"],
new_stat["start_at"] + new_stat["duration"])
max_end_time = max(
origin_stat["start_at"] + origin_stat["duration"],
new_stat["start_at"] + new_stat["duration"],
)
min_start_time = min(origin_stat["start_at"], new_stat["start_at"])
origin_stat["duration"] = max_end_time - min_start_time
else:
@@ -65,11 +66,11 @@ def get_summary(result: HtmlTestResult) -> TestCaseSummary:
time=TestCaseTime(
start_at=result.start_at,
start_at_iso_format=start_at_iso_format,
duration=result.duration
duration=result.duration,
),
name=result.name,
status=result.status,
attachment=result.attachment,
in_out=TestCaseInOut(),
step_datas=result.step_datas
step_datas=result.step_datas,
)

56
httprunner/response.py
View File

@@ -28,15 +28,28 @@ def get_uniform_comparator(comparator: Text):
return "string_equals"
elif comparator in ["len_eq", "length_equals", "count_eq"]:
return "length_equals"
elif comparator in ["len_gt", "count_gt", "length_greater_than", "count_greater_than"]:
elif comparator in [
"len_gt",
"count_gt",
"length_greater_than",
"count_greater_than",
]:
return "length_greater_than"
elif comparator in ["len_ge", "count_ge", "length_greater_than_or_equals",
"count_greater_than_or_equals"]:
elif comparator in [
"len_ge",
"count_ge",
"length_greater_than_or_equals",
"count_greater_than_or_equals",
]:
return "length_greater_than_or_equals"
elif comparator in ["len_lt", "count_lt", "length_less_than", "count_less_than"]:
return "length_less_than"
elif comparator in ["len_le", "count_le", "length_less_than_or_equals",
"count_less_than_or_equals"]:
elif comparator in [
"len_le",
"count_le",
"length_less_than_or_equals",
"count_less_than_or_equals",
]:
return "length_less_than_or_equals"
else:
return comparator
@@ -90,15 +103,10 @@ def uniform_validator(validator):
# uniform comparator, e.g. lt => less_than, eq => equals
assert_method = get_uniform_comparator(comparator)
return {
"check": check_item,
"expect": expect_value,
"assert": assert_method
}
return {"check": check_item, "expect": expect_value, "assert": assert_method}
class ResponseObject(object):
def __init__(self, resp_obj: requests.Response):
""" initialize with a requests.Response object
@@ -110,7 +118,7 @@ class ResponseObject(object):
self.resp_obj_meta = {
"status_code": resp_obj.status_code,
"headers": resp_obj.headers,
"body": resp_obj.json()
"body": resp_obj.json(),
}
self.validation_results: Dict = {}
@@ -126,10 +134,12 @@ class ResponseObject(object):
logger.info(f"extract mapping: {extract_mapping}")
return extract_mapping
def validate(self,
validators: Validators,
variables_mapping: VariablesMapping = None,
functions_mapping: FunctionsMapping = None) -> NoReturn:
def validate(
self,
validators: Validators,
variables_mapping: VariablesMapping = None,
functions_mapping: FunctionsMapping = None,
) -> NoReturn:
self.validation_results = {}
if not validators:
@@ -166,7 +176,7 @@ class ResponseObject(object):
"check": check_item,
"check_value": check_value,
"expect": expect_item,
"expect_value": expect_value
"expect_value": expect_value,
}
try:
@@ -178,11 +188,13 @@ class ResponseObject(object):
validate_pass = False
validator_dict["check_result"] = "fail"
validate_msg += "\t==> fail"
validate_msg += f"\n" \
f"check_item: {check_item}\n" \
f"check_value: {check_value}({type(check_value).__name__})\n" \
f"assert_method: {assert_method}\n" \
f"expect_value: {expect_value}({type(expect_value).__name__})"
validate_msg += (
f"\n"
f"check_item: {check_item}\n"
f"check_value: {check_value}({type(check_value).__name__})\n"
f"assert_method: {assert_method}\n"
f"expect_value: {expect_value}({type(expect_value).__name__})"
)
logger.error(validate_msg)
failures.append(validate_msg)

37
httprunner/runner.py
View File

@@ -7,7 +7,13 @@ from httprunner.client import HttpSession
from httprunner.exceptions import ValidationFailure, ParamsError
from httprunner.parser import build_url, parse_data, parse_variables_mapping
from httprunner.response import ResponseObject
from httprunner.schema import TestsConfig, TestStep, VariablesMapping, TestCase, StepData
from httprunner.schema import (
TestsConfig,
TestStep,
VariablesMapping,
TestCase,
StepData,
)
class TestCaseRunner(object):
@@ -18,7 +24,7 @@ class TestCaseRunner(object):
step_datas: List[StepData] = []
validation_results: Dict = {}
session_variables: Dict = {}
success: bool = True # indicate testcase execution result
success: bool = True # indicate testcase execution result
def init(self, testcase: TestCase) -> "TestCaseRunner":
self.config = testcase.config
@@ -35,13 +41,13 @@ class TestCaseRunner(object):
def __run_step_request(self, step: TestStep):
"""run teststep: request"""
step_data = StepData(
name=step.name
)
step_data = StepData(name=step.name)
# parse
request_dict = step.request.dict()
parsed_request_dict = parse_data(request_dict, step.variables, self.config.functions)
parsed_request_dict = parse_data(
request_dict, step.variables, self.config.functions
)
# prepare arguments
method = parsed_request_dict.pop("method")
@@ -109,13 +115,11 @@ class TestCaseRunner(object):
def __run_step_testcase(self, step):
"""run teststep: referenced testcase"""
step_data = StepData(
name=step.name
)
step_data = StepData(name=step.name)
step_variables = step.variables
testcase: TestCaseRunner = step.testcase() # TODO: fix
testcase: TestCaseRunner = step.testcase() # TODO: fix
case_result = testcase.with_variables(**step_variables).run()
step_data.data = case_result.step_datas # list of step data
step_data.data = case_result.step_datas # list of step data
step_data.export = case_result.get_export_variables()
step_data.success = case_result.success
self.success &= case_result.success
@@ -131,7 +135,9 @@ class TestCaseRunner(object):
elif step.testcase:
step_data = self.__run_step_testcase(step)
else:
raise ParamsError(f"teststep is neither a request nor a referenced testcase: {step.dict()}")
raise ParamsError(
f"teststep is neither a request nor a referenced testcase: {step.dict()}"
)
self.step_datas.append(step_data)
return step_data.export
@@ -146,7 +152,9 @@ class TestCaseRunner(object):
# update with session variables extracted from former step
step.variables.update(self.session_variables)
# parse variables
step.variables = parse_variables_mapping(step.variables, self.config.functions)
step.variables = parse_variables_mapping(
step.variables, self.config.functions
)
# run step
extract_mapping = self.__run_step(step)
# save extracted variables to session variables
@@ -163,7 +171,8 @@ class TestCaseRunner(object):
for var_name in self.config.export:
if var_name not in self.session_variables:
raise ParamsError(
f"failed to export variable {var_name} from session variables {self.session_variables}")
f"failed to export variable {var_name} from session variables {self.session_variables}"
)
export_vars_mapping[var_name] = self.session_variables[var_name]

9
httprunner/schema.py
View File

@@ -20,8 +20,8 @@ Env = Dict[Text, Any]
class MethodEnum(Text, Enum):
GET = 'GET'
POST = 'POST'
GET = "GET"
POST = "POST"
PUT = "PUT"
DELETE = "DELETE"
HEAD = "HEAD"
@@ -44,6 +44,7 @@ class TestsConfig(BaseModel):
class Request(BaseModel):
"""requests.Request model"""
method: MethodEnum = MethodEnum.GET
url: Url
params: Dict[Text, Text] = {}
@@ -125,6 +126,7 @@ class ReqRespData(BaseModel):
class SessionData(BaseModel):
"""request session data, including request, response, validators and stat data"""
success: bool = False
# in most cases, req_resps only contains one request & response
# while when 30X redirect occurs, req_resps will contain multiple request & response
@@ -135,8 +137,9 @@ class SessionData(BaseModel):
class StepData(BaseModel):
"""teststep data, each step maybe corresponding to one request or one testcase"""
success: bool = False
name: Text = "" # teststep name
name: Text = "" # teststep name
data: Union[SessionData, List[SessionData]] = None
export: Dict = {}

16
httprunner/utils.py
View File

@@ -80,10 +80,7 @@ def lower_dict_keys(origin_dict):
if not origin_dict or not isinstance(origin_dict, dict):
return origin_dict
return {
key.lower(): value
for key, value in origin_dict.items()
}
return {key.lower(): value for key, value in origin_dict.items()}
def deepcopy_dict(data):
@@ -230,6 +227,7 @@ def omit_long_data(body, omit_len=512):
def dump_json_file(json_data: Union[dict, list], json_file_abs_path: str) -> None:
""" dump json data to file
"""
class PythonObjectEncoder(json.JSONEncoder):
def default(self, obj):
try:
@@ -242,14 +240,14 @@ def dump_json_file(json_data: Union[dict, list], json_file_abs_path: str) -> Non
os.makedirs(file_foder_path)
try:
with io.open(json_file_abs_path, 'w', encoding='utf-8') as outfile:
with io.open(json_file_abs_path, "w", encoding="utf-8") as outfile:
json.dump(
json_data,
outfile,
indent=4,
separators=(',', ':'),
separators=(",", ":"),
ensure_ascii=False,
cls=PythonObjectEncoder
cls=PythonObjectEncoder,
)
msg = f"dump file: {json_file_abs_path}"
@@ -268,7 +266,9 @@ def prepare_log_file_abs_path(test_path: str, file_name: str) -> str:
if not test_path:
# running passed in testcase/testsuite data structure
dump_file_name = f"tests_mapping.{file_name}"
dumped_json_file_abs_path = os.path.join(current_working_dir, "logs", dump_file_name)
dumped_json_file_abs_path = os.path.join(
current_working_dir, "logs", dump_file_name
)
return dumped_json_file_abs_path
# both test_path and pwd_dir_path are absolute path

90
httprunner/utils_test.py
View File

@@ -6,18 +6,16 @@ from httprunner import loader, utils
class TestUtils(unittest.TestCase):
def test_set_os_environ(self):
self.assertNotIn("abc", os.environ)
variables_mapping = {
"abc": "123"
}
variables_mapping = {"abc": "123"}
utils.set_os_environ(variables_mapping)
self.assertIn("abc", os.environ)
self.assertEqual(os.environ["abc"], "123")
def current_validators(self):
from httprunner.builtin import comparators
functions_mapping = loader.load.load_module_functions(comparators)
functions_mapping["equals"](None, None)
@@ -35,17 +33,17 @@ class TestUtils(unittest.TestCase):
functions_mapping["not_equals"](123, "123")
functions_mapping["length_equals"]("123", 3)
# Because the Numbers in a CSV file are by default treated as strings,
# Because the Numbers in a CSV file are by default treated as strings,
# you need to convert them to Numbers, and we'll test that out here.
functions_mapping["length_equals"]("123", '3')
functions_mapping["length_equals"]("123", "3")
with self.assertRaises(AssertionError):
functions_mapping["length_equals"]("123", 'abc')
functions_mapping["length_equals"]("123", "abc")
functions_mapping["length_greater_than"]("123", 2)
functions_mapping["length_greater_than_or_equals"]("123", 3)
functions_mapping["contains"]("123abc456", "3ab")
functions_mapping["contains"](['1', '2'], "1")
functions_mapping["contains"]({'a':1, 'b':2}, "a")
functions_mapping["contains"](["1", "2"], "1")
functions_mapping["contains"]({"a": 1, "b": 2}, "a")
functions_mapping["contained_by"]("3ab", "123abc456")
functions_mapping["regex_match"]("123abc456", "^123\w+456$")
@@ -72,10 +70,7 @@ class TestUtils(unittest.TestCase):
request_dict = {
"url": "http://127.0.0.1:5000",
"METHOD": "POST",
"Headers": {
"Accept": "application/json",
"User-Agent": "ios/9.3"
}
"Headers": {"Accept": "application/json", "User-Agent": "ios/9.3"},
}
new_request_dict = utils.lower_dict_keys(request_dict)
self.assertIn("method", new_request_dict)
@@ -93,64 +88,43 @@ class TestUtils(unittest.TestCase):
def test_deepcopy_dict(self):
license_path = os.path.join(
os.path.dirname(os.path.dirname(__file__)),
"LICENSE"
os.path.dirname(os.path.dirname(__file__)), "LICENSE"
)
data = {
'a': 1,
'b': [2, 4],
'c': lambda x: x+1,
'd': open(license_path),
'f': {
'f1': {'a1': 2},
'f2': io.open(license_path, 'rb'),
}
"a": 1,
"b": [2, 4],
"c": lambda x: x + 1,
"d": open(license_path),
"f": {"f1": {"a1": 2}, "f2": io.open(license_path, "rb"),},
}
new_data = utils.deepcopy_dict(data)
data["a"] = 0
self.assertEqual(new_data["a"], 1)
data["f"]["f1"] = 123
self.assertEqual(new_data["f"]["f1"], {'a1': 2})
self.assertEqual(new_data["f"]["f1"], {"a1": 2})
self.assertNotEqual(id(new_data["b"]), id(data["b"]))
self.assertEqual(id(new_data["c"]), id(data["c"]))
# self.assertEqual(id(new_data["d"]), id(data["d"]))
def test_cartesian_product_one(self):
parameters_content_list = [
[
{"a": 1},
{"a": 2}
]
]
parameters_content_list = [[{"a": 1}, {"a": 2}]]
product_list = utils.gen_cartesian_product(*parameters_content_list)
self.assertEqual(
product_list,
[
{"a": 1},
{"a": 2}
]
)
self.assertEqual(product_list, [{"a": 1}, {"a": 2}])
def test_cartesian_product_multiple(self):
parameters_content_list = [
[
{"a": 1},
{"a": 2}
],
[
{"x": 111, "y": 112},
{"x": 121, "y": 122}
]
[{"a": 1}, {"a": 2}],
[{"x": 111, "y": 112}, {"x": 121, "y": 122}],
]
product_list = utils.gen_cartesian_product(*parameters_content_list)
self.assertEqual(
product_list,
[
{'a': 1, 'x': 111, 'y': 112},
{'a': 1, 'x': 121, 'y': 122},
{'a': 2, 'x': 111, 'y': 112},
{'a': 2, 'x': 121, 'y': 122}
]
{"a": 1, "x": 111, "y": 112},
{"a": 1, "x": 121, "y": 122},
{"a": 2, "x": 111, "y": 112},
{"a": 2, "x": 121, "y": 122},
],
)
def test_cartesian_product_empty(self):
@@ -159,15 +133,7 @@ class TestUtils(unittest.TestCase):
self.assertEqual(product_list, [])
def test_print_info(self):
info_mapping = {
"a": 1,
"t": (1, 2),
"b": {
"b1": 123
},
"c": None,
"d": [4, 5]
}
info_mapping = {"a": 1, "t": (1, 2), "b": {"b1": 123}, "c": None, "d": [4, 5]}
utils.print_info(info_mapping)
def test_prepare_dump_json_file_path_for_folder(self):
@@ -175,7 +141,7 @@ class TestUtils(unittest.TestCase):
test_path = os.path.join("tests", "httpbin", "a.b.c")
self.assertEqual(
utils.prepare_log_file_abs_path(test_path, "loaded.json"),
os.path.join(os.getcwd(), "logs", "tests/httpbin/a.b.c/all.loaded.json")
os.path.join(os.getcwd(), "logs", "tests/httpbin/a.b.c/all.loaded.json"),
)
def test_prepare_dump_json_file_path_for_file(self):
@@ -183,12 +149,12 @@ class TestUtils(unittest.TestCase):
test_path = os.path.join("tests", "httpbin", "a.b.c", "rpc.yml")
self.assertEqual(
utils.prepare_log_file_abs_path(test_path, "loaded.json"),
os.path.join(os.getcwd(), "logs", "tests/httpbin/a.b.c/rpc.loaded.json")
os.path.join(os.getcwd(), "logs", "tests/httpbin/a.b.c/rpc.loaded.json"),
)
def test_prepare_dump_json_file_path_for_passed_testcase(self):
test_path = ""
self.assertEqual(
utils.prepare_log_file_abs_path(test_path, "loaded.json"),
os.path.join(os.getcwd(), "logs", "tests_mapping.loaded.json")
os.path.join(os.getcwd(), "logs", "tests_mapping.loaded.json"),
)