zzhtest/apirun/extend/keywords.py

# -*- coding: utf-8 -*-

"""
    这是接口关键字驱动类，用于提供自动化接口测试的关键字方法。
    主要是实现常用的关键字内容，并定义好所有的参数内容即可
    接口中常用关键字：
        1.各种模拟请求方法：Post/get/put/delete/header/....
        2.根据需求进行断言封装：jsonpath、数据库断言
        3.集合Allure，可添加@allure.step，这样在自动化执行的时候
        Allure报告可以直接捕捉相关的执行信息，让测试报告更详细
"""
import json
import random
from apirun.Log.Log import logger
import allure
import requests
import jsonpath
from apirun.core.globalContext import g_context
from deepdiff import DeepDiff
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
import base64


class KeyWords:
    request = None

    @allure.step("参数数据：发送Post请求")
    def request_post(self, **kwargs):
        response = requests.post(**kwargs)
        # TODO: 扩展- 把对应的响应数据写到变量渲染中
        g_context().set_dict("current_response", response)  # 默认设置成变量渲染
        return response

    @allure.step("参数数据：发送Get请求")
    def request_get(self, **kwargs):
        url = kwargs.get("URL", None)
        params = kwargs.get("PARAMS", None)
        headers = kwargs.get("HEADERS", None)

        request_data = {
            "url": url,
            "params": params,
            "headers": headers,
        }

        response = requests.get(**request_data)
        g_context().set_dict("current_response", response)
        return response

    @allure.step("参数数据：发送Post请求-form_urlencoded")
    def request_post_form_urlencoded(self, **kwargs):
        """
        发送Post请求
        """
        url = kwargs.get("URL", None)
        params = kwargs.get("PARAMS", None)
        headers = kwargs.get("HEADERS", None)
        data = kwargs.get("DATA", None)

        request_data = {
            "url": url,
            "params": params,
            "headers": headers,
            "data": data,
        }

        response = requests.post(**request_data)
        g_context().set_dict("current_response", response)
        return response

    @allure.step("参数数据：发送Post请求-row_json")
    def request_post_row_json(self, **kwargs):
        """
        发送Post请求
        """
        url = kwargs.get("URL", None)
        params = kwargs.get("PARAMS", None)
        headers = kwargs.get("HEADERS", None)
        data = kwargs.get("DATA", None)

        request_data = {
            "url": url,
            "params": params,
            "headers": headers,
            "json": data,
        }

        response = requests.post(**request_data)
        g_context().set_dict("current_response", response)  # 默认设置成全局变量
        return response

    @allure.step("参数数据：发送Post请求-form_data")
    def request_post_form_data(self, **kwargs):
        """
        发送Post请求
        """
        url = kwargs.get("URL", None)
        files = kwargs.get("FILES", None)
        params = kwargs.get("PARAMS", None)
        headers = kwargs.get("HEADERS", None)
        data = kwargs.get("DATA", None)

        request_data = {
            "url": url,
            "params": params,
            "headers": headers,
            "data": data,
            "files": eval(files)  # 变成字典格式
        }

        response = requests.post(**request_data)
        g_context().set_dict("current_response", response)  # 默认设置成全局变量
        return response

    @allure.step("参数数据：发送Delete请求")
    def request_delete(self, **kwargs):
        """
        发送Post请求
        """
        url = kwargs.get("URL", None)
        params = kwargs.get("PARAMS", None)
        headers = kwargs.get("HEADERS", None)
        data = kwargs.get("DATA", None)

        request_data = {
            "url": url,
            "params": params,
            "headers": headers,
            "json": data,
        }

        response = requests.delete(**request_data)
        g_context().set_dict("current_response", response)  # 默认设置成全局变量
        return response

    # TODO: 扩展 - JAONPATH提取的方法
    @allure.step("参数数据：提取响应数据并存储")
    def ex_jsonData(self, **kwargs):
        """
        提取json数据
        EXVALUE：提取josn的表达式
        INDEX: 非必填，默认为0,all代表所有
        VARNAME：存储的变量名，方便后面使用
        """
        # 获取JsonPath的值
        EXPRESSION = kwargs.get("EXVALUE", None)
        # 获取对应的下标，非必填，默认为0
        INDEX = kwargs.get("INDEX", 0)
        if INDEX is None:
            INDEX = 0
        # 获取响应数据
        response = g_context().get_dict("current_response").json()

        if INDEX == "all":
            ex_data = jsonpath.jsonpath(response, EXPRESSION)
        else:
            ex_data = jsonpath.jsonpath(response, EXPRESSION)[INDEX]  # 通过JsonPath进行提取
        g_context().set_dict(kwargs["VARNAME"], ex_data)  # 根据变量名设置成变量渲染
        return ex_data

    # TODO: 扩展 - 数据库提取的方法
    @allure.step("参数数据：提取数据库数据并存储")
    def ex_mysqlData(self, **kwargs):
        """
        数据库 : 数据库的名称
        引用变量：数据库要存储的变量名，列表格式
        存储到全局变量：{“变量名_下标”:数据}
        """
        import pymysql
        from pymysql import cursors
        config = {"cursorclass": cursors.DictCursor}
        # 读取全局变量 - 根据选择的数据 读取指定的数据库配置 连接对应的数据库
        db_config = g_context().get_dict("_database")[kwargs["数据库"]]
        config.update(db_config)

        con = pymysql.connect(**config)
        cur = con.cursor()
        cur.execute(kwargs["SQL"])
        rs = cur.fetchall()
        cur.close()
        con.close()
        logger.info("数据库查询结果:", rs)

        var_names = kwargs["引用变量"].split(",")
        result = {}
        for i, data in enumerate(rs, start=1):
            for j, value in enumerate(var_names):
                result[f'{var_names[j]}_{i}'] = data.get(var_names[j])  # 根据变量名称找读取出来的内容
        g_context().set_by_dict(result)

    # TODO: 扩展 - 文本断言方法
    @allure.step("参数数据：断言当前文本内容")
    def assert_text_comparators(self, **kwargs):
        """
        封装断言以进行不同的比较操作。

        参数:
        value (Any): 要比较的值。
        expected (Any): 预期的值。
        op_str (str): 操作符的字符串表示（如 '>', '<', '==' 等）。
        message (str, optional): 自定义的错误消息。

        返回:
        None: 如果断言成功，则不返回任何内容。

        引发:
        AssertionError: 如果断言失败。
        """
        comparators = {
            '>': lambda a, b: a > b,
            '<': lambda a, b: a < b,
            '==': lambda a, b: a == b,
            '>=': lambda a, b: a >= b,
            '<=': lambda a, b: a <= b,
            '!=': lambda a, b: a != b,
        }

        message = kwargs.get("MESSAGE", None)

        if kwargs["OP_STR"] not in comparators:
            raise ValueError(f"没有该操作方式: {kwargs['OP_STR']}")

        if not comparators[kwargs['OP_STR']](kwargs['VALUE'], kwargs["EXPECTED"]):
            if message:
                raise AssertionError(message)
            else:
                raise AssertionError(f"{kwargs['VALUE']} {kwargs['OP_STR']} {kwargs['EXPECTED']} 失败")



    # TODO: 扩展 - 全量断言-对比两个Json的差异
    @allure.step("参数数据：全量断言-对比两个Json的差异")
    def assert_json_DeepDiff(self, **kwargs):
        """
        对比两个json的差异
        :param json1: 期望结果
        :param json2: 实际结果
        :param exclude_paths:需要排除的字段，集合的类型，比如{“id”,...}
        :param ignore_order: 忽略顺序，一般用户有序数据类型，比如列表
        :param ignore_string_case:忽略值的大小写，False
        :return: 当数据没有差异则返回空集合
        """
        json1 = kwargs["json1"]
        json2 = kwargs["json2"]

        exclude_paths = kwargs.get("过滤字段", None)
        ignore_order = kwargs.get("忽略顺序", None)
        ignore_string_case = kwargs.get("忽略大小写", False)

        screen_data = {"exclude_paths": exclude_paths, "ignore_order": ignore_order,
                       "ignore_string_case": ignore_string_case}

        diff = DeepDiff(json1, json2, **screen_data)

        assert not diff, f"全量断言失败:{diff}"

    # TODO: 扩展 - 加密处理
    @allure.step("参数数据：对数据进行AES加密处理")
    def encrypt_aes(self, **kwargs):
        """
        对数据进行AES加密
        :param data: 需要加密的数据
        :param VARNAME: 存储到全局变量的名称
        :return:
        """
        key = b"1234567812345678"  # key 密码
        data = kwargs["data"].encode('utf-8')
        cipher = AES.new(key, AES.MODE_ECB)  # 使用ECB模式
        ct_bytes = cipher.encrypt(pad(data, AES.block_size))  # 使用PKCS7填充,初始化数据块大小, 16位
        encrypt_data = base64.b64encode(ct_bytes).decode('utf-8')

        g_context().set_dict(kwargs["VARNAME"], encrypt_data)  # 根据变量名设置成变量

    # ----------------------实战扩展方法------------------------------
    @allure.step("参数数据：对数据进行AES加密处理")
    def generate_name(self, **kwargs):
        data = "hami" + str(random.randint(0, 9999))
        g_context().set_dict(kwargs["VARNAME"], data)  # 根据变量名设置成变量


    # TODO: 扩展 - JSOND断言方法
    @allure.step("参数数据：JSOND断言文本内容")
    def assert_json_comparators(self, **kwargs):
        """
        封装断言以进行不同的比较操作。

        参数:
        value (Any): 要比较的jsonPath值。
        expected (Any): 预期的值。
        op_str (str): 操作符的字符串表示（如 '>', '<', '==' 等）。
        message (str, optional): 自定义的错误消息。

        返回:
        None: 如果断言成功，则不返回任何内容。

        引发:
        AssertionError: 如果断言失败。
        """
        comparators = {
            '>': lambda a, b: a > b,
            '<': lambda a, b: a < b,
            '==': lambda a, b: a == b,
            '>=': lambda a, b: a >= b,
            '<=': lambda a, b: a <= b,
            '!=': lambda a, b: a != b,
        }

        message = kwargs.get("MESSAGE", None)

        if kwargs["OP_STR"] not in comparators:
            raise ValueError(f"没有该操作方式: {kwargs['OP_STR']}")

        # 通过jsonpath获取对应的数据
        # 获取响应数据
        response = g_context().get_dict("current_response").json()
        ex_data = jsonpath.jsonpath(response, kwargs['VALUE'])[0] # 默认就取第一个

        if not comparators[kwargs['OP_STR']](ex_data, kwargs["EXPECTED"]):
            if message:
                raise AssertionError(message)
            else:
                raise AssertionError(f"{ex_data} {kwargs['OP_STR']} {kwargs['EXPECTED']} 失败")