feat: implement upload keyword

httprunner/client.py

@@ -36,22 +36,22 @@ def get_req_resp_record(resp_obj: Response) -> ReqRespData:
         logger.debug(msg)
 
     # record actual request info
+    request_headers = dict(resp_obj.request.headers)
+    request_body = resp_obj.request.body
+
+    if request_body:
+        request_content_type = lower_dict_keys(request_headers).get("content-type")
+        if request_content_type and "multipart/form-data" in request_content_type:
+            # upload file type
+            request_body = "upload file stream (OMITTED)"
+
     request_data = RequestData(
         method=resp_obj.request.method,
         url=resp_obj.request.url,
-        headers=dict(resp_obj.request.headers),
-        body=resp_obj.request.body,
+        headers=request_headers,
+        body=request_body,
     )
 
-    request_body = resp_obj.request.body
-    if request_body:
-        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)"
-        else:
-            request_data.body = request_body
-
     # log request details in debug mode
     log_print(request_data, "request")
 

httprunner/ext/uploader/__init__.py

@@ -45,6 +45,9 @@ For compatibility, you can also write upload test script in old way:
 import os
 import sys
 
+from httprunner.parser import parse_variables_mapping
+from httprunner.schema import TStep, FunctionsMapping
+
 try:
     import filetype
     from requests_toolbelt import MultipartEncoder
@@ -57,16 +60,13 @@ $ pip install requests_toolbelt filetype
     print(msg)
     sys.exit(0)
 
-from httprunner.exceptions import ParamsError
 
-
-def prepare_upload_test(test_dict):
+def prepare_upload_step(step: TStep, functions: FunctionsMapping):
     """ preprocess for upload test
         replace `upload` info with MultipartEncoder
 
     Args:
-        test_dict (dict):
-
+        step: teststep
             {
                 "variables": {},
                 "request": {
@@ -81,26 +81,26 @@ def prepare_upload_test(test_dict):
                     }
                 }
             }
+        functions: functions mapping
 
     """
-    upload_json = test_dict["request"].pop("upload", {})
-    if not upload_json:
-        raise ParamsError(f"invalid upload info: {upload_json}")
+    if not step.request.upload:
+        return
 
     params_list = []
-    for key, value in upload_json.items():
-        test_dict["variables"][key] = value
+    for key, value in step.request.upload.items():
+        step.variables[key] = value
         params_list.append(f"{key}=${key}")
 
     params_str = ", ".join(params_list)
-    test_dict["variables"]["m_encoder"] = "${multipart_encoder(" + params_str + ")}"
+    step.variables["m_encoder"] = "${multipart_encoder(" + params_str + ")}"
 
-    test_dict["request"].setdefault("headers", {})
-    test_dict["request"]["headers"][
-        "Content-Type"
-    ] = "${multipart_content_type($m_encoder)}"
+    # parse variables
+    step.variables = parse_variables_mapping(step.variables, functions)
 
-    test_dict["request"]["data"] = "$m_encoder"
+    step.request.headers["Content-Type"] = "${multipart_content_type($m_encoder)}"
+
+    step.request.data = "$m_encoder"
 
 
 def multipart_encoder(**kwargs):

httprunner/loader_test.py

@@ -125,4 +125,3 @@ class TestLoader(unittest.TestCase):
             loader.locate_file("examples/httpbin/", "debugtalk.py"),
             os.path.join(os.getcwd(), "examples/httpbin/debugtalk.py"),
         )
-

httprunner/runner.py

@@ -8,6 +8,7 @@ from loguru import logger
 from httprunner import utils, exceptions
 from httprunner.client import HttpSession
 from httprunner.exceptions import ValidationFailure, ParamsError
+from httprunner.ext.uploader import prepare_upload_step
 from httprunner.loader import load_project_meta, load_testcase_file
 from httprunner.parser import build_url, parse_data, parse_variables_mapping
 from httprunner.response import ResponseObject
@@ -53,7 +54,9 @@ class HttpRunner(object):
         step_data = StepData(name=step.name)
 
         # parse
+        prepare_upload_step(step, self.__project_meta.functions)
         request_dict = step.request.dict()
+        request_dict.pop("upload", None)
         parsed_request_dict = parse_data(
             request_dict, step.variables, self.__project_meta.functions
         )

httprunner/runner_test.py

@@ -13,12 +13,8 @@ class TestHttpRunner(unittest.TestCase):
         )
         result = self.runner.get_summary()
         self.assertTrue(result.success)
-        self.assertEqual(
-            result.name, "request methods testcase with variables"
-        )
-        self.assertEqual(
-            result.step_datas[0].name, "get with params"
-        )
+        self.assertEqual(result.name, "request methods testcase with variables")
+        self.assertEqual(result.step_datas[0].name, "get with params")
         self.assertEqual(len(result.step_datas), 3)
 
     def test_run_testcase_by_path_ref_testcase(self):
@@ -27,10 +23,6 @@ class TestHttpRunner(unittest.TestCase):
         )
         result = self.runner.get_summary()
         self.assertTrue(result.success)
-        self.assertEqual(
-            result.name, "request methods testcase: reference testcase"
-        )
-        self.assertEqual(
-            result.step_datas[0].name, "request with variables"
-        )
+        self.assertEqual(result.name, "request methods testcase: reference testcase")
+        self.assertEqual(result.step_datas[0].name, "request with variables")
         self.assertEqual(len(result.step_datas), 1)

httprunner/schema.py

@@ -56,6 +56,7 @@ class Request(BaseModel):
     timeout: int = 120
     allow_redirects: bool = True
     verify: Verify = False
+    upload: Dict = {}  # used for upload files
 
 
 class TStep(BaseModel):
@@ -107,7 +108,7 @@ class RequestData(BaseModel):
     url: Url
     headers: Headers = {}
     # TODO: add cookies
-    body: Union[Text, Dict] = {}
+    body: Union[Text, bytes, Dict, None] = {}
 
 
 class ResponseData(BaseModel):
@@ -137,6 +138,7 @@ 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
     data: Union[SessionData, List[SessionData]] = None