• 周五. 8月 19th, 2022

5G编程聚合网

5G时代下一个聚合的编程学习网

热门标签

HttpRunner3源码阅读:3.工具文件

admin

11月 28, 2021

utils

上一篇是读的models.py那其实其他文件中除了引入models.py内容后,utils.py引入的次数也挺多

可用资料

sentry_sdk: https://docs.sentry.io/platforms/python/
os: https://docs.python.org/zh-cn/3/library/os.html?highlight=os#module-os

导包

import collections  # 数据结构之一的模块
import copy     # 复制
import json     
import os.path
import platform # 获取操作系统信息
import uuid
from multiprocessing import Queue  # 多进程, 队列
import itertools # 迭代模块
from typing import Dict, List, Any, Union, Text

import sentry_sdk # 支持自动报告错误和异常,并识别应用程序中的性能问题。
from loguru import logger # 日志库

from httprunner import __version__  # 版本信息
from httprunner import exceptions  # 自定义异常包
from httprunner.models import VariablesMapping  # 参数模型 

源码内容附注释

老实说还是第一次知道这种用法

# os.environ 一个表示字符串环境的 mapping 对象
def init_sentry_sdk():
    """索性就认为是在初始化sentry"""
    sentry_sdk.init(
        dsn="https://[email protected]/5263855",
        release="[email protected]{}".format(__version__),
    )
    with sentry_sdk.configure_scope() as scope:
        scope.set_user({"id": uuid.getnode()})


def set_os_environ(variables_mapping):
    """ set variables mapping to os.environ
    设置变量到 系统环境变量中,简单理解成提供了一个字典吧
    例子:
    import os
    print(os.environ)
    os.environ["demo"] = "aoligei"
    print(os.environ)  # 这个时候 这里面就多了demo:aoligei
    """
    for variable in variables_mapping:
        os.environ[variable] = variables_mapping[variable]
        logger.debug(f"Set OS environment variable: {variable}")


def unset_os_environ(variables_mapping):
    """ set variables mapping to os.environ
    理解成删除环境变量吧
    """
    for variable in variables_mapping:
        os.environ.pop(variable)
        logger.debug(f"Unset OS environment variable: {variable}")


def get_os_environ(variable_name):
    """ get value of environment variable.
    得到变量的值
    Args:
        variable_name(str): variable name

    Returns:
        value of environment variable.

    Raises:
        exceptions.EnvNotFound: If environment variable not found.

    """
    try:
        # 理解成从一个字典里面取值
        return os.environ[variable_name]
    except KeyError:
        raise exceptions.EnvNotFound(variable_name)


def lower_dict_keys(origin_dict):
    """ convert keys in dict to lower case

    Args:
        origin_dict (dict): mapping data structure

    Returns:
        dict: mapping with all keys lowered.

    Examples:
        >>> origin_dict = {
            "Name": "",
            "Request": "",
            "URL": "",
            "METHOD": "",
            "Headers": "",
            "Data": ""
        }
        >>> lower_dict_keys(origin_dict)
            {
                "name": "",
                "request": "",
                "url": "",
                "method": "",
                "headers": "",
                "data": ""
            }

    """
    if not origin_dict or not isinstance(origin_dict, dict):
        return origin_dict
    
    # 字典推导式把 key 转换成小写
    return {key.lower(): value for key, value in origin_dict.items()}


def print_info(info_mapping):
    """打印字典信息"""
    """ print info in mapping.

    Args:
        info_mapping (dict): input(variables) or output mapping.

    Examples:
        >>> info_mapping = {
                "var_a": "hello",
                "var_b": "world"
            }
        >>> info_mapping = {
                "status_code": 500
            }
        >>> print_info(info_mapping)
        ==================== Output ====================
        Key              :  Value
        ---------------- :  ----------------------------
        var_a            :  hello
        var_b            :  world
        ------------------------------------------------

    """
    if not info_mapping:
        return

    content_format = "{:<16} : {:<}
"
    content = "
==================== Output ====================
"
    content += content_format.format("Variable", "Value")
    content += content_format.format("-" * 16, "-" * 29)

    for key, value in info_mapping.items():
        # 判断value 是 元组 或者 deque: 类似列表(list)的容器,实现了在两端快速添加(append)和弹出(pop)
        if isinstance(value, (tuple, collections.deque)):
            continue
        # 判断value 是 字典 或者 列表
        elif isinstance(value, (dict, list)):
            value = json.dumps(value)
        elif value is None:
            value = "None"

        content += content_format.format(key, value)

    content += "-" * 48 + "
"
    logger.info(content)


def omit_long_data(body, omit_len=512):
    """ omit too long str/bytes
    处理过长的数据
    """
    if not isinstance(body, (str, bytes)):
        return body

    body_len = len(body)
    if body_len <= omit_len:
        return body

    omitted_body = body[0:omit_len]

    appendix_str = f" ... OMITTED {body_len - omit_len} CHARACTORS ..."
    if isinstance(body, bytes):
        appendix_str = appendix_str.encode("utf-8")

    return omitted_body + appendix_str


def get_platform():
    """返回框架信息
    platform.platform(): 系统版本信息
    platform.python_version(): Python信息
    python_implementation(): python 解释器版本?CPython
    """
    return {
        "httprunner_version": __version__,
        "python_version": "{} {}".format(
            platform.python_implementation(), platform.python_version()
        ),
        "platform": platform.platform(),
    }


def sort_dict_by_custom_order(raw_dict: Dict, custom_order: List):
    def get_index_from_list(lst: List, item: Any):
        try:
            # 返回元素首次出现的下标
            return lst.index(item)
        except ValueError:
            # item is not in lst
            return len(lst) + 1
    
    # 排序之后返回的是列表 然后转字典
    return dict(
        sorted(raw_dict.items(), key=lambda i: get_index_from_list(custom_order, i[0]))
    )


class ExtendJSONEncoder(json.JSONEncoder):
    """ especially used to safely dump json data with python object, such as MultipartEncoder
    JSON dump 异常
    """

    def default(self, obj):
        try:
            return super(ExtendJSONEncoder, self).default(obj)
        except (UnicodeDecodeError, TypeError):
            return repr(obj)


def merge_variables(
    variables: VariablesMapping, variables_to_be_overridden: VariablesMapping
) -> VariablesMapping:
    """ merge two variables mapping, the first variables have higher priority
    """
    step_new_variables = {}
    for key, value in variables.items():
        if f"${key}" == value or "${" + key + "}" == value:
            # e.g. {"base_url": "$base_url"}
            # or {"base_url": "${base_url}"}
            continue

        step_new_variables[key] = value
    
    # 浅复制了字典 并把其中的内容都弄过来了
    merged_variables = copy.copy(variables_to_be_overridden)
    # 更新了复制出来的字典, 原字典不会改变
    merged_variables.update(step_new_variables)
    return merged_variables


def is_support_multiprocessing() -> bool:
    try:
        Queue()
        return True
    except (ImportError, OSError):
        # system that does not support semaphores(dependency of multiprocessing), like Android termux
        return False


def gen_cartesian_product(*args: List[Dict]) -> List[Dict]:
    """ generate cartesian product for lists
    生成笛卡尔积,估计是参数化用的

    Args:
        args (list of list): lists to be generated with cartesian product

    Returns:
        list: cartesian product in list

    Examples:

        >>> arg1 = [{"a": 1}, {"a": 2}]
        >>> arg2 = [{"x": 111, "y": 112}, {"x": 121, "y": 122}]
        >>> args = [arg1, arg2]
        >>> gen_cartesian_product(*args)
        >>> # same as below
        >>> gen_cartesian_product(arg1, arg2)
            [
                {'a': 1, 'x': 111, 'y': 112},
                {'a': 1, 'x': 121, 'y': 122},
                {'a': 2, 'x': 111, 'y': 112},
                {'a': 2, 'x': 121, 'y': 122}
            ]

    """
    if not args:
        return []
    elif len(args) == 1:
        return args[0]

    product_list = []
    # product 笛卡尔积,相当于嵌套的for循环
    for product_item_tuple in itertools.product(*args):
        product_item_dict = {}
        for item in product_item_tuple:
            product_item_dict.update(item)

        product_list.append(product_item_dict)

    return product_list

作者:zy7y
本文版权归作者和博客园共有,欢迎转载,但未经作者同意必须在文章页面给出原文链接,否则保留追究法律责任的权利。

发表回复

您的电子邮箱地址不会被公开。