feat: add 'upload' keyword for upload test

docs/CHANGELOG.md

@@ -4,6 +4,7 @@
 
 **Added**
 
+- feat: add `upload` keyword for upload test
 - test: pip install package 
 - test: hrun command
 

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

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)
 

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
+        pass
-        raise exceptions.FunctionNotFound("{} is not found.".format(function_name))
+
+    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"])

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