diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index a2ccfe53..43f322a9 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -4,6 +4,7 @@ **Added** +- feat: add `upload` keyword for upload test - test: pip install package - test: hrun command diff --git a/httprunner/builtin/functions.py b/httprunner/builtin/functions.py index 0cca0295..d5b31c7a 100644 --- a/httprunner/builtin/functions.py +++ b/httprunner/builtin/functions.py @@ -3,19 +3,13 @@ Built-in functions used in YAML/JSON testcases. """ import datetime -import os import random import string import time -import filetype -from requests_toolbelt import MultipartEncoder - from httprunner.compat import builtin_str, integer_types from httprunner.exceptions import ParamsError -PWD = os.getcwd() - def gen_random_string(str_len): """ generate random string with specified length @@ -44,62 +38,3 @@ def sleep(n_secs): """ time.sleep(n_secs) - -""" -upload files with requests-toolbelt -e.g. - - - test: - name: upload file - variables: - file_path: "data/test.env" - multipart_encoder: ${multipart_encoder(file=$file_path)} - request: - url: /post - method: POST - headers: - Content-Type: ${multipart_content_type($multipart_encoder)} - data: $multipart_encoder - validate: - - eq: ["status_code", 200] - - startswith: ["content.files.file", "UserName=test"] -""" - - -def multipart_encoder(**kwargs): - """ initialize MultipartEncoder with uploading fields. - """ - - def get_filetype(file_path): - file_type = filetype.guess(file_path) - if file_type: - return file_type.mime - else: - return "text/html" - - fields_dict = {} - for key, value in kwargs.items(): - - if os.path.isabs(value): - _file_path = value - is_file = True - else: - global PWD - _file_path = os.path.join(PWD, value) - is_file = os.path.isfile(_file_path) - - if is_file: - filename = os.path.basename(_file_path) - with open(_file_path, 'rb') as f: - mime_type = get_filetype(_file_path) - fields_dict[key] = (filename, f.read(), mime_type) - else: - fields_dict[key] = value - - return MultipartEncoder(fields=fields_dict) - - -def multipart_content_type(multipart_encoder): - """ prepare Content-Type for request headers - """ - return multipart_encoder.content_type diff --git a/httprunner/loader/buildup.py b/httprunner/loader/buildup.py index deb3352a..1a1096c2 100644 --- a/httprunner/loader/buildup.py +++ b/httprunner/loader/buildup.py @@ -2,7 +2,6 @@ import importlib import os from httprunner import exceptions, logger, utils -from httprunner.builtin import functions from httprunner.loader.load import load_module_functions, load_folder_content, load_file, load_dot_env_file, \ load_folder_files from httprunner.loader.locate import init_project_working_directory, get_project_working_directory @@ -481,7 +480,6 @@ def load_project_data(test_path, dot_env_path=None): # locate PWD and load debugtalk.py functions project_mapping["PWD"] = project_working_directory - functions.PWD = project_working_directory # TODO: remove project_mapping["functions"] = debugtalk_functions project_mapping["test_path"] = os.path.abspath(test_path) diff --git a/httprunner/parser.py b/httprunner/parser.py index dccb5ed6..cc6f13aa 100644 --- a/httprunner/parser.py +++ b/httprunner/parser.py @@ -428,6 +428,11 @@ def get_mapping_function(function_name, functions_mapping): elif function_name in ["environ", "ENV"]: return utils.get_os_environ + elif function_name in ["multipart_encoder", "multipart_content_type"]: + # plugin for upload test + from httprunner.plugins import uploader + return getattr(uploader, function_name) + try: # check if HttpRunner builtin functions built_in_functions = loader.load_builtin_functions() @@ -439,8 +444,9 @@ def get_mapping_function(function_name, functions_mapping): # check if Python builtin functions return getattr(builtins, function_name) except AttributeError: - # is not builtin function - raise exceptions.FunctionNotFound("{} is not found.".format(function_name)) + pass + + raise exceptions.FunctionNotFound("{} is not found.".format(function_name)) def parse_function_params(params): @@ -1146,13 +1152,18 @@ def __prepare_testcase_tests(tests, config, project_mapping, session_variables_s api_def_dict = test_dict.pop("api_def") _extend_with_api(test_dict, api_def_dict) + # verify priority: testcase teststep > testcase config + if "request" in test_dict: + if "verify" not in test_dict["request"]: + test_dict["request"]["verify"] = config_verify + + if "upload" in test_dict["request"]: + from httprunner.plugins.uploader import prepare_upload_test + prepare_upload_test(test_dict) + # current teststep variables teststep_variables_set |= set(test_dict.get("variables", {}).keys()) - # verify priority: testcase teststep > testcase config - if "request" in test_dict and "verify" not in test_dict["request"]: - test_dict["request"]["verify"] = config_verify - # move extracted variable to session variables if "extract" in test_dict: extract_mapping = utils.ensure_mapping_format(test_dict["extract"]) diff --git a/httprunner/plugins/uploader/__init__.py b/httprunner/plugins/uploader/__init__.py new file mode 100644 index 00000000..17bcf5f7 --- /dev/null +++ b/httprunner/plugins/uploader/__init__.py @@ -0,0 +1,126 @@ +""" upload test plugin. + +If you want to use this plugin, you should install the following dependencies first. + +- requests_toolbelt +- filetype + +Then you can write upload test script as below: + + - test: + name: upload file + request: + url: http://httpbin.org/upload + method: POST + headers: + Cookie: session=AAA-BBB-CCC + upload: + file: "data/file_to_upload" + field1: "value1" + field2: "value2" + validate: + - eq: ["status_code", 200] + +""" + +import os +import sys + +try: + import filetype + from requests_toolbelt import MultipartEncoder +except ImportError: + msg = """ +uploader plugin dependencies uninstalled, install first and try again. +install with pip: +$ pip install requests_toolbelt filetype +""" + print(msg) + sys.exit(0) + +from httprunner.exceptions import ParamsError + +PWD = os.getcwd() + + +def prepare_upload_test(test_dict): + """ preprocess for upload test + replace `upload` info with MultipartEncoder + + Args: + test_dict (dict): + + { + "variables": {}, + "request": { + "url": "http://httpbin.org/upload", + "method": "POST", + "headers": { + "Cookie": "session=AAA-BBB-CCC" + }, + "upload": { + "file": "data/file_to_upload" + "md5": "123" + } + } + } + + + Returns: + (dict, dict): + - variables: prepared variables for upload test + - request: prepared request for upload test + + """ + upload_json = test_dict["request"].pop("upload", {}) + if not upload_json: + raise ParamsError("invalid upload info: {}".format(upload_json)) + + params_list = [] + for key, value in upload_json.items(): + test_dict["variables"][key] = value + params_list.append("{}=${}".format(key, key)) + + params_str = ", ".join(params_list) + test_dict["variables"]["m_encoder"] = "${multipart_encoder(" + params_str + ")}" + test_dict["request"]["headers"]["Content-Type"] = "${multipart_content_type($m_encoder)}" + test_dict["request"]["data"] = "$m_encoder" + + +def multipart_encoder(**kwargs): + """ initialize MultipartEncoder with uploading fields. + """ + + def get_filetype(file_path): + file_type = filetype.guess(file_path) + if file_type: + return file_type.mime + else: + return "text/html" + + fields_dict = {} + for key, value in kwargs.items(): + + if os.path.isabs(value): + _file_path = value + is_file = True + else: + global PWD + _file_path = os.path.join(PWD, value) + is_file = os.path.isfile(_file_path) + + if is_file: + filename = os.path.basename(_file_path) + with open(_file_path, 'rb') as f: + mime_type = get_filetype(_file_path) + fields_dict[key] = (filename, f.read(), mime_type) + else: + fields_dict[key] = value + + return MultipartEncoder(fields=fields_dict) + + +def multipart_content_type(m_encoder): + """ prepare Content-Type for request headers + """ + return m_encoder.content_type