支持上传文件接口功能

README.md

@@ -77,6 +77,7 @@
 ├────outputs/
 │    └────report/  保存测试报告的目录
 │     └────log/   保存日志文件的目录
+├────files 存放测试过程中需要上传的文件
 ├────Pipfile
 ├────pytest.ini
 ├────README.md
@@ -95,7 +96,6 @@
 
 ## 四、依赖库
 ```
-python_version = "3.9"
 pymysql = "*"
 loguru = "*"
 requests-toolbelt = "*"
@@ -112,6 +112,8 @@ pytest = "==6.2.5"
 pytest-html = "==2.1.1"
 pytest-rerunfailures = "*"
 allure-pytest = "==2.9.45"
+pydantic = "*"
+xpinyin = "*"
 ```
 
 
@@ -475,6 +477,162 @@ wechat_content = """
 
 ```
 
+### 7. 上传文件接口支持
+#### 熟悉接口
+- 确定上传文件接口的URL
+- 确定上传文件接口的METHOD
+- 确定上传文件接口请求头里面的Content-Type， 我这边调试的接口都是：multipart/form-data;
+- 确定上传文件接口请求参数， 我调试的接口有两种参数形式：
+
+```python
+# 第一种
+file: 文件二进制内容
+	
+	
+# 第二种
+file: 文件二进制内容
+language: zh
+```
+#### 文件上传的逻辑
+`common_utils/base_request.py` 中封装的request请求，是使用`from requests_toolbelt import MultipartEncoder`进行文件上传的。
+
+- 针对单文件上传，我们需要传递一个字典，参考如下：
+```python
+field = {
+	{
+		'file': (filename, file_content),   # file是接口中文件参数的名称， filename是文件名，file_content是文件二进制内容
+		"key": v    # 这里是文件上传的其他参数
+	}
+}
+```
+
+- 针对多文件上传，我们需要传递一个列表嵌套元祖，参考如下：
+```python
+field =[
+	('file', (filename, file_content)),  # file是接口中文件参数的名称， filename是文件名，file_content是文件二进制内容
+	('file', (filename, file_content)), # 这里是文件上传的其他参数
+	(k, v)  
+	]
+```
+
+
+#### 上传文件，不带其他参数
+- 我们需要设置： request_type=file
+- 然后在files中按照如下格式书写：{接口中文件参数的名称:"文件路径地址"/["文件地址1", "文件地址2"]}
+
+- 参考如下：
+
+```yaml
+# 公共参数
+case_common:
+  allure_epic: GitLink接口（手动编写用例）
+  allure_feature: 上传文件模块
+  allure_story: 上传文件
+
+# 用例数据
+case_upload_demo_01:
+  feature: 上传文件
+  title: 测试单文件上传
+  run: True
+  url: /api/attachments.json
+  method: POST
+  headers:
+    cookies: ${login_cookie}
+  cookies:
+  request_type: file
+  payload:
+  files:
+    file: TOC出库订单导入模板(2).xlsx   # 此处file对应接口中文件参数的名称
+  extract:
+    file_id: $.id
+  assert_response:
+    eq:
+      http_code: 200
+  assert_sql:
+
+case_upload_demo_02:
+  feature: 上传文件
+  title: 测试多文件上传(该接口不支持多文件上传，这是一个示例)
+  run: False
+  url: /api/attachments.json
+  method: POST
+  headers:
+    cookies: ${login_cookie}
+  cookies:
+  request_type: file
+  payload:
+  files:
+    file:
+      - 导入TOC订单.xls
+      - toc.xls
+  extract:
+    file_id: $.id
+  assert_response:
+    eq:
+      http_code: 200
+  assert_sql:
+```
+
+#### 上传文件，带其他参数
+- 我们需要设置： request_type=file
+- 然后在files中按照如下格式书写：{接口中文件参数的名称:"文件路径地址"/["文件地址1", "文件地址2"]}
+- 由于请求参数里面还传递了`language:zh`， 因此我们需要写在`payload`中
+- 在`common_utils/base_request.py` 中，我们会将`language:zh`以元祖形式处理到files里面
+
+- 参考如下：
+
+```yaml
+# 公共参数
+case_common:
+  allure_epic: OWMS系统（自动生成用例）  # 敏捷里面的概念，定义史诗，相当于module级的标签, 往下是 feature
+  allure_feature: 出库模块  # 功能点的描述，相当于class级的标签, 理解成模块往下是 story
+  allure_story: TOC扫描签出接口    # 故事，可以理解为场景，相当于method级的标签, 往下是 title
+
+# 用例数据
+case_import_toc_01:
+  feature: OMS系统
+  title: 导入TOC订单（01）
+  run: True
+  url:  /oms/retailGoodsTmp/selfImportExcel
+  method: POST
+  headers:
+    Cookietoken: ${oms_cookieToken}
+  request_type: file
+  payload:
+    language: zh
+  files:
+    file: TOC出库订单导入模板(2).xlsx
+  extract:
+  assert_response:
+    eq:
+      $.msg: 成功
+  assert_sql:
+
+case_import_toc_02:
+  feature: OMS系统
+  title: 导入TOC订单（02）
+  run: True
+  url:  /oms/retailGoodsTmp/selfImportExcel
+  method: POST
+  headers:
+    Cookietoken: ${oms_cookieToken}
+  request_type: file
+  payload:
+    language: zh
+  files:
+    file:
+      - 导入TOC订单.xls
+      - toc.xls
+  extract:
+  assert_response:
+    eq:
+      $.msg: 成功
+  assert_sql:
+```
+
+
+
+
 ## 初始化项目可能遇到的问题
 ### 1. 测试机安装的是python3.7，但是本框架要求3.9.5，怎么办？
 方法一：建议采纳此方法

case_utils/request_data_handle.py

@@ -6,6 +6,8 @@
 # @Software: PyCharm
 # @Desc: 处理request请求前后的用例数据
 import json
+import os.path
+from common_utils.files_handle import get_file_field
 from common_utils.data_handle import eval_data_process, data_replace
 from config.global_vars import GLOBAL_VARS
 from requests import Response
@@ -13,6 +15,7 @@ from loguru import logger
 from common_utils.data_handle import json_extractor, re_extract
 from case_utils.allure_handle import allure_step
 from common_utils.base_request import BaseRequest
+from config.path_config import FILES_DIR
 
 
 # ---------------------------------------- 请求前的数据处理----------------------------------------#
@@ -107,29 +110,31 @@ class RequestPreDataHandle:
         """
         requests模块中，cookies参数要求是Dict or CookieJar object
         """
+        cookies = self.request_data.get("cookies", None)
         try:
             # 从用例数据中获取cookies， 处理cookies
-            if self.request_data.get("cookies", None):
+            if cookies:
                 # 通过全局变量替换cookies，得到的是一个str类型
-                cookies = data_replace(content=self.request_data.get("cookies"), source=GLOBAL_VARS)
+                cookies = data_replace(content=cookies, source=GLOBAL_VARS)
                 if isinstance(cookies, str):
                     # 如果是字符串类型，就转成字典
                     self.request_data["cookies"] = json.loads(cookies)
                 else:
                     self.request_data["cookies"] = cookies
         except Exception as e:
-            logger.error(f"处理cookies报错了：{e}")
+            logger.error(f"处理{cookies}报错了：{e}")
-            print(f"处理cookies报错了：{e}")
+            print(f"处理{cookies}报错了：{e}")
 
     def headers_handle(self):
         """
         headers里面传cookies，要求cookies类型是str
         """
+        headers = self.request_data.get("headers", None)
         try:
             # 从用例数据中获取header， 处理header
-            if self.request_data.get("headers", None):
+            if headers:
                 self.request_data["headers"] = eval_data_process(
-                    data_replace(content=self.request_data.get("headers"), source=GLOBAL_VARS))
+                    data_replace(content=headers, source=GLOBAL_VARS))
                 # 如果请求头中有cookies，需要进行单独处理
                 if self.request_data["headers"].get("cookies", None):
                     cookies = self.request_data["headers"]["cookies"]
@@ -139,44 +144,75 @@ class RequestPreDataHandle:
                     else:
                         self.request_data["headers"]["cookies"] = cookies
         except Exception as e:
-            logger.error(f"处理header报错了：{e}")
+            logger.error(f"处理{headers}报错了：{e}")
-            print(f"处理header报错了：{e}")
+            print(f"处理{headers}报错了：{e}")
 
     def payload_handle(self):
+        # 处理请求参数payload
+        payload = self.request_data.get("payload", None)
         try:
-            # 处理请求参数payload
+            if payload:
-            if self.request_data.get("payload", None):
                 self.request_data["payload"] = eval_data_process(
-                    data_replace(content=self.request_data.get("payload"), source=GLOBAL_VARS))
+                    data_replace(content=payload, source=GLOBAL_VARS))
         except Exception as e:
-            logger.error(f"处理payload报错了：{e}")
+            logger.error(f"处理{payload}报错了：{e}")
-            print(f"处理payload报错了：{e}")
+            print(f"处理{payload}报错了：{e}")
 
     def files_handle(self):
-        # 处理文件
+        """
-        # TODO 暂时还没想好怎么处理
+        格式：接口中文件参数的名称:"文件路径地址"/["文件地址1", "文件地址2"]
-        pass
+        例如：{"file": "test_demo.py"}
+        或者
+        {"file": ["test_demo_01.py", "test_demo_02.py"]}
+        """
+        # 处理请求参数files参数
+        files = self.request_data.get("files", None)
+        try:
+            if files:
+                for k, v in files.items():
+                    # ------------------ 处理多文件的情况 ------------------
+                    # 这里需要注意：不一定所有接口都支持多文件上传
+                    _files = []
+                    if isinstance(v, list):
+                        for file in v:
+                            # 处理文件绝对路径
+                            file_path = os.path.join(FILES_DIR, file)
+                            # 多文件上传需要是元祖[('file', (filename, file_content)), ('file', (filename, file_content))]
+                            _files.append((k, get_file_field(file_path)))
+                        self.request_data["files"] = _files
+                    else:
+                        # ------------------ 处理单文件的情况 ------------------
+                        # 处理文件绝对路径
+                        file_path = os.path.join(FILES_DIR, v)
+                        # 单文件上传需要是字典{'file': (filename, file_content)}
+                        self.request_data["files"] = {k: get_file_field(file_path)}
+                logger.debug(f"处理完成后的file:{self.request_data['files']}")
+        except Exception as e:
+            logger.error(f"处理{files}报错了：{e}")
+            print(f"处理{files}报错了：{e}")
 
     def extract_handle(self):
+        # 处理后置提取参数
+        extract = self.request_data.get("extract", None)
         try:
-            # 处理后置提取参数
+            if extract:
-            if self.request_data.get("extract", None):
                 # 仅提取参数中的python表达式，不需要进行数据替换
-                self.request_data["extract"] = eval_data_process(self.request_data.get("extract"))
+                self.request_data["extract"] = eval_data_process(extract)
         except Exception as e:
-            logger.error(f"处理extract报错了：{e}")
+            logger.error(f"处理{extract}报错了：{e}")
-            print(f"处理extract报错了：{e}")
+            print(f"处理{extract}报错了：{e}")
 
     def assert_handle(self):
+        # 处理响应断言参数
+        assert_response = self.request_data.get("assert_response", None)
         try:
-            # 处理响应断言参数
+            if assert_response:
-            if self.request_data.get("assert_response", None):
                 self.request_data["assert_response"] = eval_data_process(
-                    data_replace(content=self.request_data.get("assert_response"), source=GLOBAL_VARS))
+                    data_replace(content=assert_response, source=GLOBAL_VARS))
             # 由于数据库断言里面的变量需要请求响应后进行提取，因此目前不进行处理
         except Exception as e:
-            logger.error(f"处理assert报错了：{e}")
+            logger.error(f"处理{assert_response}报错了：{e}")
-            print(f"处理assert报错了：{e}")
+            print(f"处理{assert_response}报错了：{e}")
 
 
 # ---------------------------------------- 进行请求，请求后的参数提取处理----------------------------------------#
@@ -272,3 +308,7 @@ def response_type(response: Response) -> str:
         return "json"
     except:
         return "str"
+
+if __name__ == '__main__':
+    test = RequestPreDataHandle(request_data={"files": {"file": ["导入TOC订单.xls", "toc.xls"]}})
+    test.files_handle()

common_utils/base_request.py

@@ -1,15 +1,18 @@
-import requests
-from requests import Response
-from requests_toolbelt import MultipartEncoder  # 第三方模块：pip install requests_toolbelt
-from typing import Dict, Union
 import time
+from typing import Dict, Union
+import requests
 from loguru import logger
+from requests import Response
+from requests_toolbelt import MultipartEncoder
 
 
 class BaseRequest:
     """
     进行请求
     """
+
+    TIMEOUT = 5
+
     session = None
 
     @classmethod
@@ -30,28 +33,28 @@ class BaseRequest:
         :return: 响应对象
         """
         try:
-            logger.debug("\n======================================================\n" \
+            logger.debug("\n" + "=" * 80
-                         "-------------Start：请求前--------------------\n"
+                         + "\n-------------Start：请求前--------------------\n"
-                         f"用例标题: {req_data.get('title', None)}\n" \
+                           f"用例标题: {req_data.get('title', None)}\n"
-                         f"请求路径: {req_data.get('url', None)}\n" \
+                           f"请求路径: {req_data.get('url', None)}\n"
-                         f"请求方式: {req_data.get('method', None)}\n" \
+                           f"请求方式: {req_data.get('method', None)}\n"
-                         f"请求头:   {req_data.get('headers', None)}\n" \
+                           f"请求头:   {req_data.get('headers', None)}\n"
-                         f"请求Cookies:   {req_data.get('cookies', None)}\n" \
+                           f"请求Cookies:   {req_data.get('cookies', None)}\n"
-                         f"请求关键字: {req_data.get('request_type', None)}\n" \
+                           f"请求关键字: {req_data.get('request_type', None)}\n"
-                         f"请求内容: {req_data.get('payload', None)}\n" \
+                           f"请求内容: {req_data.get('payload', None)}\n"
-                         f"请求文件: {req_data.get('files', None)}\n" \
+                           f"请求文件: {req_data.get('files', None)}\n"
-                         "=====================================================")
+                         + "=" * 80)
-            print("\n======================================================\n" \
+            print("\n" + "=" * 80
-                  "-------------Start：请求前--------------------\n"
+                  + "\n-------------Start：请求前--------------------\n"
-                  f"用例标题: {req_data.get('title', None)}\n" \
+                    f"用例标题: {req_data.get('title', None)}\n"
-                  f"请求路径: {req_data.get('url', None)}\n" \
+                    f"请求路径: {req_data.get('url', None)}\n"
-                  f"请求方式: {req_data.get('method', None)}\n" \
+                    f"请求方式: {req_data.get('method', None)}\n"
-                  f"请求头:   {req_data.get('headers', None)}\n" \
+                    f"请求头:   {req_data.get('headers', None)}\n"
-                  f"请求Cookies:   {req_data.get('cookies', None)}\n" \
+                    f"请求Cookies:   {req_data.get('cookies', None)}\n"
-                  f"请求关键字: {req_data.get('request_type', None)}\n" \
+                    f"请求关键字: {req_data.get('request_type', None)}\n"
-                  f"请求内容: {req_data.get('payload', None)}\n" \
+                    f"请求内容: {req_data.get('payload', None)}\n"
-                  f"请求文件: {req_data.get('files', None)}\n" \
+                    f"请求文件: {req_data.get('files', None)}\n"
-                  "=====================================================")
+                  + "=" * 80)
             res = cls.send_api_request(
                 url=req_data.get("url"),
                 method=req_data.get("method").lower(),
@@ -61,16 +64,16 @@ class BaseRequest:
                 files=req_data.get("files", None),
                 cookies=req_data.get("cookies", None)
             )
-            logger.debug("\n======================================================\n" \
+            logger.debug("\n" + "=" * 80
-                         "-------------End：请求后--------------------\n"
+                         + "\n-------------End：请求后--------------------\n"
-                         f"响应数据: {res.text}\n" \
+                           f"响应数据: {res.text}\n"
-                         f"响应码: {res.status_code}\n" \
+                           f"响应码: {res.status_code}\n"
-                         "=====================================================")
+                         + "=" * 80)
-            print("\n======================================================\n" \
+            print("\n" + "=" * 80
-                  "-------------End：请求后--------------------\n"
+                  + "\n-------------End：请求后--------------------\n"
-                  f"响应数据: {res.text}\n" \
+                    f"响应数据: {res.text}\n"
-                  f"响应码: {res.status_code}\n" \
+                    f"响应码: {res.status_code}\n"
-                  "=====================================================")
+                  + "=" * 80)
         except requests.exceptions.RequestException as e:
             logger.error(f"请求出错，{str(e)}")
             print(f"请求出错，{str(e)}")
@@ -95,33 +98,44 @@ class BaseRequest:
         headers = header or {}
         session = cls.get_session()
 
-        if request_type and request_type.lower() == 'params':
+        if request_type:
-            res = session.request(method=method, url=url, params=payload, headers=headers, cookies=cookies, timeout=5)
+            if request_type.lower() == 'params':
-        elif request_type and request_type.lower() == 'data':
+                res = session.request(method=method, url=url, params=payload, headers=headers, cookies=cookies,
-            if files:
+                                      timeout=cls.TIMEOUT)
-                if not isinstance(files, dict):
+                return res
-                    raise ValueError('data参数必须为dict')
+            elif request_type.lower() == 'data':
-                encoder = MultipartEncoder(fields=files, boundary='------------------------' + str(time.time()))
+                res = session.request(method=method, url=url, data=payload, headers=headers, cookies=cookies,
-                headers['Content-Type'] = encoder.content_type
+                                      timeout=cls.TIMEOUT)
-                res = session.request(method=method, url=url, data=encoder.to_string(), headers=headers,
+                return res
-                                      cookies=cookies, timeout=5)
+            elif request_type.lower() == 'json':
+                res = session.request(method=method, url=url, json=payload, headers=headers, cookies=cookies,
+                                      timeout=cls.TIMEOUT)
+                return res
+            elif request_type.lower() == 'file':
+                if files:
+                    if payload:
+                        if isinstance(files, dict):
+                            for k, v in payload.items():
+                                files[k] = v
+                        elif isinstance(files, list):
+                            # TODO 这里是应对多文件上传的情况，暂时没有接口帮助验证是否真正上传了多个文件，有可能只上传成功了一个
+                            for k, v in payload.items():
+                                files.append((k, v))
+                    encoder = MultipartEncoder(fields=files, boundary='------------------------' + str(time.time()))
+                    headers['Content-Type'] = encoder.content_type
+                    res = session.request(method=method, url=url, data=encoder.to_string(), headers=headers,
+                                          cookies=cookies, timeout=cls.TIMEOUT)
+                    return res
+                else:
+                    logger.error('上传的文件不能为空')
+                    print('上传的文件不能为空')
             else:
-                headers['Content-Type'] = 'application/x-www-form-urlencoded; charset=UTF-8'
+                logger.error('request_type可选关键字为params, json, data, file')
-                res = session.request(method=method, url=url, data=payload, headers=headers, cookies=cookies, timeout=5)
+                print('request_type可选关键字为params, json, data, file')
-        elif request_type and request_type.lower() == 'json':
+                raise ValueError('request_type可选关键字为params, json, data, file')
-            if files:
-                if not isinstance(files, dict):
-                    raise ValueError('json参数必须为dict')
-                encoder = MultipartEncoder(fields=files, boundary='------------------------' + str(time.time()))
-                headers['Content-Type'] = encoder.content_type
-                res = session.request(method=method, url=url, json=encoder.to_string(), headers=headers,
-                                      cookies=cookies, timeout=5)
-            else:
-                headers['Content-Type'] = 'application/json'
-                res = session.request(method=method, url=url, json=payload, headers=headers, cookies=cookies, timeout=5)
         else:
-            logger.error('request_type可选关键字为params, json, data')
+            logger.error('request_type参数不能为空')
-            print('request_type可选关键字为params, json, data')
+            print('request_type参数不能为空')
-            raise ValueError('request_type可选关键字为params, json, data')
+            raise ValueError('request_type参数不能为空')
+
 
-        return res

common_utils/files_handle.py

@@ -133,3 +133,16 @@ def copy_file(src_file_path, dest_dir_path):
         return "复制成功"
     except Exception as e:
         return f"复制失败：{e}"
+
+
+def get_file_field(file_path):
+    """
+    获取文件名称和二进制内容
+    :param: file_path: 文件路径
+    """
+    # 处理文件绝对路径
+    file_name = os.path.basename(file_path)
+    # 获取文件二进制内容
+    with open(file_path, 'rb') as f:
+        file_content = f.read()
+    return (file_name, file_content)

config/path_config.py

@@ -17,9 +17,12 @@ COMMON_DIR = os.path.join(BASE_DIR, "common_utils")
 # 配置模块目录
 CONF_DIR = os.path.join(BASE_DIR, "config")
 
-# 数据模块目录
+# 测试数据模块目录
 DATA_DIR = os.path.join(BASE_DIR, "data")
 
+# 测试文件模块目录
+FILES_DIR = os.path.join(BASE_DIR, "files")
+
 # 日志/报告保存目录
 OUT_DIR = os.path.join(BASE_DIR, "outputs")
 if not os.path.exists(OUT_DIR):