(2023-05-14) 优化封装requests方法，将BaseRequest类移动到公共方法中

case_utils/data_handle.py

@@ -9,10 +9,35 @@ from loguru import logger
 from jsonpath import jsonpath
 import re
 from faker import Faker
+from string import Template
+from config.global_vars import GLOBAL_VARS
 
 faker = Faker()
 
 
+def case_data_replace(content):
+    """
+    用例数据替换的方法
+    :param content: 原始的字符串内容
+    return content： 替换表达式后的字符串
+    """
+    if content is None:
+        return None
+    logger.debug(f"-----Start-----开始进行字符串替换: 初始字符串为：{content}")
+    if len(content) != 0:
+        # safe_substitute() 方法会保留没有被替换的占位符，不会抛出 KeyError 异常。
+        # 所以，如果 content 中不存在占位符，使用 safe_substitute() 方法进行替换后，得到的结果和原始字符串是一样的。
+        content = Template(str(content)).safe_substitute(GLOBAL_VARS)
+        for func in re.findall('\\${(.*?)}', content):
+            content = content.replace('${%s}' % func, exec_func(func))
+            try:
+                content = content.replace('${%s}' % func, exec_func(func))
+            except Exception as e:
+                logger.error(f"-----END-----替换数据时出现了异常：{e}")
+        logger.debug(f"-----END-----字符串替换完成: 新字符串为：{content}")
+        return content
+
+
 def json_extractor(obj: dict, expr: str = '.'):
     """
     :param obj :json/dict类型数据

case_utils/request_data_handle.py

@@ -4,15 +4,14 @@
 # @Author  : chenyinhua
 # @File    : request_data_handle.py
 # @Software: PyCharm
-# @Desc: 处理request请求前的用例数据
+# @Desc: 处理request请求前后的用例数据
 
 
-import re
+from case_utils.data_handle import eval_data_process, case_data_replace
-from string import Template
 from config.global_vars import GLOBAL_VARS
+from requests import Response
 from loguru import logger
-from case_utils.data_handle import exec_func
+from case_utils.data_handle import json_extractor, re_extract
-from case_utils.data_handle import eval_data_process
 
 
 # ---------------------------------------- 请求前的数据处理----------------------------------------#
@@ -29,26 +28,21 @@ class RequestPreDataHandle:
         """
         针对用例数据进行处理，识别用例数据中的关键字${xxxx}，使用全局变量进行替换或者执行关键字中的方法替换为具体值
         """
-        logger.info(f"-----Start-----初始获取的测试用例数据：----- {type(self.request_data)} || {self.request_data}")
+        self.url_handle()
+        self.headers_handle()
+        self.payload_handle()
+        self.extract_handle()
+        self.assert_handle()
+        return self.request_data
 
-        # 从用例数据中获取url，如果键url不存在，则返回空字符串
+    def url_handle(self):
-        url = self.request_data.get("url", "")
-        self.request_data["url"] = self.url_handle(url)
-
-        # 从用例数据中获取files，如果键files不存在，则返回None
-        files = self.request_data.get("files", None)
-        self.request_data["files"] = self.files_handle(files)
-
-        request_data = eval_data_process(case_data_replace(self.request_data))
-
-        logger.info(f"-----End-----处理完成后的测试用例数据：-----{type(request_data)} ||  {request_data}")
-        return request_data
-
-    def url_handle(self, url):
         """
         用例数据中获取到的url(一般是不带host的，个别特殊的带有host，则不进行处理)
         """
         host = GLOBAL_VARS.get("host", "")
+        url = self.request_data.get("url", "")
+        logger.info(
+            f"-----Start-----处理前的host： {GLOBAL_VARS.get('host', '')} || 处理前的url： {self.request_data.get('url', '')}")
         # 从用例数据中获取url，如果键url不存在，则返回空字符串
         # 如果url是以http开头的，则直接使用该url，不与host进行拼接
         if url.lower().startswith("http"):
@@ -66,47 +60,78 @@ class RequestPreDataHandle:
             else:
                 # 如果host不以/结尾 且 url不以/开头，则将host和url拼接起来的时候增加/，组成新的url
                 full_url = host + "/" + url
-        return full_url
+        self.request_data["url"] = full_url
+        logger.info(f"-----End-----处理完成后的full_url：{self.request_data['url']}")
 
-    def files_handle(self, files):
+    def headers_handle(self):
-        """
+        # 从用例数据中获取header， 处理header
-        格式：接口中文件参数的名称:"文件路径地址"/["文件地址1", "文件地址2"]
+        logger.info(
-        """
+            f"-----Start-----处理前的headers： {type(self.request_data.get('headers', None))} ||  {self.request_data.get('headers', None)}")
+        if self.request_data.get("headers", None):
+            self.request_data["headers"] = eval_data_process(case_data_replace(self.request_data.get("headers", None)))
+            logger.info(
+                f"-----End-----处理完成后的headers： {type(self.request_data['headers'])} ||  {self.request_data['headers']}")
 
-        if files is None:
+    def payload_handle(self):
-            return
+        # 处理请求参数payload
+        logger.info(
+            f"-----Start-----处理前的payload： {type(self.request_data.get('payload', None))} ||  {self.request_data.get('payload', None)}")
+        if self.request_data.get("payload", None):
+            self.request_data["payload"] = eval_data_process(case_data_replace(self.request_data.get("payload", None)))
+            logger.info(
+                f"-----End-----处理完成后的payload： {type(self.request_data['payload'])} ||  {self.request_data['payload']}")
 
-        if files != "" and files is not None:
+    def extract_handle(self):
-            files = eval(files)
+        # 处理后置提取参
-            for k, v in files.items():
+        logger.info(
-                # 多文件上传
+            f"-----Start-----处理前的extract： {type(self.request_data.get('extract', None))} ||  {self.request_data.get('extract', None)}")
-                if isinstance(v, list):
+        if self.request_data.get("extract", None):
-                    files = []
+            # 仅提取参数中的python表达式，不需要进行数据替换
-                    for path in v:
+            self.request_data["extract"] = eval_data_process(self.request_data.get("extract", None))
-                        files.append((k, (open(path, 'rb'))))
+            logger.info(
-                else:
+                f"-----End-----处理完成后的extract： {type(self.request_data['extract'])} ||  {self.request_data['extract']}")
-                    # 单文件上传
-                    files = {k: open(v, 'rb')}
 
-        return files
+    def assert_handle(self):
+        # 处理响应断言参数
+        logger.info(
+            f"-----Start-----处理前的assert_response： {type(self.request_data.get('assert_response', None))} ||  {self.request_data.get('assert_response', None)}")
+        if self.request_data.get("assert_response", None):
+            self.request_data["assert_response"] = eval_data_process(
+                case_data_replace(self.request_data.get("assert_response", None)))
+            logger.info(
+                f"-----End-----处理完成后的assert_response： {type(self.request_data['assert_response'])} ||  {self.request_data['assert_response']}")
 
 
-def case_data_replace(content):
+# ---------------------------------------- 请求后的参数提取处理----------------------------------------#
+def after_extract(response: Response, extract) -> None:
     """
-    用例数据替换的方法
+    从响应数据中提取请求后的参数，并保存到全局变量中
-    :param content: 原始的字符串内容
+    :param response: request 响应对象
-    return content： 替换表达式后的字符串
+    :param extract: 需要提取的参数字典 '{"k1": "$.data"}' 或 '{"k1": "data:(.*?)$"}'
+    :return:
     """
-    if content is None:
+    logger.info(f"-----Start-----请求后的参数提取处理，需要提取的参数：{extract}-----")
-        return None
+    if extract:
-    if len(content) != 0:
+        if response_type(response) == "json":
-        logger.debug(f"开始进行字符串替换: 替换字符串为：{content}")
+            # 如果响应数据是json格式，则将按照json方式对后置提取参数进行处理
-        content = Template(str(content)).safe_substitute(GLOBAL_VARS)
+            res = response.json()
-        logger.debug(f"使用模板函数Template替换字符串完成。 替换后的字符串如下：{content}")
+            for k, v in extract.items():
-        for func in re.findall('\\${(.*?)}', content):
+                GLOBAL_VARS[k] = json_extractor(res, v)
-            try:
+        else:
-                content = content.replace('${%s}' % func, exec_func(func))
+            # 如果响应数据是str格式，则将按照str方式对后置提取参数进行处理
-                logger.debug(f"通过执行函数替换用例数据值 替换字符串后为：{content}")
+            res = response.text
-            except Exception as e:
+            for k, v in extract.items():
-                logger.exception(e)
+                GLOBAL_VARS[k] = re_extract(res, v)
-        return content
+    logger.info(f"-----End-----参数提取后，打印当前的全局变量:{GLOBAL_VARS}")
+
+
+def response_type(response: Response) -> str:
+    """
+    :param response: requests 返回
+    :return: 返回响应数据类型 json或者str
+    """
+    try:
+        response.json()
+        return "json"
+    except:
+        return "str"

case_utils/requests_handle.py

@@ -1,115 +0,0 @@
-# -*- coding: utf-8 -*-
-# @Version: Python 3.9
-# @Time    : 2023/1/31 14:31
-# @Author  : chenyinhua
-# @File    : requests_handle.py
-# @Software: PyCharm
-# @Desc: 处理request请求
-
-
-import requests
-from config.global_vars import GLOBAL_VARS
-from requests import Response
-from loguru import logger
-from case_utils.data_handle import json_extractor, re_extract
-
-
-# ---------------------------------------- 进行请求----------------------------------------#
-class BaseRequest:
-    """
-    进行请求
-    """
-    session = None
-
-    @classmethod
-    def get_session(cls):
-        """
-        单例模式保证测试过程中使用的都是一个session对象；
-        requests.session可以自动处理cookies，做状态保持。
-        """
-        if cls.session is None:
-            cls.session = requests.Session()
-        return cls.session
-
-    @classmethod
-    def send_request(cls, req_data: dict) -> Response:
-        """
-        处理case数据，转换成可用数据发送请求
-        :param case: 读取出来的每一行用例内容
-        return: 响应对象
-        """
-        logger.info(f"-----Start-----进行接口请求，并获取接口响应数据，请求数据：{type(req_data)} || {req_data}-----")
-        # 进行接口请求，并获取接口响应数据
-        res = cls.send_api(
-            url=req_data["url"],
-            method=req_data["method"],
-            pk=req_data["pk"],
-            header=req_data.get("headers", None),
-            data=req_data.get("payload", None),
-            file=req_data.get("files", None)
-        )
-        logger.info(f"-----End-----接口请求结束，请求响应数据：{res.text}-----")
-        # 对用例数据中需要提取的后置参数，基于接口响应数据进行处理
-        after_extract(res, req_data.get("extract", None))
-
-        return res
-
-    @classmethod
-    def send_api(cls, url, method, pk, header=None, data=None, file=None) -> Response:
-        """
-        根据pk参数的不同，决定请求参数是使用params,data还是json
-        :param method: 请求方法
-        :param url: 请求url
-        :param pk: 入参关键字， params(查询参数类型，明文传输，一般在url?参数名=参数值), data(一般用于form表单类型参数)
-        json(一般用于json类型请求参数)
-        :param data: 参数数据，默认等于None
-        :param file: 文件对象
-        :param header: 请求头
-        :return: 返回res对象
-        """
-        session = cls.get_session()
-        pk = pk.lower()
-        if pk == 'params':
-            res = session.request(method=method, url=url, params=data, headers=header)
-        elif pk == 'data':
-            res = session.request(method=method, url=url, data=data, files=file, headers=header)
-        elif pk == 'json':
-            res = session.request(method=method, url=url, json=data, files=file, headers=header)
-        else:
-            raise ValueError('pk可选关键字为params, json, data')
-        return res
-
-
-# ---------------------------------------- 请求后的参数提取处理----------------------------------------#
-def after_extract(response: Response, extract: dict) -> None:
-    """
-    从响应数据中提取请求后的参数，并保存到全局变量中
-    :param response: request 响应对象
-    :param extract: 需要提取的参数字典 '{"k1": "$.data"}' 或 '{"k1": "data:(.*?)$"}'
-    :return:
-    """
-    logger.info("-----Start-----请求后的参数提取处理-----")
-    if extract:
-        if response_type(response) == "json":
-            # 如果响应数据是json格式，则将按照json方式对后置提取参数进行处理
-            res = response.json()
-            for k, v in extract.items():
-                GLOBAL_VARS[k] = json_extractor(res, v)
-        else:
-            # 如果响应数据是str格式，则将按照str方式对后置提取参数进行处理
-            res = response.text
-            for k, v in extract.items():
-                GLOBAL_VARS[k] = re_extract(res, v)
-        logger.info(f"-----End-----参数提取后，打印当前的全局变量:{GLOBAL_VARS}")
-
-
-def response_type(response: Response) -> str:
-    """
-    :param response: requests 返回
-    :return: 返回响应数据类型 json或者str
-    """
-    try:
-        response.json()
-        return "json"
-    except:
-        return "str"

common_utils/base_request.py

@@ -0,0 +1,89 @@
+import requests
+from requests import Response
+from requests_toolbelt import MultipartEncoder
+from typing import Dict, Union
+import time
+from loguru import logger
+
+
+class BaseRequest:
+    """
+    进行请求
+    """
+    session = None
+
+    @classmethod
+    def get_session(cls):
+        """
+        单例模式保证测试过程中使用的都是一个session对象；
+        requests.session可以自动处理cookies，做状态保持。
+        """
+        if cls.session is None:
+            cls.session = requests.Session()
+        return cls.session
+
+    @classmethod
+    def send_request(cls, req_data: Dict[str, Union[str, Dict, MultipartEncoder]]) -> Response:
+        """
+        处理请求数据，转换成可用数据发送请求
+        :param req_data: 请求数据
+        :return: 响应对象
+        """
+        try:
+            logger.info(f"-----Start-----进行接口请求，并获取接口响应数据，请求数据：{type(req_data)} || {req_data}-----")
+            res = cls.send_api_request(
+                url=req_data.get("url"),
+                method=req_data.get("method").lower(),
+                pk=req_data.get("pk", None),
+                header=req_data.get("header"),
+                payload=req_data.get("payload"),
+                files=req_data.get("files")
+            )
+            logger.info(f"-----End-----接口请求结束，请求响应数据：{res.text}-----")
+        except requests.exceptions.RequestException as e:
+            raise ValueError(f"请求出错，{str(e)}")
+
+        return res
+
+    @classmethod
+    def send_api_request(cls, url: str, method: str, pk: str, header: Dict[str, str] = None, payload=None,
+                         files=None) -> Response:
+        """
+        发送请求
+        :param method: 请求方法
+        :param url: 请求url
+        :param pk: 请求参数类型，可选值为params，json，data
+        :param payload: 请求数据，对于不同请求类型，可以为dict，MultipartEncoder等
+        :param files: 请求上传的文件
+        :param header: 请求头
+        :return: 返回res对象
+        """
+        headers = header or {}
+        session = cls.get_session()
+
+        if pk.lower() == 'params':
+            res = session.request(method=method, url=url, params=payload, headers=headers)
+        elif pk.lower() == 'data':
+            if files:
+                if not isinstance(files, dict):
+                    raise ValueError('data参数必须为dict')
+                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)
+            else:
+                headers['Content-Type'] = 'application/x-www-form-urlencoded; charset=UTF-8'
+                res = session.request(method=method, url=url, data=payload, headers=headers)
+        elif pk.lower() == 'json':
+            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)
+            else:
+                headers['Content-Type'] = 'application/json'
+                res = session.request(method=method, url=url, json=payload, headers=headers)
+        else:
+            raise ValueError('pk可选关键字为params, json, data')
+
+        return res

test_case/test_manual_case/test_login_demo.py

@@ -10,7 +10,7 @@
 import pytest
 import os
 from common_utils.yaml_handle import HandleYaml
-from config.settings import DATA_DIR
+from config.project_path import DATA_DIR
 from common_utils.base_request import BaseRequest
 from case_utils.assert_handle import assert_response, assert_sql
 from loguru import logger