利用 Pytest Cache Fixture 实现测试结果缓存

news2024/10/7 17:32:58

前言

接口自动关过程中，经常会遇到这样一些场景，"请求2需要用到请求1响应的数据"，常见的做法，进行用例依赖或者将请求1的响应结果写入一个文件，用到的时候读取文件。当然这都不是这篇文章的重点，本片文章主要介绍cache写入和读取缓存数据。

`request.config.cache`

还不了解request fixture的同学可以先看看这篇文章，pytest 的 request fixture：实现个性化测试需求

我们先看看使用案例：

def test_01(cache):
    cache.set("token", "uiouoouoiou")

def test_02(cache):
    r = cache.get("token", None)

这样段代码在执行test_01会将token值缓存，任何执行test_02时就可以从缓存中读取token值。那Cache是如何实现的呢？我们一起来看看源码。源码直达

实现原理

def test_01(cache):
    cache.set("token", {"token": "1212121"})

我们在cache.set()这一行进行断点，debug执行后，debug结果为

cache = Cache()
_CACHE_PREFIX_DIRS = 'd'
_CACHE_PREFIX_VALUES = 'v'
_cachedir = /PycharmProjects/panda-test/org/.pytest_cache
_config = <_pytest.config.Config object at 0x109e80d60>

可以看到会自动创建一个缓存实例，而且初始化了一些数据，默认应该缓存文件会在.pytest_cache目录下

/_pytest/cacheprovider.py

@fixture
def cache(request: FixtureRequest) -> Cache:
    """Return a cache object that can persist state between testing sessions.

    cache.get(key, default)
    cache.set(key, value)

    Keys must be ``/`` separated strings, where the first part is usually the
    name of your plugin or application to avoid clashes with other cache users.

    Values can be any object handled by the json stdlib module.
    """
    assert request.config.cache is not None
    return request.config.cache

可以看到，cache返回的是Cache对象，我们看看Cache对象是如何实现的

def set(self, key: str, value: object) -> None:
        path = self._getvaluepath(key)
        try:
            if path.parent.is_dir():
                cache_dir_exists_already = True
            else:
                cache_dir_exists_already = self._cachedir.exists()
                path.parent.mkdir(exist_ok=True, parents=True)
        except OSError:
            self.warn("could not create cache path {path}", path=path, _ispytest=True)
            return
        if not cache_dir_exists_already:
            self._ensure_supporting_files()
        data = json.dumps(value, ensure_ascii=False, indent=2)
        try:
            f = path.open("w", encoding="UTF-8")
        except OSError:
            self.warn("cache could not write path {path}", path=path, _ispytest=True)
        else:
            with f:
                f.write(data)

这段源码就是用来将键值对保存到缓存中。代码比较简单，简单解释一下

获取要保存的键值对的路径：通过调用 _getvaluepath() 方法，根据给定的键（key）获取值（value）在缓存中的路径（path）。这里的路径是一个字符串，使用/分隔不同的层级，通常第一个名称是插件或应用程序的名称。
检查路径是否存在：通过判断路径的父目录是否为目录来确定是否需要创建路径。如果父目录已经存在，则 cache_dir_exists_already 设置为 True；否则，它会检查缓存目录是否存在，并且如果缓存目录已经存在，则 cache_dir_exists_already 设置为 True，否则创建缓存目录。
确保支持文件已存在：如果缓存目录是新创建的，则调用 _ensure_supporting_files() 方法确保支持文件存在。这个方法可能是用来创建其他与缓存相关的文件或目录。
序列化数据并写入文件：将值（value）使用 JSON 格式进行序列化，以确保它是基本的 Python 类型或包含了嵌套类型（例如列表和字典）。然后，尝试打开路径对应的文件（使用 UTF-8 编码），并将序列化后的数据写入文件中。

def get(self, key: str, default):
        
        path = self._getvaluepath(key)
        try:
            with path.open("r", encoding="UTF-8") as f:
                return json.load(f)
        except (ValueError, OSError):
            return default

这段源码用来从缓存中获取指定键的值，简单解释一下：

获取要获取值的路径：通过调用 _getvaluepath() 方法，根据给定的键（key）获取值在缓存中的路径（path）。这里的路径是一个字符串，使用 / 分隔不同的层级，通常第一个名称是插件或应用程序的名称。
尝试读取文件并返回已缓存的值：使用路径对应的文件（使用 UTF-8 编码）打开，并使用 json.load(f) 将文件中的数据加载为 Python 对象。然后将加载的值返回。
处理异常情况：如果无法将文件中的内容解析为有效的 JSON 数据或者打开文件失败，则捕获异常（ValueError 和 OSError），并返回默认值（default）。

这里还是学习到了一种新奇的写法，以前没用过with path.open("r", encoding="UTF-8") as f:等价于open(path, "r", encoding="UTF-8")

这是两个常用的方法，当然还提供了更多方法，这里简单介绍一下：

__init__(self, cachedir: Path, config: Config, *, _ispytest: bool = False) -> None：
- 初始化方法，用于设置类的属性 _cachedir 和 _config。
for_config(cls, config: Config, *, _ispytest: bool = False) -> "Cache"：
- 类方法，根据给定的配置信息创建并返回 Cache 实例。
- 如果配置项 cacheclear 设置为 True，并且缓存目录存在，则调用 clear_cache 方法清空缓存。
- 最后返回一个新的 Cache 实例。
clear_cache(cls, cachedir: Path, _ispytest: bool = False) -> None：
- 类方法，清空缓存目录下的子目录。
- 根据参数 cachedir 构建子目录路径，并使用 rm_rf 函数递归删除该目录。
cache_dir_from_config(config: Config, *, _ispytest: bool = False) -> Path：
- 静态方法，从给定的配置信息中获取缓存目录的路径。
- 首先从配置中获取缓存目录的字符串表示，然后使用 resolve_from_str 函数将其解析为 Path 对象返回。
warn(self, fmt: str, *, _ispytest: bool = False, **args: object) -> None：
- 发出缓存警告的方法。
- 使用 warnings.warn 函数发出警告信息，并指定警告类型为 PytestCacheWarning。
- 如果存在参数 args，则将其作为格式化参数替换格式字符串中的占位符。
mkdir(self, name: str) -> Path：
- 创建一个目录路径对象，并在缓存目录下创建该目录。
- 参数 name 是要创建的目录名称。
- 检查目录名是否包含路径分隔符 /，如果有则抛出异常。
- 使用 _cachedir.joinpath 方法构建完整的目录路径，并使用 mkdir 方法创建目录。
- 返回创建的目录路径对象。
_getvaluepath(self, key: str) -> Path：
- 根据给定的键生成值文件的路径。
- 在缓存目录下构建值文件路径，使用 _CACHE_PREFIX_VALUES 作为子目录前缀。
_ensure_supporting_files(self) -> None：
- 创建缓存目录中的支持文件。
- 创建 README.md 文件，用于说明缓存目录的用途。
- 创建 .gitignore 文件，忽略缓存目录下的所有文件。
- 创建 CACHEDIR.TAG 文件，用于标记缓存目录。