pytest-twisted 1.14.3

注意：Python 3.8 支持asyncio

在 Python 3.8 中，asyncio 将默认循环实现改为使用它们的 proactor。Proactor 不实现 Twisted 的 asyncio 支持中使用的一些方法。结果是如下所示的 NotImplementedError 异常。

<snip>
  File "c:\projects\pytest-twisted\.tox\py38-asyncioreactor\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
    reactor = AsyncioSelectorReactor(eventloop)
  File "c:\projects\pytest-twisted\.tox\py38-asyncioreactor\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
    super().__init__()
  File "c:\projects\pytest-twisted\.tox\py38-asyncioreactor\lib\site-packages\twisted\internet\base.py", line 571, in __init__
    self.installWaker()
  File "c:\projects\pytest-twisted\.tox\py38-asyncioreactor\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
    self.addReader(self.waker)
  File "c:\projects\pytest-twisted\.tox\py38-asyncioreactor\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
    self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
  File "C:\Python38-x64\Lib\asyncio\events.py", line 501, in add_reader
    raise NotImplementedError
NotImplementedError

之前的默认值，选择器循环仍然有效，但您必须明确设置它并尽早设置。以下 conftest.py 提供供参考。

import sys

import pytest
import pytest_twisted


@pytest.hookimpl(tryfirst=True)
def pytest_configure(config):
    # https://twistedmatrix.com/trac/ticket/9766
    # https://github.com/pytest-dev/pytest-twisted/issues/80

    if (
        config.getoption("reactor", "default") == "asyncio"
        and sys.platform == 'win32'
        and sys.version_info >= (3, 8)
    ):
        import asyncio

        selector_policy = asyncio.WindowsSelectorEventLoopPolicy()
        asyncio.set_event_loop_policy(selector_policy)

Python 2 支持计划

在某个时候，保持 Python 2 的支持可能变得不切实际。鉴于其规模小且开发量非常低，这似乎不太可能成为近期问题。虽然我本人没有 Python 2 的支持需求，但我尽量提供帮助，因此不会明确删除支持，以免考虑它。如果报告了重大问题，并且我自己或社区没有时间解决它们，那么将考虑其他选项。

安装

按照以下步骤安装插件。

pip install pytest-twisted

使用插件

安装后，插件可用，可以使用 -p no:twisted 禁用。

默认情况下，使用 twisted.internet.default 安装 reactor。这会创建与 import twisted.internet.reactor 相同的 reactor。可以使用 --reactor 选项指定替代 reactor。目前支持 qt5reactor 以与 pyqt5 和 pytest-qt 一起使用，以及 asyncio。此 指南 描述了如何添加对新 reactor 的支持。

reactor 在第一个测试之前自动创建，但可以通过调用 pytest_twisted.init_default_reactor() 或对应于所需替代 reactor 的相应函数来显式安装。

inlineCallbacks

将 twisted.internet.defer.inlineCallbacks 作为装饰器用于使用 fixtures 的测试函数不起作用。请使用 pytest_twisted.inlineCallbacks 代替。

@pytest_twisted.inlineCallbacks
def test_some_stuff(tmpdir):
    res = yield threads.deferToThread(os.listdir, tmpdir.strpath)
    assert res == []

ensureDeferred

将 twisted.internet.defer.ensureDeferred 作为装饰器用于使用 fixtures 的测试函数不起作用。请使用 pytest_twisted.ensureDeferred 代替。

@pytest_twisted.ensureDeferred
async def test_some_stuff(tmpdir):
    res = await threads.deferToThread(os.listdir, tmpdir.strpath)
    assert res == []

在 fixtures 中等待 deferreds

pytest_twisted.blockon 允许 fixtures 等待 deferreds。

@pytest.fixture
def val():
    d = defer.Deferred()
    reactor.callLater(1.0, d.callback, 10)
    return pytest_twisted.blockon(d)

async/await fixtures

async/await fixtures 可以与 yield 一起使用，以配合 pytest fixture 的常规语义进行设置、值和拆卸。目前仅支持函数和模块范围。

# No yield (coroutine function)
#   -> use pytest_twisted.async_fixture()
@pytest_twisted.async_fixture()
async def foo():
    d = defer.Deferred()
    reactor.callLater(0.01, d.callback, 42)
    value = await d
    return value

# With yield (asynchronous generator)
#   -> use pytest_twisted.async_yield_fixture()
@pytest_twisted.async_yield_fixture()
async def foo_with_teardown():
    d1, d2 = defer.Deferred(), defer.Deferred()
    reactor.callLater(0.01, d1.callback, 42)
    reactor.callLater(0.02, d2.callback, 37)
    value = await d1
    yield value
    await d2

假设

pytest-twisted可以与Hypothesis一起使用。

@hypothesis.given(x=hypothesis.strategies.integers())
@pytest_twisted.ensureDeferred
async def test_async(x):
    assert isinstance(x, int)

Twisted greenlet

某些库（例如corotwine）需要知道运行Twisted反应器的greenlet。它可以从twisted_greenlet fixture中获得。以下代码可用于使corotwine与pytest-twisted一起工作。

@pytest.fixture(scope="session", autouse=True)
def set_MAIN(request, twisted_greenlet):
    from corotwine import protocol
    protocol.MAIN = twisted_greenlet

这（几乎）就是全部。

弃用

v1.9

pytest.blockon

使用pytest_twisted.blockon

pytest.inlineCallbacks

使用pytest_twisted.inlineCallbacks