时间戳

首页 > 解决方案 > Python pytest pytest_exception_interact 从 VCR.py 异常自定义异常信息

python - Python pytest pytest_exception_interact 从 VCR.py 异常自定义异常信息

问题描述

语境

我已经开始使用pytest-vcr它是一个pytest插件包装VCR.py，我在这篇关于 Advanced Python Testing 的博客文章中记录了它。

cassettes/*.yml它会在第一次测试运行时记录所有到文件的 HTTP 流量以保存快照。类似于Web 组件的Jest快照测试。

在随后的测试运行中，如果请求格式错误，它将找不到匹配项并抛出异常，指出禁止记录新请求并且未找到现有记录。

问题

VCR.py提出 aCannotOverwriteExistingCassetteException对于它为什么不匹配的原因并不是特别有用。

如何利用 pytestpytest_exception_interact钩子将这个异常替换为一个更丰富的利用夹具信息的异常？

我深入研究了我的site-packages位置VCR.py并pip installed重写了我希望它如何处理异常。我只需要知道如何让这个pytest_exception_interact钩子正常工作以从该测试节点访问固定装置（在它被清理之前）并引发不同的异常。

例子

让我们获取依赖项。

$ pip install pytest pytest-vcr requests

test_example.py：

import pytest
import requests

@pytest.mark.vcr
def test_example():
    r = requests.get("https://www.stackoverflow.com")
    assert r.status_code == 200

$ pytest test_example.py --vcr-record=once
...
test_example.py::test_example PASSED 
...
$ ls cassettes/
cassettes/test_example.yml
$ head cassettes/test_example.yml
interactions:
- request:
    uri: https://wwwstackoverflow.com
    body: null
    headers:
        Accept:
        - '*/*'
$ pytest test_example.py --vcr-record=none
...
test_example.py::test_example PASSED 
...

现在将测试中的 URI 更改为“ https://www.google.com ”：

test_example.py：

import pytest
import requests

@pytest.mark.vcr
def test_example():
    r = requests.get("https://www.google.com")
    assert r.status_code == 200

并再次运行测试以检测回归：

$ pytest test_example.py --vcr-record=none
E               vcr.errors.CannotOverwriteExistingCassetteException: No match for the request (<Request (GET) https://www.google.com/>)
...

我可以将一个conftest.py文件添加到我的测试结构的根目录以创建一个本地插件，并且我可以验证我是否可以拦截异常并使用以下方法注入我自己的：

conftest.py

import pytest
from vcr.errors import CannotOverwriteExistingCassetteException
from vcr.config import VCR
from vcr.cassette import Cassette

class RequestNotFoundCassetteException(CannotOverwriteExistingCassetteException):
    ...

@pytest.fixture(autouse=True)
def _vcr_marker(request):
    marker = request.node.get_closest_marker("vcr")
    if marker:
        cassette = request.getfixturevalue("vcr_cassette")
        vcr = request.getfixturevalue("vcr")
        request.node.__vcr_fixtures = dict(vcr_cassette=cassette, vcr=vcr)
    yield


@pytest.hookimpl(hookwrapper=True)
def pytest_exception_interact(node, call, report):
    excinfo = call.excinfo
    if report.when == "call" and isinstance(excinfo.value, CannotOverwriteExistingCassetteException):
        # Safely check for fixture pass through on this node
        cassette = None
        vcr = None
        if hasattr(node, "__vcr_fixtures"):
            for fixture_name, fx in node.__vcr_fixtures.items():
                vcr = fx if isinstance(fx, VCR)
                cassette = fx if isinstance(fx, Cassette)

        # If we have the extra fixture context available...
        if cassette and vcr:
            match_properties = [f.__name__ for f in cassette._match_on]
            cassette_reqs = cassette.requests
            #  filtered_req = cassette.filter_request(vcr._vcr_request)
            #  this_req, req_str = __format_near_match(filtered_req, cassette_reqs, match_properties)

            # Raise and catch a new excpetion FROM existing one to keep the traceback
            # https://stackoverflow.com/a/24752607/622276
            # https://docs.python.org/3/library/exceptions.html#built-in-exceptions
            try:
                raise RequestNotFoundCassetteException(
                    f"\nMatching Properties: {match_properties}\n" f"Cassette Requests: {cassette_reqs}\n"
                ) from excinfo.value
            except RequestNotFoundCassetteException as e:
                excinfo._excinfo = (type(e), e)
                report.longrepr = node.repr_failure(excinfo)

这是互联网上的文档变得非常薄的部分。

如何访问vcr_cassette夹具并返回不同的异常？

我想要做的是获取filtered_request试图被请求的内容，cassette_requests并使用Python difflib标准库的列表产生针对分歧信息的增量。

PyTest 代码探索

使用 pytest 触发器运行单个测试的内部结构，pytest_runtest_protocol它有效地运行以下三个call_and_report调用以获取报告集合。

src/_pytest/runner.py:L77-L94

def runtestprotocol(item, log=True, nextitem=None):
    # Abbreviated
    reports = []
    reports.append(call_and_report(item, "setup", log))
    reports.append(call_and_report(item, "call", log))
    reports.append(call_and_report(item, "teardown", log))
    return reports

所以我在通话阶段修改了报告......但仍然不知道如何访问夹具信息。

src/_pytest/runner.py:L166-L174

def call_and_report(item, when, log=True, **kwds):
    call = call_runtest_hook(item, when, **kwds)
    hook = item.ihook
    report = hook.pytest_runtest_makereport(item=item, call=call)
    if log:
        hook.pytest_runtest_logreport(report=report)
    if check_interactive_exception(call, report):
        hook.pytest_exception_interact(node=item, call=call, report=report)
    return report

看起来有一些辅助方法可以生成新的 ExceptionRepresentation，所以我更新了conftest.py示例。

src/_pytest/reports.py:L361

longrepr = item.repr_failure(excinfo)

更新＃1 2019-06-26：感谢@hoefling在评论中的一些指示，我更新了我的conftest.py。

更新 #2 2019-06-26

在创建磁带上下文管理器时修补的 VCRHTTPConnections 似乎是不可能的。我已经打开了以下拉取请求，以便在抛出异常时作为参数传递，然后在下游任意捕获和处理。

https://github.com/kevin1024/vcrpy/pull/445

有关的

提供信息但仍未回答此问题的相关问题。

标签： pythonpython-3.xpytest

解决方案

感谢@hoefling评论中的评论和指导。

我可以将cassette夹具附加到覆盖标记request.node的本地插件中......conftest.pypytest-vcr

@pytest.fixture(autouse=True)
def _vcr_marker(request):
    marker = request.node.get_closest_marker("vcr")
    if marker:
        cassette = request.getfixturevalue("vcr_cassette")
        vcr = request.getfixturevalue("vcr")
        request.node.__vcr_fixtures = dict(vcr_cassette=cassette, vcr=vcr)
    yield

但我需要的不仅仅是磁带来获得我的解决方案。

原料

食谱

获取最新的 VCRpy

这些补丁在vcrpyv2.1.0中发布

pip install vcrpy==2.1.0

覆盖pytest_exception_interact钩子

在测试目录的根目录中创建一个conftest.py以创建覆盖hook的本地插件。pytest_exception_interact

@pytest.hookimpl(hookwrapper=True)
def pytest_exception_interact(node, call, report):
    """Intercept specific exceptions from tests."""
    if report.when == "call" and isinstance(call.excinfo.value, CannotOverwriteExistingCassetteException):
        __handle_cassette_exception(node, call, report)

    yield

从异常中提取Cassette和。Request

# Define new exception to throw
class RequestNotFoundCassetteException(Exception):
   ...

def __handle_cassette_exception(node, call, report):
    # Safely check for attributes attached to exception
    vcr_request = None
    cassette = None
    if hasattr(call.excinfo.value, "cassette"):
        cassette = call.excinfo.value.cassette
    if hasattr(call.excinfo.value, "failed_request"):
        vcr_request = call.excinfo.value.failed_request

    # If we have the extra context available...
    if cassette and vcr_request:

        match_properties = [f.__name__ for f in cassette._match_on]
        this_req, req_str = __format_near_match(cassette.requests, vcr_request, match_properties)

        try:
            raise RequestNotFoundCassetteException(f"{this_req}\n\n{req_str}\n") from call.excinfo.value
        except RequestNotFoundCassetteException as e:
            call.excinfo._excinfo = (type(e), e)
            report.longrepr = node.repr_failure(call.excinfo)

推荐阅读